feat(plugins): Per-plugin config + /readyz aggregation + /stats merge

authbridge/CLAUDE.md

@@ -111,25 +111,24 @@ with protocol-specific listeners in `cmd/authbridge/listener/`:
 - Routes file is loaded once at startup; restart the pod to pick up changes
 
 **Configuration loading:**
-- YAML config with `${ENV_VAR}` expansion, mode presets, and startup validation
-- Supports `keycloak_url` + `keycloak_realm` derivation for operator compatibility
-- Waits up to 60s for credential files from client-registration
-- Reads `CLIENT_ID` from `/shared/client-id.txt` (file) or `CLIENT_ID` env var (fallback)
-- Reads `CLIENT_SECRET` from `/shared/client-secret.txt` (file) or `CLIENT_SECRET` env var (fallback)
-- `TOKEN_URL`: explicit env var, or auto-derived from `KEYCLOAK_URL` + `KEYCLOAK_REALM` (i.e. `{KEYCLOAK_URL}/realms/{KEYCLOAK_REALM}/protocol/openid-connect/token`)
-- `ISSUER`: explicit env var, or auto-derived from `KEYCLOAK_URL` + `KEYCLOAK_REALM` (i.e. `{KEYCLOAK_URL}/realms/{KEYCLOAK_REALM}`)
-- Inbound audience validation uses `CLIENT_ID` (from `/shared/client-id.txt` or `CLIENT_ID` env var) -- automatic, per-agent
-- Outbound route config from `/etc/authproxy/routes.yaml` (default; override with `ROUTES_CONFIG_PATH` env var in standalone deployments). Target audience and scopes are configured per-route only.
-- Default outbound policy from `DEFAULT_OUTBOUND_POLICY` env var: `"passthrough"` (default) or `"exchange"`
-- JWKS URL is derived from TOKEN_URL: replaces `/token` suffix with `/certs`
+- YAML config with `${ENV_VAR}` expansion, mode presets, and startup validation.
+- Plugin settings are local to each plugin under `pipeline.*.plugins[].config`; the runtime YAML itself only carries `mode`, `listener`, `session`, `stats`, and the pipeline composition. See [`cmd/authbridge/README.md`](cmd/authbridge/README.md) for the per-mode YAML shape and [`authlib/plugins/CONVENTIONS.md`](authlib/plugins/CONVENTIONS.md) for the per-plugin decode pattern.
+- The operator-supplied env vars (`KEYCLOAK_URL`, `KEYCLOAK_REALM`, `TOKEN_URL`, `ISSUER`, `DEFAULT_OUTBOUND_POLICY`, `CLIENT_ID`) are consumed by the default `authbridge-combined.yaml` via `${VAR}` expansion — they land inside the appropriate plugin's `config:` block rather than a top-level section.
+- `jwt-validation` derives `jwks_url` from `issuer` when omitted (appends `/protocol/openid-connect/certs`).
+- `token-exchange` derives `token_url` from `keycloak_url + keycloak_realm` when omitted (Keycloak convention).
+- Credential files: `client-registration` writes `/shared/client-id.txt` and `/shared/client-secret.txt`; `spiffe-helper` writes `/opt/jwt_svid.token`. `jwt-validation` reads the audience from `/shared/client-id.txt` via `audience_file`; `token-exchange` reads client credentials via `client_id_file` / `client_secret_file` / `jwt_svid_path`. Each plugin attempts a synchronous read at Configure time and falls back to a background poll from its `Init` goroutine if the file isn't yet readable.
+- Outbound route config: `token-exchange` reads `/etc/authproxy/routes.yaml` by default (path is per-plugin, configured via `routes.file` in its config block); inline rules can be declared under `routes.rules`.
+- Outbound `default_policy`: `passthrough` (default) or `exchange`, configured per-plugin (no top-level `DEFAULT_OUTBOUND_POLICY` field anymore; the env var is still expanded into the plugin config by `authbridge-combined.yaml`).
 
 **Key library packages (authlib/):**
-- `authlib/validation/` -- JWKS-backed JWT verifier
-- `authlib/exchange/` -- RFC 8693 token exchange client
+- `authlib/validation/` -- JWKS-backed JWT verifier (used internally by `jwt-validation` plugin)
+- `authlib/exchange/` -- RFC 8693 token exchange client (used internally by `token-exchange` plugin)
 - `authlib/cache/` -- SHA-256 keyed token cache
