diff --git a/cli/internal/daemon/client.go b/cli/internal/daemon/client.go index 23d34b1..ec92eab 100644 --- a/cli/internal/daemon/client.go +++ b/cli/internal/daemon/client.go @@ -78,6 +78,9 @@ type ChatClient struct { mu sync.Mutex afterSeq int argumentsByCall map[string]string + // Tail, when positive, asks a reset to replay only the newest rows of the + // transcript; History pages back through the rest. + Tail int } func readBounded(r io.Reader, limit int64) ([]byte, error) { @@ -350,6 +353,59 @@ func (c *ChatClient) Interrupt(ctx context.Context) (bool, error) { // ContextWindow reads the model's context window from the session's last // prepared request. It is nil when neither a catalog nor the configuration // knows it, or when no request has been prepared yet. +// HistoryPage is older transcript, rendered as the stream renders a reset. +type HistoryPage struct { + Events []StreamEvent + Before int64 // first row of the page: the cursor for the next older one + More bool +} + +// History returns up to rows transcript rows before the given row, oldest +// first, widened back to the start of a turn. +func (c *ChatClient) History(ctx context.Context, before int64, rows int) (*HistoryPage, error) { + reqCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + route := fmt.Sprintf("/history?rows=%d", rows) + if before > 0 { + route += fmt.Sprintf("&before=%d", before) + } + req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, c.agentURL(route), nil) + if err != nil { + return nil, err + } + if c.token != "" { + req.Header.Set("Authorization", "Bearer "+c.token) + } + res, err := c.httpClient.Do(req) + if err != nil { + return nil, err + } + defer res.Body.Close() + if res.StatusCode != http.StatusOK { + return nil, c.parseResponseError(res) + } + body, err := readBounded(res.Body, 32*1024*1024) + if err != nil { + return nil, err + } + var data struct { + Events []map[string]any `json:"events"` + Before int64 `json:"before"` + More bool `json:"more"` + } + if err := json.Unmarshal(body, &data); err != nil { + return nil, err + } + page := &HistoryPage{Before: data.Before, More: data.More} + for _, evMap := range data.Events { + if ev := normalizeEvent(evMap); ev != nil { + ev.Replayed = true + page.Events = append(page.Events, *ev) + } + } + return page, nil +} + func (c *ChatClient) ContextWindow(ctx context.Context) (*int, error) { reqCtx, cancel := context.WithTimeout(ctx, 20*time.Second) defer cancel() @@ -435,6 +491,28 @@ func (c *ChatClient) GetStatus(ctx context.Context) (*AgentStatus, error) { }, nil } +func int64Field(raw map[string]any, key string) int64 { + if n, ok := raw[key].(float64); ok && n > 0 { + return int64(n) + } + return 0 +} + +// normalizeEvent parses one daemon event map, decoding a tool's JSON-encoded args. +func normalizeEvent(evMap map[string]any) *StreamEvent { + if evMap["type"] == "tool" { + if strArgs, ok := evMap["args"].(string); ok { + var parsedArgs map[string]any + if err := json.Unmarshal([]byte(strArgs), &parsedArgs); err == nil { + evMap["args"] = parsedArgs + } else { + evMap["args"] = make(map[string]any) + } + } + } + return parseStreamEvent(evMap) +} + func parseStreamEvent(raw map[string]any) *StreamEvent { typ, _ := raw["type"].(string) if typ == "" { @@ -512,7 +590,14 @@ func parseStreamEvent(raw map[string]any) *StreamEvent { return &StreamEvent{Type: EventRetry} case "reset": - return &StreamEvent{Type: EventReset} + return &StreamEvent{Type: EventReset, Before: int64Field(raw, "before"), More: raw["more"] == true} + + case "committed": + seq := int64Field(raw, "seq") + if seq <= 0 { + return nil + } + return &StreamEvent{Type: EventCommitted, Seq: seq} case "note": txt, _ := raw["text"].(string) @@ -618,6 +703,9 @@ func (c *ChatClient) Stream(ctx context.Context, onOpen func(), onEvent func(Str var route string if c.agentID != "" { route = fmt.Sprintf("/stream?after_seq=%d", afterSeq) + if c.Tail > 0 { + route += fmt.Sprintf("&tail=%d", c.Tail) + } } else { if afterSeq > 0 { route = fmt.Sprintf("/awp/stream?after_seq=%d", afterSeq) @@ -744,7 +832,7 @@ func (c *ChatClient) Stream(ctx context.Context, onOpen func(), onEvent func(Str if itemType == "reset" { snapshot = true - if err := onEvent(StreamEvent{Type: EventReset}); err != nil { + if err := onEvent(*parseStreamEvent(evMap)); err != nil { return err } continue @@ -819,18 +907,7 @@ func (c *ChatClient) Stream(ctx context.Context, onOpen func(), onEvent func(Str } } - if itemType == "tool" { - if strArgs, ok := evMap["args"].(string); ok { - var parsedArgs map[string]any - if err := json.Unmarshal([]byte(strArgs), &parsedArgs); err == nil { - evMap["args"] = parsedArgs - } else { - evMap["args"] = make(map[string]any) - } - } - } - - if ev := parseStreamEvent(evMap); ev != nil { + if ev := normalizeEvent(evMap); ev != nil { ev.Replayed = snapshot if err := onEvent(*ev); err != nil { return err diff --git a/cli/internal/daemon/events.go b/cli/internal/daemon/events.go index dcb310b..5843415 100644 --- a/cli/internal/daemon/events.go +++ b/cli/internal/daemon/events.go @@ -118,6 +118,8 @@ const ( EventToolProgress EventType = "tool_progress" EventTool EventType = "tool" EventUsage EventType = "usage" + // EventCommitted says transcript rows up to Seq now cover what was shown. + EventCommitted EventType = "committed" ) type Usage struct { @@ -149,6 +151,12 @@ type StreamEvent struct { ToolResult string `json:"result,omitempty"` ToolTrace *ToolTrace `json:"trace,omitempty"` Usage *Usage `json:"usage,omitempty"` + // Seq is the newest transcript row an EventCommitted covers. + Seq int64 `json:"seq,omitempty"` + // Before and More describe the history a reset or page carried: Before is + // its first row (the cursor for the next older page), More whether one exists. + Before int64 `json:"before,omitempty"` + More bool `json:"more,omitempty"` // Replayed marks an event from the transcript snapshot that follows a // reset: history, which says nothing about what the session does now. Replayed bool `json:"-"` diff --git a/cli/internal/daemon/history_test.go b/cli/internal/daemon/history_test.go new file mode 100644 index 0000000..8c631a9 --- /dev/null +++ b/cli/internal/daemon/history_test.go @@ -0,0 +1,73 @@ +package daemon + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func TestStreamAsksForATailAndReadsItsCursor(t *testing.T) { + var query string + page := formatPage(3, []any{ + map[string]any{"type": "reset", "before": 40, "more": true}, + map[string]any{"type": "user", "text": "hi", "source": "chat"}, + map[string]any{"type": "committed", "seq": 41}, + }) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + query = r.URL.RawQuery + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte(page)) + })) + defer server.Close() + + client := NewChatClient(ChatClientOptions{BaseURL: server.URL, AgentID: "s"}) + client.Tail = 120 + var events []StreamEvent + if err := client.Stream(context.Background(), nil, func(e StreamEvent) error { + events = append(events, e) + return nil + }); err != nil { + t.Fatal(err) + } + if query != "after_seq=0&tail=120" && query != "after_seq=-1&tail=120" { + t.Fatalf("stream query %q does not ask for a tail", query) + } + if len(events) != 3 || events[0].Type != EventReset || events[0].Before != 40 || !events[0].More { + t.Fatalf("reset cursor not read: %+v", events) + } + if events[2].Type != EventCommitted || events[2].Seq != 41 || !events[2].Replayed { + t.Fatalf("committed marker not read: %+v", events[2]) + } +} + +func TestHistoryReadsAnOlderPage(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/sessions/s/history" || r.URL.Query().Get("before") != "40" || r.URL.Query().Get("rows") != "120" { + t.Errorf("unexpected request %s", r.URL) + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "events": []any{ + map[string]any{"type": "user", "text": "older", "source": "chat"}, + map[string]any{"type": "tool", "callId": "c", "name": "python", "args": `{"code":"1"}`, "result": "1"}, + map[string]any{"type": "committed", "seq": 39}, + }, + "before": 12, + "more": true, + }) + })) + defer server.Close() + + client := NewChatClient(ChatClientOptions{BaseURL: server.URL, AgentID: "s"}) + page, err := client.History(context.Background(), 40, 120) + if err != nil { + t.Fatal(err) + } + if page.Before != 12 || !page.More || len(page.Events) != 3 { + t.Fatalf("unexpected page: %+v", page) + } + if page.Events[1].ToolArgs["code"] != "1" || !page.Events[0].Replayed { + t.Fatalf("events not normalized: %+v", page.Events) + } +} diff --git a/cli/internal/tui/bounded_history.go b/cli/internal/tui/bounded_history.go index 549e8e2..d1c77c6 100644 --- a/cli/internal/tui/bounded_history.go +++ b/cli/internal/tui/bounded_history.go @@ -4,6 +4,7 @@ import ( "albedo/cli/internal/daemon" "encoding/json" "fmt" + "slices" ) type EntryKind string @@ -37,6 +38,9 @@ type HistoryEntry struct { Tools int // Pending is a message of yours the daemon has not echoed back yet. Pending pending + // Seq is the newest transcript row this entry came from; 0 until a + // `committed` event says which rows cover it. + Seq int64 } // pending is how far a message of yours has got: sent and waiting for the @@ -82,6 +86,9 @@ type BoundedHistory struct { totalBytes int64 evictedEntries int evictedBytes int64 + // evictedThrough is the newest transcript row among evicted entries: the + // daemon holds everything up to it, and older pages resume after it. + evictedThrough int64 } func NewBoundedHistory(maxEntries int, maxBytes int64) *BoundedHistory { @@ -125,20 +132,55 @@ func (h *BoundedHistory) TruncationNotice() string { return "" } +// EvictedThrough is the newest transcript row among evicted entries, or 0. +func (h *BoundedHistory) EvictedThrough() int64 { + return h.evictedThrough +} + +// Stamp marks the trailing entries no row covered yet as covered by seq. +func (h *BoundedHistory) Stamp(seq int64) { + for i := len(h.entries) - 1; i >= 0 && h.entries[i].Seq == 0; i-- { + h.entries[i].Seq = seq + } +} + +// Prepend puts older entries before the retained ones. It does not evict: +// they were asked for, and the next Append trims as usual. +func (h *BoundedHistory) Prepend(older []HistoryEntry) { + if len(older) == 0 { + return + } + for i := range older { + h.totalBytes += older[i].ComputeSize() + } + h.entries = append(slices.Clone(older), h.entries...) + h.evictedThrough = 0 +} + func (h *BoundedHistory) enforceBounds() { for len(h.entries) > 0 && (len(h.entries) > h.MaxEntries || h.totalBytes > h.MaxBytes) { - removed := h.entries[0] - h.entries[0] = HistoryEntry{} // Explicitly zero out evicted slot to drop GC references - h.entries = h.entries[1:] - h.totalBytes -= removed.SizeBytes - if h.totalBytes < 0 { - h.totalBytes = 0 + h.evictFirst() + // A row's entries leave together, so re-reading from the next row + // never repeats what is still shown. + for seq := h.evictedThrough; seq > 0 && len(h.entries) > 0 && h.entries[0].Seq == seq; { + h.evictFirst() } - h.evictedEntries++ - h.evictedBytes += removed.SizeBytes } } +func (h *BoundedHistory) evictFirst() { + removed := h.entries[0] + h.entries[0] = HistoryEntry{} // Explicitly zero out evicted slot to drop GC references + h.entries = h.entries[1:] + h.totalBytes -= removed.SizeBytes + if h.totalBytes < 0 { + h.totalBytes = 0 + } + h.evictedEntries++ + h.evictedBytes += removed.SizeBytes + h.evictedThrough = max(h.evictedThrough, removed.Seq) +} + func (h *BoundedHistory) Append(entry HistoryEntry) { entry.ComputeSize() h.entries = append(h.entries, entry) @@ -180,4 +222,5 @@ func (h *BoundedHistory) Clear() { h.totalBytes = 0 h.evictedEntries = 0 h.evictedBytes = 0 + h.evictedThrough = 0 } diff --git a/cli/internal/tui/chat.go b/cli/internal/tui/chat.go index aa2d926..d4f43fe 100644 --- a/cli/internal/tui/chat.go +++ b/cli/internal/tui/chat.go @@ -24,6 +24,14 @@ import ( type ChatBackToSessionsMsg struct{} type ChatQuitMsg struct{} + +// ChatOlderLoadedMsg carries a page of history from before what is shown. +type ChatOlderLoadedMsg struct { + SessionID string + Generation int64 + Page *daemon.HistoryPage + Err error +} type ChatNewSessionMsg struct{} type ChatOpenModelPickerMsg struct{} type ChatOpenExtensionPickerMsg struct{} @@ -137,8 +145,14 @@ const ( MaxPendingUsers = 10 MaxSettledLines = 1000 MaxSettledLinesBytes = 256 * 1024 - fnvOffset64 = 14695981039346656037 - fnvPrime64 = 1099511628211 + // readingSlack multiplies the caps while you are scrolled up, so output + // arriving as you read does not shrink the transcript above you. + readingSlack = 8 + // olderPageRows is how many transcript rows a reset replays and each + // older page adds; the daemon widens both back to a turn's start. + olderPageRows = 120 + fnvOffset64 = 14695981039346656037 + fnvPrime64 = 1099511628211 ) func fnv1a(h uint64, s string) uint64 { @@ -261,6 +275,13 @@ type ChatModel struct { settledLines []string settledLinesBytes int64 droppedSettledLines int + // rebuilding defers trimming until a re-render has placed the reading position. + rebuilding bool + // olderBefore is the first transcript row the reset or the last older page + // carried; olderMore says rows before it exist. loadingOlder is a fetch in flight. + olderBefore int64 + olderMore bool + loadingOlder bool scrollOffset int scrollLimit int @@ -343,6 +364,10 @@ func NewChatModel(session *daemon.Session, client *daemon.ChatClient) ChatModel streamCancel: cancel, eventChan: make(chan daemon.StreamEvent, 16), } + if client != nil { + // A reset replays the newest rows; loadOlder pages back from there. + client.Tail = olderPageRows + } return m } @@ -398,16 +423,57 @@ func (m *ChatModel) SetSize(width, height int) { } func (m *ChatModel) rebuildSettledLines() { - // Re-rendering retained history must not move the reading position. - offset := m.scrollOffset + // Re-rendering retained history must not move the reading position: it + // keeps its distance from the end of the settled rows, and trimming waits + // until it is placed. + fromEnd := m.settledOffset + len(m.settledLines) - m.scrollOffset m.settledLines = nil m.settledLinesBytes = 0 + m.droppedSettledLines = 0 m.userRows = nil + m.rebuilding = true entries := m.History.Entries() for i := range entries { m.appendBlock(entries[:i], entries[i]) } - m.scrollOffset = offset + m.rebuilding = false + m.settledOffset = len(m.headerLines()) + m.scrollOffset = max(0, m.settledOffset+len(m.settledLines)-fromEnd) + m.trimSettledLines() +} + +// headerLines lead the transcript: where older history stands. Their count +// only changes with what is loaded, and refreshViewportContent keeps the +// reading position steady when it does. +func (m ChatModel) headerLines() []string { + indent := m.Renderer.rail(laneNone) + switch { + case m.loadingOlder: + return []string{indent + m.Styles.Faint.Render("↑ loading earlier messages…"), ""} + case m.hasOlder(): + return []string{indent + m.Styles.Faint.Render("↑ earlier messages load as you scroll up"), ""} + } + if notice := m.History.TruncationNotice(); notice != "" { + return []string{indent + m.Styles.Warning.Render(notice), ""} + } + return nil +} + +// hasOlder is whether scrolling past the top can show more: rows trimmed +// from the rendering, or transcript the daemon has not sent. +func (m ChatModel) hasOlder() bool { + _, more := m.olderCursor() + return m.droppedSettledLines > 0 || more +} + +// olderCursor is the row the next older page ends before. Evicted entries +// move it forward: everything up to the newest evicted row is re-read. +func (m ChatModel) olderCursor() (int64, bool) { + before, more := m.olderBefore, m.olderMore + if through := m.History.EvictedThrough(); through > 0 && through >= before { + before, more = through+1, true + } + return before, more && before > 0 } // appendBlock renders entry after the settled entries before it. @@ -423,10 +489,21 @@ func (m *ChatModel) appendBlock(before []HistoryEntry, entry HistoryEntry) { m.trimSettledLines() } +// trimSettledLines drops the oldest rows past the caps. While you are scrolled +// up the caps stretch by readingSlack and the row you are reading is never +// dropped, so output arriving meanwhile cannot push your place off the top; +// the normal caps apply again once the transcript follows its end. func (m *ChatModel) trimSettledLines() { + if m.rebuilding { + return + } + maxLines, maxBytes := MaxSettledLines, int64(MaxSettledLinesBytes) + if !m.Follow { + maxLines, maxBytes = maxLines*readingSlack, maxBytes*readingSlack + } dropped := 0 - for len(m.settledLines) > MaxSettledLines || m.settledLinesBytes > MaxSettledLinesBytes { - if len(m.settledLines) == 0 { + for len(m.settledLines) > maxLines || m.settledLinesBytes > maxBytes { + if len(m.settledLines) == 0 || !m.Follow && dropped >= m.scrollOffset-m.settledOffset { break } removed := m.settledLines[0] @@ -487,16 +564,11 @@ func (m *ChatModel) refreshViewportContent() int { m.SetSize(m.Width, m.Height) return m.scrollLimit } - var allLines []string - - indent := m.Renderer.rail(laneNone) - if notice := m.History.TruncationNotice(); notice != "" { - allLines = append(allLines, indent+m.Styles.Warning.Render(notice), "") - } - if m.droppedSettledLines > 0 { - allLines = append(allLines, indent+m.Styles.Faint.Render(fmt.Sprintf("[%d scrollback lines truncated]", m.droppedSettledLines)), "") + allLines := m.headerLines() + if !m.Follow { + // a header growing or shrinking above must not move what you read + m.scrollOffset = max(0, m.scrollOffset+len(allLines)-m.settledOffset) } - m.settledOffset = len(allLines) allLines = append(allLines, m.settledLines...) @@ -731,7 +803,70 @@ func (m ChatModel) Init() tea.Cmd { return tea.Batch(m.startStreamSubscription(), m.statusCmd()) } +// Update handles msg, then fetches older history when it left you at the top. func (m ChatModel) Update(msg tea.Msg) (ChatModel, tea.Cmd) { + if loaded, ok := msg.(ChatOlderLoadedMsg); ok { + return m.showOlder(loaded), nil + } + m, cmd := m.update(msg) + if older := m.loadOlder(); older != nil { + return m, tea.Batch(cmd, older) + } + return m, cmd +} + +// loadOlder runs when you are scrolled to the top of the transcript: rows +// trimmed from the rendering come back from retained history first, then a +// page is asked of the daemon. +func (m *ChatModel) loadOlder() tea.Cmd { + if m.Follow || m.scrollOffset > 0 || m.loadingOlder { + return nil + } + if m.droppedSettledLines > 0 { + m.rebuildSettledLines() + m.refreshViewportContent() + return nil + } + before, more := m.olderCursor() + if !more || m.Client == nil { + return nil + } + m.loadingOlder = true + m.refreshViewportContent() + client, id, generation := m.Client, m.SessionID, m.Generation + return func() tea.Msg { + page, err := client.History(context.Background(), before, olderPageRows) + return ChatOlderLoadedMsg{SessionID: id, Generation: generation, Page: page, Err: err} + } +} + +// showOlder puts a fetched page above what is shown without moving it. +func (m ChatModel) showOlder(msg ChatOlderLoadedMsg) ChatModel { + if msg.SessionID != m.SessionID || msg.Generation != m.Generation || !m.loadingOlder { + return m + } + m.loadingOlder = false + if msg.Err != nil { + m.ErrorNotice = fmt.Sprintf("could not load earlier messages: %v", msg.Err) + m.refreshViewportContent() + return m + } + // The page renders exactly as a reset would, in a scratch transcript. + scratch := NewChatModel(&daemon.Session{ID: m.SessionID}, nil) + scratch.AgentName = m.AgentName + scratch.History = NewBoundedHistory(1<<30, 1<<40) + for _, evt := range msg.Page.Events { + scratch.handleStreamEvent(evt) + } + scratch.settleActiveStream() + m.History.Prepend(scratch.History.Entries()) + m.olderBefore, m.olderMore = msg.Page.Before, msg.Page.More && msg.Page.Before > 0 + m.rebuildSettledLines() + m.refreshViewportContent() + return m +} + +func (m ChatModel) update(msg tea.Msg) (ChatModel, tea.Cmd) { var cmds []tea.Cmd switch msg := msg.(type) { @@ -1345,6 +1480,10 @@ func (m *ChatModel) handleStreamEvent(evt daemon.StreamEvent) { m.scrollOffset = 0 m.pendingUsers = nil m.turn = nil + m.olderBefore, m.olderMore, m.loadingOlder = evt.Before, evt.More, false + + case daemon.EventCommitted: + m.History.Stamp(evt.Seq) case daemon.EventRetry: m.activeKind = StreamKindNone diff --git a/cli/internal/tui/chat_navigation_test.go b/cli/internal/tui/chat_navigation_test.go index 60758af..bf42279 100644 --- a/cli/internal/tui/chat_navigation_test.go +++ b/cli/internal/tui/chat_navigation_test.go @@ -158,3 +158,35 @@ func TestCompactTranscriptEntries(t *testing.T) { t.Fatalf("failure should remain visible without flooding transcript: %q", collapsed) } } + +func TestReadingPositionSurvivesOutputPastTheLineCap(t *testing.T) { + m := NewChatModel(&daemon.Session{ID: "s"}, nil) + m.SetSize(80, 20) + block := func(name string) HistoryEntry { + return HistoryEntry{Kind: EntryAssistant, Text: name + "\n" + strings.Repeat("line\n", 15)} + } + for i := range 60 { + m.appendSettledEntry(block(fmt.Sprintf("block %d", i))) + } + m.Follow = false + m.scrollOffset = len(m.settledLines) / 2 + m.refreshViewportContent() + view, offset := m.Viewport.View(), m.scrollOffset + // Enough output to trim everything above the reading row at the normal cap. + for i := range 40 { + m.appendSettledEntry(block(fmt.Sprintf("new %d", i))) + m.refreshViewportContent() + if m.Viewport.View() != view || m.scrollOffset != offset { + t.Fatalf("after %d blocks the view moved: offset %d → %d", i+1, offset, m.scrollOffset) + } + } + m.SetSize(80, 20) + if m.Viewport.View() != view { + t.Fatal("re-rendering moved the reading position") + } + m.scrollBy(1 << 20) + m.appendSettledEntry(block("tail")) + if !m.Follow || len(m.settledLines) > MaxSettledLines { + t.Fatalf("caps not restored when following: follow=%v lines=%d", m.Follow, len(m.settledLines)) + } +} diff --git a/cli/internal/tui/older_history_test.go b/cli/internal/tui/older_history_test.go new file mode 100644 index 0000000..690e057 --- /dev/null +++ b/cli/internal/tui/older_history_test.go @@ -0,0 +1,123 @@ +package tui + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "albedo/cli/internal/daemon" + tea "github.com/charmbracelet/bubbletea" + "github.com/charmbracelet/x/ansi" +) + +// turn is one user message and its reply, committed as rows seq and seq+1. +func turn(name string, seq int64) []daemon.StreamEvent { + return []daemon.StreamEvent{ + {Type: daemon.EventUser, Text: name, Source: "chat", Replayed: true}, + {Type: daemon.EventCommitted, Seq: seq, Replayed: true}, + {Type: daemon.EventMessage, Role: "assistant", Text: "reply to " + name + "\n" + strings.Repeat("more\n", 8), Replayed: true}, + {Type: daemon.EventCommitted, Seq: seq + 1, Replayed: true}, + } +} + +func rowOf(view, text string) int { + for i, line := range strings.Split(ansi.Strip(view), "\n") { + if strings.Contains(line, text) { + return i + } + } + return -1 +} + +func TestScrollingToTheTopLoadsOlderHistoryInPlace(t *testing.T) { + var asked []string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + asked = append(asked, r.URL.RawQuery) + _ = json.NewEncoder(w).Encode(map[string]any{ + "events": []any{ + map[string]any{"type": "user", "text": "older question", "source": "chat"}, + map[string]any{"type": "committed", "seq": 8}, + map[string]any{"type": "message", "text": "older answer"}, + map[string]any{"type": "committed", "seq": 9}, + }, + "before": 8, + "more": false, + }) + })) + defer server.Close() + client := daemon.NewChatClient(daemon.ChatClientOptions{BaseURL: server.URL, AgentID: "s"}) + m := NewChatModel(&daemon.Session{ID: "s"}, client) + if client.Tail != olderPageRows { + t.Fatalf("stream does not ask for a tail: %d", client.Tail) + } + m.SetSize(80, 20) + m.handleStreamEvent(daemon.StreamEvent{Type: daemon.EventReset, Before: 10, More: true, Replayed: true}) + for i, name := range []string{"first", "second", "third"} { + for _, evt := range turn(name, int64(10+2*i)) { + m.handleStreamEvent(evt) + } + } + m.refreshViewportContent() + + m.Follow = false + m.scrollOffset = 1 + m.refreshViewportContent() + m, cmd := m.Update(tea.KeyMsg{Type: tea.KeyUp}) + if cmd == nil || !m.loadingOlder { + t.Fatal("reaching the top did not ask for older history") + } + before := rowOf(m.Viewport.View(), "first") + if before < 0 { + t.Fatal("the first turn is not on screen before loading") + } + m, _ = m.Update(cmd()) + if len(asked) != 1 || !strings.Contains(asked[0], "before=10") { + t.Fatalf("asked %v", asked) + } + if got := rowOf(m.Viewport.View(), "first"); got != before { + t.Fatalf("loading older history moved the view: row %d → %d", before, got) + } + entries := m.History.Entries() + if entries[0].Text != "older question" || entries[0].Seq != 8 { + t.Fatalf("older page not prepended: %+v", entries[0]) + } + if _, more := m.olderCursor(); more { + t.Fatal("history start not recorded") + } + m.scrollOffset = 0 + m.refreshViewportContent() + if !strings.Contains(ansi.Strip(m.Viewport.View()), "older question") { + t.Fatal("older page not shown above") + } + if _, cmd := m.Update(tea.KeyMsg{Type: tea.KeyUp}); cmd != nil { + t.Fatal("asked again with nothing older") + } +} + +func TestEvictedHistoryResumesAfterItsNewestRow(t *testing.T) { + h := NewBoundedHistory(3, 1<<20) + for seq := int64(1); seq <= 3; seq++ { + h.Append(HistoryEntry{Kind: EntryAssistant, Text: fmt.Sprint(seq)}) + h.Stamp(seq) + } + // Two entries from row 4: the first eviction takes row 1 only. + h.Append(HistoryEntry{Kind: EntryThinking, Text: "4a"}) + h.Append(HistoryEntry{Kind: EntryAssistant, Text: "4b"}) + h.Stamp(4) + if h.EvictedThrough() != 2 || h.Entries()[0].Text != "3" { + t.Fatalf("evicted through %d, first %q", h.EvictedThrough(), h.Entries()[0].Text) + } + // Evicting into row 4 takes both of its entries. + h.Append(HistoryEntry{Kind: EntryAssistant, Text: "5"}) + h.Append(HistoryEntry{Kind: EntryAssistant, Text: "6"}) + if h.EvictedThrough() != 4 || h.Entries()[0].Text != "5" { + t.Fatalf("row 4 split: evicted through %d, first %q", h.EvictedThrough(), h.Entries()[0].Text) + } + h.Prepend([]HistoryEntry{{Kind: EntryUser, Text: "old", Seq: 4}}) + if h.EvictedThrough() != 0 || h.Entries()[0].Text != "old" { + t.Fatal("prepend did not take over the cursor") + } +} diff --git a/src/albedo/daemon/conversation.gleam b/src/albedo/daemon/conversation.gleam index 2c25ca7..3ae899a 100644 --- a/src/albedo/daemon/conversation.gleam +++ b/src/albedo/daemon/conversation.gleam @@ -5,6 +5,7 @@ import albedo/daemon/transcript import albedo/daemon/usage import albedo/openai_api/types import gleam/dynamic/decode +import gleam/int import gleam/list import gleam/option.{type Option, None, Some} import gleam/result @@ -595,6 +596,144 @@ fn load_source_pages( } } +/// Rows read per step when a tail that starts on a tool result is widened. +const tail_step_rows = 32 + +/// At most this many extra rows widen a tail. +const tail_step_limit = 512 + +/// The newest `rows` transcript rows before `before` (exclusive; None reads +/// from the end), chronological. A page never starts on a tool result, whose +/// call would be on the page before: it is widened back to the nearest row +/// that is not one. The flag says whether older rows remain. Clients page +/// history with this instead of loading it whole. +pub fn load_tail( + store: store.Store, + id: String, + before: Option(Int), + rows: Int, +) -> Result(#(List(transcript.SourcedEntry), Bool), String) { + let upper = option.unwrap(before, 9_223_372_036_854_775_807) + use #(newest, full) <- result.try(read_before( + store, + id, + upper, + int.max(1, rows), + )) + widen(store, id, newest, full, 0) +} + +fn widen( + store: store.Store, + id: String, + entries: List(transcript.SourcedEntry), + full: Bool, + extra: Int, +) -> Result(#(List(transcript.SourcedEntry), Bool), String) { + case entries { + [] -> Ok(#([], False)) + [first, ..] -> + case + is_tool_result(first.entry.input) && full && extra < tail_step_limit + { + False -> Ok(#(entries, full)) + True -> { + use #(older, older_full) <- result.try(read_before( + store, + id, + first.source.seq, + tail_step_rows, + )) + // Keep only what is needed: from the newest row that is not a + // tool result. Older rows left out mean more remain. + let skipped = + older + |> list.reverse + |> list.take_while(fn(item) { is_tool_result(item.entry.input) }) + |> list.length + case list.length(older) - skipped - 1 { + head if head >= 0 -> + Ok(#( + list.append(list.drop(older, head), entries), + head > 0 || older_full, + )) + _ -> + widen( + store, + id, + list.append(older, entries), + older_full, + extra + tail_step_rows, + ) + } + } + } + } +} + +fn is_tool_result(input: types.Input) -> Bool { + case input { + types.ToolOutput(..) -> True + _ -> False + } +} + +/// Up to `limit` rows before `upper`, chronological, and whether the read +/// was full (so older rows may remain). +fn read_before( + store: store.Store, + id: String, + upper: Int, + limit: Int, +) -> Result(#(List(transcript.SourcedEntry), Bool), String) { + store.query(store, fn(db) { + use rows <- result.try( + sqlight.query( + "SELECT seq,payload,timestamp,provider FROM transcript WHERE session=? AND seq result.map_error(fn(e) { e.message }), + ) + use entries <- result.try( + list.try_map(list.reverse(rows), fn(row) { + use input <- result.try( + unpack(row.1) + |> result.replace_error("invalid saved transcript item"), + ) + Ok(transcript.SourcedEntry( + transcript.SourceRef(id, row.0), + transcript.Entry(input, row.2, row.3), + )) + }), + ) + Ok(#(entries, list.length(rows) == limit)) + }) +} + +/// The session's newest transcript sequence, or 0 when it has none. +pub fn last_seq(store: store.Store, id: String) -> Result(Int, String) { + store.query(store, fn(db) { + sqlight.query( + "SELECT COALESCE(MAX(seq),0) FROM transcript WHERE session=?", + db, + [sqlight.text(id)], + decode.field(0, decode.int, decode.success), + ) + |> result.map_error(fn(e) { e.message }) + |> result.try(fn(rows) { + list.first(rows) |> result.replace_error("session not found") + }) + }) +} + /// Resolve one reference by both session and sequence. A missing row is an /// ordinary result (for example, a reference from a different database). pub fn source( diff --git a/src/albedo/daemon/events.gleam b/src/albedo/daemon/events.gleam index 835025e..fb1bb3f 100644 --- a/src/albedo/daemon/events.gleam +++ b/src/albedo/daemon/events.gleam @@ -161,47 +161,91 @@ pub fn snapshot( entries: List(transcript.Entry), latest_usage: Option(usage.Metadata), ) -> List(String) { - let tool_calls = - list.flat_map(entries, fn(entry) { calls(entry.input) }) - |> list.map(fn(call) { #(call.id, call) }) - |> dict.from_list - let rendered = - list.flat_map(entries, fn(entry) { - case entry.input { - types.User(value) -> - case note.parse(value) { - Some(#(origin, body)) -> [user(body, origin, None, entry.timestamp)] - None -> [user(value, "chat", None, entry.timestamp)] - } - types.UserImage(value, image) -> [ - user_image(value, "user", None, entry.timestamp, image), - ] - types.Assistant(_) -> assistant_message(entry.input, entry.timestamp) - types.ToolOutput(id, output, images) -> { - let call = dict.get(tool_calls, id) - case call { - Ok(call) -> [tool(store, call, output, images)] - Error(_) -> [] - } - } - types.Replay(item) -> { - let thinking = thinking_text(item) - list.append( - case thinking { - "" -> [] - _ -> [event("thinking", [#("text", json.string(thinking))])] - }, - assistant_message(entry.input, entry.timestamp), - ) - } - } - }) + let tool_calls = calls_by_id(entries) + let rendered = list.flat_map(entries, render(store, tool_calls, _)) case latest_usage { Some(metadata) -> list.append(rendered, [usage.event(metadata)]) None -> rendered } } +/// Rendered transcript rows, each followed by a `committed` marker naming +/// its row: a client stamps what it shows with the rows it came from, and so +/// knows where to resume when it asks for older history. +pub fn rows( + store: store.Store, + entries: List(transcript.SourcedEntry), +) -> List(String) { + let tool_calls = calls_by_id(list.map(entries, fn(item) { item.entry })) + list.flat_map(entries, fn(item) { + case render(store, tool_calls, item.entry) { + [] -> [] + events -> list.append(events, [committed(item.source.seq)]) + } + }) +} + +/// Where a rendered page starts: `before` is its first row, the cursor for +/// the next older page, and `more` says whether one exists. +pub fn page_fields( + entries: List(transcript.SourcedEntry), + more: Bool, +) -> List(#(String, json.Json)) { + case entries { + [first, ..] -> [ + #("before", json.int(first.source.seq)), + #("more", json.bool(more)), + ] + [] -> [#("more", json.bool(False))] + } +} + +/// Everything shown so far is covered by transcript rows up to `seq`. +pub fn committed(seq: Int) -> String { + event("committed", [#("seq", json.int(seq))]) +} + +fn calls_by_id( + entries: List(transcript.Entry), +) -> dict.Dict(String, types.ToolCall) { + list.flat_map(entries, fn(entry) { calls(entry.input) }) + |> list.map(fn(call) { #(call.id, call) }) + |> dict.from_list +} + +fn render( + store: store.Store, + tool_calls: dict.Dict(String, types.ToolCall), + entry: transcript.Entry, +) -> List(String) { + case entry.input { + types.User(value) -> + case note.parse(value) { + Some(#(origin, body)) -> [user(body, origin, None, entry.timestamp)] + None -> [user(value, "chat", None, entry.timestamp)] + } + types.UserImage(value, image) -> [ + user_image(value, "user", None, entry.timestamp, image), + ] + types.Assistant(_) -> assistant_message(entry.input, entry.timestamp) + types.ToolOutput(id, output, images) -> + case dict.get(tool_calls, id) { + Ok(call) -> [tool(store, call, output, images)] + Error(_) -> [] + } + types.Replay(item) -> { + let thinking = thinking_text(item) + list.append( + case thinking { + "" -> [] + _ -> [event("thinking", [#("text", json.string(thinking))])] + }, + assistant_message(entry.input, entry.timestamp), + ) + } + } +} + pub fn output_text(item: types.ReplayItem) -> String { case types.replay_protocol(item) { types.ChatCompletions -> diff --git a/src/albedo/daemon/history.gleam b/src/albedo/daemon/history.gleam index 293db9c..149fca9 100644 --- a/src/albedo/daemon/history.gleam +++ b/src/albedo/daemon/history.gleam @@ -7,6 +7,7 @@ import albedo/harness/extensions/lcm/graph as lcm_graph import albedo/openai_api/types import gleam/dynamic/decode import gleam/int +import gleam/json import gleam/list import gleam/option.{type Option, None, Some} import gleam/result @@ -82,6 +83,30 @@ pub fn page( }) } +/// Up to `rows` transcript rows before `before` (None: the newest), rendered +/// as the stream renders them, with `committed` markers, as a JSON body: +/// {"events":[...],"before":first row,"more":older rows remain}. +pub fn rendered( + ledger: store.Store, + session_id: String, + before: Option(Int), + rows: Int, +) -> Result(String, String) { + use #(entries, more) <- result.try(conversation.load_tail( + ledger, + session_id, + before, + rows, + )) + let page = json.object(events.page_fields(entries, more)) |> json.to_string + Ok( + "{\"events\":[" + <> string.join(events.rows(ledger, entries), ",") + <> "]," + <> string.drop_start(page, 1), + ) +} + pub type Recent { Recent(items: List(Item), total: Int) } diff --git a/src/albedo/daemon/server.gleam b/src/albedo/daemon/server.gleam index 4d70f47..c7fa25e 100644 --- a/src/albedo/daemon/server.gleam +++ b/src/albedo/daemon/server.gleam @@ -53,6 +53,7 @@ type Message { Host(Subject(runtime.Runtime)) ReadTree(String, Int, Int, Subject(Result(history.Page, String))) ReadRecent(String, Int, Subject(Result(history.Recent, String))) + ReadHistory(String, Option(Int), Int, Subject(Result(String, String))) Fork(String, Int, Subject(Result(conversation.Info, String))) DeleteSession(String, Subject(Result(Nil, String))) SetWorkspace(String, String, Subject(Result(conversation.Info, String))) @@ -253,6 +254,13 @@ fn handle(state: State, message: Message) { process.send(reply, history.recent(runtime.ledger(state.host), id, limit)) actor.continue(state) } + ReadHistory(id, before, rows, reply) -> { + process.send( + reply, + history.rendered(runtime.ledger(state.host), id, before, rows), + ) + actor.continue(state) + } Fork(id, checkpoint, reply) -> { let forked = history.fork(runtime.ledger(state.host), id, new_id(), checkpoint) @@ -888,6 +896,27 @@ fn daemon_route( Error(e) -> error(400, e) } } + // Older transcript as rendered events, for a client scrolled past what + // its reset carried: `before` is the `before` of the page it holds. + Get, ["sessions", id, "history"] -> { + let query = request.get_query(req) |> result.unwrap([]) + let before = + list.key_find(query, "before") + |> result.try(int.parse) + |> option.from_result + let rows = + list.key_find(query, "rows") + |> result.try(int.parse) + |> result.unwrap(history_page_rows) + |> int.clamp(1, 400) + case actor.call(registry, 10_000, ReadHistory(id, before, rows, _)) { + Ok(body) -> + response.new(200) + |> response.set_header("content-type", "application/json") + |> response.set_body(mist.Bytes(bytes_tree.from_string(body))) + Error(e) -> error(400, e) + } + } Get, ["sessions", id, "context", section, page] -> case int.parse(page) @@ -1100,13 +1129,22 @@ fn when_running( } } +/// Rows a client asks for per history page when it does not say. +const history_page_rows = 120 + fn stream(req, worker) { + let query = request.get_query(req) |> result.unwrap([]) let after = - request.get_query(req) - |> result.unwrap([]) - |> list.key_find("after_seq") + list.key_find(query, "after_seq") |> result.try(int.parse) |> result.unwrap(-1) + // With `tail`, a reset replays only the newest rows; older ones are paged + // from /sessions/:id/history. Without it, the whole transcript as before. + let tail = + list.key_find(query, "tail") + |> result.try(int.parse) + |> option.from_result + |> option.map(int.clamp(_, 1, 400)) mist.server_sent_events( req, response.new(200), @@ -1125,7 +1163,7 @@ fn stream(req, worker) { // A closing daemon stops session workers while clients are still attached, // so a dead worker ends this stream instead of failing a call into it. use <- when_running(worker) - let page = session.read(worker, state.1) + let page = session.read(worker, state.1, tail) case message, page.events { Wake, [] -> actor.continue(state) _, _ -> { diff --git a/src/albedo/daemon/session.gleam b/src/albedo/daemon/session.gleam index e11bb6e..5d454d4 100644 --- a/src/albedo/daemon/session.gleam +++ b/src/albedo/daemon/session.gleam @@ -98,7 +98,7 @@ pub type Message { ReadEffort(Subject(Result(json.Json, String))) ChangeEffort(String, Subject(Result(json.Json, String))) Status(Subject(String)) - Read(Int, Subject(Page)) + Read(Int, Option(Int), Subject(Page)) Watch(process.Pid, fn() -> Nil) Publish(String, String, Subject(Bool)) Commit( @@ -256,8 +256,11 @@ pub fn watch(session: Session, owner: process.Pid, notify: fn() -> Nil) -> Nil { process.send(session, Watch(owner, notify)) } -pub fn read(session: Session, after: Int) -> Page { - actor.call(session, 5000, Read(after, _)) +/// Events after `after`. A reset (a new or lagging client) replays the +/// transcript: whole, or with `tail` only its newest rows, which the client +/// pages back from with `history.rendered`. +pub fn read(session: Session, after: Int, tail: Option(Int)) -> Page { + actor.call(session, 5000, Read(after, tail, _)) } pub fn context(session: Session) -> json.Json { @@ -666,7 +669,7 @@ fn handle(state: State, message: Message) { }) ]), ) - Read(after, reply) -> { + Read(after, tail, reply) -> { let oldest = list.last(state.events) |> result.map(fn(e) { e.0 }) @@ -682,6 +685,34 @@ fn handle(state: State, message: Message) { process.send(reply, Page(state.sequence, False, events)) actor.continue(state) } + True if tail != None -> { + let rows = option.unwrap(tail, 0) + let events = case + conversation.load_tail( + runtime.ledger(state.host), + state.info.id, + None, + rows, + ) + { + Error(error) -> [ + view.event("reset", []), + view.text("error", "could not load transcript: " <> error), + ] + Ok(#(entries, more)) -> [ + view.event("reset", view.page_fields(entries, more)), + ..list.append( + view.rows(runtime.ledger(state.host), entries), + case state.latest_usage { + Some(metadata) -> [usage.event(metadata)] + None -> [] + }, + ) + ] + } + process.send(reply, Page(state.sequence, True, events)) + actor.continue(state) + } True -> case session_history.ensure_history(state) { Error(error) -> { diff --git a/src/albedo/daemon/session_history.gleam b/src/albedo/daemon/session_history.gleam index 5168da5..09f7dec 100644 --- a/src/albedo/daemon/session_history.gleam +++ b/src/albedo/daemon/session_history.gleam @@ -96,7 +96,7 @@ pub fn remember( // An unloaded transcript stays unloaded: the entries are already durable and // the next load reads them. Starting a list here would pass a transcript of // only these entries off as the whole conversation. - case state.history { + let state = case state.history { Some(history) -> session_state.State( ..state, @@ -104,4 +104,19 @@ pub fn remember( ) None -> state } + // A zero timestamp is a candidate projection, not a commit. + case timestamp, inputs { + 0, _ | _, [] -> state + _, _ -> mark_committed(state) + } +} + +/// Tells clients which rows now cover what they have shown (see view.rows). +fn mark_committed( + state: session_state.State(message), +) -> session_state.State(message) { + case conversation.last_seq(runtime.ledger(state.host), state.info.id) { + Ok(seq) -> session_state.emit(state, view.committed(seq)) + Error(_) -> state + } } diff --git a/test/daemon/history_paging_test.gleam b/test/daemon/history_paging_test.gleam new file mode 100644 index 0000000..b349187 --- /dev/null +++ b/test/daemon/history_paging_test.gleam @@ -0,0 +1,145 @@ +import albedo/daemon/conversation +import albedo/daemon/history +import albedo/daemon/session +import albedo/daemon/transcript +import albedo/harness/runtime +import albedo/openai_api/types +import gleam/list +import gleam/option.{None, Some} +import gleam/string +import gleeunit/should + +@external(erlang, "albedo_runtime_test_support", "temporary_database") +fn temporary_database() -> String + +@external(erlang, "albedo_runtime_test_support", "cleanup") +fn cleanup(path: String) -> Nil + +fn info(id: String) -> conversation.Info { + conversation.Info( + id, + "new session", + "/tmp", + "provider", + "model", + types.Responses, + conversation.Idle, + None, + None, + ) +} + +/// Rows 1..7: u1 a1 u2 a2 a3 u3 a4. +fn seeded() { + let path = temporary_database() + let assert Ok(host) = runtime.start(path) + let ledger = runtime.ledger(host) + let assert Ok(_) = conversation.initialise(ledger) + let assert Ok(_) = conversation.create(ledger, info("s")) + let assert Ok(_) = + conversation.commit( + ledger, + "s", + [ + types.User("u1"), + types.Assistant("a1"), + types.User("u2"), + types.Assistant("a2"), + types.Assistant("a3"), + types.User("u3"), + types.Assistant("a4"), + ], + conversation.Idle, + ) + #(path, host) +} + +fn texts(entries: List(transcript.SourcedEntry)) -> List(String) { + list.map(entries, fn(item) { + case item.entry.input { + types.User(text) | types.Assistant(text) -> text + _ -> "?" + } + }) +} + +pub fn a_tail_starts_at_a_user_message_and_reports_older_rows_test() { + let #(path, host) = seeded() + let ledger = runtime.ledger(host) + + let assert Ok(#(newest, more)) = conversation.load_tail(ledger, "s", None, 2) + texts(newest) |> should.equal(["u3", "a4"]) + more |> should.be_true + + // A page may start on any row but a tool result. + let assert Ok(#(older, more)) = + conversation.load_tail(ledger, "s", Some(6), 2) + texts(older) |> should.equal(["a2", "a3"]) + more |> should.be_true + + // Reading to the transcript's start reports nothing older. + let assert Ok(#(first, more)) = + conversation.load_tail(ledger, "s", Some(4), 10) + texts(first) |> should.equal(["u1", "a1", "u2"]) + more |> should.be_false + + let assert Ok(#([], False)) = conversation.load_tail(ledger, "s", Some(1), 5) + runtime.stop(host) + cleanup(path) +} + +pub fn a_page_never_starts_on_a_tool_result_test() { + let path = temporary_database() + let assert Ok(host) = runtime.start(path) + let ledger = runtime.ledger(host) + let assert Ok(_) = conversation.initialise(ledger) + let assert Ok(_) = conversation.create(ledger, info("t")) + let assert Ok(_) = + conversation.commit( + ledger, + "t", + [ + types.User("u"), + types.Assistant("a"), + types.ToolOutput("c1", "one", []), + types.ToolOutput("c2", "two", []), + ], + conversation.Idle, + ) + // The newest row is a tool result: the page widens back to "a" only. + let assert Ok(#(page, more)) = conversation.load_tail(ledger, "t", None, 1) + list.map(page, fn(item) { item.source.seq }) |> should.equal([2, 3, 4]) + more |> should.be_true + runtime.stop(host) + cleanup(path) +} + +pub fn a_rendered_page_marks_its_rows_and_cursor_test() { + let #(path, host) = seeded() + let assert Ok(body) = history.rendered(runtime.ledger(host), "s", Some(6), 2) + body |> string.contains("\"text\":\"a2\"") |> should.be_true + body |> string.contains("\"text\":\"u3\"") |> should.be_false + body + |> string.contains("{\"type\":\"committed\",\"seq\":5}") + |> should.be_true + body |> string.ends_with("\"before\":4,\"more\":true}") |> should.be_true + runtime.stop(host) + cleanup(path) +} + +pub fn a_tail_reset_replays_only_the_newest_turn_test() { + let #(path, host) = seeded() + let assert Ok(worker) = session.start(host, info("s"), "/tmp") + let page = session.read(worker, -1, Some(2)) + let assert [reset, ..events] = page.events + reset |> should.equal("{\"type\":\"reset\",\"before\":6,\"more\":true}") + list.any(events, string.contains(_, "\"u3\"")) |> should.be_true + list.any(events, string.contains(_, "\"u2\"")) |> should.be_false + list.any(events, string.contains(_, "\"committed\"")) |> should.be_true + // Without a tail the whole transcript replays, as older clients expect. + let full = session.read(worker, -1, None) + list.any(full.events, string.contains(_, "\"u1\"")) |> should.be_true + session.close(worker) + runtime.stop(host) + cleanup(path) +} diff --git a/test/daemon/session_lazy_test.gleam b/test/daemon/session_lazy_test.gleam index c78fd5c..ec64064 100644 --- a/test/daemon/session_lazy_test.gleam +++ b/test/daemon/session_lazy_test.gleam @@ -68,7 +68,7 @@ pub fn history_is_read_from_the_durable_transcript_and_not_kept_test() { conversation.Idle, ) let assert Ok(worker) = session.start(host, info("reload"), "/tmp") - let first = session.read(worker, -1) + let first = session.read(worker, -1, None) list.any(first.events, string.contains(_, "first durable turn")) |> should.be_true // A reset renders the transcript from disk and keeps none of it resident. @@ -82,7 +82,7 @@ pub fn history_is_read_from_the_durable_transcript_and_not_kept_test() { [types.Assistant("written while detached")], conversation.Idle, ) - let reloaded = session.read(worker, -1) + let reloaded = session.read(worker, -1, None) list.any(reloaded.events, string.contains(_, "written while detached")) |> should.be_true session.report(worker).history_loaded |> should.be_false