Forward panic handler through FilterError/IgnoreError wrappers

`efStage` (the wrapper returned by `FilterError` and
`IgnoreError`) embeds the `Stage` interface, which only exposes
the four Stage methods. So when `Pipeline.Start()` checks
`if phs, ok := s.(StagePanicHandlerAware); ok`, the assertion
silently fails for any wrapped stage — even if the underlying
stage is a `goStage` that implements `SetPanicHandler`.

The result: a configured `WithStagePanicHandler` is bypassed
when the panicking Function is wrapped in `IgnoreError`. The
goroutine inside `goStage.Start` sees `panicHandler == nil`
and returns without calling `recover()`, letting the panic
propagate up the runtime and crash the host process.

This was a pre-existing bug in main (not introduced by the Stage
interface redesign), but we're already touching the panic-handler
surface to restore it for bare Function stages, so fix it in the
same series.

Implement `SetPanicHandler` on `efStage` and forward it to the
embedded stage if it implements `StagePanicHandlerAware`. Add a
`var _ StagePanicHandlerAware = efStage{}` compile-time
assertion and a regression test that wraps a panicking Function
in `IgnoreError` and verifies the handler runs.

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

Author: znull

pipe/filter-error.go

@@ -26,6 +26,21 @@ func (s efStage) Wait() error {
 	return s.filter(s.Stage.Wait())
 }
 
+// SetPanicHandler forwards the handler to the wrapped stage if it
+// implements `StagePanicHandlerAware`. Without this, wrapping a
+// panicking `Function` in `IgnoreError` / `FilterError` would
+// silently bypass `WithStagePanicHandler` (the type assertion in
+// `Pipeline.Start()` only sees this wrapper's methods, not the
+// embedded stage's `SetPanicHandler`), letting the panic propagate
+// out of the goroutine and crash the host process.
+func (s efStage) SetPanicHandler(ph StagePanicHandler) {
+	if phs, ok := s.Stage.(StagePanicHandlerAware); ok {
+		phs.SetPanicHandler(ph)
+	}
+}
+
+var _ StagePanicHandlerAware = efStage{}
+
 // ErrorMatcher decides whether its argument matches some class of
 // errors (e.g., errors that we want to ignore). The function will
 // only be invoked for non-nil errors.

pipe/pipeline_test.go

@@ -532,6 +532,35 @@ func TestFunction(t *testing.T) {
 		assert.ErrorContains(t, err, "panic handled")
 		assert.Empty(t, out)
 	})
+
+	t.Run("panic with handler through IgnoreError", func(t *testing.T) {
+		// Regression: efStage (the FilterError/IgnoreError wrapper)
+		// previously did not implement StagePanicHandlerAware, so
+		// the type assertion in Pipeline.Start() silently failed
+		// and the wrapped goStage never received the panic handler.
+		// The result was an unrecovered panic crashing the host.
+		p := pipe.New(
+			pipe.WithStagePanicHandler(func(p any) error {
+				return fmt.Errorf("panic handled: %v", p)
+			}),
+		)
+		p.Add(
+			pipe.Print("hello world"),
+			pipe.IgnoreError(
+				pipe.Function(
+					"farewell",
+					func(_ context.Context, _ pipe.Env, _ io.Reader, _ io.Writer) error {
+						panic("this is a panic")
+					},
+				),
+				func(_ error) bool { return false },
+			),
+		)
+
+		out, err := p.Output(ctx)
+		assert.ErrorContains(t, err, "panic handled")
+		assert.Empty(t, out)
+	})
 }
 
 func TestPipelineWithFunction(t *testing.T) {