feat(abctl): In-place pipeline editor with hot-reload

authbridge/cmd/abctl/README.md

@@ -89,9 +89,115 @@ The UI has three panes. `Enter` drills in; `Esc` backs out.
 | `p` | any | pause/resume stream |
 | `y` | detail | yank event JSON to `/tmp` |
 | `g` / `G` | lists | jump to top / bottom |
+| `e` | pipeline | edit pipeline subtree in `$EDITOR` |
+| `y` | edit/diff | apply the edit |
+| `N` | edit/diff | abort the edit |
+| `r` | edit/error | retry: re-open the editor (post-edit failure) or refetch (fetch failure) |
+| `Esc` | edit/{fetching,editing,applying} | abort the edit, return to Pipeline pane |
+| `Esc` | edit/{waiting,rollback} | background the watch; result lands as a footer flash |
 | `?` | any | (reserved for future help overlay) |
 | `q` / `Ctrl+C` | any | quit |
 
+## Editing the pipeline
+
+Press `e` on the Pipeline pane to edit the agent's runtime `pipeline:`
+subtree in `$EDITOR` (or `vi` if unset). On save, abctl shows a diff
+and asks `apply this change? (y/N)`. Confirming runs
+`kubectl apply --server-side` against the per-agent ConfigMap with
+`--field-manager=abctl --force-conflicts=true` (taking ownership of
+`data.config.yaml` from the kagenti-operator's webhook on first
+edit), then polls the framework's `/reload/status` until the reload
+completes (success or failure).
+
+The single edit flow covers four operations:
+- **Edit a value** — change a config field of an existing plugin
+- **Reorder** — move a plugin's lines up or down
+- **Remove** — delete a plugin's entry from `inbound:` or `outbound:`
+- **Add** — append a new plugin entry
+
+All four work because they're all just lines you change inside the
+pipeline subtree.
+
+`e` is only available in picker mode. With `--endpoint`, the cluster
+fields needed to fetch and apply aren't populated; pressing `e`
+flashes a hint instead of opening a broken edit.
+
+### Agent-name resolution
+
+The per-agent ConfigMap is named `authbridge-config-<agent>`. abctl
+resolves `<agent>` from the selected pod's `app.kubernetes.io/name`
+label (kagenti-operator sets this). If the label is absent, abctl
+falls back to stripping the last two dash-separated segments of the
+pod name (the ReplicaSet hash + pod suffix).
+
+### Auto-rollback on reload failure
+
+If `kubectl apply` succeeds but the in-pod reload fails (unknown
+plugin name, malformed config, validation error), the framework
+keeps the previous in-memory pipeline serving requests. The on-disk
+ConfigMap, however, now holds the bad YAML. abctl detects this via
+`/reload/status` and re-applies the original ConfigMap content
+captured at Fetch time, reconciling the on-disk state back to what's
+actually running. The error overlay then reports
+`reload failed: <reason>; rolled back to previous ConfigMap`.
+
+The rollback is best-effort — with `--force-conflicts=true`, if a
+third party (controller, kubectl edit, kustomize) modified the
+ConfigMap between Fetch and the failed reload, the rollback
+overwrites their change. The running pipeline is unaffected.
+
+### Backgrounding the watch
+
+Pressing `Esc` while waiting for hot-reload (or during rollback)
+moves the watch to the background instead of aborting it. The
+overlay closes, the footer flashes
+`hot-reload watch moved to background; you'll be notified`, and you
+can resume navigating the TUI. When the watch terminates, the
+result lands as a one-line flash:
+
+- `hot-reload succeeded`
+- `hot-reload failed: <reason>; rolled back to previous ConfigMap`
+- `hot-reload failed: <reason>; rollback failed: <err>` (rare)
+
+Flashes auto-dismiss after a few seconds; if you miss one, query
+`/reload/status` directly via the port-forward.
+
+### Permissions
+
+abctl shells out to `kubectl`; kubectl uses your kubeconfig. Editing
+requires `update` on `configmaps` in the agent's namespace (in
+addition to `get pods` which the picker already needs). RBAC denial
+surfaces verbatim in the overlay.
+
+### Tempfile lifecycle
+
+abctl writes the editable pipeline subtree to `$TMPDIR/abctl-pipeline-*.yaml`
+on every edit. The tempfile is **left in place on every exit path**
+(success, error, abort) so an interrupted edit is recoverable. On
+abctl launch, files older than 24h in this glob are swept
+automatically — no manual cleanup needed.
+
+### Hot-reload window
+
+The framework reloads via a config-file watcher; kubelet syncs
+ConfigMap edits into the pod's mount within ~60s, then the framework
+debounces and reloads. Total wall-clock from `apply` to reload is
+typically under 90s. abctl shows a spinner during the wait.
+
+The poller terminates with one of:
+
+- **Success** — `/reload/status.last_success` advances past the apply
+  time.
+- **Failure** — `reloads_failed` increments past its baseline; the
+  framework's `last_error` is shown.
+- **Unreachable** — 5 consecutive transport errors against
+  `:9093/reload/status` (port-forward dropped, framework crashed,
+  etc.) surface as `reload status endpoint unreachable` after a few
+  seconds rather than waiting the full deadline.
+- **Timeout** — none of the above within 120s. Triggers an
+  auto-rollback so the on-disk ConfigMap doesn't drift from the
+  running pipeline.
+
 ## Trust model
 
 `abctl` does no authentication — same as the server. Use only against

