Add a pipeline whose output can be read as it arrives

Rendering a diff through a renderer on Windows means running the
renderer ourselves, since ConPTY mangles the metadata records it emits
and git only invokes a renderer of its own when it talks to a terminal.
That needs a chain of commands whose output a view can be filled from
while it runs, where PipeCommands runs a chain to completion and reports
what it said afterwards.

StartPipeline starts such a chain and hands back the reader for its
output, along with a handle offering exactly what a render task asks of
a command: something to wait for, something to name it by, and a way to
stop it. Every command's stderr joins the output, so a renderer that
objects to its input says so where the diff would have been.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

pkg/commands/oscommands/os.go

@@ -201,7 +201,7 @@ func (c *OSCommand) FileExists(path string) (bool, error) {
 
 // PipeCommands runs a heap of commands and pipes their inputs/outputs together like A | B | C
 func (c *OSCommand) PipeCommands(cmdObjs ...*CmdObj) error {
-	c.LogCommand(pipelineString(cmdObjs), true)
+	c.logPipeline(cmdObjs)
 
 	cmds, parentEnds, err := wirePipeline(cmdObjs)
 	if err != nil {

pkg/commands/oscommands/pipeline.go

@@ -1,6 +1,7 @@
 package oscommands
 
 import (
+	"fmt"
 	"io"
 	"os"
 	"os/exec"
@@ -9,6 +10,100 @@ import (
 	"github.com/samber/lo"
 )
 
+// Pipeline is a chain of running commands, each one's output feeding the next
+// one's input, with the last one's output going somewhere the caller reads. Its
+// method set is the one a render task expects of a command (see tasks.Cmd), so
+// a pipeline can render a view just as a single command can.
+type Pipeline struct {
+	cmds   []*exec.Cmd
+	cmdStr string
+}
+
+// StartPipeline starts the given commands wired A | B | C and returns the
+// pipeline together with the reader for its output.
+//
+// Every command's stderr goes to that same output, so whatever a command
+// complains about is part of what the caller reads. A diff renderer's error
+// message belongs on screen with the diff it failed to render.
+//
+// Closing the reader is how a pipeline is brought down. The last command's next
+// write fails, so it exits, and the failure travels back up the chain as each
+// command in turn writes into a pipe whose reader is gone.
+func (c *OSCommand) StartPipeline(cmdObjs ...*CmdObj) (*Pipeline, io.ReadCloser, error) {
+	c.logPipeline(cmdObjs)
+
+	cmds, parentEnds, err := wirePipeline(cmdObjs)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	reader, writer, err := os.Pipe()
+	if err != nil {
+		closeAll(parentEnds)
+		return nil, nil, err
+	}
+	for _, cmd := range cmds {
+		cmd.Stderr = writer
+	}
+	cmds[len(cmds)-1].Stdout = writer
+	parentEnds = append(parentEnds, writer)
+
+	started, err := startPipeline(cmds, parentEnds)
+	if err != nil {
+		for _, cmd := range cmds[:started] {
+			_ = cmd.Wait()
+		}
+		_ = reader.Close()
+		return nil, nil, err
+	}
+
+	return &Pipeline{cmds: cmds, cmdStr: pipelineString(cmdObjs)}, reader, nil
+}
+
+func (self *Pipeline) String() string {
+	return self.cmdStr
+}
+
+// Wait waits for every command to exit and reports the failure nearest the end
+// of the pipeline. A command that fails leaves the ones before it writing into
+// a pipe nobody reads, so their own broken-pipe failures are consequences of it
+// rather than the cause worth reporting, while a command that fails early
+// leaves the ones after it with nothing to read and no reason to fail at all.
+func (self *Pipeline) Wait() error {
+	var lastErr error
+	for _, cmd := range self.cmds {
+		if err := cmd.Wait(); err != nil {
+			lastErr = fmt.Errorf("%s: %w", cmd.String(), err)
+		}
+	}
+
+	return lastErr
+}
+
+// Terminate asks every command to stop, without waiting for any of them. On
+// platforms where that does nothing, the pipeline comes down when its output
+// reader is closed; see StartPipeline.
+func (self *Pipeline) Terminate() error {
+	var firstErr error
+	for _, cmd := range self.cmds {
+		if err := TerminateProcessGracefully(cmd.Process); err != nil && firstErr == nil {
+			firstErr = err
+		}
+	}
+
+	return firstErr
+}
+
+// logPipeline enters a chain of commands into the command log, unless the first
+// command was marked not to be logged; it speaks for the pipeline. A render runs
+// its pipeline again on every selection change, so a caller has to be able to
+// keep it out of the log.
+func (c *OSCommand) logPipeline(cmdObjs []*CmdObj) {
+	if cmdObjs[0].ShouldLog() {
+		c.LogCommand(pipelineString(cmdObjs), true)
+	}
+}
+
 // pipelineString names a chain of commands the way a shell would write it.
 func pipelineString(cmdObjs []*CmdObj) string {
 	return strings.Join(

pkg/commands/oscommands/pipeline_test.go

@@ -56,6 +56,75 @@ func TestPipelineMember(t *testing.T) {
 	os.Exit(0)
 }
 
+func TestStartPipelineStreamsTheOutputOfTheLastCommand(t *testing.T) {
+	pipeline, reader, err := NewDummyOSCommand().StartPipeline(
+		pipelineMember("count"),
+		pipelineMember("upcase"),
+	)
+	assert.NoError(t, err)
+
+	output, err := io.ReadAll(reader)
+	assert.NoError(t, err)
+	assert.Equal(t, "LINE 1\nLINE 2\nLINE 3\n", string(output))
+
+	assert.NoError(t, pipeline.Wait())
+	assert.NoError(t, reader.Close())
+}
+
+func TestStartPipelineReadsWhatTheCommandsComplainAbout(t *testing.T) {
+	pipeline, reader, err := NewDummyOSCommand().StartPipeline(
+		pipelineMember("count"),
+		pipelineMember("complain"),
+	)
+	assert.NoError(t, err)
+
+	output, err := io.ReadAll(reader)
+	assert.NoError(t, err)
+	assert.Equal(t, "something went wrong\n", string(output))
+
+	// The failure of the command nearest the output is the one reported, even
+	// though the one feeding it was left writing into a pipe nobody reads.
+	assert.ErrorContains(t, pipeline.Wait(), "exit status 3")
+
+	assert.NoError(t, reader.Close())
+}
+
+func TestClosingAPipelinesOutputBringsItDown(t *testing.T) {
+	pipeline, reader, err := NewDummyOSCommand().StartPipeline(
+		pipelineMember("flood"),
+		pipelineMember("copy"),
+	)
+	assert.NoError(t, err)
+
+	// Read some output first, so that both commands are past their startup and
+	// really running when the reader goes.
+	buf := make([]byte, len("line 1\n"))
+	_, err = io.ReadFull(reader, buf)
+	assert.NoError(t, err)
+	assert.Equal(t, "line 1\n", string(buf))
+
+	assert.NoError(t, reader.Close())
+
+	done := make(chan error, 1)
+	go func() { done <- pipeline.Wait() }()
+
+	select {
+	case <-done:
+	case <-time.After(10 * time.Second):
+		t.Fatal("the pipeline was still running long after its output was closed")
+	}
+}
+
+func TestStartPipelineReportsACommandItCannotStart(t *testing.T) {
+	osCommand := NewDummyOSCommand()
+
+	_, _, err := osCommand.StartPipeline(
+		pipelineMember("count"),
+		osCommand.Cmd.New([]string{"lazygit-no-such-command"}),
+	)
+	assert.Error(t, err)
+}
+
 func TestPipeCommandsReturnsWhenALaterCommandDiesEarly(t *testing.T) {
 	done := make(chan error, 1)
 	go func() {