feat: implement replica_rightsizing, rightsize_pvc, volume_delete agent_task remediations

charts/nudgebee-agent/values.yaml

@@ -61,7 +61,7 @@ runnerServiceAccount:
 runner:
   image:
     repository: ghcr.io/nudgebee/nudgebee-agent
-    tag: 2026-06-15T07-37-24_c32d436823190c10409d9382076601c3bce0e930
+    tag: 2026-06-15T10-47-06_f694c523c6ce9a8c5d2fc3c923866e6bbed0b0c8
   # Image template the pod_profiler action launches debugger pods from.
   # The agent substitutes `{}` for the variant (bpf, jvm, python, perf, ruby).
   # Surfaces as PROFILER_IMAGE; leave empty to fall back to the binary default.

runner/cmd/actionctl/main.go

@@ -0,0 +1,112 @@
+// Command actionctl is a local dev harness for invoking the agent's
+// agent_task remediation handlers against a real cluster, without standing up
+// the relay / backend / poller. It builds the same mutate.Handlers map the
+// agent registers and calls one handler by action_name with a params map —
+// exactly what the task poller does via HandleTrusted.
+//
+// Cluster creds come from in-cluster config or, locally, $KUBECONFIG /
+// ~/.kube/config (k8sclient.New). Use it to exercise replica_rightsizing,
+// rightsize_pvc (expand + downsize migration), and volume_delete.
+//
+// Examples:
+//
+//	go run ./cmd/actionctl -action replica_rightsizing -kind Deployment -namespace demo -name web -replicas 0
+//	go run ./cmd/actionctl -action rightsize_pvc -namespace demo -name data -size 3Gi     # expand
+//	go run ./cmd/actionctl -action rightsize_pvc -namespace demo -name data -size 1Gi     # downsize migration
+//	go run ./cmd/actionctl -action volume_delete -pv pvc-abc123                            # PV name
+//
+// This is a mutation tool. It acts on whatever cluster your kubeconfig points
+// at — double-check the context first.
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"os"
+	"time"
+
+	"k8s.io/client-go/dynamic"
+
+	"github.com/nudgebee/nudgebee-agent/internal/k8sclient"
+	"github.com/nudgebee/nudgebee-agent/pkg/mutate"
+	"github.com/nudgebee/nudgebee-agent/pkg/podexec"
+)
+
+func main() {
+	var (
+		action     = flag.String("action", "", "replica_rightsizing | rightsize_pvc | volume_delete")
+		kubeconfig = flag.String("kubeconfig", "", "kubeconfig path (default: $KUBECONFIG or ~/.kube/config)")
+		namespace  = flag.String("namespace", "", "namespace (rightsize_pvc, replica_rightsizing)")
+		name       = flag.String("name", "", "PVC name (rightsize_pvc) or workload name (replica_rightsizing)")
+		size       = flag.String("size", "", "target size for rightsize_pvc, e.g. 3Gi")
+		kind       = flag.String("kind", "", "Deployment | StatefulSet | Rollout (replica_rightsizing)")
+		replicas   = flag.Int("replicas", -1, "target replica count (replica_rightsizing)")
+		pv         = flag.String("pv", "", "PersistentVolume name (volume_delete)")
+		timeout    = flag.Duration("timeout", 55*time.Minute, "overall deadline (downsize copies can be slow)")
+	)
+	flag.Parse()
+	if *action == "" {
+		fmt.Fprintln(os.Stderr, "missing -action")
+		flag.Usage()
+		os.Exit(2)
+	}
+
+	cs, restCfg, err := k8sclient.New(*kubeconfig)
+	if err != nil {
+		fatal("build kube client", err)
+	}
+	dyn, err := dynamic.NewForConfig(restCfg)
+	if err != nil {
+		fatal("build dynamic client", err)
+	}
+
+	mut := mutate.New(cs, "", nil)
+	mut.SetDynamic(dyn)
+	mut.SetExec(podexec.New(cs, restCfg))
+
+	handlers := mutate.Handlers(mut)
+	h, ok := handlers[*action]
+	if !ok {
+		fatal("unknown action", fmt.Errorf("%q not registered (have it spelled right?)", *action))
+	}
+
+	var params map[string]any
+	switch *action {
+	case "replica_rightsizing":
+		if *replicas < 0 {
+			fatal("params", fmt.Errorf("replica_rightsizing needs -replicas >= 0"))
+		}
+		params = map[string]any{"kind": *kind, "namespace": *namespace, "name": *name, "replica_count": *replicas}
+	case "rightsize_pvc":
+		params = map[string]any{"namespace": *namespace, "name": *name, "size": *size}
+	case "volume_delete":
+		params = map[string]any{"name": *pv}
+	default:
+		fatal("params", fmt.Errorf("no param mapping for action %q", *action))
+	}
+
+	fmt.Printf("→ %s %s\n", *action, mustJSON(params))
+	ctx, cancel := context.WithTimeout(context.Background(), *timeout)
+
+	start := time.Now()
+	data, err := h(ctx, params)
+	cancel() // done with ctx; explicit so the os.Exit paths below don't skip a defer
+	elapsed := time.Since(start).Round(time.Millisecond)
+	if err != nil {
+		fmt.Printf("✗ FAILED in %s: %v\n", elapsed, err)
+		os.Exit(1)
+	}
+	fmt.Printf("✓ OK in %s: %s\n", elapsed, mustJSON(data))
+}
+
+func fatal(ctx string, err error) {
+	fmt.Fprintf(os.Stderr, "%s: %v\n", ctx, err)
+	os.Exit(1)
+}
+
+func mustJSON(v any) string {
+	b, _ := json.Marshal(v)
+	return string(b)
+}

