fix(tui): refresh agents after stream overflow
Subscribe before loading the tree and discard stale refresh replies. Recovery clears unfinished previews while keeping drafts and pending action results.
2 unmodified lines3456789101125 unmodified lines373839404142434445464748495035 unmodified lines868788898490919293948895969798991001 unmodified line102103104105981061071081091101111121131141041151161171181191201211221231101241251261271281141291301311321331341351361371382 unmodified linesimport ( "bufio" "context" "encoding/json" "errors" "fmt" "net/http" "strings")25 unmodified lines } if err := json.Unmarshal(raw, &header); err != nil { return nil, fmt.Errorf("invalid agent event: %w", err) } if header.Type == "" { return nil, errors.New("invalid agent event: missing type") } switch header.Type { case "overflow": return &AgentEvent{Type: "overflow"}, nil case "spawn", "gone", "mail", "running", "text", "thinking", "arguments_delta", "tool_progress", "tool", "user", "message", "error", "interrupted", "progress", "renamed", "closed": default: return nil, nil }35 unmodified lines } return &event, nil}
// StreamAgents delivers nonempty batches until the stream ends or onBatch fails.// StreamAgents delivers the first batch, including an empty readiness batch,// then nonempty batches until the stream ends or onBatch fails.// The caller controls cancellation, including any blocking work in onBatch.func StreamAgents(ctx context.Context, conn *Connection, onBatch func([]AgentEvent) error) error { operation := Operation{Name: "stream agents", Method: http.MethodGet, Path: "/agents/stream", Policy: ReadRecovery} return scanEventStream(ctx, conn, operation, streamLimits{lineBytes: 8 * 1024 * 1024}, func(scanner *bufio.Scanner) error { first := true err := scanEventStream(ctx, conn, operation, streamLimits{requireSSE: true, lineBytes: 8 * 1024 * 1024}, func(scanner *bufio.Scanner) error { for scanner.Scan() { line := scanner.Text() if !strings.HasPrefix(line, "data:") { continue1 unmodified line var batch struct { Events []json.RawMessage `json:"events"` } if err := json.Unmarshal([]byte(strings.TrimSpace(line[5:])), &batch); err != nil { return fmt.Errorf("invalid agent batch: %w", err) return streamFailure(StreamProtocol, fmt.Errorf("invalid agent batch: %w", err)) } if batch.Events == nil { return streamFailure(StreamProtocol, errors.New("invalid agent batch: missing events")) } events := make([]AgentEvent, 0, len(batch.Events)) for _, raw := range batch.Events { event, err := decodeAgentEvent(raw) if err != nil { return err return streamFailure(StreamProtocol, err) } if event != nil { if event.Type == "overflow" && len(batch.Events) != 1 { return streamFailure(StreamProtocol, errors.New("overflow must be the sole agent event")) } events = append(events, *event) } } if len(events) == 0 { if len(events) == 0 && !first { continue } first = false if err := onBatch(events); err != nil { return err return streamFailure(StreamTerminal, err) } if len(events) == 1 && events[0].Type == "overflow" { return nil } } return nil }) return classifyStreamFailure(err)}123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657// Controlled peers can send malformed batches and data after overflow, which// the daemon's agent bus cannot produce through the E2E provider.package daemon
import ( "errors" "io" "net" "net/http" "net/http/httptest" "testing")
func TestAgentStreamReadinessAndOverflowStopDelivery(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/event-stream") _, _ = io.WriteString(w, "data: "+`{"events":[]}`+"\n\n"+ "data: "+`{"events":[]}`+"\n\n"+ "data: "+`{"events":[{"type":"overflow"}]}`+"\n\n"+ "data: "+`{"events":[{"type":"text","session":"s","text":"must stay hidden"}]}`+"\n\n") })) defer server.Close() conn := NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, "") var delivered [][]AgentEvent err := StreamAgents(t.Context(), conn, func(events []AgentEvent) error { delivered = append(delivered, events); return nil }) if err != nil { t.Fatal(err) } if len(delivered) != 2 || len(delivered[0]) != 0 || len(delivered[1]) != 1 || delivered[1][0].Type != "overflow" { t.Fatalf("readiness or overflow delivery changed: %+v", delivered) }}
func TestAgentStreamRejectsMalformedBatchBeforeDelivery(t *testing.T) { for name, batch := range map[string]string{ "missing events": `{}`, "null events": `{"events":null}`, "missing type": `{"events":[{"session":"s"}]}`, "empty type": `{"events":[{"type":""}]}`, "partial batch": `{"events":[{"type":"text","session":"s","text":"must stay hidden"},{"type":"running","session":"s","running":"false"}]}`, "mixed overflow": `{"events":[{"type":"overflow"},{"type":"text","session":"s","text":"must stay hidden"}]}`, } { t.Run(name, func(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/event-stream") _, _ = io.WriteString(w, "data: "+batch+"\n\n") })) defer server.Close() conn := NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, "") err := StreamAgents(t.Context(), conn, func([]AgentEvent) error { t.Error("malformed batch partially delivered"); return nil }) failure, ok := errors.AsType[*StreamError](err) if !ok || failure.Kind != StreamProtocol { t.Fatalf("malformed batch classified incorrectly: %v", err) } }) }}123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197// A controlled peer holds snapshots across stream updates and delivers replies// late. Real provider timing cannot reliably create these response orderings.package tui
import ( "context" "fmt" "net" "net/http" "net/http/httptest" "strings" "sync/atomic" "testing" "time"
"albedo/cli/internal/daemon" tea "charm.land/bubbletea/v2" "github.com/charmbracelet/x/ansi")
func TestAgentsRefreshDiscardsDirtyAndPreviousSubscriptionReplies(t *testing.T) { ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) defer cancel() feeds := []chan string{make(chan string, 4), make(chan string, 4), make(chan string, 4)} failureRelease, dirtyRelease := make(chan struct{}), make(chan struct{}) var streams, snapshots, seeds atomic.Int32 changed := make(chan struct{}, 16) notify := func() { select { case changed <- struct{}{}: default: } } server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch { case r.URL.Path == "/agents/stream": number := streams.Add(1) notify() if number > int32(len(feeds)) { t.Error("unexpected extra subscription") return } w.Header().Set("Content-Type", "text/event-stream") _, _ = fmt.Fprint(w, "data: {\"events\":[]}\n\n") w.(http.Flusher).Flush() for { select { case batch := <-feeds[number-1]: _, _ = fmt.Fprintf(w, "data: %s\n\n", batch) w.(http.Flusher).Flush() if strings.Contains(batch, "overflow") { return } case <-r.Context().Done(): return case <-ctx.Done(): return } } case r.URL.Path == "/agents": number := snapshots.Add(1) notify() if streams.Load() == 0 { t.Error("snapshot preceded subscription readiness") } name := "LIVE_TREE" switch number { case 1: select { case <-failureRelease: case <-ctx.Done(): return } w.WriteHeader(http.StatusServiceUnavailable) _, _ = fmt.Fprint(w, `{"error":"snapshot refused"}`) return case 2: select { case <-dirtyRelease: case <-ctx.Done(): return } name = "DIRTY_TREE" case 4: name = "OLD_SUBSCRIPTION_TREE" case 5: name = "FINAL_TREE" } _, _ = fmt.Fprintf(w, `{"root":"lead","nodes":[{"session":{"id":"lead","model":"test"},"name":%q,"depth":0,"running":false,"closed":false}]}`, name) case strings.HasSuffix(r.URL.Path, "/preview"): preview := "OLD_SUBSCRIPTION_SEED" if seeds.Add(1) > 1 { preview = "FINAL_SEED" } _, _ = fmt.Fprintf(w, `{"items":[{"type":"assistant","preview":%q}]}`, preview) default: t.Errorf("unexpected request: %s", r.URL.Path) w.WriteHeader(http.StatusNotFound) } })) defer server.Close() defer cancel() connection := daemon.NewConnection(daemon.ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port}, "") defer connection.HTTPClient().CloseIdleConnections() model := NewAgentsViewModel(connection, "lead") defer func() { model.Close() }() model.SetSize(120, 30) inbox := make(chan tea.Msg, 16) var run func(tea.Cmd) run = func(cmd tea.Cmd) { if cmd == nil { return } go func() { message := cmd() if batch, ok := message.(tea.BatchMsg); ok { for _, child := range batch { run(child) } return } switch message.(type) { case agentsSnapshotMsg, agentsEventsMsg, agentsStreamClosedMsg, agentsSnapshotRetryMsg, agentsSeedMsg, agentsSeedErrMsg: select { case inbox <- message: case <-ctx.Done(): } } }() } update := func(message tea.Msg) { var cmd tea.Cmd; model, cmd = model.Update(message); run(cmd) } view := func() string { return ansi.Strip(model.View()) } pump := func(done func() bool) { t.Helper() for !done() { select { case message := <-inbox: update(message) case <-changed: case <-ctx.Done(): t.Fatalf("refresh did not settle:\n%s", view()) } } } take := func(matches func(tea.Msg) bool) tea.Msg { t.Helper() for { select { case message := <-inbox: if matches(message) { return message } update(message) case <-ctx.Done(): t.Fatalf("expected response never arrived:\n%s", view()) return nil } } } run(model.Init()) pump(func() bool { return snapshots.Load() == 1 }) feeds[0] <- `{"events":[{"type":"spawn","session":"lead","parent":"lead","name":"first","model":"test","depth":0}]}` // The event must be consumed before releasing the failed snapshot. takeEvent := take(func(message tea.Msg) bool { batch, ok := message.(agentsEventsMsg); return ok && len(batch.Events) > 0 }) update(takeEvent) close(failureRelease) pump(func() bool { return strings.Contains(view(), "snapshot refused") }) pump(func() bool { return snapshots.Load() == 2 }) feeds[0] <- `{"events":[{"type":"renamed","session":"lead","name":"renamed-during-fetch"}]}` update(take(func(message tea.Msg) bool { batch, ok := message.(agentsEventsMsg); return ok && len(batch.Events) > 0 })) close(dirtyRelease) dirty := take(func(message tea.Msg) bool { snapshot, ok := message.(agentsSnapshotMsg) return ok && snapshot.Err == nil }) update(dirty) if strings.Contains(view(), "DIRTY_TREE") { t.Fatal("dirty snapshot replaced the live graph") } pump(func() bool { return strings.Contains(view(), "LIVE_TREE") }) oldSeed := take(func(message tea.Msg) bool { _, ok := message.(agentsSeedMsg); return ok }) feeds[0] <- `{"events":[{"type":"overflow"}]}` oldSnapshot := take(func(message tea.Msg) bool { snapshot, ok := message.(agentsSnapshotMsg) return ok && snapshot.Err == nil }) feeds[1] <- `{"events":[{"type":"overflow"}]}` pump(func() bool { return strings.Contains(view(), "FINAL_SEED") }) update(oldSnapshot) update(oldSeed) if strings.Contains(view(), "OLD_SUBSCRIPTION") || !strings.Contains(view(), "FINAL_TREE") || !strings.Contains(view(), "FINAL_SEED") { t.Fatalf("late success restored stale membership or history:\n%s", view()) } if streams.Load() != 3 || snapshots.Load() != 5 { t.Fatalf("recovery made unexpected requests: streams=%d snapshots=%d", streams.Load(), snapshots.Load()) }}43 unmodified lines4445464748484950515253545556575354585960616263646566616263676869707172735 unmodified lines79808182777879808182838483848586878889909192939465 unmodified lines1601611621631581591601611621631641651661671641651661671681691701711721731741751761771781791801811825 unmodified lines18818919019119219319419519628 unmodified lines2252262272282292302312192322332342352362372382392402412422432442452462472482492502512522533 unmodified lines2572582592602412612622632642655 unmodified lines27127227327425527527627727827928028128226026126228328428528626628728828929029129229329429529629729829930027630130230330430530630730830931031131231331431531631731831932032132228232332432532632717 unmodified lines3453463473483493503513523103533543553563573583593603613623633643653663673683693703713723733743753765 unmodified lines38238338438538638738838939039139239339439539639739839940040140240340440521 unmodified lines42742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246337646446546637938046746846947047147247347447547647747838347948048148248348448548648748848949049149249349449549649749849950050150234 unmodified lines53753853954054154254354454554646 unmodified lines59359459559648959759859960060160260360460560660749660860949861061150061261361461561615 unmodified lines6326336346355246366376386396402 unmodified lines643644645646535647648649650651652653654543655656657658659843 unmodified linesExpand all150315041505150613951507150815091510151116 unmodified lines43 unmodified lines Events []daemon.AgentEvent Gen int}
type agentsStreamClosedMsg struct{ Gen int }type agentsStreamClosedMsg struct { Gen int Err error}type agentsSnapshotRetryMsg struct{ Gen int }
// agentsSeedMsg is an agent's recent history, so its tail starts from where// it is rather than empty.type agentsSeedMsg struct { Snapshot int ID string Items []struct { ID string Items []struct { Type string `json:"type"` Preview string `json:"preview"` } Gen int}type agentsSeedErrMsg struct { Snapshot int Err error ID string Gen int Err error ID string Gen int}
type agentsFrameMsg struct{ Gen int }type agentsSentMsg struct {5 unmodified linestype agentsReconnectMsg struct{ Gen int }
type agentsGenMsg interface{ gen() int }
func (m agentsSnapshotMsg) gen() int { return m.Gen }func (m agentsSeedMsg) gen() int { return m.Gen }func (m agentsSeedErrMsg) gen() int { return m.Gen }func (m agentsEventsMsg) gen() int { return m.Gen }func (m agentsStreamClosedMsg) gen() int { return m.Gen }func (m agentsReconnectMsg) gen() int { return m.Gen }func (m agentsFrameMsg) gen() int { return m.Gen }func (m agentsSentMsg) gen() int { return m.Gen }func (m agentsSnapshotRetryMsg) gen() int { return m.Gen }func (m agentsSnapshotMsg) gen() int { return m.Gen }func (m agentsSeedMsg) gen() int { return m.Gen }func (m agentsSeedErrMsg) gen() int { return m.Gen }func (m agentsEventsMsg) gen() int { return m.Gen }func (m agentsStreamClosedMsg) gen() int { return m.Gen }func (m agentsReconnectMsg) gen() int { return m.Gen }func (m agentsFrameMsg) gen() int { return m.Gen }
type agentWire struct { Parent *string `json:"parent"` Address *string `json:"address"`65 unmodified lines t float64}
type AgentsViewModel struct { last time.Time err error heat map[string]float64 events chan []daemon.AgentEvent cancel context.CancelFunc Conn *daemon.Connection nodes map[string]*agentNode SessionID string root string selected string last time.Time err error heat map[string]float64 events chan tea.Msg streamCtx context.Context streamReady bool snapshotInFlight bool snapshotDirty bool snapshotRevision int cancel context.CancelFunc Conn *daemon.Connection nodes map[string]*agentNode SessionID string root string selected string
notice string confirm string // the agent waiting for y to delete it packets []agentPacket5 unmodified lines Width int clock float64 Height int Gen int viewGen int // User operation outcomes survive stream replacement. noticeT float64 ticking bool // a frame tick is in flight}
28 unmodified lines m.refreshTail()}
func (m *AgentsViewModel) Init() tea.Cmd { m.viewGen++ m.Gen++ m.ticking = true return tea.Batch(m.snapshotCmd(m.Gen), m.startStream(m.Gen), m.frameCmd(m.Gen), textinput.Blink) return tea.Batch(m.startStream(m.Gen), m.frameCmd(m.Gen), textinput.Blink)}
// Close stops the stream and invalidates batches, completions, and timers// already queued for this view.func (m *AgentsViewModel) Close() { m.viewGen++ m.stopStream()}
func (m *AgentsViewModel) stopStream() { if m.cancel != nil { m.cancel() m.cancel = nil } m.Gen++ m.streamReady = false m.snapshotInFlight, m.snapshotDirty = false, false}
func (m AgentsViewModel) snapshotCmd(gen int) tea.Cmd { conn, id := m.Conn, m.SessionID3 unmodified lines } tree, err := daemon.RequestOperation[struct { Root string `json:"root"` Nodes []agentWire `json:"nodes"` }](context.Background(), conn, daemon.Operation{Name: "snapshot", Method: http.MethodGet, Path: "/agents?session=" + url.QueryEscape(id), Body: nil, Policy: daemon.ReadRecovery}) }](m.streamCtx, conn, daemon.Operation{Name: "snapshot", Method: http.MethodGet, Path: "/agents?session=" + url.QueryEscape(id), Body: nil, Policy: daemon.ReadRecovery}) return agentsSnapshotMsg{Gen: gen, Root: tree.Root, Nodes: tree.Nodes, Err: err} }}
5 unmodified lines m.cancel() } ctx, cancel := context.WithCancel(context.Background()) m.cancel = cancel events := make(chan []daemon.AgentEvent, 64) m.streamCtx = ctx m.streamReady = false m.snapshotInFlight, m.snapshotDirty = false, false events := make(chan tea.Msg, 1) m.events = events conn := m.Conn go func() { defer close(events) // EOF and failures both reconnect through queue closure, so the view // deliberately discards the transport error here. _ = daemon.StreamAgents(ctx, conn, func(batch []daemon.AgentEvent) error { err := daemon.StreamAgents(ctx, conn, func(batch []daemon.AgentEvent) error { // Leaving the view stops its consumer. Cancellation must release a // producer blocked on a full queue so it can close the HTTP body. select { case events <- batch: case events <- agentsEventsMsg{Gen: gen, Events: batch}: return nil case <-ctx.Done(): return ctx.Err() } }) select { case events <- agentsStreamClosedMsg{Gen: gen, Err: err}: case <-ctx.Done(): } }() return waitAgents(events, gen)}
func waitAgents(events <-chan []daemon.AgentEvent, gen int) tea.Cmd {func (m *AgentsViewModel) restartStream() tea.Cmd { m.stopStream() m.snapshotRevision++ for _, n := range m.nodes { n.preview = agentPreview{} n.tail = agentTail{} n.call = "" n.lineKind = tailText n.seeded = false n.revision++ } m.packets, m.floats = nil, nil m.ticking = false return m.startStream(m.Gen)}
func waitAgents(events <-chan tea.Msg, gen int) tea.Cmd { return func() tea.Msg { batch, ok := <-events if !ok { return agentsStreamClosedMsg{Gen: gen} } return agentsEventsMsg{Gen: gen, Events: batch} return batch }}
func (m AgentsViewModel) frameCmd(gen int) tea.Cmd {17 unmodified lines return m, nil } switch msg := msg.(type) { case agentsSnapshotMsg: m.snapshotInFlight = false if msg.Err != nil { m.snapshotDirty = false m.err = msg.Err return m, nil m.say("Could not refresh agents; retrying: " + msg.Err.Error()) return m, tea.Tick(time.Second, func(time.Time) tea.Msg { return agentsSnapshotRetryMsg{Gen: m.Gen} }) } if m.snapshotDirty { m.snapshotDirty = false m.snapshotInFlight = true return m, m.snapshotCmd(m.Gen) } m.err, m.root = nil, msg.Root m.snapshotRevision++ previous := m.nodes m.nodes = make(map[string]*agentNode, len(msg.Nodes)) for _, wire := range msg.Nodes { if n := previous[wire.Session.ID]; n != nil { m.nodes[wire.Session.ID] = n } n := m.node(wire.Session.ID, wire.Name) n.name = wire.Name n.parent, n.address = "", "" n.seeded = false n.session = wire.Session n.model = wire.Session.Model n.depth = wire.Depth n.running = wire.Running5 unmodified lines if wire.Parent != nil { n.parent = *wire.Parent } } if m.nodes[m.selected] == nil { m.selected = m.root } if m.nodes[m.confirm] == nil { m.confirm = "" } if m.nodes[m.rename.id] == nil { m.rename = renameField{} } m.layout() return m, m.seedCmd()
case agentsSeedMsg: if msg.Snapshot != m.snapshotRevision { return m, nil } n := m.nodes[msg.ID] if n == nil { return m, nil }21 unmodified lines n.revision++ return m, nil
case agentsSeedErrMsg: if msg.Snapshot != m.snapshotRevision { return m, nil } // The seed died; unseed its node so the next selection tries again. if n := m.nodes[msg.ID]; n != nil { n.seeded = false m.say("Could not load recent messages for " + m.label(msg.ID, "this agent") + ". Select it again to retry: " + msg.Err.Error()) } return m, nil
case agentsEventsMsg: if len(msg.Events) == 1 && msg.Events[0].Type == "overflow" { return m, m.restartStream() } var snapshot tea.Cmd if !m.streamReady { m.streamReady = true m.snapshotInFlight = true snapshot = m.snapshotCmd(m.Gen) } relayout := false for _, event := range msg.Events { if m.snapshotInFlight { switch event.Type { case "spawn", "gone", "running", "renamed", "closed": m.snapshotDirty = true } } relayout = m.apply(event) || relayout } if relayout { m.layout() } return m, waitAgents(m.events, m.Gen) return m, tea.Batch(snapshot, waitAgents(m.events, m.Gen))
case agentsStreamClosedMsg: // The daemon restarted or the connection dropped: reconnect shortly. return m, tea.Tick(time.Second, func(time.Time) tea.Msg { return agentsReconnectMsg(msg) }) if msg.Err != nil { m.err = msg.Err m.say("Agent stream failed; ctrl+l retries: " + msg.Err.Error()) if failure, ok := errors.AsType[*daemon.StreamError](msg.Err); ok && failure.Kind != daemon.StreamTransient { m.stopStream() m.ticking = false return m, nil } } return m, tea.Tick(time.Second, func(time.Time) tea.Msg { return agentsReconnectMsg{Gen: m.Gen} })
case agentsReconnectMsg: return m, tea.Batch(m.snapshotCmd(m.Gen), m.startStream(m.Gen)) return m, m.restartStream()
case agentsSnapshotRetryMsg: if !m.streamReady || m.snapshotInFlight { return m, nil } m.snapshotInFlight = true return m, m.snapshotCmd(m.Gen)
case agentsFrameMsg: m.step() if m.ticking = m.moving(); !m.ticking { return m, nil } return m, m.frameCmd(m.Gen)
case agentsSentMsg: if msg.Gen != m.viewGen { return m, nil } if msg.Err != nil { m.say(operationError(msg.Err, "Could not "+msg.Action+": ", "The request to "+msg.Action+" may have been accepted; check the agent before trying again.")) } else if msg.Notice != "" { m.say(msg.Notice)34 unmodified lines m.say("kept") return m, nil } switch { case s == "ctrl+l": return m, m.restartStream() case s == "ctrl+r": if n := m.nodes[m.selected]; n != nil && n.id != agentsYou { m.rename.open(n.id, n.name, "name this agent") }46 unmodified lines
// seedCmd loads the selected agent's recent history once; failures retry.func (m *AgentsViewModel) seedCmd() tea.Cmd { n := m.nodes[m.selected] if n == nil || n.seeded || n.id == agentsYou || m.Conn == nil { if !m.streamReady || m.snapshotInFlight || m.err != nil || n == nil || n.seeded || n.id == agentsYou || m.Conn == nil { return nil } n.seeded = true conn, gen, id := m.Conn, m.Gen, n.id snapshot, ctx := m.snapshotRevision, m.streamCtx if ctx == nil { ctx = context.Background() } return func() tea.Msg { path := fmt.Sprintf("/sessions/%s/preview?limit=10", url.PathEscape(id)) res, err := daemon.RequestOperation[agentsSeedMsg](context.Background(), conn, daemon.Operation{Name: "seed", Method: http.MethodGet, Path: path, Body: nil, Policy: daemon.ReadRecovery}) res, err := daemon.RequestOperation[agentsSeedMsg](ctx, conn, daemon.Operation{Name: "seed", Method: http.MethodGet, Path: path, Body: nil, Policy: daemon.ReadRecovery}) if err != nil { return agentsSeedErrMsg{Gen: gen, ID: id, Err: err} return agentsSeedErrMsg{Gen: gen, Snapshot: snapshot, ID: id, Err: err} } res.Gen, res.ID = gen, id res.Gen, res.ID, res.Snapshot = gen, id, snapshot return res }}
15 unmodified lines return count}
func (m AgentsViewModel) deleteCmd(id string) tea.Cmd { conn, gen, name := m.Conn, m.Gen, m.label(id, "") conn, gen, name := m.Conn, m.viewGen, m.label(id, "") return func() tea.Msg { res, err := daemon.DeleteSession(context.Background(), conn, id, true) if err != nil { return agentsSentMsg{Gen: gen, Action: "delete " + name, Err: err}2 unmodified lines }}
func (m AgentsViewModel) sendCmd(id, text string) tea.Cmd { conn, gen, target := m.Conn, m.Gen, m.label(id, "agent") conn, gen, target := m.Conn, m.viewGen, m.label(id, "agent") return func() tea.Msg { _, err := daemon.Submit(context.Background(), conn, id, map[string]any{"content": text}) return agentsSentMsg{Gen: gen, Action: "send a message to " + target, Err: err} }}
func (m AgentsViewModel) spawnCmd(parent, name, task string) tea.Cmd { conn, gen := m.Conn, m.Gen conn, gen := m.Conn, m.viewGen return func() tea.Msg { if name == "" || task == "" { return agentsSentMsg{Gen: gen, Action: "start an agent", Err: errors.New("use /spawn <name> <task>")} }843 unmodified linesExpand all if n := m.nodes[m.rename.id]; n != nil && n.address != "" { status += DefaultStyles.Faint.Render(" · its family can still mail it at ") + DefaultStyles.Muted.Render(n.address) } case m.err != nil: status = DefaultStyles.Error.Render(m.err.Error()) status = DefaultStyles.Error.Render(m.err.Error() + " · ctrl+l retries") case m.noticeT > 0: status = DefaultStyles.Muted.Render(m.notice) } if len(confirmRows) > 0 {16 unmodified lines1123456789101011121314154 unmodified lines2021222324252627284 unmodified lines33343536373839404124 unmodified lines66676869677071727374122 unmodified linesExpand all197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259// Four invariants the e2e suite cannot reach, because each lives inside the// Invariants the e2e suite cannot reach, because each lives inside the// tui package or needs a failure the daemon never produces://// - frame pacing: an idle graph must not keep scheduling animation frames,// a running one must resume them exactly once (wasted CPU, double speed);// - the preview decoder must reveal streamed code from incomplete JSON arguments,// which a real stream only ever shows mid-flight;// - a pending delete confirm must not outlive the agent it names;// - a failed history seed must retry on the next selection, and a late// error from an older generation must not unseed the live node.// error from an older generation must not unseed the live node;// - overflow during a blocked mutation must preserve its later outcome and draft.package tui
import ( "encoding/json"4 unmodified lines "slices" "strconv" "strings" "testing" "time"
"albedo/cli/internal/daemon"
tea "charm.land/bubbletea/v2"4 unmodified linesfunc agentsFixture(t *testing.T) AgentsViewModel { t.Helper() m := NewAgentsViewModel(nil, "lead") m.Gen = 1 m.streamReady = true m.SetSize(120, 30) lead := "lead" coder := "coder" m, _ = m.Update(agentsSnapshotMsg{Gen: 1, Root: "lead", Nodes: []agentWire{24 unmodified lines}
func TestAgentsFramesStopWhenStillAndResumeOnce(t *testing.T) { m := agentsFixture(t) closed := make(chan []daemon.AgentEvent) closed := make(chan tea.Msg) close(closed) m.events = closed running := func(id string, on bool) agentsEventsMsg { return agentsEventsMsg{Gen: 1, Events: []daemon.AgentEvent{{Type: "running", Session: id, Running: on}}}122 unmodified linesExpand all if cmd := m.seedCmd(); cmd != nil { t.Fatal("a seeded node should not fetch again") }}
func TestAgentsOverflowPreservesPendingOperationOutcomeAndDraft(t *testing.T) { started, release := make(chan struct{}), make(chan struct{}) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path == "/agents/stream" { <-r.Context().Done() return } if r.URL.Path != "/sessions/lead/children" { t.Errorf("unexpected request: %s", r.URL.Path) w.WriteHeader(404) return } close(started) <-release w.WriteHeader(http.StatusForbidden) _, _ = w.Write([]byte(`{"error":"spawn refused"}`)) })) defer server.Close() defer func() { select { case <-release: default: close(release) } }() address, _ := url.Parse(server.URL) port, _ := strconv.Atoi(address.Port()) m := agentsFixture(t) m.Conn = daemon.NewConnection(daemon.ConnectionSnapshot{Port: port}, "") for _, char := range "/spawn helper inspect files" { m, _ = m.Update(tea.KeyPressMsg{Code: char, Text: string(char)}) } var cmd tea.Cmd m, cmd = m.Update(tea.KeyPressMsg{Code: tea.KeyEnter}) finished := make(chan tea.Msg, 1) go func() { finished <- cmd() }() select { case <-started: case <-time.After(5 * time.Second): t.Fatal("spawn never reached server") } for _, char := range "next draft" { m, _ = m.Update(tea.KeyPressMsg{Code: char, Text: string(char)}) } m, _ = m.Update(agentsEventsMsg{Gen: m.Gen, Events: []daemon.AgentEvent{{Type: "overflow"}}}) defer func() { m.Close() }() close(release) select { case outcome := <-finished: m, _ = m.Update(outcome) case <-time.After(5 * time.Second): t.Fatal("spawn did not complete") } rendered := ansi.Strip(m.View()) if !strings.Contains(rendered, "spawn refused") || !strings.Contains(rendered, "next draft") { t.Fatalf("overflow lost pending outcome or draft: %s", rendered) }}123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156//go:build unix
// The agents view must replace stale membership and previews after a lost feed.package e2e
import ( "context" "encoding/json" "fmt" "net" "net/http" "net/http/httptest" "net/http/httputil" "net/url" "strings" "sync/atomic" "testing" "time"
"albedo/cli/internal/daemon" "albedo/cli/internal/tui" tea "charm.land/bubbletea/v2")
func TestTUIAgentsOverflowRefreshesMembershipAndPreviews(t *testing.T) { entered, release := make(chan struct{}), make(chan struct{}) defer close(release) profile := providerRoute(t, func(request map[string]any) string { if lastUserText(request) == "held agent refresh turn" { close(entered) <-release } return echoReply(request) }) root := daemonSession(t, newSession(t, t.TempDir())) create := func(name string) daemon.Session { t.Helper() child, err := daemon.CreateChild(t.Context(), conn(t), root.ID, map[string]any{"name": name, "task": "finish child setup"}) if err != nil { t.Fatal(err) } waitIdle(t, child.Session.ID, profile, 1) return child.Session } removed := create("removed") retained := create("old-name") waitIdle(t, root.ID, profile, 2) if _, err := daemon.NewChatClient(conn(t), root.ID).Send(t.Context(), "INITIAL_PREVIEW_SEED", nil); err != nil { t.Fatal(err) } waitIdle(t, root.ID, profile, 3)
destination, err := url.Parse(conn(t).BaseURL()) if err != nil { t.Fatal(err) } forward := httputil.NewSingleHostReverseProxy(destination) forward.FlushInterval = -1 var streams atomic.Int32 seedReady := make(chan struct{}, 1) fragments, gap := make(chan struct{}), make(chan struct{}) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path == "/agents/stream" && streams.Add(1) == 1 { w.Header().Set("Content-Type", "text/event-stream") _, _ = fmt.Fprint(w, "data: {\"events\":[]}\n\n") w.(http.Flusher).Flush() select { case <-fragments: case <-r.Context().Done(): return } events, _ := json.Marshal(map[string]any{"events": []map[string]any{ {"type": "text", "session": root.ID, "text": "OLD_PARTIAL_TEXT"}, {"type": "arguments_delta", "session": root.ID, "name": "python", "callId": "old-call", "text": "{\"code\":\"OLD_PARTIAL_CODE"}, }}) _, _ = fmt.Fprintf(w, "data: %s\n\n", events) w.(http.Flusher).Flush() select { case <-gap: _, _ = fmt.Fprint(w, "data: {\"events\":[{\"type\":\"overflow\"}]}\n\n") w.(http.Flusher).Flush() case <-r.Context().Done(): } return } forward.ServeHTTP(w, r) if r.URL.Path == "/sessions/"+root.ID+"/preview" { select { case seedReady <- struct{}{}: default: } } })) t.Cleanup(server.Close) snapshot := conn(t).Snapshot() snapshot.Port = server.Listener.Addr().(*net.TCPAddr).Port connection := daemon.NewConnection(snapshot, "") t.Cleanup(connection.HTTPClient().CloseIdleConnections) driver := driveTUIWithConnection(t, &root, connection) driver.Update(tea.WindowSizeMsg{Width: 140, Height: 50}) t.Cleanup(driver.App.Chat.Close) t.Cleanup(func() { driver.App.Agents.Close() }) ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second) defer cancel() inbox := make(chan tea.Msg, 64) streamMessages(ctx, driver.Update(tui.ChatOpenAgentsMsg{}), inbox) pump := func(condition func() bool) { t.Helper() for !condition() { select { case message := <-inbox: streamMessages(ctx, driver.Update(message), inbox) case <-ctx.Done(): t.Fatalf("agent view did not recover:\n%s", driver.View()) } } } pump(func() bool { return strings.Contains(driver.View(), "INITIAL_PREVIEW_SEED") }) select { case <-seedReady: case <-ctx.Done(): t.Fatal("initial preview request never completed") } close(fragments) pump(func() bool { return strings.Contains(driver.View(), "OLD_PARTIAL_CODE") }) if !strings.Contains(driver.View(), "removed") || !strings.Contains(driver.View(), "old-name") { t.Fatalf("initial membership did not load:\n%s", driver.View()) } if _, err := daemon.RenameSession(t.Context(), conn(t), retained.ID, "renamed"); err != nil { t.Fatal(err) } if _, err := daemon.DeleteSession(t.Context(), conn(t), removed.ID, false); err != nil { t.Fatal(err) } create("new-agent") if _, err := daemon.NewChatClient(conn(t), root.ID).Send(t.Context(), "held agent refresh turn", nil); err != nil { t.Fatal(err) } select { case <-entered: case <-ctx.Done(): t.Fatal("held model turn did not start") } for _, character := range "draft survives refresh" { driver.Update(tea.KeyPressMsg{Code: character, Text: string(character)}) } close(gap) pump(func() bool { view := driver.View() return strings.Contains(view, "renamed") && strings.Contains(view, "new-agent") && strings.Contains(view, "1 running") && !strings.Contains(view, "removed") && !strings.Contains(view, "old-name") }) view := driver.View() if strings.Contains(view, "OLD_PARTIAL") || !strings.Contains(view, "draft survives refresh") { t.Fatalf("refresh retained stale fragments or discarded the input draft:\n%s", view) }}