git.iko.soy/crush
on main help sign in
feat(bang): stream results in
6814ccfKieran Klukas
internal/shell/stream.gonew file+57
compare with main
unifiedsplit
@@ -0,0 +1,57 @@
1+package shell
2+
3+import (
4+ "bytes"
5+ "context"
6+ "log/slog"
7+ "os"
8+)
9+
10+// progressWriter wraps an io.Writer and calls onProgress with each write.
11+type progressWriter struct {
12+ buf bytes.Buffer
13+ onProgress func(string)
14+}
15+
16+func (w *progressWriter) Write(p []byte) (int, error) {
17+ n, err := w.buf.Write(p)
18+ if n > 0 && w.onProgress != nil {
19+ slog.Debug("Shell stream progress", "bytes", n)
20+ w.onProgress(string(p[:n]))
21+ }
22+ return n, err
23+}
24+
25+// RunAndCaptureStream executes a shell command and streams output chunks
26+// to onProgress as they arrive. Returns the complete output and exit code.
27+func RunAndCaptureStream(ctx context.Context, opts RunOptions, onProgress func(string)) (CaptureResult, error) {
28+ if opts.Env == nil {
29+ opts.Env = os.Environ()
30+ }
31+ opts.Env = append(opts.Env, ptyColorEnvVars...)
32+
33+ stdout := &progressWriter{onProgress: onProgress}
34+ stderr := &progressWriter{onProgress: onProgress}
35+ opts.Stdout = stdout
36+ opts.Stderr = stderr
37+
38+ runErr := Run(ctx, opts)
39+
40+ exitCode := 0
41+ if runErr != nil {
42+ exitCode = ExitCode(runErr)
43+ }
44+
45+ output := stdout.buf.String()
46+ if stderr.buf.Len() > 0 {
47+ if output != "" {
48+ output += "\n"
49+ }
50+ output += stderr.buf.String()
51+ }
52+
53+ return CaptureResult{
54+ Output: output,
55+ ExitCode: exitCode,
56+ }, nil
57+}
internal/ui/chat/shell.go+43−11
@@ -101,6 +101,12 @@ func (s *ShellItem) Complete(output string, exitCode int) {
101 s.Bump()
101 s.Bump()
102}
102}
103
103
104+// AppendOutput appends incremental output to a pending ShellItem.
105+func (s *ShellItem) AppendOutput(chunk string) {
106+ s.output += chunk
107+ s.Bump()
108+}
109+
104func (s *ShellItem) ID() string { return s.id }
110func (s *ShellItem) ID() string { return s.id }
105func (s *ShellItem) FilterValue() string { return s.command }
111func (s *ShellItem) FilterValue() string { return s.command }
106func (s *ShellItem) Finished() bool { return !s.pending }
112func (s *ShellItem) Finished() bool { return !s.pending }
@@ -199,10 +205,11 @@ func (s *ShellItem) RawRender(width int) string {
199 header := prompt + " " + highlighted
205 header := prompt + " " + highlighted
200
206
201 if s.pending {
207 if s.pending {
208+ if s.output == "" {
209+ // Nothing streamed yet: show the spinner under the header.
202− return header + "\n" + s.anim.Render()
210+ return header + "\n" + s.anim.Render()
203− }
211+ }
204−
205− if s.exitCode != 0 {
212+ } else if s.exitCode != 0 {
206 header += " " + s.sty.Messages.ShellExitCode.Render(fmt.Sprintf("(exit %d)", s.exitCode))
213 header += " " + s.sty.Messages.ShellExitCode.Render(fmt.Sprintf("(exit %d)", s.exitCode))
207 }
214 }
208
215
@@ -213,14 +220,23 @@ func (s *ShellItem) RawRender(width int) string {
213 output := strings.TrimRight(s.output, "\n")
220 output := strings.TrimRight(s.output, "\n")
214 lines := strings.Split(output, "\n")
221 lines := strings.Split(output, "\n")
215
222
223+ // While streaming, show the tail of the output so the most recent
224+ // lines stay visible without forcing the user to expand.
216 maxLines := shellMaxCollapsedLines
225 maxLines := shellMaxCollapsedLines
217 if s.expandedContent {
226 if s.expandedContent {
218 maxLines = len(lines)
227 maxLines = len(lines)
219 }
228 }
220
229
221 displayLines := lines
230 displayLines := lines
231+ truncatedCount := 0
222 if len(lines) > maxLines {
232 if len(lines) > maxLines {
233+ if s.pending {
234+ // Show the most recent lines while still running.
235+ displayLines = lines[len(lines)-maxLines:]
236+ } else {
223− displayLines = lines[:maxLines]
237+ displayLines = lines[:maxLines]
238+ }
239+ truncatedCount = len(lines) - maxLines
224 }
240 }
225
241
226 // Compute max line width for scroll clamping.
242 // Compute max line width for scroll clamping.
@@ -234,6 +250,16 @@ func (s *ShellItem) RawRender(width int) string {
234 s.maxLineWidth = max(0, maxW-cappedWidth)
250 s.maxLineWidth = max(0, maxW-cappedWidth)
235
251
236 var body strings.Builder
252 var body strings.Builder
253+
254+ // When streaming, hidden lines are above the visible tail, so show
255+ // the "more lines" notice before the output.
256+ if truncatedCount > 0 && s.pending {
257+ body.WriteString(s.sty.Messages.ShellTruncation.Render(
258+ fmt.Sprintf("… %d earlier lines", truncatedCount),
259+ ))
260+ body.WriteString("\n")
261+ }
262+
237 for _, ln := range displayLines {
263 for _, ln := range displayLines {
238 scrolled := ansi.GraphemeWidth.Cut(ln, s.xOffset, len(ln))
264 scrolled := ansi.GraphemeWidth.Cut(ln, s.xOffset, len(ln))
239 truncated := ansi.Truncate(scrolled, cappedWidth, "…")
265 truncated := ansi.Truncate(scrolled, cappedWidth, "…")
@@ -244,14 +270,20 @@ func (s *ShellItem) RawRender(width int) string {
244 body.WriteString("\n")
270 body.WriteString("\n")
245 }
271 }
246
272
273+ // When finished, hidden lines are below, so show the notice after.
247− if len(lines) > maxLines && !s.expandedContent {
274+ if truncatedCount > 0 && !s.pending && !s.expandedContent {
248 body.WriteString(s.sty.Messages.ShellTruncation.Render(
275 body.WriteString(s.sty.Messages.ShellTruncation.Render(
249− fmt.Sprintf("… %d more lines", len(lines)-maxLines),
276+ fmt.Sprintf("… %d more lines", truncatedCount),
250 ))
277 ))
251− } else {
278+ return header + "\n" + body.String()
252− result := body.String()
279+ }
280+
253− return header + "\n" + strings.TrimRight(result, "\n")
281+ result := header + "\n" + strings.TrimRight(body.String(), "\n")
282+
283+ // While streaming, keep the spinner pinned below the latest output.
284+ if s.pending {
285+ result += "\n" + s.anim.Render()
254 }
286 }
255
287
256− return header + "\n" + strings.TrimRight(body.String(), "\n")
288+ return result
257}
289}
internal/ui/chat/shell_test.gonew file+59
@@ -0,0 +1,59 @@
1+package chat
2+
3+import (
4+ "strconv"
5+ "testing"
6+
7+ "github.com/charmbracelet/crush/internal/ui/styles"
8+ "github.com/charmbracelet/x/ansi"
9+ "github.com/stretchr/testify/require"
10+)
11+
12+// TestPendingShellItemRendersStreamedOutput verifies that a pending shell
13+// item surfaces output appended during execution, rather than hiding it
14+// behind the spinner until completion.
15+func TestPendingShellItemRendersStreamedOutput(t *testing.T) {
16+ t.Parallel()
17+
18+ sty := styles.CharmtonePantera()
19+ item := NewPendingShellItem(&sty, "ping -c 3 localhost")
20+
21+ require.NotContains(t, ansi.Strip(item.Render(80)), "bytes from",
22+ "freshly created pending item should have no output yet")
23+
24+ item.AppendOutput("64 bytes from localhost: icmp_seq=0\n")
25+ item.AppendOutput("64 bytes from localhost: icmp_seq=1\n")
26+
27+ rendered := ansi.Strip(item.Render(80))
28+ require.Contains(t, rendered, "icmp_seq=0",
29+ "streamed output must be visible while the command is still running")
30+ require.Contains(t, rendered, "icmp_seq=1")
31+}
32+
33+// TestPendingShellItemShowsTail verifies that while streaming, the most
34+// recent lines are shown (tail) rather than the first lines.
35+func TestPendingShellItemShowsTail(t *testing.T) {
36+ t.Parallel()
37+
38+ sty := styles.CharmtonePantera()
39+ item := NewPendingShellItem(&sty, "seq 30")
40+
41+ for i := 1; i <= 30; i++ {
42+ item.AppendOutput(strconv.Itoa(i) + "\n")
43+ }
44+
45+ rendered := ansi.Strip(item.Render(80))
46+ require.Contains(t, rendered, "30", "the newest line must be visible while streaming")
47+ require.Contains(t, rendered, "earlier lines", "older lines should be summarized as a count")
48+}
49+
50+// TestCompletedShellItemRendersOutput verifies completion still shows output.
51+func TestCompletedShellItemRendersOutput(t *testing.T) {
52+ t.Parallel()
53+
54+ sty := styles.CharmtonePantera()
55+ item := NewPendingShellItem(&sty, "echo hi")
56+ item.Complete("hi\n", 0)
57+
58+ require.Contains(t, ansi.Strip(item.Render(80)), "hi")
59+}
internal/ui/model/ui.go+53−2
@@ -119,6 +119,13 @@ type shellResultMsg struct {
119 ExitCode int
119 ExitCode int
120}
120}
121
121
122+// shellStreamMsg carries incremental output from a streaming shell command.
123+type shellStreamMsg struct {
124+ PendingID string
125+ Chunk string
126+ streamCh <-chan string // unexported; used to continue draining
127+}
128+
122type (
129type (
123 // cancelTimerExpiredMsg is sent when the cancel timer expires.
130 // cancelTimerExpiredMsg is sent when the cancel timer expires.
124 cancelTimerExpiredMsg struct{}
131 cancelTimerExpiredMsg struct{}
@@ -953,6 +960,27 @@ func (m *UI) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
953 m.textarea.SetValue(msg.Text)
960 m.textarea.SetValue(msg.Text)
954 m.textarea.MoveToEnd()
961 m.textarea.MoveToEnd()
955 cmds = append(cmds, m.updateTextareaWithPrevHeight(msg, prevHeight))
962 cmds = append(cmds, m.updateTextareaWithPrevHeight(msg, prevHeight))
963+ case shellStreamMsg:
964+ if item := m.chat.MessageItem(msg.PendingID); item != nil {
965+ if shellItem, ok := item.(*chat.ShellItem); ok {
966+ shellItem.AppendOutput(msg.Chunk)
967+ if cmd := m.chat.ScrollToBottomAndAnimate(); cmd != nil {
968+ cmds = append(cmds, cmd)
969+ }
970+ }
971+ }
972+ // Continue draining the stream channel.
973+ if msg.streamCh != nil {
974+ ch := msg.streamCh
975+ pid := msg.PendingID
976+ cmds = append(cmds, func() tea.Msg {
977+ chunk, ok := <-ch
978+ if !ok {
979+ return nil
980+ }
981+ return shellStreamMsg{PendingID: pid, Chunk: chunk, streamCh: ch}
982+ })
983+ }
956 case shellResultMsg:
984 case shellResultMsg:
957 // Complete the pending shell item if it exists, otherwise create a new one.
985 // Complete the pending shell item if it exists, otherwise create a new one.
958 if msg.PendingID != "" {
986 if msg.PendingID != "" {
@@ -3447,8 +3475,31 @@ func (m *UI) runShellCommand(command string) tea.Cmd {
3447 }
3475 }
3448 }
3476 }
3449
3477
3478+ // Stream output via channel. The progress callback writes chunks
3479+ // to streamCh; a reader cmd converts them to shellStreamMsg values.
3480+ streamCh := make(chan string, 64)
3481+ pendingID := pendingItem.ID()
3482+
3483+ onProgress := func(chunk string) {
3484+ select {
3485+ case streamCh <- chunk:
3486+ default:
3487+ // Drop if UI can't keep up.
3488+ }
3489+ }
3490+
3491+ // Reader cmd: drains streamCh into shellStreamMsg until closed.
3492+ cmds = append(cmds, func() tea.Msg {
3493+ chunk, ok := <-streamCh
3494+ if !ok {
3495+ return nil
3496+ }
3497+ return shellStreamMsg{PendingID: pendingID, Chunk: chunk, streamCh: streamCh}
3498+ })
3499+
3450 cmds = append(cmds, func() tea.Msg {
3500 cmds = append(cmds, func() tea.Msg {
3451− resp, err := m.com.Workspace.AgentRunShellCommand(context.Background(), sessionID, command, contentWidth)
3501+ resp, err := m.com.Workspace.AgentRunShellCommand(context.Background(), sessionID, command, contentWidth, onProgress)
3502+ close(streamCh)
3452 if err != nil {
3503 if err != nil {
3453 return util.InfoMsg{
3504 return util.InfoMsg{
3454 Type: util.InfoTypeError,
3505 Type: util.InfoTypeError,
@@ -3456,7 +3507,7 @@ func (m *UI) runShellCommand(command string) tea.Cmd {
3456 }
3507 }
3457 }
3508 }
3458 return shellResultMsg{
3509 return shellResultMsg{
3459− PendingID: pendingItem.ID(),
3510+ PendingID: pendingID,
3460 Command: command,
3511 Command: command,
3461 Output: resp.Output,
3512 Output: resp.Output,
3462 ExitCode: resp.ExitCode,
3513 ExitCode: resp.ExitCode,
internal/workspace/app_workspace.go+23−4
@@ -4,6 +4,7 @@ import (
4 "context"
4 "context"
5 "errors"
5 "errors"
6 "fmt"
6 "fmt"
7+ "log/slog"
7 "time"
8 "time"
8
9
9 tea "charm.land/bubbletea/v2"
10 tea "charm.land/bubbletea/v2"
@@ -107,7 +108,7 @@ func (w *AppWorkspace) AgentRun(ctx context.Context, sessionID, prompt string, a
107 return err
108 return err
108}
109}
109
110
110−func (w *AppWorkspace) AgentRunShellCommand(ctx context.Context, sessionID, command string, termWidth int) (proto.ShellCommandResponse, error) {
111+func (w *AppWorkspace) AgentRunShellCommand(ctx context.Context, sessionID, command string, termWidth int, onProgress func(string)) (proto.ShellCommandResponse, error) {
111 var isFirstMessage bool
112 var isFirstMessage bool
112 var persist shell.PersistFunc
113 var persist shell.PersistFunc
113 if sessionID != "" {
114 if sessionID != "" {
@@ -135,15 +136,33 @@ func (w *AppWorkspace) AgentRunShellCommand(ctx context.Context, sessionID, comm
135 }
136 }
136 }
137 }
137
138
138− result, err := shell.RunAndPersist(ctx, shell.RunOptions{
139+ opts := shell.RunOptions{
139 Command: command,
140 Command: command,
140 Cwd: w.store.WorkingDir(),
141 Cwd: w.store.WorkingDir(),
141 TermWidth: termWidth,
142 TermWidth: termWidth,
142− }, persist)
143+ }
144+
145+ var result shell.CaptureResult
146+ var err error
147+
148+ if onProgress != nil {
149+ // Use streaming execution with progress callback.
150+ result, err = shell.RunAndCaptureStream(ctx, opts, onProgress)
151+ } else {
152+ result, err = shell.RunAndPersist(ctx, opts, persist)
153+ }
154+
143− if err != nil {
155+ if err != nil && onProgress == nil {
144 return proto.ShellCommandResponse{}, err
156 return proto.ShellCommandResponse{}, err
145 }
157 }
146
158
159+ // Persist if we used streaming path (persist wasn't called by RunAndPersist).
160+ if onProgress != nil && persist != nil {
161+ if persistErr := persist(command, result.Output, result.ExitCode); persistErr != nil {
162+ slog.Error("Failed to persist shell command output", "error", persistErr, "command", command)
163+ }
164+ }
165+
147 // Generate a title from the shell command if it was the first message.
166 // Generate a title from the shell command if it was the first message.
148 if isFirstMessage {
167 if isFirstMessage {
149 go func() {
168 go func() {
internal/workspace/client_workspace.go+1−1
@@ -186,7 +186,7 @@ func (w *ClientWorkspace) AgentRun(ctx context.Context, sessionID, prompt string
186 return w.client.SendMessage(ctx, w.workspaceID(), sessionID, "", prompt, attachments...)
186 return w.client.SendMessage(ctx, w.workspaceID(), sessionID, "", prompt, attachments...)
187}
187}
188
188
189−func (w *ClientWorkspace) AgentRunShellCommand(ctx context.Context, sessionID, command string, termWidth int) (proto.ShellCommandResponse, error) {
189+func (w *ClientWorkspace) AgentRunShellCommand(ctx context.Context, sessionID, command string, termWidth int, _ func(string)) (proto.ShellCommandResponse, error) {
190 return w.client.RunShellCommand(ctx, w.workspaceID(), sessionID, command, termWidth)
190 return w.client.RunShellCommand(ctx, w.workspaceID(), sessionID, command, termWidth)
191}
191}
192
192
internal/workspace/workspace.go+1−1
@@ -83,7 +83,7 @@ type Workspace interface {
83
83
84 // Agent
84 // Agent
85 AgentRun(ctx context.Context, sessionID, prompt string, attachments ...message.Attachment) error
85 AgentRun(ctx context.Context, sessionID, prompt string, attachments ...message.Attachment) error
86− AgentRunShellCommand(ctx context.Context, sessionID, command string, termWidth int) (proto.ShellCommandResponse, error)
86+ AgentRunShellCommand(ctx context.Context, sessionID, command string, termWidth int, onProgress func(string)) (proto.ShellCommandResponse, error)
87 AgentCancel(sessionID string)
87 AgentCancel(sessionID string)
88 AgentIsBusy() bool
88 AgentIsBusy() bool
89 AgentIsSessionBusy(sessionID string) bool
89 AgentIsSessionBusy(sessionID string) bool