runner/cmd/agent/main.go

@@ -538,6 +538,12 @@ func run(ctx context.Context, logger *slog.Logger, cfg *config.Config) error {
 		}
 		mut := mutate.New(typedKube, cfg.AlertManagerURL, amHeaders)
 		mut.SetDynamic(dynamicKube) // unlocks PrometheusRule CRUD actions
+		// SPDY exec capability for the rightsize_pvc downsize migration's
+		// data-mover pod. Without a REST config the downsize path errors at
+		// request time; expansion / replica / volume_delete still work.
+		if kubeRestCfg != nil {
+			mut.SetExec(podexec.New(typedKube, kubeRestCfg))
+		}
 		// INSTALLATION_NAMESPACE is set by the chart via the downward API.
 		// Required by the legacy alert-rule path; falls back to the scanner
 		// namespace (also chart-set) so a hand-rolled deployment without
@@ -638,8 +644,27 @@ func run(ctx context.Context, logger *slog.Logger, cfg *config.Config) error {
 	lightActions["refresh_playbook"] = struct{}{}
 	validator.SetLightActions(lightActions)
 
+	// longActions run far past the 180s default when invoked through the
+	// trusted agent_task poller — currently only the rightsize_pvc downsize
+	// migration (copies volume data via a mover pod). The ceiling stays under
+	// the server's 60-min PROCESSING→TIMEOUT reap so a task isn't force-failed
+	// mid-flight. Override with LONG_TASK_TIMEOUT_SECONDS.
+	longActions := map[string]struct{}{"rightsize_pvc": {}}
+	longTaskTimeout := 50 * time.Minute
+	if v := os.Getenv("LONG_TASK_TIMEOUT_SECONDS"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			longTaskTimeout = time.Duration(n) * time.Second
+		} else {
+			logger.Warn("invalid LONG_TASK_TIMEOUT_SECONDS, using default", "value", v)
+		}
+	}
+
 	mreg := metrics.New()
-	disp := dispatch.New(dispatch.Config{Logger: logger}, validator, handlers)
+	disp := dispatch.New(dispatch.Config{
+		Logger:          logger,
+		LongTaskTimeout: longTaskTimeout,
+		LongActions:     longActions,
+	}, validator, handlers)
 	disp.SetMetrics(mreg)
 
 	// Pod-shell session manager (TerminalRequest WS shape, output_type=Terminal).
