feat: add configurable per-task panic handler

config.go

@@ -1,4 +1,4 @@
-package agilepool
+package agilepool
 
 import (
 	"time"
@@ -43,10 +43,25 @@ type Config struct {
 	statsWindowSize    int           // number of windows for median calculation
 	scalerPeriod       time.Duration // scaler tick interval (e.g. 50ms)
 	backlogDecayFactor float64       // queue backlog weight in scaler target (0-1)
+	panicHandler       PanicHandler  // custom task-panic handler (nil keeps default logging)
 }
 
 type ConfigOption func(*Config)
 
+// PanicHandler is invoked when a task panics while running inside a worker.
+// Setting one with WithPanicHandler replaces the pool's default behaviour of
+// logging the recovered value and stack trace, letting callers route panics
+// to their own error reporting.
+//
+//   - task:      the task whose execution panicked
+//   - recovered: the value passed to panic
+//   - stack:     formatted call stack captured at the recovery point
+//
+// A nil handler keeps the default logging behaviour. A handler that panics is
+// itself recovered and logged, so it can never crash the worker goroutine or
+// skip task-completion bookkeeping; see Pool.handlePanic.
+type PanicHandler func(task Task, recovered any, stack []byte)
+
 func NewConfig(opts ...ConfigOption) *Config {
 	config := &Config{
 		cleanPeriod:        defaultCleanPeriod,
@@ -148,3 +163,12 @@ func WithBacklogDecayFactor(factor float64) ConfigOption {
 		}
 	}
 }
+
+// WithPanicHandler installs a custom handler invoked when a task panics inside
+// a worker, replacing the default logger-based reporting. Pass nil to keep the
+// default behaviour.
+func WithPanicHandler(handler PanicHandler) ConfigOption {
+	return func(c *Config) {
+		c.panicHandler = handler
+	}
+}

pool.go

@@ -100,22 +100,24 @@ type Pool struct {
 	consumeHist *histogram // consume count distribution per window
 	exitHist    *histogram // exit count distribution per window
 
-	hooks Hooks // lifecycle callbacks registered via SetHook
+	hooks        Hooks        // lifecycle callbacks registered via SetHook
+	panicHandler PanicHandler // custom task-panic handler (nil keeps default logging)
 }
 
 func NewPool(c *Config) *Pool {
 	if c == nil {
 		c = NewConfig()
 	}
 	p := &Pool{
-		closePoolCn: make(chan struct{}),
-		config:      c,
-		lock:        &sync.Mutex{},
-		muIdle:      &sync.Mutex{}, // default value, overridden below based on config
-		logger:      log.Default(),
-		capacity:    c.workerNumCapacity,
-		taskQueue:   make(chan Task, c.taskQueueSize),
-		taskBuf:     newChunkedTaskBuffer(),
+		closePoolCn:  make(chan struct{}),
+		config:       c,
+		lock:         &sync.Mutex{},
+		muIdle:       &sync.Mutex{}, // default value, overridden below based on config
+		logger:       log.Default(),
+		capacity:     c.workerNumCapacity,
+		taskQueue:    make(chan Task, c.taskQueueSize),
+		taskBuf:      newChunkedTaskBuffer(),
+		panicHandler: c.panicHandler,
 	}
 
 	// Select muIdle lock implementation based on config: SpinLock or MutexLock (sync.Mutex)
@@ -625,3 +627,23 @@ func (p *Pool) dispatchHook(fn func(h Hooks)) {
 	}()
 	fn(p.hooks)
 }
+
+// handlePanic reports a task panic that was recovered by the worker's runTask
+// defer. When a PanicHandler is configured it takes over; otherwise the
+// recovered value and stack trace are written to the pool logger, preserving
+// the historical default. A panicking handler is itself recovered and logged
+// so it cannot crash the worker goroutine or skip the task-completion hook
+// that runTask dispatches afterwards.
+func (p *Pool) handlePanic(task Task, recovered any, stack []byte) {
+	handler := p.panicHandler
+	if handler == nil {
+		p.logger.Printf("worker exits from panic: %v\n%s\n", recovered, stack)
+		return
+	}
+	defer func() {
+		if r := recover(); r != nil {
+			p.logger.Printf("panic handler panicked: %v\n%s\n", r, Stack(1))
+		}
+	}()
+	handler(task, recovered, stack)
+}

pool_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"io"
 	"log"
+	"reflect"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -451,6 +452,100 @@ func TestAgilePoolTaskPanicDoesNotBreakPool(t *testing.T) {
 	assert.Equal(t, int64(2), atomic.LoadInt64(&executed))
 }
 
+// TestAgilePoolCustomPanicHandler verifies that a handler registered via
+// WithPanicHandler receives the panicking task, the recovered value and the
+// captured stack trace, and that the worker keeps servicing subsequent tasks.
+func TestAgilePoolCustomPanicHandler(t *testing.T) {
+	var (
+		calls    int64
+		gotTask  atomic.Value
+		gotValue atomic.Value
+		gotStack atomic.Value
+	)
+
+	agilePool := agilepool.NewPool(agilepool.NewConfig(
+		agilepool.WithWorkerNumCapacity(1),
+		agilepool.WithTaskQueueSize(10),
+		agilepool.WithPanicHandler(func(task agilepool.Task, recovered any, stack []byte) {
+			atomic.AddInt64(&calls, 1)
+			gotTask.Store(task)
+			gotValue.Store(recovered)
+			gotStack.Store(string(stack))
+		}),
+	))
+	defer agilePool.Close()
+
+	panicked := agilepool.TaskFunc(func() error {
+		panic("custom-handler-boom")
+	})
+	agilePool.Submit(panicked)
+
+	var executed int64
+	agilePool.Submit(agilepool.TaskFunc(func() error {
+		atomic.AddInt64(&executed, 1)
+		return nil
+	}))
+
+	done := make(chan struct{})
+	go func() {
+		agilePool.Wait()
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("pool.Wait() timed out after a task panic")
+	}
+
+	assert.Equal(t, int64(1), atomic.LoadInt64(&calls))
+	assert.Equal(t, "custom-handler-boom", gotValue.Load())
+	assert.Equal(t, int64(1), atomic.LoadInt64(&executed))
+
+	// The exact task that panicked must be forwarded, not a wrapper.
+	assert.Equal(t,
+		reflect.ValueOf(panicked).Pointer(),
+		reflect.ValueOf(gotTask.Load()).Pointer(),
+	)
+
+	// A non-empty, formatted stack trace must be provided.
+	stack, _ := gotStack.Load().(string)
+	assert.Contains(t, stack, "worker.go")
+}
+
+// TestAgilePoolPanicHandlerPanicIsRecovered guards Pool.handlePanic: a
+// PanicHandler that itself panics must be recovered so the worker goroutine
+// survives and Wait() still returns.
+func TestAgilePoolPanicHandlerPanicIsRecovered(t *testing.T) {
+	agilePool := agilepool.NewPool(agilepool.NewConfig(
+		agilepool.WithWorkerNumCapacity(1),
+		agilepool.WithTaskQueueSize(10),
+		agilepool.WithPanicHandler(func(agilepool.Task, any, []byte) {
+			panic("handler boom")
+		}),
+	))
+	agilePool.SetLogger(log.New(io.Discard, "", 0))
+	defer agilePool.Close()
+
+	var executed int64
+	agilePool.Submit(agilepool.TaskFunc(func() error { panic("task boom") }))
+	agilePool.Submit(agilepool.TaskFunc(func() error {
+		atomic.AddInt64(&executed, 1)
+		return nil
+	}))
+
+	done := make(chan struct{})
+	go func() {
+		agilePool.Wait()
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Wait() blocked after the panic handler panicked")
+	}
+	assert.Equal(t, int64(1), atomic.LoadInt64(&executed))
+}
+
 // TestAgilePoolBatchWithScaler verifies that the scaler spawns workers to
 // drain tasks submitted in a burst, even when no worker goroutines existed
 // at submission time (replaces the old safety-net / race-stuck-task test

readme.md

@@ -93,6 +93,7 @@ defer pool.Close()
 | `WithStatsWindowSize` | `10` | Number of sliding windows for median calculation. |
 | `WithScalerPeriod` | `10ms` | How often the scaler evaluates whether to spawn workers. |
 | `WithBacklogDecayFactor` | `0.3` | Weight (0–1) for backlog in the scaler target formula. Higher values make the scaler more aggressive at draining queues. |
+| `WithPanicHandler` | `nil` | Custom `func(task Task, recovered any, stack []byte)` invoked when a task panics, replacing the default log line. Defaults to logging with `SetLogger`. |
 
 > **Note**: `WithTaskQueueSize` controls the capacity of the internal handoff channel. Tasks beyond that capacity are stored in the dynamically growing chunked buffer.
 

worker.go

@@ -124,13 +124,14 @@ func (w *worker) runTask(task Task) {
 	}
 
 	// Capture task panics so the Completed hook can observe the recovered
-	// value. dispatchHook guards the hook call itself; w.pool.done() above
-	// runs afterwards regardless of what the hooks do.
+	// value. The configured PanicHandler (or the default logger) is invoked
+	// via w.pool.handlePanic; w.pool.done() above runs afterwards regardless
+	// of what the hooks or handler do.
 	var recovered any
 	defer func() {
 		if p := recover(); p != nil {
 			recovered = p
-			w.pool.logger.Printf("worker exits from panic: %v\n%s\n", p, Stack(1))
+			w.pool.handlePanic(task, p, Stack(1))
 		}
 		w.pool.dispatchHook(func(h Hooks) {
 			h.DispatchTaskCompleted(hookCtx, recovered)