Restore identity-copy behavior for empty pipelines

Prior to the Stage interface redesign, an empty pipeline with a
configured `stdout` ran a synthetic ioCopier that copied `p.stdin` (if
any) to `p.stdout` and closed the destination if it came from
`WithStdoutCloser()`.

Restore that behavior by synthesizing an identity-copy Function stage
when the pipeline has no stages but does have a configured output. The
empty/no-output case remains a no-op as before.

This affects callers like `pipe.New(WithStdin(r)).Output(ctx)`.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: znull

pipe/pipeline.go

@@ -269,6 +269,28 @@ func (p *Pipeline) Start(ctx context.Context) error {
 	atomic.StoreUint32(&p.started, 1)
 	ctx, p.cancel = context.WithCancel(ctx)
 
+	if len(p.stages) == 0 {
+		if p.stdout == nil {
+			// No stages and no destination: there is nothing to do
+			// and nowhere to put `p.stdin` even if it was set.
+			return nil
+		}
+		// No stages but a destination was configured: synthesize an
+		// identity-copy stage so that `WithStdin()` is drained into
+		// `WithStdout()`/`WithStdoutCloser()` and the destination
+		// closer (if any) is invoked.
+		p.stages = append(p.stages, Function(
+			"identity",
+			func(_ context.Context, _ Env, stdin io.Reader, stdout io.Writer) error {
+				if stdin == nil {
+					return nil
+				}
+				_, err := io.Copy(stdout, stdin)
+				return err
+			},
+		))
+	}
+
 	// We need to decide how to start the stages, especially what
 	// pipes to use to connect adjacent stages (`os.Pipe()` vs.
 	// `io.Pipe()`) based on the two stages' preferences.

pipe/pipeline_test.go

@@ -26,6 +26,60 @@ func TestMain(m *testing.M) {
 	goleak.VerifyTestMain(m)
 }
 
+func TestPipelineEmpty(t *testing.T) {
+	t.Parallel()
+	p := pipe.New()
+	assert.NoError(t, p.Run(context.Background()))
+}
+
+func TestPipelineEmptyWithStdinAndStdout(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	stdout := &bytes.Buffer{}
+	p := pipe.New(
+		pipe.WithStdin(strings.NewReader("hello world\n")),
+		pipe.WithStdout(stdout),
+	)
+	if assert.NoError(t, p.Run(ctx)) {
+		assert.Equal(t, "hello world\n", stdout.String())
+	}
+}
+
+func TestPipelineEmptyOutput(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	p := pipe.New(pipe.WithStdin(strings.NewReader("hello world\n")))
+	out, err := p.Output(ctx)
+	if assert.NoError(t, err) {
+		assert.Equal(t, "hello world\n", string(out))
+	}
+}
+
+func TestPipelineEmptyWithStdoutCloser(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	stdout := &closeTrackingWriter{}
+	p := pipe.New(
+		pipe.WithStdin(strings.NewReader("hello world\n")),
+		pipe.WithStdoutCloser(stdout),
+	)
+	if assert.NoError(t, p.Run(ctx)) {
+		assert.Equal(t, "hello world\n", stdout.buf.String())
+		assert.True(t, stdout.closed, "WithStdoutCloser destination should be closed")
+	}
+}
+
+type closeTrackingWriter struct {
+	buf    bytes.Buffer
+	closed bool
+}
+
+func (w *closeTrackingWriter) Write(p []byte) (int, error) { return w.buf.Write(p) }
+func (w *closeTrackingWriter) Close() error {
+	w.closed = true
+	return nil
+}
+
 func TestPipelineFirstStageFailsToStart(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()