Fix: Pin IBAC user intent against FIFO eviction; add listener.skip_hosts

authbridge/CLAUDE.md

@@ -579,6 +579,19 @@ 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 and IBAC user intent**: The session store is FIFO with a default cap of 100 events per session. Two layered defenses keep the inbound A2A user intent visible to IBAC even when an agent generates dozens of outbound events per turn:
+
+    - **Primary fix — `listener.skip_hosts`**: list infrastructure destinations (OTel collectors, metrics endpoints, log shippers) whose traffic should bypass the pipeline AND session recording entirely. Matched requests are forwarded as a transparent proxy: no plugin runs, no event is appended. 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.
+
+    - **Backstop — intent pin in the eviction policy**: even with `skip_hosts` empty, the session store now pins the most-recent inbound A2A request event against FIFO eviction. If the buffer overflows, every other event evicts in normal chronological order; the protected intent stays at its original timestamp, leaving a visible time gap in the timeline. Older intents from earlier turns are NOT pinned — only the latest one — so a multi-turn conversation with huge fan-out can't pile up stale intents and starve the buffer. The pin protects against FIFO eviction only: IBAC's `LastIntent()` survives buffer overflow as long as the session is still alive and an inbound A2A request has landed in it, but can still return nil after session expiry, explicit deletion, or before the first inbound request arrives. The pin is defense-in-depth; reach for `skip_hosts` first when the offending traffic is identifiable infrastructure.
+
 ## 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,67 @@ 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.
+	//
+	// Trust model — the value matched against SkipHosts is the
+	// destination Host as observed at the listener boundary, which is
+	// agent-influenceable in two of the three deployment shapes:
+	//
+	//   - ext_proc / envoy-sidecar: matches Envoy's `:authority`
+	//     (fallback `host` header). The agent sets these; Envoy may
+	//     rewrite them per its config, but ultimately the value is
+	//     "what the agent told Envoy it wanted to talk to."
+	//   - HTTP forward-proxy / proxy-sidecar: matches `r.Host` from
+	//     the HTTP request. The request is then dialed against
+	//     `r.URL`, so a forged Host that diverges from the real URL
+	//     host would skip-match yet send to the actual upstream.
+	//   - CONNECT-tunnel / proxy-sidecar: safer-by-construction —
+	//     `r.Host` IS the dial target. A forged Host cannot
+	//     skip-match while dialing elsewhere.
+	//
+	// Implication: do NOT list a destination here that you'd want
+	// IBAC / token-exchange to deny on. Skip means "the operator
+	// trusts every flow to this host enough to bypass the entire
+	// outbound enforcement pipeline." Limit entries to infrastructure
+	// destinations the agent should not be making policy decisions
+	// against in the first place (collector sidecars, log shippers).
+	//
+	// Each skip is logged at INFO with the matched host and pattern
+	// so an operator reviewing logs can see when a pattern fired and
+	// catch unexpected matches early. The `transparentproxy` listener
+	// (proxy-sidecar enforce-redirect mode) intentionally does NOT
+	// consult SkipHosts — that is the hard egress guard and must not
+	// be self-exemptable via the agent's outbound destination.
+	//
+	// Match-all patterns ("*", "**", whitespace-only) and patterns
+	// containing ":port" are rejected at startup so a single
+	// misconfigured entry can't silently disable all outbound
+	// enforcement. Mirrors the bypass-pattern guard added to ibac
+	// in #496.
+	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,18 @@ 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 pat, matched := s.SkipHosts.MatchPattern(pctx.Host); matched {
+		slog.Info("ext_proc: skip_hosts match (body phase) — bypassing pipeline + session recording",
+			"host", pctx.Host, "pattern", pat, "path", pctx.Path)
+		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())
+	}
+}