feat: add WithAgentInactivityTimeout option for terminating stalled a… (#315)

Adds a new `RequestHandlerOption` that bounds how long the server will
wait for the next event from an agent's producer before terminating its
execution. The timer is reset whenever the agent writes an event to the
event pipe, so steady producers are unaffected.

Approach was discussed and confirmed on the issue: #78 (comment).

## Implementation

- New `ErrAgentInactivityTimeout` sentinel in `internal/taskexec`,
re-exported from `a2asrv` so callers can detect the condition with
`errors.Is`.
- New `activityTrackingWriter` wraps the producer's `eventpipe.Writer`
and signals on every successful `Write`. Wrapping is a no-op when the
option is not configured, so existing users pay no cost.
- New watcher goroutine inside `runProducerConsumer`, started only when
the inactivity timeout is positive. On firing, it returns
`ErrAgentInactivityTimeout` from the errgroup which becomes the
cancellation cause on the producer and consumer contexts via
`context.Cause`, exactly like the existing
`TestRunProducerConsumer_CausePropagation` pattern.
- Plumbed through `LocalManagerConfig.AgentInactivityTimeout` and
`DistributedManagerConfig.AgentInactivityTimeout` for cluster-mode
parity. Heartbeats from `workqueue.Heartbeater` remain independent.

## Usage

```go
handler := a2asrv.NewHandler(
    myExecutor,
    a2asrv.WithAgentInactivityTimeout(30 * time.Second),
)
```

When the agent stops emitting events for the configured duration, the
executor's context is canceled and the task is finalized via the
existing failure path. Callers can detect the cause:

```go
if errors.Is(err, a2asrv.ErrAgentInactivityTimeout) {
    // handle inactivity termination
}
```

## Defaults and compatibility

- Default `0` disables the watcher, preserving prior behavior. Existing
users see no change.
- Applies to both `localManager` and the cluster-mode `workQueueHandler`
for parity.
- `workqueue.Heartbeater` heartbeats are independent — they continue to
fire on their own timer, unaffected by the inactivity watcher.

## Tests

- **7 unit tests** in `execution_handler_test.go`:
- Producer stall → `ErrAgentInactivityTimeout` returned from
`runProducerConsumer` and surfaced as `context.Cause` on the producer
ctx
- Activity signals reset the timer (producer emits at half the timeout
interval, watcher never fires)
  - Zero timeout does not start the watcher goroutine
- `activityTrackingWriter`: nil tracker is a no-op, successful write
signals, failed write does not signal, non-blocking when signal channel
is full
- **1 end-to-end test** in `manager_test.go`
(`TestManager_AgentInactivityTimeout`) that constructs a `localManager`
with the option configured, runs a stalled `AgentExecutor`, and asserts
the executor sees `ErrAgentInactivityTimeout` via `context.Cause`.

All existing tests in `internal/taskexec/...` and `a2asrv/...` continue
to pass. `go vet ./...` clean. Full repo `go test ./...` green.

Fixes #78

---------

Co-authored-by: Yaroslav Shevchuk <yarolegovich@gmail.com>

Author: ravyg

a2asrv/handler.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"iter"
 	"log/slog"
+	"time"
 
 	"github.com/a2aproject/a2a-go/v2/a2a"
 	"github.com/a2aproject/a2a-go/v2/a2asrv/eventqueue"
@@ -30,6 +31,12 @@ import (
 	"github.com/a2aproject/a2a-go/v2/log"
 )
 
+// ErrAgentInactivityTimeout is the cancellation cause set on an agent's
+// execution context when no events are written to the event pipe for the
+// duration configured via [WithAgentInactivityTimeout]. Callers can detect
+// this condition with errors.Is.
+var ErrAgentInactivityTimeout = taskexec.ErrAgentInactivityTimeout
+
 // RequestHandler defines a transport-agnostic interface for handling incoming A2A requests.
 type RequestHandler interface {
 	// GetTask handles the 'GetTask' protocol method.
@@ -81,6 +88,8 @@ type defaultRequestHandler struct {
 	workQueue              workqueue.Queue
 	reqContextInterceptors []ExecutorContextInterceptor
 
+	agentInactivityTimeout time.Duration
+
 	authenticatedCardProducer ExtendedAgentCardProducer
 	capabilities              *a2a.AgentCapabilities
 }
@@ -122,6 +131,22 @@ func WithExecutionPanicHandler(handler func(r any) error) RequestHandlerOption {
 	}
 }
 
+// WithAgentInactivityTimeout configures the maximum time the server will wait
+// for the next event from an agent's producer before terminating its execution.
+// The timer is reset whenever the agent writes an event to the event pipe.
+// When the timeout fires the producer and consumer contexts are canceled with
+// [ErrAgentInactivityTimeout] as the cause, and the task is finalized via the
+// existing failure path.
+//
+// A value of 0 (default) or negative disables the inactivity watcher and
+// preserves prior behavior. The timeout applies in both single-process and
+// cluster modes.
+func WithAgentInactivityTimeout(d time.Duration) RequestHandlerOption {
+	return func(ih *InterceptedHandler, h *defaultRequestHandler) {
+		h.agentInactivityTimeout = d
+	}
+}
+
 // WithConcurrencyConfig allows to set limits on the number of concurrent executions.
 func WithConcurrencyConfig(config limiter.ConcurrencyConfig) RequestHandlerOption {
 	return func(ih *InterceptedHandler, h *defaultRequestHandler) {
@@ -186,12 +211,13 @@ func NewHandler(executor AgentExecutor, options ...RequestHandlerOption) Request
 			panic("TaskStore and QueueManager must be provided for cluster mode")
 		}
 		h.execManager = taskexec.NewDistributedManager(taskexec.DistributedManagerConfig{
-			WorkQueue:         h.workQueue,
-			TaskStore:         h.taskStore,
-			QueueManager:      h.queueManager,
-			ConcurrencyConfig: h.concurrencyConfig,
-			Factory:           execFactory,
-			PanicHandler:      h.panicHandler,
+			WorkQueue:              h.workQueue,
+			TaskStore:              h.taskStore,
+			QueueManager:           h.queueManager,
+			ConcurrencyConfig:      h.concurrencyConfig,
+			Factory:                execFactory,
+			PanicHandler:           h.panicHandler,
+			AgentInactivityTimeout: h.agentInactivityTimeout,
 		})
 	} else {
 		if h.queueManager == nil {
@@ -204,11 +230,12 @@ func NewHandler(executor AgentExecutor, options ...RequestHandlerOption) Request
 			execFactory.taskStore = h.taskStore
 		}
 		h.execManager = taskexec.NewLocalManager(taskexec.LocalManagerConfig{
-			QueueManager:      h.queueManager,
-			ConcurrencyConfig: h.concurrencyConfig,
-			Factory:           execFactory,
-			TaskStore:         h.taskStore,
-			PanicHandler:      h.panicHandler,
+			QueueManager:           h.queueManager,
+			ConcurrencyConfig:      h.concurrencyConfig,
+			Factory:                execFactory,
+			TaskStore:              h.taskStore,
+			PanicHandler:           h.panicHandler,
+			AgentInactivityTimeout: h.agentInactivityTimeout,
 		})
 	}
 

internal/taskexec/api.go

@@ -16,13 +16,20 @@ package taskexec
 
 import (
 	"context"
+	"errors"
 	"iter"
 
 	"github.com/a2aproject/a2a-go/v2/a2a"
 	"github.com/a2aproject/a2a-go/v2/a2asrv/taskstore"
 	"github.com/a2aproject/a2a-go/v2/internal/eventpipe"
 )
 
+// ErrAgentInactivityTimeout is the cancellation cause set on the producer
+// context when an agent fails to write any events to the event pipe within
+// the configured inactivity timeout window. Callers can check for this
+// condition with errors.Is.
+var ErrAgentInactivityTimeout = errors.New("agent inactivity timeout")
+
 // Manager provides an API for executing and canceling tasks.
 type Manager interface {
 	// Resubscribe is used to resubscribe to events of an active execution.

internal/taskexec/distributed_manager.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"time"
 
 	"github.com/a2aproject/a2a-go/v2/a2a"
 	"github.com/a2aproject/a2a-go/v2/a2asrv/eventqueue"
@@ -38,6 +39,11 @@ type DistributedManagerConfig struct {
 	ConcurrencyConfig limiter.ConcurrencyConfig
 	Logger            *slog.Logger
 	PanicHandler      PanicHandlerFn
+	// AgentInactivityTimeout, if positive, terminates an execution when the
+	// agent's producer has not written any events to the pipe for the
+	// configured duration. The terminating cause is [ErrAgentInactivityTimeout].
+	// A value of 0 disables the watcher, preserving prior behavior.
+	AgentInactivityTimeout time.Duration
 }
 
 type distributedManager struct {

internal/taskexec/execution_handler.go

@@ -82,6 +82,71 @@ func (h *executionHandler) processEvents(ctx context.Context) (a2a.SendMessageRe
 type eventProducerFn func(context.Context) error
 type eventConsumerFn func(context.Context) (a2a.SendMessageResult, error)
 
+// inactivityTracker bundles the inactivity watcher's configuration with the
+// activity-recording channel. Callers obtain one via [newInactivityTracker]
+// (returns nil when disabled), wrap the producer's writer with
+// [newActivityTrackingWriter], and pass the tracker to [runProducerConsumer]
+// to start the watcher. A nil tracker disables tracking end-to-end: the writer
+// wrapper is a no-op and the watcher goroutine is not started.
+type inactivityTracker struct {
+	config        inactivityConfig
+	writeRecorded chan struct{}
+}
+
+// newInactivityTracker returns a tracker for the given timeout. A non-positive
+// timeout returns nil, disabling inactivity tracking.
+func newInactivityTracker(timeout time.Duration) *inactivityTracker {
+	if timeout <= 0 {
+		return nil
+	}
+	signal := make(chan struct{}, 1)
+	return &inactivityTracker{
+		config:        inactivityConfig{timeout: timeout},
+		writeRecorded: signal,
+	}
+}
+
+// record signals that a producer write just succeeded. Non-blocking: the
+// watcher only needs an "activity happened" hint, so a full channel is fine.
+func (t *inactivityTracker) record() {
+	if t == nil {
+		return
+	}
+	select {
+	case t.writeRecorded <- struct{}{}:
+	default:
+	}
+}
+
+// newActivityTrackingWriter wraps an [eventpipe.Writer] so each successful
+// write signals the provided tracker. A nil tracker returns the writer
+// unchanged so callers without an inactivity timeout configured pay no cost.
+func newActivityTrackingWriter(inner eventpipe.Writer, tracker *inactivityTracker) eventpipe.Writer {
+	if tracker == nil {
+		return inner
+	}
+	return &activityTrackingWriter{inner: inner, tracker: tracker}
+}
+
+type activityTrackingWriter struct {
+	inner   eventpipe.Writer
+	tracker *inactivityTracker
+}
+
+func (w *activityTrackingWriter) Write(ctx context.Context, event a2a.Event) error {
+	if err := w.inner.Write(ctx, event); err != nil {
+		return err
+	}
+	w.tracker.record()
+	return nil
+}
+
+// inactivityConfig configures the inactivity watcher started by
+// [runProducerConsumer]. A zero or negative timeout disables the watcher.
+type inactivityConfig struct {
+	timeout time.Duration
+}
+
 // runProducerConsumer starts producer and consumer goroutines in an error group and waits
 // for both of them to finish or one of them to fail. If both complete successfuly and consumer produces a result,
 // the result is returned, otherwise an error is returned.
@@ -91,9 +156,28 @@ func runProducerConsumer(
 	consumer eventConsumerFn,
 	heartbeater workqueue.Heartbeater,
 	panicHandler PanicHandlerFn,
+	inactivity *inactivityTracker,
 ) (a2a.SendMessageResult, error) {
 	group, ctx := errgroup.WithContext(ctx)
 
+	if inactivity != nil && inactivity.config.timeout > 0 && inactivity.writeRecorded != nil {
+		cfg := inactivity.config
+		group.Go(func() error {
+			timer := time.NewTimer(cfg.timeout)
+			defer timer.Stop()
+			for {
+				select {
+				case <-ctx.Done():
+					return ctx.Err()
+				case <-inactivity.writeRecorded:
+					timer.Reset(cfg.timeout)
+				case <-timer.C:
+					return ErrAgentInactivityTimeout
+				}
+			}
+		})
+	}
+
 	if heartbeater != nil {
 		group.Go(func() error {
 			timer := time.NewTicker(heartbeater.HeartbeatInterval())