fix(authbridge): Add listener.skip_hosts to bypass infrastructure traffic

A chatty outbound destination (e.g., an OpenTelemetry collector sidecar
that exports dozens of times per agent turn) fills the per-session FIFO
event buffer and evicts the inbound A2A user intent that IBAC needs to
align tool calls. Once evicted, IBAC's LastIntent() returns nil and the
plugin falls through to the no_user_context skip path with sub_reason
no_intent -- every tool call after the first is allowed without LLM-
judged alignment.

Add a listener-level skip list. Hosts matching listener.skip_hosts are
forwarded as a transparent proxy: no pipeline run, no session event
recorded. The matcher uses the same dot-delimited gobwas/glob semantics
as authproxy-routes; ports are stripped before matching. The skip
applies symmetrically to both proxy-sidecar (forward + reverse proxy)
and envoy-sidecar (ext_proc) deployments, and to both HTTP and CONNECT-
tunnel paths in proxy-sidecar mode.

Wired through the runtime config; required a switch from struct == to
reflect.DeepEqual in the reloader's listener-comparison guard since
ListenerConfig now contains a slice.

Includes per-listener tests covering: skipped host bypasses pipeline
plus recording while still forwarding upstream, non-matching host runs
the pipeline normally, and a nil matcher preserves today's behavior.

Assisted-By: Claude (Anthropic AI) <noreply@anthropic.com>
Signed-off-by: Kelly Abuelsaad <kaymar@gmail.com>

Author: kellyaa

authbridge/CLAUDE.md

@@ -579,6 +579,15 @@ See [`docs/framework-architecture.md`](docs/framework-architecture.md#9-config-h
 
 10. **Outbound passthrough is the safe default**: The `DEFAULT_OUTBOUND_POLICY` defaults to `passthrough`, which means outbound traffic to LLM inference endpoints (e.g., Ollama via `host.docker.internal`) passes through without token exchange. If this were set to `exchange`, all outbound HTTP calls would attempt token exchange and fail for non-Keycloak destinations.
 
+11. **Chatty observability traffic can evict the IBAC user intent**: The session store is FIFO with a default cap of 100 events per session. If an outbound destination is high-volume (e.g., an OpenTelemetry collector sidecar that exports dozens of times per agent turn), each export becomes a session event and the original inbound A2A user intent rolls out the back of the buffer within seconds. IBAC then sees `LastIntent() == nil` and falls through to the `no_user_context` skip path with `sub_reason: no_intent`, allowing every tool call after the first without LLM-judged alignment. Mitigate by listing the offending hosts under `listener.skip_hosts` in the runtime config — matched destinations bypass the pipeline AND session recording entirely (transparent forward), so they no longer compete with agent-meaningful events for buffer slots. Patterns use the same `.`-delimited glob semantics as `authproxy-routes`; ports are stripped before matching. Example:
+    ```yaml
+    listener:
+      skip_hosts:
+        - "otel-collector.*.svc.cluster.local"
+        - "*.metrics.local"
+    ```
+    Any change to `listener.skip_hosts` requires a pod restart (same rule as other `listener.*` fields). Do NOT add hosts here that need IBAC / token-exchange policy applied — bypass means bypass.
+
 ## DCO Sign-Off (Mandatory)
 
 All commits **must** include a `Signed-off-by` trailer (Developer Certificate of Origin).

authbridge/authlib/config/config.go

@@ -334,6 +334,31 @@ type ListenerConfig struct {
 	// (JSON snapshots + SSE stream consumed by abctl or curl). Default per
 	// mode preset is ":9094". Set to empty string to disable the endpoint.
 	SessionAPIAddr string `yaml:"session_api_addr" json:"session_api_addr"`
+
+	// SkipHosts lists outbound destination host patterns whose traffic
+	// bypasses the plugin pipeline AND session recording entirely. The
+	// listener forwards matched requests as a transparent proxy without
+	// running plugins or appending events to any session bucket.
+	//
+	// Intended for high-volume infrastructure traffic that competes
+	// with agent-meaningful events for session-buffer slots. The
+	// canonical example: an OpenTelemetry collector sidecar that emits
+	// dozens of exports per agent turn — without this gate, those
+	// exports occupy the session buffer's FIFO eviction window and
+	// silently push out the inbound A2A user intent that IBAC needs
+	// to align tool calls against, causing IBAC to fall through to
+	// the no_intent skip path on every call after the first.
+	//
+	// Patterns use `.`-delimited glob semantics (same library as
+	// `authproxy-routes`): "otel-collector*" matches the short
+	// service name, "otel-collector.kagenti-system.svc.cluster.local"
+	// matches the FQDN, "*-collector" matches any single-label name
+	// ending in -collector. Port is stripped before matching, so
+	// patterns must NOT include `:port`.
+	//
+	// Empty list (default) preserves current behavior: every outbound
+	// host runs the pipeline and is eligible for session recording.
+	SkipHosts []string `yaml:"skip_hosts" json:"skip_hosts"`
 }
 
 // StatsConfig represents the configuration for reporting config and statistics

authbridge/authlib/listener/extproc/server.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/auth"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/listener/internal/sseframe"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/listener/skiphost"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/session"
 )
