diff --git a/authbridge/CLAUDE.md b/authbridge/CLAUDE.md
index 8ba9985aa..38a7a7001 100644
--- a/authbridge/CLAUDE.md
+++ b/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).
diff --git a/authbridge/authlib/config/config.go b/authbridge/authlib/config/config.go
index 349b9970a..c9ca9854c 100644
--- a/authbridge/authlib/config/config.go
+++ b/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
diff --git a/authbridge/authlib/listener/extproc/server.go b/authbridge/authlib/listener/extproc/server.go
index d733d0e4c..466a9bcd0 100644
--- a/authbridge/authlib/listener/extproc/server.go
+++ b/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)
diff --git a/authbridge/authlib/listener/extproc/skiphost_test.go b/authbridge/authlib/listener/extproc/skiphost_test.go
new file mode 100644
index 000000000..a54399930
--- /dev/null
+++ b/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())
+	}
+}
diff --git a/authbridge/authlib/listener/forwardproxy/server.go b/authbridge/authlib/listener/forwardproxy/server.go
index cc2e6e356..788834f2e 100644
--- a/authbridge/authlib/listener/forwardproxy/server.go
+++ b/authbridge/authlib/listener/forwardproxy/server.go
@@ -19,6 +19,7 @@ import (
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/auth"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/listener/httpx"
 	"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"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/spiffe"
@@ -48,6 +49,15 @@ type Server struct {
 	Sessions         *session.Store       // nil when session tracking is disabled
 	Shared           pipeline.SharedStore // process-scoped store; set by main, may be nil
 	Client           *http.Client
+
+	// SkipHosts, when non-nil and matching the request Host, causes
+	// the listener to forward the request as a transparent proxy:
+	// no pipeline run, no session recording. Applies to both HTTP
+	// (handleRequest) and CONNECT-tunnel (handleConnect) paths so
+	// matched destinations behave identically regardless of scheme.
+	// See authlib/config/config.go ListenerConfig.SkipHosts for
+	// motivation.
+	SkipHosts *skiphost.Matcher
 }
 
 // MTLSOptions configures outbound mTLS for the forward proxy. When
@@ -195,15 +205,37 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) {
 		StartedAt: time.Now(),
 	}
 
+	// SkipHosts short-circuit: forward as a transparent proxy. No
+	// pipeline run, no body buffering, no session recording, no
+	// response-phase work. RunFinish is also skipped (no defer
+	// registered) because the pipeline never ran and has nothing to
+	// finalize. See ListenerConfig.SkipHosts for motivation.
+	//
+	// Audit log: Match keys on r.Host (the agent-supplied Host header
+	// at the listener boundary), and the request is then dialed against
+	// r.URL via s.Client.Do(r). A forged Host that diverges from the
+	// dial target would skip-match yet send to a different upstream —
+	// the same trust shape as ext_proc's :authority. Logging the host
+	// + matched pattern at INFO leaves a per-skip audit trail so
+	// successful self-exemption isn't invisible.
+	pat, skipped := s.SkipHosts.MatchPattern(pctx.Host)
+	if skipped {
+		slog.Info("forward-proxy: skip_hosts match — bypassing pipeline + session recording",
+			"host", pctx.Host, "pattern", pat, "method", r.Method, "path", r.URL.Path)
+	}
+
 	// Finisher dispatch runs after every exit path. RunFinish is a
 	// no-op when pctx.dispatched is empty (pre-pipeline rejects), so
 	// this defer is safe on every path including the body-too-large
-	// early return.
-	defer func() {
-		s.OutboundPipeline.RunFinish(r.Context(), pctx, pipeline.OutcomeFromContext(pctx))
-	}()
+	// early return. Suppressed when skipped because no plugin saw
+	// this request and there is nothing to finalize.
+	if !skipped {
+		defer func() {
+			s.OutboundPipeline.RunFinish(r.Context(), pctx, pipeline.OutcomeFromContext(pctx))
+		}()
+	}
 