-- `authlib/routing/` -- Host-to-audience route resolver
-- `authlib/auth/` -- `HandleInbound` + `HandleOutbound` composition
-- `authlib/config/` -- Mode presets, YAML config, validation
+- `authlib/routing/` -- Host-to-audience route resolver (used internally by `token-exchange` plugin)
+- `authlib/auth/` -- `HandleInbound` + `HandleOutbound` composition; each plugin instance constructs its own `auth.Auth` from its own local config
+- `authlib/config/` -- Mode presets, YAML config loader, credential-file waiters, top-level (mode + listener + session) validation
+- `authlib/pipeline/` -- Plugin interface + lifecycle (`Configurable`, `Initializer`, `Shutdowner`); see [`authlib/pipeline/README.md`](authlib/pipeline/README.md)
+- `authlib/plugins/` -- The concrete plugins + registry; see [`authlib/plugins/CONVENTIONS.md`](authlib/plugins/CONVENTIONS.md) for the per-plugin config convention
 
 ### init-iptables.sh
 
@@ -396,7 +395,7 @@ Set `session.enabled: false` in the runtime config to turn off the store (and im
 
 ## Gotchas and Known Issues
 
-1. **Credential file race condition**: Authbridge waits up to 60s for `/shared/client-id.txt` and `/shared/client-secret.txt`. If client-registration takes longer (e.g., Keycloak slow to start), authbridge will fall back to env vars which may be empty.
+1. **Credential file race condition**: Each plugin that reads a credential file (jwt-validation's `audience_file`, token-exchange's `client_id_file` / `client_secret_file` / `jwt_svid_path`) tries a synchronous read at Configure time and, on miss, spawns an Init goroutine that polls indefinitely — emitting a WARN every ~60s while the file is still missing. OnRequest returns 503 until the file arrives. If the file never shows up (wrong path, missing volume mount), the pod stays unready for outbound traffic; follow the WARN lines to the misconfigured path.
 
 2. **ISSUER vs TOKEN_URL**: `ISSUER` must be the Keycloak **frontend URL** (what appears in the `iss` claim of tokens), while `TOKEN_URL` is the **internal service URL**. These are often different in Kubernetes (e.g., `http://keycloak.localtest.me:8080` vs `http://keycloak-service.keycloak.svc:8080`).
 

authbridge/authlib/README.md

@@ -12,32 +12,39 @@ A pure Go library providing reusable building blocks for JWT validation, OAuth 2
 | `bypass/` | Path pattern matcher for public endpoints (health, agent card) |
 | `spiffe/` | SPIFFE credential sources (file-based JWT-SVID) |
 | `routing/` | Host-to-audience router with glob pattern matching |
-| `auth/` | Composition layer: `HandleInbound` + `HandleOutbound` |
-| `config/` | YAML config, mode presets, startup validation, URL derivation |
-| `pipeline/` | Plugin pipeline — see [pipeline/README.md](./pipeline/README.md) for the full interface spec and bridge-plugin integration guide |
+| `auth/` | Composition layer: `HandleInbound` + `HandleOutbound` — used internally by `jwt-validation` and `token-exchange` plugins |
+| `config/` | YAML config loader, mode presets, credential-file waiters, top-level validation |
+| `pipeline/` | Plugin pipeline + lifecycle (`Configurable`, `Initializer`, `Shutdowner`) — see [pipeline/README.md](./pipeline/README.md) |
 | `session/` | In-memory session store + `SessionSummary` aggregation, backing the `:9094` API |
 | `sessionapi/` | HTTP API (`/v1/sessions`, `/v1/events` SSE, `/v1/pipeline`) exposing the session store |
-| `plugins/` | Built-in protocol parsers: `a2a-parser`, `mcp-parser`, `inference-parser`, `jwt-validation` |
+| `plugins/` | Built-in plugins: `jwt-validation`, `token-exchange`, `a2a-parser`, `mcp-parser`, `inference-parser`. See [plugins/CONVENTIONS.md](./plugins/CONVENTIONS.md) for the per-plugin config convention |
 | `observe/` | OTEL + metrics helpers |
 
 ## Usage
 
+Plugins own their own configuration and construct any `auth.Auth` they need internally from their per-plugin config (see [plugins/CONVENTIONS.md](./plugins/CONVENTIONS.md)). Host processes (cmd/authbridge) just load the YAML, call `plugins.Build`, run the pipelines, and let each plugin handle its own dependencies.
+
 ```go
 import (
-    "github.com/kagenti/kagenti-extensions/authbridge/authlib/auth"
     "github.com/kagenti/kagenti-extensions/authbridge/authlib/config"
+    "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins"
 )
 
-// Load and resolve config
 cfg, _ := config.Load("config.yaml")
-resolved, _ := config.Resolve(ctx, cfg)
-handler := auth.New(*resolved)
+config.ApplyPreset(cfg)
+_ = config.Validate(cfg) // mode + listener combo only; plugins self-validate
+
+inbound,  _ := plugins.Build(cfg.Pipeline.Inbound.Plugins)
+outbound, _ := plugins.Build(cfg.Pipeline.Outbound.Plugins)
 
-// Handle requests
-inResult := handler.HandleInbound(ctx, authHeader, path, "")
-outResult := handler.HandleOutbound(ctx, authHeader, host)
+_ = inbound.Start(ctx)   // invokes Init on plugins that implement pipeline.Initializer
+_ = outbound.Start(ctx)
+
+// Listeners drive the pipelines — see cmd/authbridge/listener/*
 ```
 
+For direct use of the `auth/` composition layer (outside of plugins), see `auth/auth.go` — `auth.New(cfg)` takes an `auth.Config` containing the specific building blocks a caller needs.
+
 ## Go Module
 
 ```

authbridge/authlib/auth/auth.go

@@ -521,6 +521,41 @@ func NewStats() *Stats {
 	}
 }
 
+// MergeStats returns a new Stats whose counters are the sum of all
+// srcs. Each src's mutex is taken independently; the returned Stats
+// has its own storage and no relationship to the sources. Safe to
+// call concurrently — sources are only read under their own locks.
+//
+// Used by the /stats aggregator to fold per-plugin counters into a
+// single response per HTTP request without leaking plugin-local
+// Stats instances into the presentation layer.
+func MergeStats(srcs ...*Stats) *Stats {
+	out := NewStats()
+	for _, s := range srcs {
+		if s == nil {
+			continue
+		}
+		s.mu.Lock()
+		for k, v := range s.inboundApprovals {
+			out.inboundApprovals[k] += v
+		}
+		for k, v := range s.inboundDenials {
+			out.inboundDenials[k] += v
+		}
+		for k, v := range s.outboundApprovals {
+			out.outboundApprovals[k] += v
+		}
+		for k, v := range s.outboundDenials {
+			out.outboundDenials[k] += v
+		}
+		for k, v := range s.outboundReplaceTokens {
+			out.outboundReplaceTokens[k] += v
+		}
+		s.mu.Unlock()
+	}
+	return out
+}
+
 // IncInboundApprove records a new approval (for statistics)
 func (a *Auth) IncInboundApprove(reason InboundApprovalReason) {
 	a.Stats.mu.Lock()

authbridge/authlib/auth/auth_test.go

@@ -453,6 +453,56 @@ func TestIncOutboundReplaceToken(t *testing.T) {
 	}
 }
 
+// MergeStats sums counters from multiple Stats objects into a fresh
+// one. Each source's mutex is taken independently — no deadlock even
+// if sources are being mutated concurrently.
+func TestMergeStats_SumsCounters(t *testing.T) {
+	a := New(Config{})
+	a.IncInboundApprove(APPROVE_AUTHORIZED)
+	a.IncInboundApprove(APPROVE_AUTHORIZED)
+	a.IncInboundDeny(DENY_NO_HEADER)
+
+	b := New(Config{})
+	b.IncInboundApprove(APPROVE_AUTHORIZED) // +1, total 3
+	b.IncInboundApprove(APPROVE_PASSTHROUGH)
+	b.IncOutboundApprove(OUTBOUND_PASSTHROUGH)
+
+	merged := MergeStats(a.Stats, b.Stats)
+	if got, want := merged.inboundApprovals[APPROVE_AUTHORIZED], 3; got != want {
+		t.Errorf("APPROVE_AUTHORIZED = %d, want %d", got, want)
+	}
+	if got, want := merged.inboundApprovals[APPROVE_PASSTHROUGH], 1; got != want {
+		t.Errorf("APPROVE_PASSTHROUGH = %d, want %d", got, want)
+	}
+	if got, want := merged.inboundDenials[DENY_NO_HEADER], 1; got != want {
+		t.Errorf("DENY_NO_HEADER = %d, want %d", got, want)
+	}
+	if got, want := merged.outboundApprovals[OUTBOUND_PASSTHROUGH], 1; got != want {
+		t.Errorf("OUTBOUND_PASSTHROUGH = %d, want %d", got, want)
+	}
+
+	// Merging must not mutate source counters — the aggregator runs
+	// on every /stats probe, and we can't have each probe double-
+	// counting.
+	if got, want := a.Stats.inboundApprovals[APPROVE_AUTHORIZED], 2; got != want {
+		t.Errorf("source a mutated: APPROVE_AUTHORIZED = %d, want %d", got, want)
+	}
+}
+
+func TestMergeStats_TolerantOfNilSources(t *testing.T) {
+	merged := MergeStats(nil, nil)
+	if merged == nil {
+		t.Fatal("MergeStats(nil, nil) returned nil")
+	}
+}
+
+func TestMergeStats_Empty(t *testing.T) {
+	merged := MergeStats()
+	if merged == nil {
+		t.Fatal("MergeStats() returned nil")
+	}
+}
+
 func TestStatsMarshalJSON_Empty(t *testing.T) {
 	s := NewStats()
 	data, err := json.Marshal(s)