@@ -41,6 +42,15 @@ type Server struct {
 	OutboundPipeline *pipeline.Holder
 	Sessions         *session.Store       // nil when session tracking is disabled
 	Shared           pipeline.SharedStore // process-scoped store; set by main, may be nil
+
+	// SkipHosts, when non-nil and matching pctx.Host on an outbound
+	// request, causes the listener to return passResponse() / nil pctx
+	// immediately — bypassing the pipeline AND session recording for
+	// that request. Forward the bytes; do nothing else. See
+	// authlib/config/config.go ListenerConfig.SkipHosts for the
+	// motivating case (OTel-collector traffic evicting the inbound
+	// A2A intent from the session buffer's FIFO window).
+	SkipHosts *skiphost.Matcher
 }
 
 // Process handles the bidirectional ext_proc stream.
@@ -467,6 +477,15 @@ func (s *Server) handleOutbound(stream extprocv3.ExternalProcessor_ProcessServer
 		pctx.Host = getHeader(headers, "host")
 	}
 
+	// SkipHosts short-circuit: forward the request as a transparent
+	// proxy without running the pipeline or recording a session event.
+	// pctx=nil signals the response handlers (handleResponseHeaders,
+	// handleResponseBody) and the deferred RunFinish to no-op as well —
+	// all four phases are skipped consistently. See ListenerConfig.SkipHosts.
+	if s.SkipHosts.Match(pctx.Host) {
+		return passResponse(), nil
+	}
+
 	if s.Sessions != nil {
 		if aid := s.Sessions.ActiveSession(); aid != "" {
 			pctx.Session = s.Sessions.View(aid)
@@ -506,6 +525,16 @@ func (s *Server) handleOutboundBody(stream extprocv3.ExternalProcessor_ProcessSe
 		pctx.Host = getHeader(headers, "host")
 	}
 
+	// SkipHosts short-circuit: see handleOutbound for rationale. The
+	// body-phase entry point needs the same gate because Envoy may
+	// deliver the body in a separate ProcessingRequest message even
+	// when the headers were already passed through — without checking
+	// here, a skip-listed host whose request carries a body would still
+	// run the pipeline on the body phase.
+	if s.SkipHosts.Match(pctx.Host) {
+		return allowBodyResponse(), nil
+	}
+
 	if s.Sessions != nil {
 		if aid := s.Sessions.ActiveSession(); aid != "" {
 			pctx.Session = s.Sessions.View(aid)

authbridge/authlib/listener/extproc/skiphost_test.go

@@ -0,0 +1,174 @@
+package extproc
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
+
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/listener/skiphost"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/plugintesting"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/session"
+)
+
+// markerPlugin records one Invocation per OnRequest call so tests can
+// assert whether the outbound pipeline ran. Mirrors the helper in the
+// forwardproxy skiphost tests; kept local to avoid a public test-only
+// type in plugintesting.
+type markerPlugin struct {
+	calls atomic.Int32
+}
+
+func (p *markerPlugin) Name() string                              { return "marker" }
+func (p *markerPlugin) Capabilities() pipeline.PluginCapabilities { return pipeline.PluginCapabilities{} }
+func (p *markerPlugin) OnResponse(context.Context, *pipeline.Context) pipeline.Action {
+	return pipeline.Action{Type: pipeline.Continue}
+}
+
+func (p *markerPlugin) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action {
+	p.calls.Add(1)
+	pctx.Record(pipeline.Invocation{
+		Plugin: "marker",
+		Action: pipeline.ActionObserve,
+		Phase:  pipeline.InvocationPhaseRequest,
+		Reason: "ran",
+	})
+	return pipeline.Action{Type: pipeline.Continue}
+}
+
+func newSkipServer(t *testing.T, store *session.Store, skip *skiphost.Matcher) (*Server, *markerPlugin) {
+	t.Helper()
+	inbound, err := plugintesting.BuildPipeline([]pipeline.Plugin{})
+	if err != nil {
+		t.Fatalf("building inbound pipeline: %v", err)
+	}
+	mp := &markerPlugin{}
+	outbound, err := plugintesting.BuildPipeline([]pipeline.Plugin{mp})
+	if err != nil {
+		t.Fatalf("building outbound pipeline: %v", err)
+	}
+	return &Server{
+		InboundPipeline:  pipeline.NewHolder(inbound),
+		OutboundPipeline: pipeline.NewHolder(outbound),
+		Sessions:         store,
+		SkipHosts:        skip,
+	}, mp
+}
+
+// TestExtProc_SkipHosts_OutboundBypass asserts the headers-only outbound
+// path: a SkipHosts-matched destination produces zero plugin invocations
+// and zero session events, and the response is a plain pass-through.
+// The motivating case is OTel-collector traffic in envoy-sidecar
+// deployments — without this gate, every export from the agent would
+// run the pipeline and append a session event, evicting the inbound
+// A2A user intent from the FIFO buffer.
+func TestExtProc_SkipHosts_OutboundBypass(t *testing.T) {
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+
+	// Match the agent's exgentic-style FQDN pattern: a leading-* glob
+	// against the fixed suffix is the operator-friendly way to write
+	// this and matches the hostname after net.SplitHostPort strips the
+	// :8335 from pctx.Host.
+	skip, err := skiphost.New([]string{"otel-collector.*.svc.cluster.local"})
+	if err != nil {
+		t.Fatalf("skiphost.New: %v", err)
+	}
+
+	srv, mp := newSkipServer(t, store, skip)
+
+	stream := &mockStream{
+		ctx: context.Background(),
+		requests: []*extprocv3.ProcessingRequest{
+			outboundRequest(makeHeaders(
+				":authority", "otel-collector.kagenti-system.svc.cluster.local:8335",
+				":path", "/v1/traces",
+			)),
+		},
+	}
+
+	_ = srv.Process(stream)
+
+	if len(stream.responses) != 1 {
+		t.Fatalf("expected 1 response, got %d", len(stream.responses))
+	}
+	rh := stream.responses[0].GetRequestHeaders()
+	if rh == nil {
+		t.Fatal("expected RequestHeaders pass-through response")
+	}
+	if rh.Response != nil && rh.Response.HeaderMutation != nil &&
+		len(rh.Response.HeaderMutation.SetHeaders) > 0 {
+		t.Error("skipped host must not have header mutations (pipeline did not run)")
+	}
+	if mp.calls.Load() != 0 {
+		t.Errorf("pipeline ran %d times; want 0 — SkipHosts must short-circuit before pipeline.Run", mp.calls.Load())
+	}
+	if sessions := store.ListSessions(); len(sessions) != 0 {
+		t.Errorf("%d session(s) recorded; want 0 — SkipHosts must skip recording entirely", len(sessions))
+	}
+}
+
+// TestExtProc_SkipHosts_NonMatchingRunsPipeline is the regression guard:
+// with a SkipHosts list set, hosts that don't match must still run the
+// pipeline and have their Invocation recorded. Without this pairing,
+// the bypass test above could pass trivially with a globally disabled
+// pipeline.
+func TestExtProc_SkipHosts_NonMatchingRunsPipeline(t *testing.T) {
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+
+	skip, err := skiphost.New([]string{"otel-collector*"})
+	if err != nil {
+		t.Fatalf("skiphost.New: %v", err)
+	}
+
+	srv, mp := newSkipServer(t, store, skip)
+
+	stream := &mockStream{
+		ctx: context.Background(),
+		requests: []*extprocv3.ProcessingRequest{
+			outboundRequest(makeHeaders(
+				":authority", "github-tool-mcp:8000",
+				":path", "/mcp",
+			)),
+		},
+	}
+
+	_ = srv.Process(stream)
+
+	if mp.calls.Load() != 1 {
+		t.Errorf("pipeline ran %d times; want 1 — host did not match skip list", mp.calls.Load())
+	}
+	if sessions := store.ListSessions(); len(sessions) != 1 {
+		t.Errorf("session count = %d; want 1 — Invocation should drive recording for non-skipped hosts", len(sessions))
+	}
+}
+
+// TestExtProc_SkipHosts_NilMatcherPreservesBehavior asserts the
+// upgrade-safety contract: a Server without SkipHosts (nil Matcher)
+// behaves identically to today's code. Pipeline runs, sessions record.
+func TestExtProc_SkipHosts_NilMatcherPreservesBehavior(t *testing.T) {
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+
+	srv, mp := newSkipServer(t, store, nil)
+
+	stream := &mockStream{
+		ctx: context.Background(),
+		requests: []*extprocv3.ProcessingRequest{
+			outboundRequest(makeHeaders(
+				":authority", "any-service",
+				":path", "/",
+			)),
+		},
+	}
+
+	_ = srv.Process(stream)
+
+	if mp.calls.Load() != 1 {
+		t.Errorf("nil SkipHosts: pipeline ran %d times, want 1", mp.calls.Load())
+	}
+}