6814ccf · crush · git.iko.soy

git.iko.soy/crush

feat(bang): stream results in

commit6814ccfca6715ac018ac1ff7046146a913b24824 authorKieran Klukas <kieran@dunkirk.sh> dateFri 19 Jun 2026, 11:22 (4 months ago) parent6db54b3

7 files changed+237−19

compare with main

unifiedsplit

internal/shell/stream.go+57 internal/ui/chat/shell.go+43−11 internal/ui/chat/shell_test.go+59 internal/ui/model/ui.go+53−2 internal/workspace/app_workspace.go+23−4 internal/workspace/client_workspace.go+1−1 internal/workspace/workspace.go+1−1

internal/shell/stream.gonew file+57

@@ -0,0 +1,57 @@

1+package shell

3+import (

4+ "bytes"

5+ "context"

6+ "log/slog"

7+ "os"

8+)

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) {

101101 s.Bump()

102102}

103103

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+

104110func (s *ShellItem) ID() string { return s.id }

105111func (s *ShellItem) FilterValue() string { return s.command }

106112func (s *ShellItem) Finished() bool { return !s.pending }

@@ -199,10 +205,11 @@ func (s *ShellItem) RawRender(width int) string {

199205 header := prompt + " " + highlighted

200206

201207 if s.pending {

202− return header + "\n" + s.anim.Render()

203− }

204−

205− if s.exitCode != 0 {

208+ if s.output == "" {

209+ // Nothing streamed yet: show the spinner under the header.

210+ return header + "\n" + s.anim.Render()

211+ }

212+ } else if s.exitCode != 0 {

206213 header += " " + s.sty.Messages.ShellExitCode.Render(fmt.Sprintf("(exit %d)", s.exitCode))

207214 }

208215

@@ -213,14 +220,23 @@ func (s *ShellItem) RawRender(width int) string {

213220 output := strings.TrimRight(s.output, "\n")

214221 lines := strings.Split(output, "\n")

215222

223+ // While streaming, show the tail of the output so the most recent

224+ // lines stay visible without forcing the user to expand.

216225 maxLines := shellMaxCollapsedLines

217226 if s.expandedContent {

218227 maxLines = len(lines)

219228 }

220229

221230 displayLines := lines

231+ truncatedCount := 0

222232 if len(lines) > maxLines {

223− displayLines = lines[:maxLines]

233+ if s.pending {

234+ // Show the most recent lines while still running.

235+ displayLines = lines[len(lines)-maxLines:]

236+ } else {

237+ displayLines = lines[:maxLines]

238+ }

239+ truncatedCount = len(lines) - maxLines

224240 }

225241

226242 // Compute max line width for scroll clamping.

@@ -234,6 +250,16 @@ func (s *ShellItem) RawRender(width int) string {

234250 s.maxLineWidth = max(0, maxW-cappedWidth)

235251

236252 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+

237263 for _, ln := range displayLines {

238264 scrolled := ansi.GraphemeWidth.Cut(ln, s.xOffset, len(ln))

239265 truncated := ansi.Truncate(scrolled, cappedWidth, "…")

@@ -244,14 +270,20 @@ func (s *ShellItem) RawRender(width int) string {

244270 body.WriteString("\n")

245271 }

246272

247− if len(lines) > maxLines && !s.expandedContent {

273+ // When finished, hidden lines are below, so show the notice after.

274+ if truncatedCount > 0 && !s.pending && !s.expandedContent {

248275 body.WriteString(s.sty.Messages.ShellTruncation.Render(

249− fmt.Sprintf("… %d more lines", len(lines)-maxLines),

276+ fmt.Sprintf("… %d more lines", truncatedCount),

250277 ))

251− } else {

252− result := body.String()

253− return header + "\n" + strings.TrimRight(result, "\n")

278+ return header + "\n" + body.String()

279+ }

280+

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()

254286 }

255287

256− return header + "\n" + strings.TrimRight(body.String(), "\n")

288+ return result

257289}

internal/ui/chat/shell_test.gonew file+59

@@ -0,0 +1,59 @@

1+package chat

3+import (

4+ "strconv"

5+ "testing"

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 {

119119 ExitCode int

120120}

121121

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+

122129type (

123130 // cancelTimerExpiredMsg is sent when the cancel timer expires.

124131 cancelTimerExpiredMsg struct{}

@@ -953,6 +960,27 @@ func (m *UI) Update(msg tea.Msg) (tea.Model, tea.Cmd) {

953960 m.textarea.SetValue(msg.Text)

954961 m.textarea.MoveToEnd()

955962 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+ }

956984 case shellResultMsg:

957985 // Complete the pending shell item if it exists, otherwise create a new one.

958986 if msg.PendingID != "" {

@@ -3447,8 +3475,31 @@ func (m *UI) runShellCommand(command string) tea.Cmd {

34473475 }

34483476 }

34493477

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+

34503500 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)

34523503 if err != nil {

34533504 return util.InfoMsg{

34543505 Type: util.InfoTypeError,

@@ -3456,7 +3507,7 @@ func (m *UI) runShellCommand(command string) tea.Cmd {

34563507 }

34573508 }

34583509 return shellResultMsg{

3459− PendingID: pendingItem.ID(),

3510+ PendingID: pendingID,

34603511 Command: command,

34613512 Output: resp.Output,

34623513 ExitCode: resp.ExitCode,

internal/workspace/app_workspace.go+23−4

@@ -4,6 +4,7 @@ import (

44 "context"

55 "errors"

66 "fmt"

7+ "log/slog"

78 "time"

910 tea "charm.land/bubbletea/v2"

@@ -107,7 +108,7 @@ func (w *AppWorkspace) AgentRun(ctx context.Context, sessionID, prompt string, a

107108 return err

108109}

109110

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) {

111112 var isFirstMessage bool

112113 var persist shell.PersistFunc

113114 if sessionID != "" {

@@ -135,15 +136,33 @@ func (w *AppWorkspace) AgentRunShellCommand(ctx context.Context, sessionID, comm

135136 }

136137 }

137138

138− result, err := shell.RunAndPersist(ctx, shell.RunOptions{

139+ opts := shell.RunOptions{

139140 Command: command,

140141 Cwd: w.store.WorkingDir(),

141142 TermWidth: termWidth,

142− }, persist)

143− if err != nil {

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+

155+ if err != nil && onProgress == nil {

144156 return proto.ShellCommandResponse{}, err

145157 }

146158

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+

147166 // Generate a title from the shell command if it was the first message.

148167 if isFirstMessage {

149168 go func() {

internal/workspace/client_workspace.go+1−1

@@ -186,7 +186,7 @@ func (w *ClientWorkspace) AgentRun(ctx context.Context, sessionID, prompt string

186186 return w.client.SendMessage(ctx, w.workspaceID(), sessionID, "", prompt, attachments...)

187187}

188188

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) {

190190 return w.client.RunShellCommand(ctx, w.workspaceID(), sessionID, command, termWidth)

191191}

192192

internal/workspace/workspace.go+1−1

@@ -83,7 +83,7 @@ type Workspace interface {

8383

8484 // Agent

8585 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)

8787 AgentCancel(sessionID string)

8888 AgentIsBusy() bool

8989 AgentIsSessionBusy(sessionID string) bool