authbridge/cmd/abctl/cluster/portforward.go

@@ -59,8 +59,10 @@ type PortForwarder interface {
 // PortForward is a live tunnel to a pod. The caller MUST Close it
 // exactly once.
 type PortForward interface {
-	// Endpoint is the URL abctl points its apiclient at.
+	// Endpoint is the URL abctl points its apiclient at (:9094 session API).
 	Endpoint() string
+	// StatusEndpoint is the URL of the agent's stat server (:9093 /reload/status).
+	StatusEndpoint() string
 	// Close terminates the tunnel and waits for it to exit.
 	Close() error
 }
@@ -81,13 +83,30 @@ func (k *kubectlPortForwarder) Start(ctx context.Context, namespace, pod string)
 	if err != nil {
 		return nil, err
 	}
+	// freeLocalPort closes the listener it picks, leaving a tiny window
+	// where two consecutive calls can return the same port (TIME_WAIT
+	// cleared, kernel re-issues). Loop until we get a distinct one.
+	var statusPort int
+	for attempts := 0; attempts < 8; attempts++ {
+		statusPort, err = freeLocalPort()
+		if err != nil {
+			return nil, err
+		}
+		if statusPort != port {
+			break
+		}
+	}
+	if statusPort == port {
+		return nil, fmt.Errorf("port-forward: could not allocate two distinct local ports")
+	}
 	// We do NOT bind ctx to the subprocess — kubectl port-forward should
 	// outlive the per-call context (which is just for the readiness
 	// check). The subprocess is terminated explicitly via Close.
 	cmd := exec.Command("kubectl", "port-forward",
 		"-n", namespace,
 		"pod/"+pod,
 		strconv.Itoa(port)+":9094",
+		strconv.Itoa(statusPort)+":9093",
 	)
 	stderr, err := cmd.StderrPipe()
 	if err != nil {
@@ -96,7 +115,7 @@ func (k *kubectlPortForwarder) Start(ctx context.Context, namespace, pod string)
 	if err := cmd.Start(); err != nil {
 		return nil, fmt.Errorf("start kubectl port-forward: %w", err)
 	}
-	pf := &kubectlPortForward{cmd: cmd, port: port, stderr: stderr}
+	pf := &kubectlPortForward{cmd: cmd, port: port, statusPort: statusPort, stderr: stderr}
 	pf.startStderrDrain()
 
 	readyCtx, cancel := context.WithTimeout(ctx, pfReadyTimeout)
@@ -110,13 +129,22 @@ func (k *kubectlPortForwarder) Start(ctx context.Context, namespace, pod string)
 		}
 		return nil, fmt.Errorf("port-forward not ready: %w", err)
 	}
+	if err := waitForAccept(readyCtx, statusPort); err != nil {
+		_ = pf.Close()
+		stderrTail := pf.stderrTail()
+		if stderrTail != "" {
+			return nil, fmt.Errorf("port-forward (:9093) not ready: %w (stderr: %s)", err, stderrTail)
+		}
+		return nil, fmt.Errorf("port-forward (:9093) not ready: %w", err)
+	}
 	return pf, nil
 }
 
 type kubectlPortForward struct {
-	cmd    *exec.Cmd
-	port   int
-	stderr io.ReadCloser
+	cmd        *exec.Cmd
+	port       int // local 127.0.0.1 port forwarding to pod :9094
+	statusPort int // local 127.0.0.1 port forwarding to pod :9093
+	stderr     io.ReadCloser
 
 	mu          sync.Mutex
 	stderrLines []string
@@ -127,6 +155,13 @@ func (p *kubectlPortForward) Endpoint() string {
 	return "http://127.0.0.1:" + strconv.Itoa(p.port)
 }
 
+// StatusEndpoint is the URL of the agent's stat server (:9093) reached
+// via the picker's port-forward. abctl's edit flow polls /reload/status
+// here.
+func (p *kubectlPortForward) StatusEndpoint() string {
+	return "http://127.0.0.1:" + strconv.Itoa(p.statusPort)
+}
+
 func (p *kubectlPortForward) Close() error {
 	if p.cmd.Process == nil {
 		return nil

authbridge/cmd/abctl/cluster/portforward_test.go

@@ -72,3 +72,22 @@ func TestPortForwarderBuildOnly(t *testing.T) {
 	// kubectl.
 	var _ PortForwarder = NewPortForwarder()
 }
+
+func TestKubectlPortForward_StatusEndpoint(t *testing.T) {
+	pf := &kubectlPortForward{port: 12345, statusPort: 12346}
+	want := "http://127.0.0.1:12346"
+	if got := pf.StatusEndpoint(); got != want {
+		t.Fatalf("StatusEndpoint: got %q want %q", got, want)
+	}
+	if got := pf.Endpoint(); got != "http://127.0.0.1:12345" {
+		t.Fatalf("Endpoint: got %q", got)
+	}
+}
+
+func TestPortForwarderInterfaceHasStatusEndpoint(t *testing.T) {
+	var _ interface {
+		Endpoint() string
+		StatusEndpoint() string
+		Close() error
+	} = (*kubectlPortForward)(nil)
+}