-	if s.OutboundPipeline.NeedsBody() && r.Body != nil {
+	if !skipped && s.OutboundPipeline.NeedsBody() && r.Body != nil {
 		r.Body = http.MaxBytesReader(w, r.Body, maxBodySize)
 		body, err := io.ReadAll(r.Body)
 		if err != nil {
@@ -216,22 +248,24 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) {
 		slog.Debug("forward-proxy: buffered request body", "host", r.Host, "bodyLen", len(body))
 	}
 
-	if s.Sessions != nil {
+	if !skipped && s.Sessions != nil {
 		if aid := s.Sessions.ActiveSession(); aid != "" {
 			pctx.Session = s.Sessions.View(aid)
 		}
 	}
 
 	originalAuth := pctx.Headers.Get("Authorization")
-	action := s.OutboundPipeline.Run(r.Context(), pctx)
+	if !skipped {
+		action := s.OutboundPipeline.Run(r.Context(), pctx)
 
-	if action.Type == pipeline.Reject {
-		s.recordOutboundReject(pctx, action)
-		httpx.WriteRejection(w, action)
-		return
+		if action.Type == pipeline.Reject {
+			s.recordOutboundReject(pctx, action)
+			httpx.WriteRejection(w, action)
+			return
+		}
 	}
 
-	if s.Sessions != nil {
+	if !skipped && s.Sessions != nil {
 		sid := s.Sessions.ActiveSession()
 		if sid == "" {
 			sid = session.DefaultSessionID
@@ -303,74 +337,79 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) {
 	pctx.StatusCode = resp.StatusCode
 	pctx.ResponseHeaders = resp.Header.Clone()
 
-	// Branch on Content-Type per response. The Streamable HTTP transport
-	// lets the server pick application/json vs text/event-stream per
-	// response (the client Accepts both), so the same tool may return
-	// JSON on one call and SSE on the next. Decide here rather than
-	// negotiating, and don't take the streaming path when a plugin
-	// declares WritesBody (mutating a body we've already started
-	// forwarding is incompatible with streaming) — fall back to
-	// buffered with a warning log instead.
-	if isEventStream(resp.Header.Get("Content-Type")) &&
-		s.OutboundPipeline.HasStreamingResponders() &&
-		resp.Body != nil {
-		if s.OutboundPipeline.WritesBody() {
-			slog.Warn("forward-proxy: text/event-stream response with WritesBody plugin — falling back to buffered path", "host", r.Host)
-		} else {
-			s.handleStreamingResponse(w, r, resp, pctx)
-			return
+	// SkipHosts: bypass response-phase pipeline + recording entirely.
+	// Stream the upstream body straight through to the caller. Falls
+	// out below to the unconditional header copy + io.Copy.
+	if !skipped {
+		// Branch on Content-Type per response. The Streamable HTTP transport
+		// lets the server pick application/json vs text/event-stream per
+		// response (the client Accepts both), so the same tool may return
+		// JSON on one call and SSE on the next. Decide here rather than
+		// negotiating, and don't take the streaming path when a plugin
+		// declares WritesBody (mutating a body we've already started
+		// forwarding is incompatible with streaming) — fall back to
+		// buffered with a warning log instead.
+		if isEventStream(resp.Header.Get("Content-Type")) &&
+			s.OutboundPipeline.HasStreamingResponders() &&
+			resp.Body != nil {
+			if s.OutboundPipeline.WritesBody() {
+				slog.Warn("forward-proxy: text/event-stream response with WritesBody plugin — falling back to buffered path", "host", r.Host)
+			} else {
+				s.handleStreamingResponse(w, r, resp, pctx)
+				return
+			}
 		}
-	}
 
-	if s.OutboundPipeline.NeedsBody() && resp.Body != nil {
-		respBody, err := io.ReadAll(io.LimitReader(resp.Body, maxBodySize+1))
-		if err != nil {
-			slog.Warn("forward-proxy: response body read error", "host", r.Host, "error", err)
-			http.Error(w, `{"error":"response body read error"}`, http.StatusBadGateway)
-			return
+		if s.OutboundPipeline.NeedsBody() && resp.Body != nil {
+			respBody, err := io.ReadAll(io.LimitReader(resp.Body, maxBodySize+1))
+			if err != nil {
+				slog.Warn("forward-proxy: response body read error", "host", r.Host, "error", err)
+				http.Error(w, `{"error":"response body read error"}`, http.StatusBadGateway)
+				return
+			}
+			if len(respBody) > maxBodySize {
+				slog.Warn("forward-proxy: response body too large", "host", r.Host, "len", len(respBody))
+				http.Error(w, `{"error":"response body too large"}`, http.StatusBadGateway)
+				return
+			}
+			pctx.ResponseBody = respBody
+			resp.Body = io.NopCloser(bytes.NewReader(respBody))
 		}
-		if len(respBody) > maxBodySize {
-			slog.Warn("forward-proxy: response body too large", "host", r.Host, "len", len(respBody))
-			http.Error(w, `{"error":"response body too large"}`, http.StatusBadGateway)
+
+		respAction := s.OutboundPipeline.RunResponse(r.Context(), pctx)
+		if respAction.Type == pipeline.Reject {
+			httpx.WriteRejection(w, respAction)
 			return
 		}
-		pctx.ResponseBody = respBody
-		resp.Body = io.NopCloser(bytes.NewReader(respBody))
-	}
 
-	respAction := s.OutboundPipeline.RunResponse(r.Context(), pctx)
-	if respAction.Type == pipeline.Reject {
-		httpx.WriteRejection(w, respAction)
-		return
-	}
+		// Streaming-aware plugins use a single code path for both shapes:
+		// for the buffered application/json case we deliver the whole body
+		// as one last=true frame so plugins finalize their running state.
+		// Plugins that didn't migrate — i.e. don't implement
+		// StreamingResponder — are unaffected (RunResponseFrame skips them).
+		if s.OutboundPipeline.HasStreamingResponders() && resp.Body != nil {
+			respFrameAction := s.OutboundPipeline.RunResponseFrame(r.Context(), pctx, pctx.ResponseBody, true)
+			if respFrameAction.Type == pipeline.Reject {
+				httpx.WriteRejection(w, respFrameAction)
+				return
+			}
+		}
 
-	// Streaming-aware plugins use a single code path for both shapes:
-	// for the buffered application/json case we deliver the whole body
-	// as one last=true frame so plugins finalize their running state.
-	// Plugins that didn't migrate — i.e. don't implement
-	// StreamingResponder — are unaffected (RunResponseFrame skips them).
-	if s.OutboundPipeline.HasStreamingResponders() && resp.Body != nil {
-		respFrameAction := s.OutboundPipeline.RunResponseFrame(r.Context(), pctx, pctx.ResponseBody, true)
-		if respFrameAction.Type == pipeline.Reject {
-			httpx.WriteRejection(w, respFrameAction)
-			return
+		// A plugin that called pctx.SetResponseBody flipped the mutation flag.
+		// Use the replaced bytes and rewrite Content-Length so the downstream
+		// client gets a consistent response. Content-Encoding is cleared
+		// because the framework can't know if the plugin also decompressed;
+		// safer to ship plain bytes than a broken archive.
+		if pctx.ResponseBodyMutated() {
+			resp.Body = io.NopCloser(bytes.NewReader(pctx.ResponseBody))
+			resp.ContentLength = int64(len(pctx.ResponseBody))
+			resp.Header.Set("Content-Length", fmt.Sprintf("%d", len(pctx.ResponseBody)))
+			resp.Header.Del("Content-Encoding")
 		}
-	}
 
-	// A plugin that called pctx.SetResponseBody flipped the mutation flag.
-	// Use the replaced bytes and rewrite Content-Length so the downstream
-	// client gets a consistent response. Content-Encoding is cleared
-	// because the framework can't know if the plugin also decompressed;
-	// safer to ship plain bytes than a broken archive.
-	if pctx.ResponseBodyMutated() {
-		resp.Body = io.NopCloser(bytes.NewReader(pctx.ResponseBody))
-		resp.ContentLength = int64(len(pctx.ResponseBody))
-		resp.Header.Set("Content-Length", fmt.Sprintf("%d", len(pctx.ResponseBody)))
-		resp.Header.Del("Content-Encoding")
+		s.recordOutboundResponseEvent(pctx, resp.StatusCode)
 	}
 
-	s.recordOutboundResponseEvent(pctx, resp.StatusCode)
-
 	for key, values := range resp.Header {
 		for _, value := range values {
 			w.Header().Add(key, value)
@@ -670,24 +709,48 @@ func (s *Server) handleConnect(w http.ResponseWriter, r *http.Request) {
 		Headers:   r.Header.Clone(),
 		StartedAt: time.Now(),
 	}
-	defer func() {
-		s.OutboundPipeline.RunFinish(r.Context(), pctx, pipeline.OutcomeFromContext(pctx))
-	}()
 
-	if s.Sessions != nil {
-		if aid := s.Sessions.ActiveSession(); aid != "" {
-			pctx.Session = s.Sessions.View(aid)
+	// SkipHosts short-circuit: open the tunnel without running the
+	// pipeline or recording a session event. The pipeline never ran,
+	// so there's nothing to RunFinish — defer is suppressed. Mirrors
+	// handleRequest's skip path so HTTP and CONNECT-tunnel destinations
+	// that match a skip pattern behave identically. Note the gate
+	// plugin loss this implies: if your skip-host list includes a
+	// destination you'd want IBAC or token-exchange to deny on, that
+	// denial does not happen — the SkipHosts list is a "trusted
+	// infrastructure" surface, not a generic per-route policy knob.
+	//
+	// CONNECT is safer-by-construction than the HTTP path: r.Host on
+	// CONNECT is the dial target, so a forged Host header cannot
+	// skip-match while dialing elsewhere — the proxy dials the same
+	// "host:port" it matched. We still emit an audit log so a
+	// successful skip leaves a trace.
+	pat, skipped := s.SkipHosts.MatchPattern(pctx.Host)
+	if skipped {
+		slog.Info("forward-proxy: skip_hosts match (CONNECT) — opening tunnel without pipeline + recording",
+			"host", pctx.Host, "pattern", pat)
+	}
+
+	if !skipped {
+		defer func() {
+			s.OutboundPipeline.RunFinish(r.Context(), pctx, pipeline.OutcomeFromContext(pctx))
+		}()
+
+		if s.Sessions != nil {
+			if aid := s.Sessions.ActiveSession(); aid != "" {
+				pctx.Session = s.Sessions.View(aid)
+			}
 		}
-	}
 
-	// Run the outbound pipeline. Plugins that policy on host/identity
-	// (ibac, content gates) still get to allow/deny; plugins that need
-	// HTTP body (parsers) see no body, which they handle gracefully.
-	action := s.OutboundPipeline.Run(r.Context(), pctx)
-	if action.Type == pipeline.Reject {
-		s.recordOutboundReject(pctx, action)
-		httpx.WriteRejection(w, action)
-		return
+		// Run the outbound pipeline. Plugins that policy on host/identity
+		// (ibac, content gates) still get to allow/deny; plugins that need
+		// HTTP body (parsers) see no body, which they handle gracefully.
+		action := s.OutboundPipeline.Run(r.Context(), pctx)
+		if action.Type == pipeline.Reject {
+			s.recordOutboundReject(pctx, action)
+			httpx.WriteRejection(w, action)
+			return
+		}
 	}
 
 	// Verify hijack capability BEFORE dialing upstream. If hijacking
@@ -739,7 +802,11 @@ func (s *Server) handleConnect(w http.ResponseWriter, r *http.Request) {
 	// Record a SessionRequest event so /v1/sessions and abctl show that
 	// a tunnel was opened. Mirrors the HTTP path's post-Allow recording
 	// (see handleRequest above). Shared with the transparent-redirect path.
-	s.recordTunnelOpened(pctx)
+	// Skipped when the destination matched SkipHosts: no plugin ran, so
+	// there are no Invocations to attribute the event to.
+	if !skipped {
+		s.recordTunnelOpened(pctx)
+	}
 
 	// Bidirectional copy until either side closes.
 	tunnel(clientConn, upstream)
diff --git a/authbridge/authlib/listener/forwardproxy/skiphost_test.go b/authbridge/authlib/listener/forwardproxy/skiphost_test.go
new file mode 100644
index 000000000..b110ee6ee
--- /dev/null
+++ b/authbridge/authlib/listener/forwardproxy/skiphost_test.go
@@ -0,0 +1,366 @@
+package forwardproxy
+
+import (
+	"bufio"
+	"context"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"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. Tests use it to
+// assert whether the pipeline ran on a given request — if SkipHosts
+// short-circuits correctly, calls counts AND session events stay at zero.
+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 newMarkerServer(t *testing.T, store *session.Store, skip *skiphost.Matcher) (*httptest.Server, *markerPlugin) {
+	t.Helper()
+	mp := &markerPlugin{}
+	pp, err := plugintesting.BuildPipeline([]pipeline.Plugin{mp})
+	if err != nil {
+		t.Fatalf("build pipeline: %v", err)
+	}
+	srv := &Server{
+		OutboundPipeline: pipeline.NewHolder(pp),
+		Sessions:         store,
+		Client:           http.DefaultClient,
+		SkipHosts:        skip,
+	}
+	return httptest.NewServer(srv.Handler()), mp
+}
+
+// TestForwardProxy_SkipHosts_BypassesPipeline asserts the core property:
+// a host matching SkipHosts produces NO plugin invocations and NO session
+// events, while a non-matching host on the same proxy still runs both.
+// This is the only behavioral guarantee that prevents OTel-style chatty
+// infrastructure traffic from evicting the inbound A2A user intent out
+// of the session buffer's FIFO eviction window.
+func TestForwardProxy_SkipHosts_BypassesPipeline(t *testing.T) {
+	upstreamHits := atomic.Int32{}
+	backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		upstreamHits.Add(1)
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte("upstream-ok"))
+	}))
+	defer backend.Close()
+
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+
+	// Match the loopback IP that httptest backends listen on. A pattern
+	// like "127.0.0.1" works because skiphost strips the port before
+	// matching. We deliberately do NOT use a glob so the test is brittle
+	// to behavior, not to glob semantics (those are exercised in the
+	// skiphost package's own tests).
+	skip, err := skiphost.New([]string{"127.0.0.1"})
+	if err != nil {
+		t.Fatalf("skiphost.New: %v", err)
+	}
+
+	proxy, mp := newMarkerServer(t, store, skip)
+	defer proxy.Close()
+
+	proxyClient := &http.Client{
+		Transport: &http.Transport{
+			Proxy: http.ProxyURL(mustParseURL(proxy.URL)),
+		},
+	}
+
+	// Skip path: backend URL is on 127.0.0.1, so it should match the
+	// skip pattern. Pipeline must not run; session must remain empty.
+	resp, err := proxyClient.Get(backend.URL + "/skip-me")
+	if err != nil {
+		t.Fatalf("skip request failed: %v", err)
+	}
+	body, _ := io.ReadAll(resp.Body)
+	resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		t.Errorf("skip path: status = %d, want 200", resp.StatusCode)
+	}
+	if string(body) != "upstream-ok" {
+		t.Errorf("skip path: body = %q, want upstream-ok (transparent forward must still deliver upstream bytes)", string(body))
+	}
+	if upstreamHits.Load() != 1 {
+		t.Errorf("skip path: upstream hit count = %d, want 1 (skip must not block the request)", upstreamHits.Load())
+	}
+	if mp.calls.Load() != 0 {
+		t.Errorf("skip path: pipeline plugin ran %d times, want 0 (SkipHosts must short-circuit before pipeline.Run)", mp.calls.Load())
+	}
+	if sessions := store.ListSessions(); len(sessions) != 0 {
+		t.Errorf("skip path: %d session(s) recorded, want 0 (SkipHosts must skip recording, otherwise OTel-style traffic still evicts the A2A intent)", len(sessions))
+	}
+}
+
+// TestForwardProxy_SkipHosts_NonMatchingRunsPipeline is the regression
+// guard: a Server with a SkipHosts list set must still run the pipeline
+// + record events for hosts that DO NOT match the list. Without this
+// pairing the skip test above could pass trivially with a globally
+// disabled pipeline.
+func TestForwardProxy_SkipHosts_NonMatchingRunsPipeline(t *testing.T) {
+	backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer backend.Close()
+
+	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)
+	}
+
+	proxy, mp := newMarkerServer(t, store, skip)
+	defer proxy.Close()
+
+	proxyClient := &http.Client{
+		Transport: &http.Transport{
+			Proxy: http.ProxyURL(mustParseURL(proxy.URL)),
+		},
+	}
+	resp, err := proxyClient.Get(backend.URL + "/run-me")
+	if err != nil {
+		t.Fatalf("non-skip request failed: %v", err)
+	}
+	resp.Body.Close()
+
+	if mp.calls.Load() != 1 {
+		t.Errorf("non-skip path: pipeline plugin ran %d times, want 1 (host did not match skip list)", mp.calls.Load())
+	}
+	// The marker plugin recorded an Invocation, so a session event
+	// should land. Bucket is DefaultSessionID since no inbound primed
+	// an active session for this proxy instance.
+	if sessions := store.ListSessions(); len(sessions) != 1 {
+		t.Errorf("non-skip path: session count = %d, want 1 (Invocation should drive event recording)", len(sessions))
+	}
+}
+
+// TestForwardProxy_SkipHosts_NilMatcherPreservesBehavior asserts the
+// zero-value default: a Server constructed without SkipHosts (nil
+// Matcher) behaves like today's code — pipeline runs, sessions record.
+// This is the upgrade-safety contract for existing deployments that
+// don't opt into skip_hosts.
+func TestForwardProxy_SkipHosts_NilMatcherPreservesBehavior(t *testing.T) {
+	backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer backend.Close()
+
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+
+	proxy, mp := newMarkerServer(t, store, nil) // SkipHosts: nil
+	defer proxy.Close()
+
+	proxyClient := &http.Client{
+		Transport: &http.Transport{
+			Proxy: http.ProxyURL(mustParseURL(proxy.URL)),
+		},
+	}
+	resp, err := proxyClient.Get(backend.URL + "/")
+	if err != nil {
+		t.Fatalf("request failed: %v", err)
+	}
+	resp.Body.Close()
+
+	if mp.calls.Load() != 1 {
+		t.Errorf("nil SkipHosts: pipeline ran %d times, want 1", mp.calls.Load())
+	}
+}
+
+// TestForwardProxy_SkipHosts_MatchesAgainstHostHeaderNotURL locks the
+// trust-model contract for the HTTP forward path: the skip match keys
+// on the request `Host` header (`r.Host`), not on the dial target
+// derived from `r.URL`. This is what huang195's review called out as
+// the agent-influenceable input — the test exists to make the behavior
+// observable so future maintainers see it when reading the test, and
+// so a refactor that flips this boundary (e.g. switching to r.URL.Host)
+// breaks loudly with a named test.
+//
+// Operators reading this: do NOT add a destination to skip_hosts that
+// you'd want IBAC / token-exchange to deny on. The skip is keyed on
+// agent-influenceable headers; the audit log emitted on each skip is
+// the only after-the-fact signal.
+//
+// The test scopes itself to the trust contract — "skip fires when
+// r.Host matches" — and does NOT assert what happens to the forwarded
+// request afterward. Go's http.Transport behavior with a proxy and a
+// forged Host header is implementation-detail that the trust model
+// shouldn't depend on; what matters is that the skip decision was
+// made on the agent-supplied value.
+func TestForwardProxy_SkipHosts_MatchesAgainstHostHeaderNotURL(t *testing.T) {
+	backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer backend.Close()
+
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+
+	// Skip pattern matches a hostname that is NOT the dial target.
+	// The dial target is the backend (127.0.0.1:N from backend.URL);
+	// the pattern is a fictitious hostname.
+	skip, err := skiphost.New([]string{"infrastructure-only.example"})
+	if err != nil {
+		t.Fatalf("skiphost.New: %v", err)
+	}
+
+	proxy, mp := newMarkerServer(t, store, skip)
+	defer proxy.Close()
+
+	// Send a request whose Host header is the skip-listed pattern
+	// while r.URL points at the real backend. A normal Go http.Client
+	// would set Host from URL; we override it explicitly to model the
+	// adversarial case.
+	proxyClient := &http.Client{
+		Transport: &http.Transport{
+			Proxy: http.ProxyURL(mustParseURL(proxy.URL)),
+		},
+	}
+	req, err := http.NewRequest("GET", backend.URL+"/test", nil)
+	if err != nil {
+		t.Fatalf("NewRequest: %v", err)
+	}
+	req.Host = "infrastructure-only.example" // forged Host header
+
+	// Don't assert on resp/err — the trust-model claim is solely
+	// about the skip decision. Whatever Go's transport ends up doing
+	// with the divergent Host vs URL is incidental.
+	if resp, err := proxyClient.Do(req); err == nil && resp != nil {
+		resp.Body.Close()
+	}
+
+	// The skip matched against the forged Host, so the pipeline
+	// did NOT run. That's the trust-contract assertion: the agent
+	// chose the skip-match value via the Host header, not via the
+	// real dial target.
+	if mp.calls.Load() != 0 {
+		t.Errorf("forged Host: pipeline ran %d times, want 0 — skip matched on r.Host (the agent-supplied Host header), not r.URL.Host", mp.calls.Load())
+	}
+	if sessions := store.ListSessions(); len(sessions) != 0 {
+		t.Errorf("forged Host: %d session(s) recorded, want 0 — skip path bypasses recording", len(sessions))
+	}
+}
+
+// TestForwardProxy_SkipHosts_CONNECT_BypassesPipeline asserts that the
+// CONNECT-tunnel path honors SkipHosts. CONNECT is safer-by-construction
+// than the HTTP path because r.Host IS the dial target, but skip
+// behavior must still be exercised so a regression that disables the
+// CONNECT skip path is caught.
+func TestForwardProxy_SkipHosts_CONNECT_BypassesPipeline(t *testing.T) {
+	// Minimal echo server — stand-in for an HTTPS upstream we don't
+	// want to do TLS to in tests. The skip path opens a tunnel and
+	// shuttles bytes; we just need to prove bytes flow.
+	upstream, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen upstream: %v", err)
+	}
+	defer upstream.Close()
+	go func() {
+		conn, err := upstream.Accept()
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+		buf := make([]byte, 64)
+		n, _ := conn.Read(buf)
+		_, _ = conn.Write(append([]byte("echo:"), buf[:n]...))
+	}()
+
+	upstreamAddr := upstream.Addr().String()
+	upstreamHost, _, _ := net.SplitHostPort(upstreamAddr)
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+
+	skip, err := skiphost.New([]string{upstreamHost})
+	if err != nil {
+		t.Fatalf("skiphost.New: %v", err)
+	}
+
+	proxy, mp := newMarkerServer(t, store, skip)
+	defer proxy.Close()
+
+	// Drive CONNECT manually so we can observe the 200 + tunnel.
+	proxyAddr := strings.TrimPrefix(proxy.URL, "http://")
+	tunnel, err := net.DialTimeout("tcp", proxyAddr, 5*time.Second)
+	if err != nil {
+		t.Fatalf("dial proxy: %v", err)
+	}
+	defer tunnel.Close()
+	if _, err := tunnel.Write([]byte("CONNECT " + upstreamAddr + " HTTP/1.1\r\nHost: " + upstreamAddr + "\r\n\r\n")); err != nil {
+		t.Fatalf("write CONNECT: %v", err)
+	}
+
+	br := bufio.NewReader(tunnel)
+	line, err := br.ReadString('\n')
+	if err != nil {
+		t.Fatalf("read CONNECT response: %v", err)
+	}
+	if !strings.Contains(line, "200") {
+		t.Fatalf("CONNECT response = %q, want 200 (skip path must still establish the tunnel)", line)
+	}
+	// Drain headers.
+	for {
+		hdr, err := br.ReadString('\n')
+		if err != nil {
+			t.Fatalf("read headers: %v", err)
+		}
+		if hdr == "\r\n" || hdr == "\n" {
+			break
+		}
+	}
+
+	// Tunnel up — write some bytes and confirm the echo round-trips.
+	if _, err := tunnel.Write([]byte("hello")); err != nil {
+		t.Fatalf("write through tunnel: %v", err)
+	}
+	got := make([]byte, 32)
+	_ = tunnel.SetReadDeadline(time.Now().Add(2 * time.Second))
+	n, err := br.Read(got)
+	if err != nil && err != io.EOF {
+		t.Fatalf("read tunnel: %v", err)
+	}
+	if string(got[:n]) != "echo:hello" {
+		t.Errorf("tunnel echo = %q, want echo:hello", got[:n])
+	}
+
+	// The skipped CONNECT must NOT have run the outbound pipeline
+	// and must NOT have appended a session event.
+	if mp.calls.Load() != 0 {
+		t.Errorf("skipped CONNECT: pipeline ran %d times, want 0", mp.calls.Load())
+	}
+	if sessions := store.ListSessions(); len(sessions) != 0 {
+		t.Errorf("skipped CONNECT: %d session(s) recorded, want 0", len(sessions))
+	}
+}
+
diff --git a/authbridge/authlib/listener/forwardproxy/transparent.go b/authbridge/authlib/listener/forwardproxy/transparent.go
index 5a43d0d93..d2cb1e6ef 100644
--- a/authbridge/authlib/listener/forwardproxy/transparent.go
+++ b/authbridge/authlib/listener/forwardproxy/transparent.go
@@ -42,6 +42,17 @@ import (
 // hard control against a hostile one — only the IP is ground truth. Hard
 // enforcement would need IP-set allowlists or SNI/cert cross-checks.
 //
+// SkipHosts is intentionally NOT consulted here. listener.skip_hosts is an
+// ops convenience for the cooperative-egress paths (forward proxy + ext_proc),
+// where bypassing the pipeline on infrastructure traffic is fine because the
+// agent is trusted to honor HTTP_PROXY anyway. The transparent path exists
+// precisely as the hard egress guard against agents that route around the
+// cooperative paths, and pctx.Host here is recovered from agent-controlled
+// SNI/Host bytes on the wire — making it self-exemptable would defeat the
+// reason this listener exists. If you find yourself wanting to add a
+// SkipHosts check here to "match the other listeners," don't — that's the
+// failure mode this comment is explicitly trying to prevent.
+//
 // HandleTransparentConn owns clientConn's lifecycle and always closes it.
 func (s *Server) HandleTransparentConn(clientConn net.Conn, dst string) {
 	defer func() { _ = clientConn.Close() }()
diff --git a/authbridge/authlib/listener/skiphost/skiphost.go b/authbridge/authlib/listener/skiphost/skiphost.go
new file mode 100644
index 000000000..49d463b1b
--- /dev/null
+++ b/authbridge/authlib/listener/skiphost/skiphost.go
@@ -0,0 +1,130 @@
+// Package skiphost matches an outbound destination Host against an
+// operator-configured pattern list to decide whether the listener should
+// short-circuit — bypassing the plugin pipeline and session recording —
+// and forward the request as a transparent proxy.
+//
+// Matcher semantics intentionally mirror authlib/routing: gobwas/glob with
+// `.` as the separator so "*.svc.cluster.local" matches a single label and
+// "service-*" matches anything starting with "service-". Port is stripped
+// before matching so operators write patterns against the hostname alone
+// regardless of which port the upstream listens on.
+//
+// The package is a leaf — no dependencies inside authlib — so both
+// listener implementations (extproc, forwardproxy) can import it without
+// risking an import cycle, and tests can exercise the matcher in
+// isolation from the listener machinery.
+package skiphost
+
+import (
+	"fmt"
+	"net"
+	"strings"
+
+	"github.com/gobwas/glob"
+)
+
+// Matcher answers "does this host match any configured skip pattern?".
+// A nil Matcher matches nothing (zero value is safe to call).
+type Matcher struct {
+	patterns []compiled
+}
+
+type compiled struct {
+	raw  string
+	glob glob.Glob
+}
+
+// New compiles a skip-host matcher from raw glob patterns. Returns an
+// error identifying the first invalid pattern so misconfigurations
+// surface at startup rather than at first request. An empty input is
+// valid and yields a Matcher that matches nothing.
+//
+// Construction-time guards reject footgun patterns that would silently
+// disable the entire outbound enforcement pipeline (skip_hosts bypasses
+// plugins AND session recording for matched traffic):
+//
+//   - empty / whitespace-only patterns: trivially-true matches with no
+//     intent expressed.
+//   - "*" — under our `.`-delimited glob semantics, matches every
+//     single-label hostname, which is how every short Kubernetes
+//     service name reaches the listener (`Host: github-tool-mcp`,
+//     `Host: otel-collector`, etc.). One wildcard would silently
+//     exempt every in-cluster outbound from IBAC + token-exchange.
+//   - "**" — the unambiguous match-all under gobwas/glob.
+//   - patterns containing ":" — Match strips the port from the
+//     incoming host before comparing, so colon-bearing patterns
+//     compile but never match. Almost certainly an operator typo.
+//
+// Mirrors the bypass-pattern guard added to ibac in #496. Operators
+// that mean to disable enforcement should remove the relevant plugin
+// from the pipeline rather than wildcarding it away here.
+//
+// Patterns like "*.*", "*.svc.cluster.local", or "service-*" are NOT
+// match-all under `.`-delimited glob (they require a fixed label
+// count or fixed suffix) and are accepted normally.
+func New(patterns []string) (*Matcher, error) {
+	if len(patterns) == 0 {
+		return &Matcher{}, nil
+	}
+	out := make([]compiled, 0, len(patterns))
+	for _, p := range patterns {
+		trimmed := strings.TrimSpace(p)
+		if trimmed == "" {
+			return nil, fmt.Errorf("skiphost: empty pattern in skip_hosts list")
+		}
+		if trimmed == "*" || trimmed == "**" {
+			return nil, fmt.Errorf("skiphost: pattern %q matches everything; "+
+				"if you mean to disable outbound enforcement, remove the "+
+				"relevant plugins from the pipeline instead", p)
+		}
+		if strings.Contains(p, ":") {
+			return nil, fmt.Errorf("skiphost: pattern %q must not contain a port "+
+				"(Match strips the port from the incoming host before comparing, "+
+				"so a port-bearing pattern would never match)", p)
+		}
+		g, err := glob.Compile(p, '.')
+		if err != nil {
+			return nil, fmt.Errorf("skiphost: invalid pattern %q: %w", p, err)
+		}
+		out = append(out, compiled{raw: p, glob: g})
+	}
+	return &Matcher{patterns: out}, nil
+}
+
+// Match reports whether host matches any configured pattern. Strips
+// the port (everything from the first colon) before comparing so
+// operators can write "otel-collector.kagenti-system.svc.cluster.local"
+// without worrying which port the upstream listens on. Returns false
+// for the nil Matcher and for the empty host.
+//
+// Wraps MatchPattern so callers that only want the boolean don't pay
+// for the unused string allocation in their hot path.
+func (m *Matcher) Match(host string) bool {
+	_, matched := m.MatchPattern(host)
+	return matched
+}
+
+// MatchPattern reports whether host matches any configured pattern,
+// and if it does, returns the raw operator-supplied pattern that
+// matched. The matched pattern is used by listeners to attribute
+// audit logs / counters — "this host was skipped because pattern P
+// matched" — so an operator reviewing logs can correlate skips back
+// to entries in their skip_hosts list. First-match-wins; iteration
+// order is the order patterns were configured.
+//
+// Returns ("", false) for the nil Matcher, the empty pattern list,
+// the empty host, and unmatched hosts.
+func (m *Matcher) MatchPattern(host string) (string, bool) {
+	if m == nil || len(m.patterns) == 0 || host == "" {
+		return "", false
+	}
+	if h, _, err := net.SplitHostPort(host); err == nil {
+		host = h
+	}
+	for _, p := range m.patterns {
+		if p.glob.Match(host) {
+			return p.raw, true
+		}
+	}
+	return "", false
+}
diff --git a/authbridge/authlib/listener/skiphost/skiphost_test.go b/authbridge/authlib/listener/skiphost/skiphost_test.go
new file mode 100644
index 000000000..1948a0af0
--- /dev/null
+++ b/authbridge/authlib/listener/skiphost/skiphost_test.go
@@ -0,0 +1,136 @@
+package skiphost
+
+import "testing"
+
+func TestNew_EmptyList(t *testing.T) {
+	m, err := New(nil)
+	if err != nil {
+		t.Fatalf("New(nil) err = %v", err)
+	}
+	if m.Match("any-host") {
+		t.Error("empty matcher matched a host; should match nothing")
+	}
+}
+
+func TestNew_RejectsEmptyPattern(t *testing.T) {
+	if _, err := New([]string{""}); err == nil {
+		t.Error("New([\"\"]) returned nil error; empty pattern must be rejected at boot")
+	}
+}
+
+func TestNew_RejectsInvalidPattern(t *testing.T) {
+	if _, err := New([]string{"["}); err == nil {
+		t.Error("New([\"[\"]) returned nil error; malformed glob must surface at boot")
+	}
+}
+
+func TestMatch_NilMatcher(t *testing.T) {
+	var m *Matcher
+	if m.Match("host") {
+		t.Error("nil matcher matched; zero value must be safe and match nothing")
+	}
+}
+
+func TestMatch_EmptyHost(t *testing.T) {
+	// "*-anything" matches every single-label host that ends in
+	// "-anything" — broad enough that the empty-host defense is the
+	// only thing keeping Match from returning true on an unset Host.
+	// We can't use bare "*" anymore (rejected by New as match-all).
+	m, _ := New([]string{"some-host"})
+	if m.Match("") {
+		t.Error("matcher matched empty host; empty host must never match (defensive against unset Host header)")
+	}
+}
+
+func TestNew_RejectsMatchAllStar(t *testing.T) {
+	if _, err := New([]string{"*"}); err == nil {
+		t.Error("New([\"*\"]) must reject match-all (every single-label hostname would skip enforcement)")
+	}
+}
+
+func TestNew_RejectsMatchAllDoubleStar(t *testing.T) {
+	if _, err := New([]string{"**"}); err == nil {
+		t.Error("New([\"**\"]) must reject the unambiguous match-all glob")
+	}
+}
+
+func TestNew_RejectsWhitespaceOnly(t *testing.T) {
+	if _, err := New([]string{"   "}); err == nil {
+		t.Error("New([\"   \"]) must reject whitespace-only patterns (trivial-true)")
+	}
+}
+
+func TestNew_RejectsPortInPattern(t *testing.T) {
+	if _, err := New([]string{"otel-collector:8335"}); err == nil {
+		t.Error("New([\"otel-collector:8335\"]) must reject port-bearing patterns; Match strips ports before comparing so they would never match")
+	}
+}
+
+func TestNew_AcceptsLeadingStar(t *testing.T) {
+	// "*.something" is NOT match-all under .-delimited glob — it
+	// requires the fixed suffix. Explicit guard against a future
+	// rewrite that over-rejects and breaks operator-typical FQDN
+	// patterns.
+	cases := []string{
+		"*.svc.cluster.local",
+		"*.metrics.local",
+		"otel-collector.*.svc.cluster.local",
+		"otel-collector*",
+	}
+	for _, p := range cases {
+		if _, err := New([]string{p}); err != nil {
+			t.Errorf("New([%q]) returned err = %v; non-match-all pattern must be accepted", p, err)
+		}
+	}
+}
+
+func TestMatch_StripsPort(t *testing.T) {
+	m, _ := New([]string{"otel-collector.kagenti-system.svc.cluster.local"})
+	if !m.Match("otel-collector.kagenti-system.svc.cluster.local:8335") {
+		t.Error("port-stripping failed: pattern without :port should match host with :port")
+	}
+}
+
+func TestMatch_GlobSingleLabel(t *testing.T) {
+	// `*` with `.` separator matches a single DNS label, not multi-label.
+	m, _ := New([]string{"otel-collector*"})
+	cases := []struct {
+		host string
+		want bool
+	}{
+		{"otel-collector", true},
+		{"otel-collector-v2", true},
+		{"otel-collector.kagenti-system.svc.cluster.local", false}, // separator stops at .
+		{"foo-otel-collector", false},
+	}
+	for _, tc := range cases {
+		if got := m.Match(tc.host); got != tc.want {
+			t.Errorf("Match(%q) = %v, want %v", tc.host, got, tc.want)
+		}
+	}
+}
+
+func TestMatch_GlobLeadingWildcard(t *testing.T) {
+	// `*.svc.cluster.local` → single-label prefix on a fixed suffix.
+	m, _ := New([]string{"*.kagenti-system.svc.cluster.local"})
+	if !m.Match("otel-collector.kagenti-system.svc.cluster.local") {
+		t.Error("leading-* should match a single-label prefix on the FQDN")
+	}
+	if m.Match("a.b.kagenti-system.svc.cluster.local") {
+		t.Error("leading-* must NOT match a two-label prefix (separator semantics)")
+	}
+}
+
+func TestMatch_MultiplePatterns_FirstMatchWins(t *testing.T) {
+	m, _ := New([]string{"never-matches", "otel-*", "another"})
+	if !m.Match("otel-collector") {
+		t.Error("matcher with multiple patterns should match if any pattern matches")
+	}
+}
+
+func TestMatch_NoMatch(t *testing.T) {
+	m, _ := New([]string{"otel-collector*", "*.metrics.local"})
+	if m.Match("github-tool-mcp") {
+		t.Error("Match returned true for an unrelated host")
+	}
+}
diff --git a/authbridge/authlib/reloader/reloader.go b/authbridge/authlib/reloader/reloader.go
index 88befdc15..a159c5bf0 100644
--- a/authbridge/authlib/reloader/reloader.go
+++ b/authbridge/authlib/reloader/reloader.go
@@ -28,6 +28,7 @@ import (
 	"log/slog"
 	"os"
 	"path/filepath"
+	"reflect"
 	"sync/atomic"
 	"time"
 
@@ -308,7 +309,13 @@ func validateReloadable(active, next *config.Config) error {
 	if active.Mode != next.Mode {
 		diffs = append(diffs, fmt.Sprintf("mode (%s→%s)", active.Mode, next.Mode))
 	}
-	if active.Listener != next.Listener {
+	// reflect.DeepEqual rather than `!=` because ListenerConfig now
+	// contains a SkipHosts []string field, and slices are not
+	// comparable with `!=`. The semantic stays the same: any field
+	// difference under listener.* fails reload and asks for a restart,
+	// because these addresses (and skip-host patterns) are bound at
+	// listener construction.
+	if !reflect.DeepEqual(active.Listener, next.Listener) {
 		diffs = append(diffs, "listener.*")
 	}
 	if len(diffs) > 0 {
diff --git a/authbridge/authlib/session/intent_pin_test.go b/authbridge/authlib/session/intent_pin_test.go
new file mode 100644
index 000000000..4284e20cb
--- /dev/null
+++ b/authbridge/authlib/session/intent_pin_test.go
@@ -0,0 +1,205 @@
+package session
+
+import (
+	"testing"
+	"time"
+
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
+)
+
+// intent fabricates the same shape SessionView.LastIntent recognizes:
+// inbound A2A request event. Tests below assert that this exact shape
+// survives FIFO eviction so IBAC's LastIntent never returns nil after
+// chatty traffic floods the buffer.
+func intent(method string) pipeline.SessionEvent {
+	return pipeline.SessionEvent{
+		At:        time.Now(),
+		Direction: pipeline.Inbound,
+		Phase:     pipeline.SessionRequest,
+		A2A:       &pipeline.A2AExtension{Method: method},
+	}
+}
+
+// chatter fabricates an outbound non-intent event. Stand-in for the
+// OTel exports / inference calls / tool calls that flooded the bucket
+// in the GSM8K reproducer.
+func chatter(tag string) pipeline.SessionEvent {
+	return pipeline.SessionEvent{
+		At:        time.Now(),
+		Direction: pipeline.Outbound,
+		Phase:     pipeline.SessionRequest,
+		MCP:       &pipeline.MCPExtension{Method: tag},
+	}
+}
+
+// TestStore_PinsIntentAcrossEviction reproduces the GSM8K bug at the
+// store level: an inbound A2A intent followed by enough chatty
+// outbound traffic to overflow maxEvents. Pre-fix, the intent rolled
+// out the back of the FIFO and LastIntent returned nil. Post-fix the
+// intent must survive — that's the property IBAC depends on.
+func TestStore_PinsIntentAcrossEviction(t *testing.T) {
+	s := New(5*time.Minute, 5, 0)
+
+	s.Append("sess", intent("solve-math"))
+	for i := 0; i < 20; i++ {
+		s.Append("sess", chatter("chat"))
+	}
+
+	v := s.View("sess")
+	if v == nil {
+		t.Fatal("View returned nil")
+	}
+	if len(v.Events) != 5 {
+		t.Fatalf("Events len = %d, want 5 (capped at maxEvents)", len(v.Events))
+	}
+	last := v.LastIntent()
+	if last == nil {
+		t.Fatal("LastIntent() = nil; intent must survive eviction even when buried under chatter")
+	}
+	if last.A2A == nil || last.A2A.Method != "solve-math" {
+		t.Errorf("LastIntent A2A.Method = %v, want solve-math", last.A2A)
+	}
+}
+
+// TestStore_NoIntent_FifoEvictionUnchanged is the regression guard:
+// a bucket with no intent in it must keep behaving exactly like
+// today's FIFO. Pinning logic must not change non-intent buckets.
+func TestStore_NoIntent_FifoEvictionUnchanged(t *testing.T) {
+	s := New(5*time.Minute, 3, 0)
+
+	for i := 0; i < 5; i++ {
+		ev := chatter(string(rune('a' + i)))
+		s.Append("sess", ev)
+	}
+
+	v := s.View("sess")
+	if len(v.Events) != 3 {
+		t.Fatalf("Events len = %d, want 3", len(v.Events))
+	}
+	// Oldest two evicted: surviving methods are c, d, e.
+	want := []string{"c", "d", "e"}
+	for i, w := range want {
+		if v.Events[i].MCP.Method != w {
+			t.Errorf("Events[%d].MCP.Method = %q, want %q", i, v.Events[i].MCP.Method, w)
+		}
+	}
+}
+
+// TestStore_OnlyMostRecentIntentPinned: a bucket with multiple
+// intents (multi-turn conversation) must pin only the LATEST.
+// Older intents evict via normal FIFO so pathological traffic
+// (many turns + huge fan-out) can't starve the buffer with stale
+// intents.
+func TestStore_OnlyMostRecentIntentPinned(t *testing.T) {
+	s := New(5*time.Minute, 3, 0)
+
+	s.Append("sess", intent("turn-1"))
+	s.Append("sess", chatter("c1"))
+	s.Append("sess", intent("turn-2"))
+	for i := 0; i < 10; i++ {
+		s.Append("sess", chatter("flood"))
+	}
+
+	v := s.View("sess")
+	if len(v.Events) != 3 {
+		t.Fatalf("Events len = %d, want 3", len(v.Events))
+	}
+	last := v.LastIntent()
+	if last == nil || last.A2A == nil || last.A2A.Method != "turn-2" {
+		t.Errorf("LastIntent.A2A.Method = %v, want turn-2 (only the latest intent is pinned)", last)
+	}
+	// Older intent should be gone (evicted by FIFO with only the
+	// latest pinned).
+	for _, e := range v.Events {
+		if e.A2A != nil && e.A2A.Method == "turn-1" {
+			t.Errorf("turn-1 intent survived; only the most recent intent should be pinned")
+		}
+	}
+}
+
+// TestStore_IntentInKeepWindow_TrivialFifo: when the intent is
+// already in the keep window (i.e. plain FIFO would preserve it
+// anyway), the result is byte-identical to plain FIFO. Catches
+// regressions where the pin path mutates or reorders unnecessarily.
+func TestStore_IntentInKeepWindow_TrivialFifo(t *testing.T) {
+	s := New(5*time.Minute, 3, 0)
+
+	s.Append("sess", chatter("a"))
+	s.Append("sess", chatter("b"))
+	s.Append("sess", intent("recent-intent"))
+	s.Append("sess", chatter("c"))
+	s.Append("sess", chatter("d"))
+
+	v := s.View("sess")
+	if len(v.Events) != 3 {
+		t.Fatalf("Events len = %d, want 3", len(v.Events))
+	}
+	// Plain FIFO would preserve [intent, c, d] — and the intent is
+	// already in the keep window, so the pin path takes the fast
+	// branch and produces exactly that.
+	if v.Events[0].A2A == nil || v.Events[0].A2A.Method != "recent-intent" {
+		t.Errorf("Events[0] = %v, want recent-intent (FIFO order preserved)", v.Events[0])
+	}
+	if v.Events[1].MCP == nil || v.Events[1].MCP.Method != "c" {
+		t.Errorf("Events[1].MCP.Method = %v, want c", v.Events[1].MCP)
+	}
+	if v.Events[2].MCP == nil || v.Events[2].MCP.Method != "d" {
+		t.Errorf("Events[2].MCP.Method = %v, want d", v.Events[2].MCP)
+	}
+}
+
+// TestStore_PinnedIntentChronologicalOrder: with the intent pinned
+// from the eviction prefix, the surviving slice must still be
+// chronologically ordered (At ascending). Consumers like abctl
+// render events in slice order assuming monotonic time; a
+// non-monotonic result here would break the timeline UI.
+func TestStore_PinnedIntentChronologicalOrder(t *testing.T) {
+	s := New(5*time.Minute, 3, 0)
+
+	// Append intent first so it has the earliest timestamp, then
+	// flood with chatter. Pin path keeps intent at index 0 and the
+	// most-recent two chatters at indices 1, 2.
+	intentEv := intent("first")
+	intentEv.At = time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
+	s.Append("sess", intentEv)
+	for i := 0; i < 10; i++ {
+		ev := chatter("chat")
+		ev.At = time.Date(2026, 1, 1, 12, 0, i+1, 0, time.UTC)
+		s.Append("sess", ev)
+	}
+
+	v := s.View("sess")
+	if len(v.Events) != 3 {
+		t.Fatalf("Events len = %d, want 3", len(v.Events))
+	}
+	for i := 1; i < len(v.Events); i++ {
+		if !v.Events[i].At.After(v.Events[i-1].At) {
+			t.Errorf("Events[%d].At (%v) not after Events[%d].At (%v) — pin must preserve chronological order",
+				i, v.Events[i].At, i-1, v.Events[i-1].At)
+		}
+	}
+	if v.Events[0].A2A == nil || v.Events[0].A2A.Method != "first" {
+		t.Errorf("Events[0] should be the pinned intent, got %v", v.Events[0])
+	}
+}
+
+// TestStore_AllIntentsBucket: a pathological bucket made entirely
+// of intent events must still respect maxEvents. Only the most
+// recent intent is "protected" — the rest evict via plain FIFO.
+// This bounds the worst case (intents alone cannot starve the buffer).
+func TestStore_AllIntentsBucket(t *testing.T) {
+	s := New(5*time.Minute, 3, 0)
+
+	for i := 0; i < 10; i++ {
+		s.Append("sess", intent(string(rune('a'+i))))
+	}
+
+	v := s.View("sess")
+	if len(v.Events) != 3 {
+		t.Fatalf("Events len = %d, want 3 (maxEvents must hold even for all-intent buckets)", len(v.Events))
+	}
+	last := v.LastIntent()
+	if last == nil || last.A2A == nil || last.A2A.Method != "j" {
+		t.Errorf("LastIntent.Method = %v, want j (most recent intent)", last)
+	}
+}
diff --git a/authbridge/authlib/session/store.go b/authbridge/authlib/session/store.go
index d81d610fd..89288adca 100644
--- a/authbridge/authlib/session/store.go
+++ b/authbridge/authlib/session/store.go
@@ -193,10 +193,7 @@ func (s *Store) Append(sessionID string, event pipeline.SessionEvent) {
 	s.publishLocked(event)
 
 	if s.maxEvents > 0 && len(sess.Events) > s.maxEvents {
-		excess := len(sess.Events) - s.maxEvents
-		trimmed := make([]pipeline.SessionEvent, s.maxEvents)
-		copy(trimmed, sess.Events[excess:])
-		sess.Events = trimmed
+		sess.Events = trimEventsPinIntent(sess.Events, s.maxEvents)
 	}
 
 	// Evict oldest session if cap is exceeded.
@@ -205,6 +202,77 @@ func (s *Store) Append(sessionID string, event pipeline.SessionEvent) {
 	}
 }
 
+// isIntentEvent matches the SessionView.LastIntent predicate: an
+// inbound A2A request event. IBAC and any future intent-aware
+// guardrail call LastIntent and need the most recent such event to
+// survive FIFO eviction. Defined here so the eviction policy and
+// the consumer view can never drift apart.
+func isIntentEvent(e pipeline.SessionEvent) bool {
+	return e.Direction == pipeline.Inbound &&
+		e.Phase == pipeline.SessionRequest &&
+		e.A2A != nil
+}
+
+// trimEventsPinIntent reduces events to len <= maxEvents while
+// preserving the most-recent intent event, even when that intent
+// sits in the prefix that FIFO would normally evict. All other
+// events evict in chronological order; the protected intent stays
+// at its original index, leaving a temporal gap between it and the
+// first non-evicted event after it. That gap is visible in
+// /v1/sessions and abctl as a discontinuity in the timeline, which
+// is the right shape: the intent's append time is preserved
+// (consumers can correlate against the inbound request's wall-clock
+// timestamp) and chronological order across surviving events stays
+// monotonic.
+//
+// Why pin only the MOST-RECENT intent: a session can carry several
+// user turns ("solve this", then "now do that") and only the latest
+// is what IBAC aligns subsequent tool calls against. Pinning all
+// intents would let stale ones pile up under pathological loads
+// (slow turns + huge fan-out) and starve the buffer.
+//
+// Pathological case — the buffer is mostly intents, exceeds
+// maxEvents, and only the latest is pinned: older intents evict via
+// normal FIFO. There is no scenario where this returns more than
+// maxEvents events.
+//
+// Caller guarantees len(events) > maxEvents and maxEvents > 0;
+// otherwise this is a no-op shape (returns events unchanged).
+func trimEventsPinIntent(events []pipeline.SessionEvent, maxEvents int) []pipeline.SessionEvent {
+	if maxEvents <= 0 || len(events) <= maxEvents {
+		return events
+	}
+	excess := len(events) - maxEvents
+
+	// Locate the most-recent intent. If it sits in the keep window
+	// (index >= excess) it survives a plain FIFO trim — fast path.
+	intentIdx := -1
+	for i := len(events) - 1; i >= 0; i-- {
+		if isIntentEvent(events[i]) {
+			intentIdx = i
+			break
+		}
+	}
+	if intentIdx == -1 || intentIdx >= excess {
+		// No protected event in the eviction prefix; FIFO trim.
+		trimmed := make([]pipeline.SessionEvent, maxEvents)
+		copy(trimmed, events[excess:])
+		return trimmed
+	}
+
+	// Protected intent is in the eviction prefix. Build the result
+	// as: [pinned intent] ++ [last maxEvents-1 non-intent-prefix
+	// events]. We keep the intent at the front so its original
+	// timestamp is preserved AND the resulting slice stays
+	// chronologically ordered (intent.At < kept-tail.At by construction —
+	// it's the oldest surviving event).
+	out := make([]pipeline.SessionEvent, 0, maxEvents)
+	out = append(out, events[intentIdx])
+	tailStart := len(events) - (maxEvents - 1)
+	out = append(out, events[tailStart:]...)
+	return out
+}
+
 // View returns a read-only snapshot of the session's events.
 // Returns nil if the session doesn't exist or is expired.
 func (s *Store) View(sessionID string) *pipeline.SessionView {
diff --git a/authbridge/cmd/authbridge-envoy/main.go b/authbridge/cmd/authbridge-envoy/main.go
index a35efb088..d4edfa658 100644
--- a/authbridge/cmd/authbridge-envoy/main.go
+++ b/authbridge/cmd/authbridge-envoy/main.go
@@ -45,6 +45,7 @@ import (
 	// Only the ext_proc listener is compiled in (no ext_authz, no
 	// HTTP proxies).
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/listener/extproc"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/listener/skiphost"
 
 	// Plugins. Auth gates first, then the protocol parsers that
 	// supply session-event context for abctl.
@@ -218,8 +219,17 @@ func main() {
 	store := shared.New()
 	defer store.Close() // stop the TTL janitor on normal main return
 
+	// SkipHosts: outbound destinations that bypass the pipeline AND
+	// session recording entirely. See ListenerConfig.SkipHosts for the
+	// motivating case (chatty observability sidecars evicting the
+	// inbound A2A user intent from the session FIFO).
+	skipHosts, err := skiphost.New(cfg.Listener.SkipHosts)
+	if err != nil {
+		log.Fatalf("listener.skip_hosts: %v", err)
+	}
+
 	var grpcServers []*grpc.Server
-	grpcServers = append(grpcServers, startGRPCExtProc(inboundH, outboundH, sessions, store, cfg.Listener.ExtProcAddr))
+	grpcServers = append(grpcServers, startGRPCExtProc(inboundH, outboundH, sessions, store, skipHosts, cfg.Listener.ExtProcAddr))
 
 	statsProvider := func() *auth.Stats {
 		sources := plugins.CollectStats(inboundH.Load())
@@ -302,13 +312,14 @@ func main() {
 	}
 }
 
-func startGRPCExtProc(inbound, outbound *pipeline.Holder, sessions *session.Store, store pipeline.SharedStore, addr string) *grpc.Server {
+func startGRPCExtProc(inbound, outbound *pipeline.Holder, sessions *session.Store, store pipeline.SharedStore, skipHosts *skiphost.Matcher, addr string) *grpc.Server {
 	srv := grpc.NewServer()
 	extprocv3.RegisterExternalProcessorServer(srv, &extproc.Server{
 		InboundPipeline:  inbound,
 		OutboundPipeline: outbound,
 		Sessions:         sessions,
 		Shared:           store,
+		SkipHosts:        skipHosts,
 	})
 	registerHealth(srv)
 	reflection.Register(srv)
diff --git a/authbridge/cmd/authbridge-lite/main.go b/authbridge/cmd/authbridge-lite/main.go
index abde9d5ed..26058c957 100644
--- a/authbridge/cmd/authbridge-lite/main.go
+++ b/authbridge/cmd/authbridge-lite/main.go
@@ -39,6 +39,7 @@ import (
 	// (no gRPC, no envoy types).
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/listener/forwardproxy"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/listener/reverseproxy"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/listener/skiphost"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/listener/transparentproxy"
 
 	// Auth gates only: drop the parsers and token-broker.
@@ -241,6 +242,13 @@ func main() {
 	if err != nil {
 		log.Fatalf("creating forward proxy: %v", err)
 	}
+	// SkipHosts: outbound destinations that bypass the pipeline AND
+	// session recording entirely. See ListenerConfig.SkipHosts.
+	skipHosts, err := skiphost.New(cfg.Listener.SkipHosts)
+	if err != nil {
+		log.Fatalf("listener.skip_hosts: %v", err)
+	}
+	fpSrv.SkipHosts = skipHosts
 	sharedStore := shared.New()
 	defer sharedStore.Close() // stop the TTL janitor on normal main return
 	rpSrv.Shared = sharedStore
diff --git a/authbridge/cmd/authbridge-proxy/main.go b/authbridge/cmd/authbridge-proxy/main.go
index 418a75b5a..efef83d44 100644
--- a/authbridge/cmd/authbridge-proxy/main.go
+++ b/authbridge/cmd/authbridge-proxy/main.go
@@ -39,6 +39,7 @@ import (
 	// (no gRPC, no envoy types).
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/listener/forwardproxy"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/listener/reverseproxy"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/listener/skiphost"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/listener/transparentproxy"
 
 	// Plugins. Auth gates first, then the protocol parsers that
@@ -256,6 +257,15 @@ func main() {
 	if err != nil {
 		log.Fatalf("creating forward proxy: %v", err)
 	}
+	// SkipHosts: outbound destinations that bypass the pipeline AND
+	// session recording entirely. See ListenerConfig.SkipHosts for the
+	// motivating case (chatty observability sidecars evicting the
+	// inbound A2A user intent from the session FIFO).
+	skipHosts, err := skiphost.New(cfg.Listener.SkipHosts)
+	if err != nil {
+		log.Fatalf("listener.skip_hosts: %v", err)
+	}
+	fpSrv.SkipHosts = skipHosts
 	sharedStore := shared.New()
 	defer sharedStore.Close() // stop the TTL janitor on normal main return
 	rpSrv.Shared = sharedStore