@@ -948,11 +973,12 @@ func run(ctx context.Context, logger *slog.Logger, cfg *config.Config) error {
 	// the tenant. Period defaults to 120s (matches TASK_RUNNER_WINDOW).
 	if cfg.BackendEndpoint != "" {
 		ts := &tasks.Service{
-			Endpoint:   cfg.BackendEndpoint,
-			AuthSecret: cfg.AuthSecretKey,
-			Period:     tasks.ParseTaskWindow(os.Getenv("TASK_RUNNER_WINDOW")),
-			Logger:     logger,
-			Dispatch:   disp,
+			Endpoint:    cfg.BackendEndpoint,
+			AuthSecret:  cfg.AuthSecretKey,
+			Period:      tasks.ParseTaskWindow(os.Getenv("TASK_RUNNER_WINDOW")),
+			Logger:      logger,
+			Dispatch:    disp,
+			LongActions: longActions,
 		}
 		g.Go(func() error {
 			logger.Info("starting task poller", "endpoint", cfg.BackendEndpoint, "period", ts.Period)

runner/pkg/dispatch/dispatch.go

@@ -35,6 +35,16 @@ type Config struct {
 	HighPriorityPoolSize int           // default 3  (WEBSOCKET_HIGH_PRIORITY_THREADPOOL_SIZE)
 	TaskTimeout          time.Duration // default 180s
 	Logger               *slog.Logger
+
+	// LongTaskTimeout is the deadline applied to actions in LongActions when
+	// invoked through HandleTrusted (the agent_task poller). It exists for
+	// long-running remediations — notably the rightsize_pvc downsize
+	// migration, which copies volume data via a mover pod and routinely
+	// exceeds the 180s default. 0 disables the override (LongActions then run
+	// under TaskTimeout). Only the trusted poller path honours this; the WS
+	// Handle path always uses TaskTimeout (no long action reaches it).
+	LongTaskTimeout time.Duration
+	LongActions     map[string]struct{}
 }
 
 // Metrics is the optional metrics sink. Pass *metrics.Registry from main, or
@@ -131,7 +141,13 @@ func (d *Dispatcher) HandleTrusted(ctx context.Context, actionName string, param
 	}
 	defer pool.Release(1)
 
-	taskCtx, cancel := context.WithTimeout(ctx, d.cfg.TaskTimeout)
+	timeout := d.cfg.TaskTimeout
+	if d.cfg.LongTaskTimeout > 0 {
+		if _, long := d.cfg.LongActions[actionName]; long {
+			timeout = d.cfg.LongTaskTimeout
+		}
+	}
+	taskCtx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
 
 	start := time.Now()

runner/pkg/dispatch/dispatch_test.go

@@ -439,3 +439,42 @@ func TestDispatch_RegularActionUnchanged(t *testing.T) {
 
 // silence unused-imports warning for auth (kept for future tests)
 var _ = auth.Validator{}
+
+// TestHandleTrusted_LongActionTimeout verifies the per-action timeout: a long
+// action runs under LongTaskTimeout while everything else uses the (short)
+// TaskTimeout and times out.
+func TestHandleTrusted_LongActionTimeout(t *testing.T) {
+	d := New(Config{
+		TaskTimeout:     20 * time.Millisecond,
+		LongTaskTimeout: 2 * time.Second,
+		LongActions:     map[string]struct{}{"migrate": {}},
+	}, nil, map[string]Handler{
+		"migrate": Handler(func(ctx context.Context, _ map[string]any) (any, error) {
+			select {
+			case <-time.After(80 * time.Millisecond):
+				return "ok", nil
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			}
+		}),
+		"quick": Handler(func(ctx context.Context, _ map[string]any) (any, error) {
+			select {
+			case <-time.After(80 * time.Millisecond):
+				return "ok", nil
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			}
+		}),
+	})
+
+	// long action: survives past the short TaskTimeout.
+	data, ok, err := d.HandleTrusted(context.Background(), "migrate", nil, false)
+	if !ok || err != nil || data != "ok" {
+		t.Errorf("long action: data=%v ok=%v err=%v; want ok", data, ok, err)
+	}
+	// non-long action: reaped at TaskTimeout.
+	_, ok, err = d.HandleTrusted(context.Background(), "quick", nil, false)
+	if !ok || err == nil {
+		t.Errorf("short action should hit deadline; ok=%v err=%v", ok, err)
+	}
+}

runner/pkg/mutate/handlers.go

@@ -30,7 +30,16 @@ func Handlers(m *Mutator) map[string]dispatch.Handler {
 		hs["create_or_replace_alert_rule"] = wrap(m, handleCreateOrReplacePromRule)
 		hs["delete_alert_rule"] = wrapErr(m, handleDeletePromRule)
 		hs["replace_workload"] = wrap(m, handleReplaceWorkload)
-	}
+		// replica_rightsizing scales Deployment/StatefulSet/Rollout via the
+		// dynamic client. Delivered through the agent_task poller (trusted),
+		// so — like rightsizing_resource — it is deliberately NOT a lightAction.
+		hs["replica_rightsizing"] = wrap(m, handleReplicaRightsizing)
+	}
+	// PVC remediation needs only the typed client. Same trusted-poller
+	// posture: kept out of lightActions (callers reach these via agent_task,
+	// never unsigned WS).
+	hs["rightsize_pvc"] = wrap(m, handleRightsizePVC)
+	hs["volume_delete"] = wrap(m, handleVolumeDelete)
 	if m.LokiRulesURL != "" {
 		hs["create_loki_alert_rule"] = wrap(m, handleCreateLokiRule)
 		hs["update_loki_alert_rule"] = wrap(m, handleCreateLokiRule) // upsert; same handler
@@ -270,6 +279,25 @@ func perKindBodyField(kind string) string {
 	return ""
 }
 
+// handleReplicaRightsizing scales a workload. replica_count arrives as a JSON
+// number (scale-to-zero path) or a numeric string (event-resolution path);
+// toInt64 coerces both. kind is one of Deployment/StatefulSet/Rollout.
+func handleReplicaRightsizing(ctx context.Context, m *Mutator, p map[string]any) (any, error) {
+	replicas, ok := toInt64(p["replica_count"])
+	if !ok {
+		return nil, errors.New("replica_rightsizing: replica_count required (integer or numeric string)")
+	}
+	return m.ScaleWorkload(ctx, str(p, "kind"), str(p, "namespace"), str(p, "name"), replicas)
+}
+
+func handleRightsizePVC(ctx context.Context, m *Mutator, p map[string]any) (any, error) {
+	return m.RightsizePVC(ctx, str(p, "namespace"), str(p, "name"), str(p, "size"))
+}
+
+func handleVolumeDelete(ctx context.Context, m *Mutator, p map[string]any) (any, error) {
+	return m.DeleteVolume(ctx, str(p, "name"))
+}
+
 func handleCreateLokiRule(ctx context.Context, m *Mutator, p map[string]any) (any, error) {
 	body, _ := p["body"].(string)
 	resp, err := m.CreateOrReplaceLokiAlertRule(ctx, str(p, "namespace"), body)