perf(cli): read history without the shuffle
Count turns per page and assemble the requested history once.
11 unmodified lines121314151616171819202111 unmodified lines3334353637373839404142434444454546474849505152535455565758596054616263646566676869707159607273747576774 unmodified lines11 unmodified lines
// SessionTurns is the newest turns of a session's committed transcript.type SessionTurns struct { Session string `json:"session"` Running bool `json:"running"` Events []daemon.StreamEvent `json:"events"` Running bool `json:"running"`}
// Read answers the newest turns of a session, reading older transcript pages// until it has that many. A turn that is still running is not committed yet,11 unmodified lines status, err := client.GetStatus(ctx) if err != nil { return SessionTurns{}, err } var events []daemon.StreamEvent var pages [][]daemon.StreamEvent eventCount, turnCount := 0, 0 var before int64 for { page, err := client.History(ctx, before, historyPageRows) if err != nil { return SessionTurns{}, err } events = append(page.Events, events...) if !page.More || turnStarts(events) >= turns { pages = append(pages, page.Events) eventCount += len(page.Events) turnCount += turnStarts(page.Events) if !page.More || turnCount >= turns { break } before = page.Before } events := make([]daemon.StreamEvent, 0, eventCount) for _, page := range slices.Backward(pages) { events = append(events, page...) } return SessionTurns{Session: id, Running: status.Running, Events: lastTurns(events, turns)}, nil}
func turnStarts(events []daemon.StreamEvent) int { return len(slices.DeleteFunc(slices.Clone(events), func(e daemon.StreamEvent) bool { return e.Type != daemon.EventUser })) count := 0 for _, event := range events { if event.Type == daemon.EventUser { count++ } } return count}
// lastTurns keeps the events from the start of the turns-th newest turn on.func lastTurns(events []daemon.StreamEvent, turns int) []daemon.StreamEvent { for i := len(events) - 1; i >= 0; i-- { if events[i].Type != daemon.EventUser { for i, event := range slices.Backward(events) { if event.Type != daemon.EventUser { continue } if turns--; turns == 0 { return events[i:]4 unmodified lines1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677// Controlled page boundaries catch ordering and turn selection bugs without// requiring hundreds of provider turns to cross the daemon's page limit.package app
import ( "context" "encoding/json" "net" "net/http" "net/http/httptest" "reflect" "testing"
"albedo/cli/internal/daemon")
func TestReadFlattensHistoryPagesInChronologicalOrder(t *testing.T) { var cursors []string server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.URL.Path { case "/sessions": _, _ = w.Write([]byte(`[{"id":"s"}]`)) case "/sessions/s/status": _, _ = w.Write([]byte(`{"running":true,"idle":false}`)) case "/sessions/s/history": cursor := r.URL.Query().Get("before") cursors = append(cursors, cursor) var body string switch cursor { case "": body = `{"events":[{"type":"user","text":"third"},{"type":"message","text":"third reply"}],"before":5,"more":true}` case "5": body = `{"events":[{"type":"user","text":"second"},{"type":"message","text":"second reply"}],"before":3,"more":true}` case "3": body = `{"events":[{"type":"user","text":"first"},{"type":"message","text":"first reply"}],"before":1,"more":false}` default: t.Errorf("unexpected cursor %s", cursor) } _, _ = w.Write([]byte(body)) default: t.Errorf("unexpected request %s", r.URL.Path) w.WriteHeader(404) } })) defer server.Close() conn := daemon.NewConnection(daemon.ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, "") service := Service{Connect: func(context.Context) (*daemon.Connection, error) { return conn, nil }} for _, turns := range []int{3, 2} { cursors = nil result, err := service.Read(context.Background(), "s", turns) if err != nil { t.Fatal(err) } var texts []string for _, event := range result.Events { texts = append(texts, event.Text) if !event.Replayed { t.Fatal("history event not marked replayed") } } wanted := []string{"first", "first reply", "second", "second reply", "third", "third reply"} if turns == 2 { wanted = wanted[2:] } if !reflect.DeepEqual(texts, wanted) || !result.Running { data, _ := json.Marshal(result) t.Fatalf("history changed order or running state: %s", data) } expectedCursors := []string{"", "5", "3"} if turns == 2 { expectedCursors = expectedCursors[:2] } if !reflect.DeepEqual(cursors, expectedCursors) { t.Fatalf("read fetched wrong pages: %v", cursors) } }}