okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

Aly Raffauffeat(tui): show subagent tasks and progress in agent cards61b31d6

main
branchesmainall refs →
3.0 KiB123 linesraw
1package tui
2
3import (
4	"context"
5	"errors"
6
7	"albedo/cli/internal/daemon"
8
9	tea "charm.land/bubbletea/v2"
10)
11
12// Close stops the stream and invalidates batches, completions, and timers
13// already queued for this view.
14func (m *AgentsViewModel) Close() {
15	m.viewGen++
16	m.stopStream()
17}
18
19func (m *AgentsViewModel) stopStream() {
20	if m.cancel != nil {
21		m.cancel()
22		m.cancel = nil
23	}
24	m.Gen++
25	m.seenMail = nil
26	m.streamReady = false
27	m.snapshotInFlight, m.snapshotDirty = false, false
28}
29
30func (m AgentsViewModel) snapshotCmd(gen int) tea.Cmd {
31	conn, id := m.Conn, m.SessionID
32	return func() tea.Msg {
33		if conn == nil {
34			return agentsSnapshotMsg{Gen: gen, Err: errors.New("daemon connection unavailable")}
35		}
36		tree, err := daemon.GetAgents(m.streamCtx, conn, id)
37		return agentsSnapshotMsg{Gen: gen, Root: tree.Root, Nodes: tree.Nodes, Err: err}
38	}
39}
40
41func (m *AgentsViewModel) startStream(gen int) tea.Cmd {
42	if m.Conn == nil {
43		return nil
44	}
45	if m.cancel != nil {
46		m.cancel()
47	}
48	ctx, cancel := context.WithCancel(context.Background())
49	m.cancel = cancel
50	m.streamCtx = ctx
51	m.streamReady = false
52	m.snapshotInFlight, m.snapshotDirty = false, false
53	events := make(chan tea.Msg, 1)
54	m.events = events
55	conn := m.Conn
56	go func() {
57		defer close(events)
58		err := daemon.StreamAgents(ctx, conn, func(batch []daemon.AgentEvent) error {
59			// Leaving the view stops its consumer. Cancellation must release a
60			// producer blocked on a full queue so it can close the HTTP body.
61			select {
62			case events <- agentsEventsMsg{Gen: gen, Events: batch}:
63				return nil
64			case <-ctx.Done():
65				return ctx.Err()
66			}
67		})
68		select {
69		case events <- agentsStreamClosedMsg{Gen: gen, Err: err}:
70		case <-ctx.Done():
71		}
72	}()
73	return waitAgents(events, gen)
74}
75
76func (m *AgentsViewModel) restartStream() tea.Cmd {
77	m.stopStream()
78	m.snapshotRevision++
79	for _, n := range m.nodes {
80		n.preview = agentTail{}
81		n.tail = agentTail{}
82		n.progressByCallID = nil
83		n.progressOrder = nil
84		n.previewCallID = ""
85		n.lineKind = tailText
86		n.seeded = false
87		n.revision++
88	}
89	m.packets, m.floats = nil, nil
90	m.ticking = false
91	return m.startStream(m.Gen)
92}
93
94func waitAgents(events <-chan tea.Msg, gen int) tea.Cmd {
95	return func() tea.Msg {
96		batch, ok := <-events
97		if !ok {
98			return agentsStreamClosedMsg{Gen: gen}
99		}
100		return batch
101	}
102}
103
104// seedCmd loads the selected agent's recent history once; failures retry.
105func (m *AgentsViewModel) seedCmd() tea.Cmd {
106	n := m.nodes[m.selected]
107	if !m.streamReady || m.snapshotInFlight || m.err != nil || n == nil || n.seeded || n.id == agentsYou || m.Conn == nil {
108		return nil
109	}
110	n.seeded = true
111	conn, gen, id := m.Conn, m.Gen, n.id
112	snapshot, ctx := m.snapshotRevision, m.streamCtx
113	if ctx == nil {
114		ctx = context.Background()
115	}
116	return func() tea.Msg {
117		preview, err := daemon.GetSessionPreview(ctx, conn, id, 10)
118		if err != nil {
119			return agentsSeedErrMsg{Gen: gen, Snapshot: snapshot, ID: id, Err: err}
120		}
121		return agentsSeedMsg{Gen: gen, ID: id, Snapshot: snapshot, Items: preview.Items, Session: preview.Session}
122	}
123}