diff --git a/authbridge/authlib/config/config.go b/authbridge/authlib/config/config.go
index acf11ddd1..1a83d91c4 100644
--- a/authbridge/authlib/config/config.go
+++ b/authbridge/authlib/config/config.go
@@ -198,7 +198,7 @@ type SessionConfig struct {
 	// "user didn't say" with "user said false" and silently flip the default.
 	Enabled     *bool  `yaml:"enabled" json:"enabled"`
 	TTL         string `yaml:"ttl" json:"ttl"`                   // duration string; default: 30m
-	MaxEvents   int    `yaml:"max_events" json:"max_events"`     // max events per session; default: 100
+	MaxEvents   int    `yaml:"max_events" json:"max_events"`     // max events per session; default: 500
 	MaxSessions int    `yaml:"max_sessions" json:"max_sessions"` // max concurrent sessions; default: 100 (0 = unlimited)
 }
 
diff --git a/authbridge/authlib/listener/forwardproxy/mcp_sse_repro_test.go b/authbridge/authlib/listener/forwardproxy/mcp_sse_repro_test.go
new file mode 100644
index 000000000..1fa1c43c9
--- /dev/null
+++ b/authbridge/authlib/listener/forwardproxy/mcp_sse_repro_test.go
@@ -0,0 +1,106 @@
+package forwardproxy
+
+import (
+	"bytes"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/inferenceparser"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/mcpparser"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/session"
+)
+
+// TestForwardProxy_MCP_SSEResponse_RecordsObserve reproduces the live
+// weather-tool-mcp scenario: an MCP tools/list call whose response is
+// text/event-stream (Streamable HTTP). The request is parsed (observe), and
+// the RESPONSE must also be parsed into a result + recorded as an mcp-parser
+// observe on the response event. The live cluster showed result=N,
+// invocations=[] on the response — this test pins whether the proxy+parser
+// path reproduces that.
+func TestForwardProxy_MCP_SSEResponse_RecordsObserve(t *testing.T) {
+	// Upstream mimics weather-tool-mcp: tools/list -> SSE with one data frame
+	// carrying the JSON-RPC result, exactly as FastMCP emits it.
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.WriteHeader(http.StatusOK)
+		io.WriteString(w, "event: message\ndata: {\"jsonrpc\":\"2.0\",\"id\":2,\"result\":{\"tools\":[{\"name\":\"get_weather\"}]}}\n\n")
+		if f, ok := w.(http.Flusher); ok {
+			f.Flush()
+		}
+	}))
+	defer upstream.Close()
+
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+
+	// Production builds Configurable plugins through plugins.Build, which
+	// wraps them via pipeline.WrapConfigured. mcp-parser is Configurable +
+	// StreamingResponder, so it MUST stay a StreamingResponder after wrapping
+	// — that's the bug this test guards (a bare wrapper drops OnResponseFrame
+	// and RunResponseFrame skips it, leaving the SSE response unparsed).
+	// inference-parser is not Configurable, so it's added raw, as in production.
+	pipe, err := pipeline.New([]pipeline.Plugin{
+		pipeline.WrapConfigured(mcpparser.NewMCPParser(), nil),
+		inferenceparser.NewInferenceParser(),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	srv, err := NewServer(pipeline.NewHolder(pipe), store, nil)
+	if err != nil {
+		t.Fatalf("NewServer: %v", err)
+	}
+	proxy := httptest.NewServer(srv.Handler())
+	defer proxy.Close()
+
+	reqBody := `{"jsonrpc":"2.0","id":2,"method":"tools/list"}`
+	req, _ := http.NewRequest("POST", upstream.URL+"/mcp", bytes.NewReader([]byte(reqBody)))
+	req.Header.Set("Content-Type", "application/json")
+	proxyClient := &http.Client{Transport: &http.Transport{Proxy: http.ProxyURL(mustParseURL(proxy.URL))}}
+	resp, err := proxyClient.Do(req)
+	if err != nil {
+		t.Fatalf("request failed: %v", err)
+	}
+	_, _ = io.ReadAll(resp.Body)
+	resp.Body.Close()
+
+	v := store.View(session.DefaultSessionID)
+	if v == nil {
+		t.Fatal("no session recorded")
+	}
+	var reqEv, respEv *pipeline.SessionEvent
+	for i := range v.Events {
+		e := &v.Events[i]
+		if e.MCP == nil || e.MCP.Method != "tools/list" {
+			continue
+		}
+		switch e.Phase {
+		case pipeline.SessionRequest:
+			reqEv = e
+		case pipeline.SessionResponse:
+			respEv = e
+		}
+	}
+	if reqEv == nil || respEv == nil {
+		t.Fatalf("missing req/resp events: req=%v resp=%v", reqEv != nil, respEv != nil)
+	}
+
+	// Request side parsed (sanity).
+	if reqEv.Invocations == nil || len(reqEv.Invocations.Outbound) == 0 {
+		t.Errorf("request event has no mcp-parser invocation: %+v", reqEv.Invocations)
+	}
+
+	// THE ASSERTION: the response was parsed into a result and recorded an
+	// observe. Live cluster fails both.
+	if respEv.MCP.Result == nil {
+		t.Errorf("response MCP.Result not populated (parser never saw the SSE result)")
+	}
+	respObserved := respEv.Invocations != nil && len(respEv.Invocations.Outbound) > 0
+	if !respObserved {
+		t.Errorf("response event has NO mcp-parser invocation — reproduces the bug. invocations=%+v", respEv.Invocations)
+	}
+}
diff --git a/authbridge/authlib/listener/forwardproxy/server.go b/authbridge/authlib/listener/forwardproxy/server.go
index c792e4a8a..d6c49dccf 100644
--- a/authbridge/authlib/listener/forwardproxy/server.go
+++ b/authbridge/authlib/listener/forwardproxy/server.go
@@ -307,16 +307,14 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, isBridge
 			Identity:    pipeline.SnapshotIdentity(pctx),
 			Host:        pctx.Host,
 		}
-		// Record whenever ANY protocol-or-plugin context is present —
-		// MCP/Inference (parser-emitted), Invocations (gate plugins like
-		// jwt-validation/token-exchange), or plugin-public Plugins
-		// entries. Earlier the gate was just MCP||Inference; widening
-		// it ensures auth-only outbound traffic and pure observability
-		// events show up in abctl. Don't narrow this back without
-		// understanding why each clause is necessary.
-		if ev.MCP != nil || ev.Inference != nil || ev.Invocations != nil || plugins != nil {
-			s.Sessions.Append(sid, ev)
-		}
+		// Record EVERY message that reaches the pipeline — even when no
+		// plugin acted and no parser matched (Invocations/MCP/Inference all
+		// nil). The session API is an observability surface; a request the
+		// pipeline saw but no plugin touched is still a network message the
+		// operator wants to see (it carries Host, and the paired response
+		// carries StatusCode). skip_hosts traffic never reaches here (the
+		// !skipped guard above), so it stays suppressed by design.
+		s.Sessions.Append(sid, ev)
 	}
 
 	newAuth := pctx.Headers.Get("Authorization")
@@ -519,11 +517,10 @@ func (s *Server) recordOutboundResponseEvent(pctx *pipeline.Context, statusCode
 		Error:       pipeline.DeriveError(pctx),
 		Duration:    pipeline.DurationSince(pctx.StartedAt),
 	}
-	// Same widened gate as the request side — see the request-phase
-	// comment for why each clause matters.
-	if ev.MCP != nil || ev.Inference != nil || ev.Invocations != nil || plugins != nil {
-		s.Sessions.Append(sid, ev)
-	}
+	// Always record — see the request-phase comment. This is what surfaces
+	// responses no plugin acted on (e.g. a generic 404), carrying StatusCode
+	// + Error even with empty invocations.
+	s.Sessions.Append(sid, ev)
 }
 
 // isEventStream reports whether a Content-Type header value names the
diff --git a/authbridge/authlib/listener/forwardproxy/server_test.go b/authbridge/authlib/listener/forwardproxy/server_test.go
index 0eb801ced..bd06b718e 100644
--- a/authbridge/authlib/listener/forwardproxy/server_test.go
+++ b/authbridge/authlib/listener/forwardproxy/server_test.go
@@ -567,6 +567,57 @@ func TestRecordOutboundReject_SkipsWithoutInvocations(t *testing.T) {
 	}
 }
 
+// TestForwardProxy_RecordsMessageWithNoPluginActivity locks the Part A
+// behavior: a request/response that no plugin acted on (empty pipeline,
+// no parser match) is still recorded as two session events so abctl can
+// show every network message — not just the ones a plugin touched. The
+// response event carries the upstream status even though Invocations is
+// nil.
+func TestForwardProxy_RecordsMessageWithNoPluginActivity(t *testing.T) {
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+
+	backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusNotFound)
+	}))
+	defer backend.Close()
+
+	// Empty pipeline: zero plugins, so no Invocations are ever appended.
+	p, err := pipeline.New([]pipeline.Plugin{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	srv := &Server{OutboundPipeline: pipeline.NewHolder(p), Sessions: store, Client: http.DefaultClient}
+	proxy := httptest.NewServer(srv.Handler())
+	defer proxy.Close()
+
+	req, _ := http.NewRequest("GET", backend.URL+"/missing", nil)
+	proxyClient := &http.Client{Transport: &http.Transport{Proxy: http.ProxyURL(mustParseURL(proxy.URL))}}
+	resp, err := proxyClient.Do(req)
+	if err != nil {
+		t.Fatalf("request failed: %v", err)
+	}
+	resp.Body.Close()
+	if resp.StatusCode != http.StatusNotFound {
+		t.Errorf("status = %d, want 404", resp.StatusCode)
+	}
+
+	v := store.View(session.DefaultSessionID)
+	if v == nil || len(v.Events) != 2 {
+		t.Fatalf("expected 2 events (request + response) with no plugin activity, got %+v", v)
+	}
+	reqEv, respEv := v.Events[0], v.Events[1]
+	if reqEv.Phase != pipeline.SessionRequest || reqEv.Invocations != nil {
+		t.Errorf("request event = phase %v invocations %+v, want SessionRequest / nil", reqEv.Phase, reqEv.Invocations)
+	}
+	if respEv.Phase != pipeline.SessionResponse || respEv.StatusCode != http.StatusNotFound {
+		t.Errorf("response event = phase %v status %d, want SessionResponse / 404", respEv.Phase, respEv.StatusCode)
+	}
+	if respEv.Invocations != nil {
+		t.Errorf("response event invocations = %+v, want nil (no plugin acted)", respEv.Invocations)
+	}
+}
+
 // schemeCapturePlugin captures pctx.Scheme for the scheme-wiring
 // test below.
 type schemeCapturePlugin struct {
diff --git a/authbridge/authlib/listener/forwardproxy/transparent.go b/authbridge/authlib/listener/forwardproxy/transparent.go
index 9f5ec0e36..aa0892820 100644
--- a/authbridge/authlib/listener/forwardproxy/transparent.go
+++ b/authbridge/authlib/listener/forwardproxy/transparent.go
@@ -179,10 +179,14 @@ func (s *Server) recordTunnelOpened(pctx *pipeline.Context) {
 		Plugins:     plugins,
 		Identity:    pipeline.SnapshotIdentity(pctx),
 		Host:        pctx.Host,
+		// Explicit opaque-tunnel marker so abctl can fold this CONNECT into
+		// the decrypted inner request without inferring "tunnel" from shape.
+		Tunnel: true,
 	}
-	if ev.Invocations != nil || plugins != nil {
-		s.Sessions.Append(sid, ev)
-	}
+	// Always record the tunnel-open so passthrough/non-bridged tunnels (no
+	// plugin activity) are still visible. For a TLS-bridged call abctl folds
+	// this CONNECT event into the decrypted inner-request row.
+	s.Sessions.Append(sid, ev)
 }
 
 // tunnel bidirectionally copies between two connections until either side
diff --git a/authbridge/authlib/listener/forwardproxy/transparent_test.go b/authbridge/authlib/listener/forwardproxy/transparent_test.go
index 0933b39d0..d8da40fed 100644
--- a/authbridge/authlib/listener/forwardproxy/transparent_test.go
+++ b/authbridge/authlib/listener/forwardproxy/transparent_test.go
@@ -8,8 +8,33 @@ import (
 
 	"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"
 )
 
+// TestRecordTunnelOpened_SetsTunnelMarker locks the explicit producer marker:
+// a CONNECT / transparent-redirect tunnel-open is recorded with Tunnel=true so
+// consumers (abctl) fold it into the decrypted inner request without inferring
+// "tunnel" from host/extension shape.
+func TestRecordTunnelOpened_SetsTunnelMarker(t *testing.T) {
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+	s := &Server{Sessions: store}
+
+	s.recordTunnelOpened(&pipeline.Context{Direction: pipeline.Outbound, Host: "example.com:443"})
+
+	v := store.View(session.DefaultSessionID)
+	if v == nil || len(v.Events) != 1 {
+		t.Fatalf("expected 1 tunnel-open event, got %+v", v)
+	}
+	ev := v.Events[0]
+	if !ev.Tunnel {
+		t.Error("tunnel-open event must have Tunnel=true")
+	}
+	if ev.Direction != pipeline.Outbound || ev.Phase != pipeline.SessionRequest {
+		t.Errorf("tunnel-open = %v/%v, want Outbound/SessionRequest", ev.Direction, ev.Phase)
+	}
+}
+
 // HandleTransparentConn gates then blind-tunnels: with an allow-all pipeline it
 // must dial the recovered destination and copy bytes both ways, emitting no
 // proxy-protocol bytes of its own (the agent thinks it's talking to dst).
diff --git a/authbridge/authlib/listener/reverseproxy/finisher_test.go b/authbridge/authlib/listener/reverseproxy/finisher_test.go
index 73ffeb982..aa86fbf89 100644
--- a/authbridge/authlib/listener/reverseproxy/finisher_test.go
+++ b/authbridge/authlib/listener/reverseproxy/finisher_test.go
@@ -6,11 +6,27 @@ import (
 	"net/http/httptest"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/plugintesting"
 )
 
+// waitSeen polls b until it's true, up to ~1s. OnFinish runs in the
+// reverse-proxy handler's deferred RunFinish, which fires only after the
+// streamed response (NewServer sets FlushInterval=-1) has already reached the
+// client — so http.Get can return before OnFinish has run. Poll rather than
+// read immediately, which otherwise races (flaky under -race).
+func waitSeen(b *atomic.Bool) bool {
+	for i := 0; i < 1000; i++ {
+		if b.Load() {
+			return true
+		}
+		time.Sleep(time.Millisecond)
+	}
+	return false
+}
+
 // finisherStub is a minimal Plugin + Finisher used by these tests to
 // observe OnFinish dispatch and the Outcome the listener derived.
 type finisherStub struct {
@@ -34,11 +50,13 @@ func (p *finisherStub) OnResponse(context.Context, *pipeline.Context) pipeline.A
 	return pipeline.Action{Type: pipeline.Continue}
 }
 func (p *finisherStub) OnFinish(_ context.Context, pctx *pipeline.Context) {
-	p.seen.Store(true)
 	if o := pctx.Outcome(); o != nil {
 		cp := *o
 		p.outcome.Store(&cp)
 	}
+	// Store seen LAST so a reader that observes seen==true is guaranteed to
+	// also see the outcome that was set just above.
+	p.seen.Store(true)
 }
 
 func pipelineWith(t *testing.T, plugins ...pipeline.Plugin) *pipeline.Holder {
@@ -73,7 +91,7 @@ func TestReverseProxy_Finisher_Allow(t *testing.T) {
 	}
 	resp.Body.Close()
 
-	if !f.seen.Load() {
+	if !waitSeen(&f.seen) {
 		t.Fatal("OnFinish did not fire")
 	}
 	o := f.outcome.Load()
@@ -133,12 +151,15 @@ func TestReverseProxy_Finisher_Deny(t *testing.T) {
 		t.Errorf("HTTP status = %d, want 403", resp.StatusCode)
 	}
 
-	if !before.seen.Load() {
+	if !waitSeen(&before.seen) {
 		t.Error("before-deny.OnFinish should have fired (OnRequest ran before denial)")
 	}
-	if !denier.seen.Load() {
+	if !waitSeen(&denier.seen) {
 		t.Error("denier.OnFinish should have fired (OnRequest ran and produced the deny)")
 	}
+	// before/denier have fired, so the single RunFinish dispatch is complete;
+	// after-deny was never dispatched (its OnRequest never ran), so its
+	// OnFinish must not have fired.
 	if after.seen.Load() {
 		t.Error("after-deny.OnFinish should NOT have fired (OnRequest never ran)")
 	}
diff --git a/authbridge/authlib/listener/reverseproxy/server.go b/authbridge/authlib/listener/reverseproxy/server.go
index 78e9be08b..ae1701b5c 100644
--- a/authbridge/authlib/listener/reverseproxy/server.go
+++ b/authbridge/authlib/listener/reverseproxy/server.go
@@ -256,7 +256,9 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) {
 	// as denials already do via recordInboundReject). The A2A-specific session
 	// rekey in modifyResponse stays A2A-gated.
 	plugins := pipeline.SnapshotPlugins(pctx.Extensions.Custom)
-	if s.Sessions != nil && (pctx.Extensions.A2A != nil || pctx.Extensions.Invocations != nil || plugins != nil) {
+	// Record every inbound request the pipeline saw, even with no plugin
+	// activity (skip_hosts is N/A inbound).
+	if s.Sessions != nil {
 		sid := inboundSessionID(pctx)
 		// Snapshot-copy the protocol extension and use the shared helpers
 		// for plugin invocations / observability / identity. Mirrors what
@@ -378,7 +380,8 @@ func (s *Server) modifyResponse(resp *http.Response) error {
 	// pipeline saw at this point (may be empty for streamed bodies),
 	// but the status code and plugin invocations are always meaningful.
 	plugins := pipeline.SnapshotPlugins(pctx.Extensions.Custom)
-	if s.Sessions != nil && (pctx.Extensions.A2A != nil || pctx.Extensions.Invocations != nil || plugins != nil) {
+	// Always pair every inbound request with a response row (carries StatusCode).
+	if s.Sessions != nil {
 		sid := inboundSessionID(pctx)
 		s.Sessions.Append(sid, pipeline.SessionEvent{
 			At:          time.Now(),
@@ -513,9 +516,8 @@ func (s *Server) recordInboundResponseEvent(pctx *pipeline.Context, statusCode i
 		return
 	}
 	plugins := pipeline.SnapshotPlugins(pctx.Extensions.Custom)
-	if !(pctx.Extensions.A2A != nil || pctx.Extensions.Invocations != nil || plugins != nil) {
-		return
-	}
+	// Always record the streaming response (carries StatusCode), even with
+	// no plugin activity.
 	sid := inboundSessionID(pctx)
 	// Rekey default → contextId mirroring the buffered path's behavior;
 	// streaming A2A message/stream may discover the contextId mid-stream.
diff --git a/authbridge/authlib/pipeline/configured.go b/authbridge/authlib/pipeline/configured.go
index ec6a2c437..5bcdbd3c4 100644
--- a/authbridge/authlib/pipeline/configured.go
+++ b/authbridge/authlib/pipeline/configured.go
@@ -28,6 +28,13 @@ type RawConfigProvider interface {
 // implements each of those four interfaces explicitly and forwards
 // conditionally; see the methods below.
 //
+// StreamingResponder is handled differently — see configuredStreamingPlugin
+// and WrapConfigured. It must NOT be forwarded unconditionally here, because
+// listeners type-assert StreamingResponder to choose the streaming-vs-buffered
+// response path (Pipeline.HasStreamingResponders); making every wrapped plugin
+// a no-op StreamingResponder would silently flip that path selection for
+// non-streaming pipelines.
+//
 // Side effect of unconditional forwarding: every wrapped (i.e., every
 // Configurable) plugin satisfies all four optional interfaces regardless
 // of what the inner plugin actually implements. Callers that distinguish
@@ -41,6 +48,29 @@ type configuredPlugin struct {
 	raw json.RawMessage
 }
 
+// configuredStreamingPlugin is the StreamingResponder-preserving variant of
+// configuredPlugin, returned by WrapConfigured only when the wrapped plugin
+// implements StreamingResponder. The base wrapper embeds the Plugin interface,
+// which does not promote OnResponseFrame — so without this a Configurable +
+// StreamingResponder plugin (e.g. mcp-parser, which has Configure) would lose
+// its response-frame dispatch once wrapped: RunResponseFrame type-asserts
+// StreamingResponder, fails on the bare wrapper, and skips the plugin, leaving
+// MCP/SSE responses unparsed.
+//
+// Using a distinct type (instead of an unconditional no-op OnResponseFrame on
+// configuredPlugin) keeps HasStreamingResponders exact: only plugins that
+// genuinely stream are reported as streaming responders.
+type configuredStreamingPlugin struct {
+	*configuredPlugin
+}
+
+// OnResponseFrame forwards to the wrapped plugin's StreamingResponder
+// implementation. WrapConfigured constructs this wrapper only when the inner
+// plugin implements StreamingResponder, so the assertion always holds.
+func (c *configuredStreamingPlugin) OnResponseFrame(ctx context.Context, pctx *Context, frame []byte, last bool) Action {
+	return c.Plugin.(StreamingResponder).OnResponseFrame(ctx, pctx, frame, last)
+}
+
 // WrapConfigured returns a Plugin whose dynamic type retains the raw
 // config bytes the plugin was built from, so the session API can
 // surface them on /v1/pipeline. Callers (plugins.Build) invoke this
@@ -55,7 +85,14 @@ type configuredPlugin struct {
 // and on the rare hot-reload path.
 func WrapConfigured(p Plugin, raw json.RawMessage) Plugin {
 	cp := append(json.RawMessage(nil), raw...)
-	return &configuredPlugin{Plugin: p, raw: cp}
+	base := &configuredPlugin{Plugin: p, raw: cp}
+	// Preserve StreamingResponder: the embedded Plugin interface does not
+	// promote OnResponseFrame, so a Configurable + StreamingResponder plugin
+	// (e.g. mcp-parser) would otherwise lose response-frame dispatch.
+	if _, ok := p.(StreamingResponder); ok {
+		return &configuredStreamingPlugin{configuredPlugin: base}
+	}
+	return base
 }
 
 // RawConfig returns a defensive copy of the raw config bytes the
diff --git a/authbridge/authlib/pipeline/configured_test.go b/authbridge/authlib/pipeline/configured_test.go
index 7aadf44cf..03c4d5a25 100644
--- a/authbridge/authlib/pipeline/configured_test.go
+++ b/authbridge/authlib/pipeline/configured_test.go
@@ -15,7 +15,7 @@ type fakePlugin struct {
 	responses int
 }
 
-func (f *fakePlugin) Name() string                  { return f.name }
+func (f *fakePlugin) Name() string                     { return f.name }
 func (f *fakePlugin) Capabilities() PluginCapabilities { return f.caps }
 func (f *fakePlugin) OnRequest(ctx context.Context, pctx *Context) Action {
 	f.requests++
@@ -210,3 +210,50 @@ func TestConfiguredPluginReadyDefaultsTrueForNonReadier(t *testing.T) {
 		t.Fatal("non-Readier wrapped plugin should default to ready=true")
 	}
 }
+
+// streamingFakePlugin is a Plugin that also implements StreamingResponder,
+// modeling a Configurable + streaming plugin like mcp-parser.
+type streamingFakePlugin struct {
+	fakePlugin
+	frames int
+}
+
+func (s *streamingFakePlugin) OnResponseFrame(_ context.Context, _ *Context, _ []byte, _ bool) Action {
+	s.frames++
+	return Action{Type: Continue}
+}
+
+// TestWrapConfigured_PreservesStreamingResponder is the regression test for the
+// mcp-parser "tools/list response not parsed" bug: a Configurable plugin that
+// also implements StreamingResponder must STILL satisfy StreamingResponder
+// after WrapConfigured, and OnResponseFrame must forward to the inner plugin.
+// Before the fix, the wrapper embedded only the Plugin interface, so
+// OnResponseFrame was not promoted and RunResponseFrame skipped the plugin.
+func TestWrapConfigured_PreservesStreamingResponder(t *testing.T) {
+	inner := &streamingFakePlugin{fakePlugin: fakePlugin{name: "mcp-parser"}}
+	cp := WrapConfigured(inner, json.RawMessage(`{}`))
+
+	sr, ok := cp.(StreamingResponder)
+	if !ok {
+		t.Fatal("wrapped Configurable+StreamingResponder plugin must still satisfy StreamingResponder")
+	}
+	sr.OnResponseFrame(context.Background(), nil, []byte("frame"), true)
+	if inner.frames != 1 {
+		t.Fatalf("OnResponseFrame did not forward to inner plugin: frames=%d, want 1", inner.frames)
+	}
+	// The config is still surfaced through the streaming wrapper.
+	if rc, ok := cp.(RawConfigProvider); !ok || string(rc.RawConfig()) != `{}` {
+		t.Fatalf("streaming wrapper lost RawConfig: ok=%v", ok)
+	}
+}
+
+// TestWrapConfigured_NonStreamingNotPromoted confirms the fix preserves
+// HasStreamingResponders semantics: a Configurable plugin that does NOT
+// implement StreamingResponder must NOT become one through the wrapper, so the
+// listener's streaming-vs-buffered path selection is unchanged.
+func TestWrapConfigured_NonStreamingNotPromoted(t *testing.T) {
+	cp := WrapConfigured(&fakePlugin{name: "token-exchange"}, json.RawMessage(`{}`))
+	if _, ok := cp.(StreamingResponder); ok {
+		t.Fatal("wrapping a non-streaming plugin must not make it a StreamingResponder")
+	}
+}
diff --git a/authbridge/authlib/pipeline/session.go b/authbridge/authlib/pipeline/session.go
index e24cb0796..fa853ecbf 100644
--- a/authbridge/authlib/pipeline/session.go
+++ b/authbridge/authlib/pipeline/session.go
@@ -137,6 +137,14 @@ type SessionEvent struct {
 	// and for events recorded before the TLS handshake completed.
 	// Populated by the listener layer; plugins do not write to it.
 	TLS *EventTLS
+
+	// Tunnel marks an opaque CONNECT / transparent-redirect tunnel-open: the
+	// bytes are not HTTP, so there's no protocol parse. Set only by
+	// recordTunnelOpened. Consumers (abctl) use it to fold the tunnel into the
+	// decrypted inner request a TLS bridge produces — an explicit producer
+	// signal rather than inferring "tunnel" from host/extension shape, which
+	// an ordinary unparsed request could otherwise mimic.
+	Tunnel bool
 }
 
 // EventTLS describes the TLS state of a connection that produced a
@@ -215,6 +223,7 @@ type sessionEventWire struct {
 	Host        string                     `json:"host,omitempty"`
 	DurationMs  int64                      `json:"durationMs,omitempty"`
 	TLS         *EventTLS                  `json:"tls,omitempty"`
+	Tunnel      bool                       `json:"tunnel,omitempty"`
 }
 
 func (e SessionEvent) MarshalJSON() ([]byte, error) {
@@ -234,6 +243,7 @@ func (e SessionEvent) MarshalJSON() ([]byte, error) {
 		Host:        e.Host,
 		DurationMs:  e.Duration.Milliseconds(),
 		TLS:         e.TLS,
+		Tunnel:      e.Tunnel,
 	})
 }
 
@@ -261,6 +271,7 @@ func (e *SessionEvent) UnmarshalJSON(data []byte) error {
 		Host:        w.Host,
 		Duration:    time.Duration(w.DurationMs) * time.Millisecond,
 		TLS:         w.TLS,
+		Tunnel:      w.Tunnel,
 	}
 	return nil
 }
diff --git a/authbridge/authlib/pipeline/session_test.go b/authbridge/authlib/pipeline/session_test.go
index 2680a33f7..467676321 100644
--- a/authbridge/authlib/pipeline/session_test.go
+++ b/authbridge/authlib/pipeline/session_test.go
@@ -79,6 +79,7 @@ func TestSessionEvent_JSONRoundTrip(t *testing.T) {
 		Identity:   &EventIdentity{Subject: "alice", ClientID: "agent-1"},
 		StatusCode: 200,
 		Error:      &EventError{Kind: "upstream", Message: "timeout"},
+		Tunnel:     true,
 	}
 
 	first, err := json.Marshal(orig)
diff --git a/authbridge/authlib/plugins/mcpparser/plugin.go b/authbridge/authlib/plugins/mcpparser/plugin.go
index 445560ba5..66503e676 100644
--- a/authbridge/authlib/plugins/mcpparser/plugin.go
+++ b/authbridge/authlib/plugins/mcpparser/plugin.go
@@ -304,18 +304,22 @@ func (p *MCPParser) OnResponseFrame(_ context.Context, pctx *pipeline.Context, f
 		return pipeline.Action{Type: pipeline.Continue}
 	}
 
-	var rpc jsonRPCResponse
-	if err := json.Unmarshal(frame, &rpc); err != nil {
-		// A malformed JSON-RPC message in a stream is unusual but
-		// recoverable — skip it and keep going. Don't return Reject
-		// because the listener is forwarding bytes regardless of what
-		// we say (record-only contract).
-		slog.Debug("mcp-parser: malformed frame, skipping", "error", err, "frameLen", len(frame))
-		return pipeline.Action{Type: pipeline.Continue}
-	}
-	if rpc.Result == nil && rpc.Error == nil {
-		// Notifications, heartbeats, or other JSON-RPC shapes without
-		// a result/error envelope — silently skip (no observation).
+	// Parse SSE-aware: a frame is normally one SSE event's pre-stripped
+	// JSON-RPC payload (streaming path), but the buffered dispatch hands the
+	// WHOLE response body as a single last=true frame — and for a Streamable
+	// HTTP server that body is a raw `data: {...}` SSE blob. A bare
+	// json.Unmarshal fails on that blob and silently drops the result (the
+	// tools/list-response-not-recorded bug). parseMCPResponse tries a direct
+	// decode first (clean frame), then scans `data:` lines (raw SSE blob), so
+	// both shapes record correctly.
+	rpc, ok := parseMCPResponse(frame)
+	if !ok {
+		// Notifications, heartbeats, malformed frames, or other shapes with
+		// no JSON-RPC result/error envelope — nothing to record per-frame.
+		// The end-of-stream last=true call records a Skip if no envelope was
+		// ever seen, so the response row still pairs with its request.
+		slog.Debug("mcp-parser: response frame carried no JSON-RPC result/error",
+			"frameLen", len(frame), "frame", parsercommon.Truncate(string(frame), parsercommon.DebugBodyMax))
 		return pipeline.Action{Type: pipeline.Continue}
 	}
 
diff --git a/authbridge/authlib/plugins/mcpparser/streaming_test.go b/authbridge/authlib/plugins/mcpparser/streaming_test.go
index f824ce179..fb58548c0 100644
--- a/authbridge/authlib/plugins/mcpparser/streaming_test.go
+++ b/authbridge/authlib/plugins/mcpparser/streaming_test.go
@@ -97,6 +97,32 @@ func TestMCPParser_OnResponseFrame_EmptyStreamRecordsSkip(t *testing.T) {
 	}
 }
 
+// TestMCPParser_OnResponseFrame_RawSSEBlob reproduces the live tools/list
+// bug: the forward proxy's buffered dispatch hands the WHOLE response body as
+// one last=true frame, and for a Streamable HTTP (MCP) server that body is a
+// raw "data: {...}" SSE blob — not pre-stripped JSON. The bare json.Unmarshal
+// failed on it and silently dropped the result (response recorded with no
+// result, no invocation). OnResponseFrame must parse the embedded JSON-RPC
+// result and record an observe.
+func TestMCPParser_OnResponseFrame_RawSSEBlob(t *testing.T) {
+	p := NewMCPParser()
+	pctx := &pipeline.Context{
+		Extensions: pipeline.Extensions{MCP: &pipeline.MCPExtension{Method: "tools/list"}},
+	}
+	blob := []byte("event: message\ndata: {\"jsonrpc\":\"2.0\",\"id\":2,\"result\":{\"tools\":[{\"name\":\"get_weather\"}]}}\n\n")
+	pctx.SetCurrentPlugin("mcp-parser", pipeline.InvocationPhaseResponse)
+	p.OnResponseFrame(context.Background(), pctx, blob, true)
+	pctx.ClearCurrentPlugin()
+
+	if pctx.Extensions.MCP.Result == nil {
+		t.Fatal("Result not populated from raw SSE blob — the bug")
+	}
+	inv := pctx.Extensions.Invocations
+	if inv == nil || len(inv.Inbound)+len(inv.Outbound) == 0 {
+		t.Fatal("no mcp-parser observe recorded for the SSE response")
+	}
+}
+
 func TestMCPParser_OnResponseFrame_MalformedFrameSkipped(t *testing.T) {
 	p := NewMCPParser()
 	pctx := &pipeline.Context{
diff --git a/authbridge/cmd/abctl/tui/app.go b/authbridge/cmd/abctl/tui/app.go
index 35f438f20..67904b982 100644
--- a/authbridge/cmd/abctl/tui/app.go
+++ b/authbridge/cmd/abctl/tui/app.go
@@ -193,41 +193,43 @@ type model struct {
 	filter       string
 	filtering    bool
 	paused       bool
-	// showSkips toggles whether Action=skip rows render in the events
-	// table. False (default) hides them — most operators care about
-	// allow/deny/modify/observe events. Toggle with `s`. hiddenSkips
-	// is the count from the most recent rebuildEventsTable, surfaced
-	// in the footer so a sparse-looking timeline doesn't read as
-	// data loss.
-	showSkips     bool
-	hiddenSkips   int
-	flash         string
-	flashUntil    time.Time
-	width, height int
+	// hideInactive toggles whether passthrough / skip-only messages are
+	// hidden from the events table. False (default) shows every message —
+	// the operator asked to see all network traffic, processed or not.
+	// Toggle with `s` to focus on plugin activity (deny/modify/allow/
+	// observe). hiddenInactive is the count from the most recent
+	// rebuildEventsTable, surfaced in the footer so a filtered timeline
+	// doesn't read as data loss.
+	hideInactive   bool
+	hiddenInactive int
+	flash          string
+	flashUntil     time.Time
+	width, height  int
 	// bodyHeight is the inner height available to panes (terminal height
 	// minus title + footer). Cached by layout() so rebuildEventsTable can
 	// size the events table after accounting for the IDENTITY banner.
 	bodyHeight int
 
 	// Panel components.
-	sessionsTbl  table.Model
-	eventsTbl    table.Model
-	pipelineTbl  table.Model
-	catalogTbl   table.Model
-	detailVp     viewport.Model
-	detailEvent  *pipeline.SessionEvent
-	// detailInvocation is the plugin invocation selected in the events pane
-	// when the detail view was opened. Used to re-scope the event to that
-	// plugin on resize/re-render. nil means "whole event" (no invocation).
-	detailInvocation *pipeline.Invocation
-	detailPlugin     *apiclient.PipelinePlugin
+	sessionsTbl table.Model
+	eventsTbl   table.Model
+	pipelineTbl table.Model
+	catalogTbl  table.Model
+	detailVp    viewport.Model
+	detailEvent *pipeline.SessionEvent
+	// detailRow is the full events-pane row (event + any folded CONNECT
+	// tunnel) the detail view was opened on. Kept so layout() can re-render
+	// the detail pane on resize without re-deriving the tunnel fold.
+	detailRow    eventRow
+	detailPlugin *apiclient.PipelinePlugin
 	filterInput  textinput.Model
 
-	// visibleRows holds the invocationRow spec for each rendered row in
-	// eventsTbl. Populated by rebuildEventsTable so selectedEvent can
-	// return the (event, invocation) tuple the cursor is on without
-	// re-walking the cache. Reset on every rebuild.
-	visibleRows []invocationRow
+	// visibleRows holds the eventRow for each rendered row in eventsTbl —
+	// one per network message. Populated by rebuildEventsTable so
+	// selectedEvent / selectedEventRow can return the message the cursor is
+	// on (and any folded tunnel) without re-walking the cache. Reset on
+	// every rebuild.
+	visibleRows []eventRow
 
 	// pipeline is the fetched plugin composition. nil until the initial
 	// GetPipeline response arrives; the pipeline pane shows "(loading…)"
diff --git a/authbridge/cmd/abctl/tui/detail_pane.go b/authbridge/cmd/abctl/tui/detail_pane.go
index ba5e78f79..557b9bf01 100644
--- a/authbridge/cmd/abctl/tui/detail_pane.go
+++ b/authbridge/cmd/abctl/tui/detail_pane.go
@@ -9,45 +9,9 @@ import (
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
 )
 
-// eventScopedToPlugin returns a shallow copy of e whose Invocations and
-// per-plugin Plugins map are limited to the given plugin. Event-level context
-// (protocol slot, identity) is preserved; filterForDetail still trims those by
-// phase. Returns e unchanged when plugin is empty.
-func eventScopedToPlugin(e *pipeline.SessionEvent, plugin string) *pipeline.SessionEvent {
-	if e == nil || plugin == "" {
-		return e
-	}
-	scoped := *e // shallow copy
-	if e.Invocations != nil {
-		scoped.Invocations = &pipeline.Invocations{
-			Inbound:  filterInvocationsByPlugin(e.Invocations.Inbound, plugin),
-			Outbound: filterInvocationsByPlugin(e.Invocations.Outbound, plugin),
-		}
-	}
-	if e.Plugins != nil {
-		if raw, ok := e.Plugins[plugin]; ok {
-			scoped.Plugins = map[string]json.RawMessage{plugin: raw}
-		} else {
-			scoped.Plugins = nil
-		}
-	}
-	return &scoped
-}
-
-// filterInvocationsByPlugin returns the invocations in invs whose Plugin
-// matches plugin, preserving order. Returns nil when none match.
-func filterInvocationsByPlugin(invs []pipeline.Invocation, plugin string) []pipeline.Invocation {
-	var out []pipeline.Invocation
-	for _, iv := range invs {
-		if iv.Plugin == plugin {
-			out = append(out, iv)
-		}
-	}
-	return out
-}
-
-// showDetail loads e into the detail viewport as colorized JSON and
-// remembers the focused event so yank (y) can find it.
+// showDetail loads the row's event into the detail viewport as colorized
+// JSON and remembers the focused row so yank (y) can find the event and
+// layout() can re-render on resize.
 //
 // Marshal with SessionEvent.MarshalJSON first (readable wire form — string
 // enums, durationMs), then filter inference/mcp extensions so request
@@ -55,21 +19,20 @@ func filterInvocationsByPlugin(invs []pipeline.Invocation, plugin string) []pipe
 // response-side fields (TUI readability only — wire format is unchanged,
 // and yank still writes the full JSON).
 //
-// When the event arrived over TLS (SessionEvent.TLS non-nil), a small
-// header block is prepended to the JSON so operators can see the
-// connection-level identity at a glance. Absent for plaintext events.
+// The whole event is rendered — all plugin invocations, both directions —
+// because the timeline is now one row per message; the per-plugin breakdown
+// lives here in the detail, not in separate rows.
 //
-// The events list is one row per plugin invocation, so showDetail takes the
-// selected invocation and renders a copy of the event scoped to that plugin
-// (see eventScopedToPlugin). A nil invocation renders the whole event.
-func (m *model) showDetail(e *pipeline.SessionEvent, inv *pipeline.Invocation) {
+// When a CONNECT tunnel was folded into this row (TLS bridge), a one-block
+// summary of the tunnel is prepended so the operator sees the bridged
+// origin and the connection-level decision without a separate row. When the
+// event arrived over TLS (SessionEvent.TLS non-nil), a small TLS header
+// block is prepended too. Both absent for plaintext, non-bridged events.
+func (m *model) showDetail(r eventRow) {
+	e := r.event
+	m.detailRow = r
 	m.detailEvent = e
-	m.detailInvocation = inv
-	ev := e
-	if inv != nil {
-		ev = eventScopedToPlugin(e, inv.Plugin)
-	}
-	data, err := json.Marshal(ev)
+	data, err := json.Marshal(e)
 	if err != nil {
 		m.detailVp.SetContent("error marshaling event: " + err.Error())
 		return
@@ -81,6 +44,9 @@ func (m *model) showDetail(e *pipeline.SessionEvent, inv *pipeline.Invocation) {
 		// content keeps its highlighting.
 		content = ansi.Wrap(content, w, " -")
 	}
+	if r.tunnel != nil {
+		content = tunnelHeader(r.tunnel) + "\n\n" + content
+	}
 	if header := tlsHeader(e.TLS); header != "" {
 		content = header + "\n\n" + content
 	}
@@ -88,6 +54,25 @@ func (m *model) showDetail(e *pipeline.SessionEvent, inv *pipeline.Invocation) {
 	m.detailVp.GotoTop()
 }
 
+// tunnelHeader summarizes the CONNECT tunnel folded into a TLS-bridged
+// request row: the bridged origin (host:port) and any gate invocations that
+// ran on the tunnel-open before the bytes were decrypted. Lets an operator
+// see the connection-level decision without a separate row.
+//
+//	tunnel:  CONNECT example.com:443 (TLS bridge)
+//	  jwt-validation skip (no_inbound_identity)
+func tunnelHeader(tunnel *pipeline.SessionEvent) string {
+	var b strings.Builder
+	b.WriteString(fmt.Sprintf("tunnel:  CONNECT %s (TLS bridge)", tunnel.Host))
+	for _, iv := range allInvocations(tunnel) {
+		b.WriteString(fmt.Sprintf("\n  %s %s", iv.Plugin, iv.Action))
+		if iv.Reason != "" {
+			b.WriteString(" (" + iv.Reason + ")")
+		}
+	}
+	return b.String()
+}
+
 // tlsHeader builds a one-block summary of the TLS connection state.
 // Returns the empty string when tls is nil (plaintext events) so the
 // caller can prepend unconditionally.
diff --git a/authbridge/cmd/abctl/tui/detail_pane_test.go b/authbridge/cmd/abctl/tui/detail_pane_test.go
index 48a55ac77..74d013a5f 100644
--- a/authbridge/cmd/abctl/tui/detail_pane_test.go
+++ b/authbridge/cmd/abctl/tui/detail_pane_test.go
@@ -1,7 +1,6 @@
 package tui
 
 import (
-	"encoding/json"
 	"strings"
 	"testing"
 
@@ -67,61 +66,38 @@ func TestTLSHeader_PeerOnly(t *testing.T) {
 	}
 }
 
-// eventScopedToPlugin must restrict both the Invocations slices and the
-// per-plugin Plugins map to the selected plugin, and must NOT mutate the
-// original event (the shallow copy aliases the slices/map otherwise).
-func TestEventScopedToPlugin_FiltersToSelectedPlugin(t *testing.T) {
-	ev := &pipeline.SessionEvent{
+// tunnelHeader summarizes the folded CONNECT tunnel on a TLS-bridged row:
+// the bridged origin (host:port) and any gate invocation that ran on the
+// tunnel-open before the bytes were decrypted.
+func TestTunnelHeader_OriginAndInvocations(t *testing.T) {
+	got := tunnelHeader(&pipeline.SessionEvent{
+		Host: "api.anthropic.com:443",
 		Invocations: &pipeline.Invocations{
-			Inbound: []pipeline.Invocation{
-				{Plugin: "jwt-validation", Action: pipeline.ActionAllow},
-				{Plugin: "a2a-parser", Action: pipeline.ActionObserve},
+			Outbound: []pipeline.Invocation{
+				{Plugin: "jwt-validation", Action: pipeline.ActionSkip, Reason: "no_inbound_identity"},
 			},
 		},
-		Plugins: map[string]json.RawMessage{
-			"jwt-validation": json.RawMessage("{}"),
-			"a2a-parser":     json.RawMessage("{}"),
-		},
-	}
-
-	scoped := eventScopedToPlugin(ev, "jwt-validation")
-
-	if got := len(scoped.Invocations.Inbound); got != 1 {
-		t.Fatalf("scoped inbound invocations = %d, want 1", got)
-	}
-	if got := scoped.Invocations.Inbound[0].Plugin; got != "jwt-validation" {
-		t.Errorf("scoped invocation plugin = %q, want %q", got, "jwt-validation")
-	}
-	if got := len(scoped.Plugins); got != 1 {
-		t.Fatalf("scoped Plugins entries = %d, want 1", got)
-	}
-	if _, ok := scoped.Plugins["jwt-validation"]; !ok {
-		t.Errorf("scoped Plugins missing jwt-validation key: %v", scoped.Plugins)
-	}
-	if _, ok := scoped.Plugins["a2a-parser"]; ok {
-		t.Errorf("scoped Plugins unexpectedly retained a2a-parser")
-	}
-
-	// Original event must be untouched (no aliasing of slice/map).
-	if got := len(ev.Invocations.Inbound); got != 2 {
-		t.Errorf("original inbound invocations mutated: = %d, want 2", got)
-	}
-	if got := len(ev.Plugins); got != 2 {
-		t.Errorf("original Plugins map mutated: = %d, want 2", got)
+	})
+	for _, want := range []string{
+		"CONNECT api.anthropic.com:443",
+		"TLS bridge",
+		"jwt-validation skip",
+		"no_inbound_identity",
+	} {
+		if !strings.Contains(got, want) {
+			t.Errorf("tunnelHeader missing %q\ngot:\n%s", want, got)
+		}
 	}
 }
 
-// An empty plugin string means "no specific invocation" — the helper returns
-// the event unchanged (same pointer) so old whole-event behavior is preserved.
-func TestEventScopedToPlugin_EmptyPluginReturnsUnchanged(t *testing.T) {
-	ev := &pipeline.SessionEvent{
-		Invocations: &pipeline.Invocations{
-			Inbound: []pipeline.Invocation{
-				{Plugin: "jwt-validation", Action: pipeline.ActionAllow},
-			},
-		},
-	}
-	if got := eventScopedToPlugin(ev, ""); got != ev {
-		t.Errorf("eventScopedToPlugin(ev, \"\") = %p, want original %p", got, ev)
+// A tunnel with no gate invocations (pure passthrough that was then bridged)
+// still names the origin without trailing invocation lines.
+func TestTunnelHeader_NoInvocations(t *testing.T) {
+	got := tunnelHeader(&pipeline.SessionEvent{Host: "example.com:8443"})
+	if !strings.Contains(got, "CONNECT example.com:8443") {
+		t.Errorf("tunnelHeader missing origin\ngot:\n%s", got)
+	}
+	if strings.Count(got, "\n") != 0 {
+		t.Errorf("tunnelHeader with no invocations should be one line\ngot:\n%s", got)
 	}
 }
diff --git a/authbridge/cmd/abctl/tui/events_pane.go b/authbridge/cmd/abctl/tui/events_pane.go
index f8e5c484d..574fff9d5 100644
--- a/authbridge/cmd/abctl/tui/events_pane.go
+++ b/authbridge/cmd/abctl/tui/events_pane.go
@@ -2,6 +2,7 @@ package tui
 
 import (
 	"fmt"
+	"net"
 	"sort"
 	"strconv"
 	"strings"
@@ -21,7 +22,7 @@ func newEventsTable() table.Model {
 			{Title: "#", Width: 4},
 			{Title: "TIME", Width: 12},
 			{Title: "DIR", Width: 4},
-			{Title: "PHASE", Width: 6},
+			{Title: "PHASE", Width: 7},
 			{Title: "ACTION", Width: 8},
 			{Title: "PLUGIN", Width: 18},
 			{Title: "METHOD", Width: 22},
@@ -36,6 +37,30 @@ func newEventsTable() table.Model {
 	return t
 }
 
+// eventRow is one display row — exactly one network message. The cartesian
+// "row per plugin invocation" model is gone: a message touched by N plugins
+// is a single row, with the per-plugin breakdown available in the detail
+// pane. tunnel, when non-nil, is a CONNECT tunnel-open event folded into a
+// TLS-bridged request (see buildEventRows); it contributes a summary line to
+// the detail pane but no separate row.
+type eventRow struct {
+	event  *pipeline.SessionEvent
+	tunnel *pipeline.SessionEvent
+}
+
+// invocations returns every plugin invocation the row's ACTION/PLUGIN cell and
+// the inactive filter should consider: the event's own, plus any folded
+// CONNECT tunnel's gate invocations — so a bridged row reflects activity on the
+// tunnel-open (e.g. an egress gate that allowed the CONNECT) and isn't wrongly
+// hidden or under-reported.
+func (er eventRow) invocations() []pipeline.Invocation {
+	invs := allInvocations(er.event)
+	if er.tunnel != nil {
+		invs = append(invs, allInvocations(er.tunnel)...)
+	}
+	return invs
+}
+
 // rebuildEventsTable populates the events table from the cache for the
 // currently selected session, applying filter + preserving cursor. Also
 // resizes the table height to account for the IDENTITY banner — when
@@ -59,105 +84,66 @@ func (m *model) rebuildEventsTable() {
 	prevRow := m.eventsTbl.Cursor()
 	wasAtEnd := prevRow >= len(m.eventsTbl.Rows())-1
 
-	// Flatten (event, invocation) into row specs up-front so pair-linking
-	// and filtering can run against the flat row list. Events without
-	// invocations fall back to a single pseudo-row (unusual — the listener
-	// only records events that have at least one Invocation or A2A/MCP/
-	// Inference extension, but parser-only events can still land here if
-	// the parser populated its extension without emitting an Invocation).
-	rowSpecs := flattenInvocations(events)
-
-	// Pair request/response rows by (direction, plugin) so each plugin's
-	// contribution on the request side connects to its contribution on the
-	// response side, independent of other plugins in the same pipeline.
-	pairs := pairInvocationRows(rowSpecs)
-
-	// Event-level pair IDs for the # column. Assigns each event a small
-	// integer; events that match as a (request, response) pair share one
-	// integer so the operator can scan the column for the repeated
-	// number. Derived from the same row-level pair map that drives the
-	// └resp glyph, so the two visual cues stay consistent.
-	eventIDs := computeEventPairIDs(rowSpecs, pairs)
-
-	// Per-row tree glyph (┌ / │ / └) for the PHASE column. Lets
-	// operators trace a paired (req, resp) span visually even when
-	// other rows interleave between them.
-	spanGlyphs := computeSpanGlyphs(pairs, len(rowSpecs))
-
-	rows := make([]table.Row, 0, len(rowSpecs))
+	// One display row per network message. CONNECT tunnel-opens are folded
+	// into the decrypted inner request that immediately follows them (TLS
+	// bridge), so a bridged call reads as a single request row — the same
+	// shape as a plaintext call.
+	eventRows := buildEventRows(events)
+
+	// Pair request rows with their response rows. ids drives the # column
+	// (one integer repeated across a request/response exchange); partner
+	// drives the PHASE-column span glyphs (┌/│/└) that visually bracket each
+	// exchange even when other events interleave between request and response.
+	ids, partner := computeEventPairs(eventRows)
+	glyphs := computeSpanGlyphs(partner, len(eventRows))
+
+	rows := make([]table.Row, 0, len(eventRows))
 	m.visibleRows = m.visibleRows[:0]
-	m.hiddenSkips = 0
-	var lastEvent *pipeline.SessionEvent // most-recent event already rendered (post-filter)
-	for i, rs := range rowSpecs {
-		if m.filter != "" && !matchInvocationRow(rs, m.filter) {
+	m.hiddenInactive = 0
+	for i, er := range eventRows {
+		ev := er.event
+		if m.filter != "" && !matchEventRow(er, m.filter) {
 			continue
 		}
-		// Hide plugin-didn't-act rows by default. Exception: if a skip
-		// row's pair partner (the other side of the req/resp span) is
-		// NOT a skip, keep this row visible — the partner's ┌/└ glyph
-		// would otherwise orphan, making a successful exchange read
-		// as a missing response. This is exactly what
-		// notifications/initialized produces: a request observe paired
-		// with a response skip ("no_response_body"). Both visible
-		// preserves the pair semantics; both hidden when both are
-		// skips keeps the noise down.
-		if !m.showSkips && isSkipRow(rs) {
-			partnerIdx, paired := pairs[i]
-			if !paired || isSkipRow(rowSpecs[partnerIdx]) {
-				m.hiddenSkips++
-				continue
-			}
-		}
-		// A "continuation" row is one whose event is the same as the
-		// previous RENDERED row's event (filtering-aware). We blank the
-		// event-level columns (#, TIME, DIR, PHASE, STATUS, DURATION,
-		// TOKENS, HOST) on continuation rows so an event's multi-plugin
-		// group reads as one visual block — only PLUGIN and ACTION vary.
-		// METHOD stays populated since a multi-plugin row set can still
-		// show per-plugin method context (e.g. a2a-parser observes
-		// message/stream while jwt-validation has no method at all).
-		continuation := lastEvent == rs.event
-
-		var idCell, timeCell, dirCell, phaseCell, statusC, durCell, tokC, hostC string
-		// Up to two levels of (req, resp) span nesting are surfaced as
-		// box-drawing prefixes in PHASE so operators can trace pairs
-		// even when other rows interleave. Width fits the 6-char PHASE
-		// budget: outer (1) + inner (1) + base phase ("resp"/"deny" =
-		// 4) → 6 max.
-		prefix := spanGlyphs[i].prefix()
-		// PHASE is always populated (prefix + "req"/"resp") so a quick
-		// scan down the column reveals lifecycle position even on
-		// continuation rows. Earlier the continuation case rendered
-		// only the prefix, which left operators wondering whether the
-		// row was a request or response invocation.
-		phaseCell = prefix + shortPhase(rs.event.Phase)
-		if !continuation {
-			if id, ok := eventIDs[rs.event]; ok {
-				idCell = strconv.Itoa(id)
-			}
-			timeCell = rs.event.At.Format("15:04:05.00")
-			dirCell = shortDirection(rs.event.Direction)
-			statusC = statusCell(*rs.event)
-			durCell = durationCell(*rs.event)
-			tokC = tokensCell(*rs.event)
-			hostC = truncStr(rs.event.Host, 20)
+		// hideInactive (the `s` toggle) is off by default — every message is
+		// shown, including passthrough/skip-only ones, per "I should see all
+		// network messages". Turning it on focuses the timeline on plugin
+		// activity (deny/modify/observe/allow). Both the filter and the
+		// headline consider the folded tunnel's invocations too.
+		invs := er.invocations()
+		if m.hideInactive && eventInactive(invs) {
+			m.hiddenInactive++
+			continue
 		}
 
+		action, plugin := eventAction(invs)
+		var idCell string
+		if id, ok := ids[ev]; ok {
+			idCell = strconv.Itoa(id)
+		}
+		// Prefix PHASE with the span glyph for this row's exchange. A request
+		// paired with a later response renders ┌; the response renders └;
+		// events nested between them render │ (with a second level when an
+		// inner exchange sits inside an outer one, e.g. inference calls inside
+		// an a2a message/stream). Unpaired rows get no prefix.
+		phaseCell := shortPhase(ev.Phase)
+		if p := glyphs[i].prefix(); p != "" {
+			phaseCell = p + " " + phaseCell
+		}
 		rows = append(rows, table.Row{
 			idCell,
-			timeCell,
-			dirCell,
+			ev.At.Format("15:04:05.00"),
+			shortDirection(ev.Direction),
 			phaseCell,
-			rs.actionCell(),
-			truncStr(rs.pluginCell(), 18),
-			eventMethod(*rs.event),
-			statusC,
-			durCell,
-			tokC,
-			hostC,
+			action,
+			truncStr(plugin, 18),
+			eventMethod(*ev),
+			statusCell(*ev),
+			durationCell(*ev),
+			tokensCell(*ev),
+			truncStr(ev.Host, 20),
 		})
-		m.visibleRows = append(m.visibleRows, rs)
-		lastEvent = rs.event
+		m.visibleRows = append(m.visibleRows, er)
 	}
 	m.eventsTbl.SetRows(rows)
 
@@ -171,209 +157,225 @@ func (m *model) rebuildEventsTable() {
 }
 
 // selectedEvent returns the event at the cursor row, or nil. The cursor
-// points into m.visibleRows (the flattened row list), and each row carries
-// a reference to its source event.
+// points into m.visibleRows (one entry per rendered message), and each row
+// carries a reference to its source event.
 func (m *model) selectedEvent() *pipeline.SessionEvent {
-	if len(m.visibleRows) == 0 {
+	er, ok := m.selectedEventRow()
+	if !ok {
 		return nil
 	}
-	cur := m.eventsTbl.Cursor()
-	if cur < 0 || cur >= len(m.visibleRows) {
-		return nil
-	}
-	return m.visibleRows[cur].event
+	return er.event
 }
 
-// selectedInvocation returns the plugin invocation for the highlighted events
-// row, or nil (pseudo-row for an event with no invocations, or no rows).
-func (m *model) selectedInvocation() *pipeline.Invocation {
+// selectedEventRow returns the full row (event + any folded tunnel) under the
+// cursor. ok is false when there are no rendered rows or the cursor is out of
+// range.
+func (m *model) selectedEventRow() (eventRow, bool) {
 	if len(m.visibleRows) == 0 {
-		return nil
+		return eventRow{}, false
 	}
 	cur := m.eventsTbl.Cursor()
 	if cur < 0 || cur >= len(m.visibleRows) {
-		return nil
-	}
-	return m.visibleRows[cur].inv
-}
-
-// invocationRow is one table row — the cartesian product of SessionEvent
-// × Invocation. An event with N plugin invocations produces N rows; an
-// event with no invocations produces one row with an empty invocation.
-// Rendering and filtering both work off this flat list.
-type invocationRow struct {
-	event *pipeline.SessionEvent
-	// inv may be nil when the event has no Invocation records. The
-	// pseudo-row still renders so the event is reachable in the table.
-	inv *pipeline.Invocation
-	// direction is the Invocations.{Inbound,Outbound} this row came
-	// from, disambiguating when a single event somehow carries both
-	// (doesn't happen today but cheap to be explicit).
-	direction pipeline.Direction
-}
-
-func (r invocationRow) actionCell() string {
-	if r.inv == nil {
-		return "—"
-	}
-	// Under on_error: observe the framework converts Reject to a
-	// pass-through and marks the Invocation Shadow=true. Prefix the
-	// action with an asterisk so operators scanning the timeline can
-	// spot would-have-blocked rows at a glance — the request actually
-	// passed. Width stays within the 8-char column budget (deny* fits,
-	// observe* fits, modify* fits at 7).
-	if r.inv.Shadow {
-		return string(r.inv.Action) + "*"
+		return eventRow{}, false
 	}
-	return string(r.inv.Action)
+	return m.visibleRows[cur], true
 }
 
-func (r invocationRow) pluginCell() string {
-	if r.inv == nil {
-		return "—"
+// buildEventRows turns the chronological event slice into display rows, one
+// per network message. The only folding is the TLS-bridge CONNECT pair: a
+// tunnel-open event (opaque, host:port) immediately followed by the decrypted
+// inner request for the same host is rendered as a single row keyed on the
+// inner request, with the tunnel attached for the detail pane. A non-bridged
+// (passthrough) tunnel has no inner request following it, so it stands as its
+// own row — it IS the whole message.
+func buildEventRows(events []pipeline.SessionEvent) []eventRow {
+	rows := make([]eventRow, 0, len(events))
+	for i := 0; i < len(events); i++ {
+		e := &events[i]
+		if i+1 < len(events) && isTunnelOpen(e) {
+			inner := &events[i+1]
+			if isBridgedInner(e, inner) {
+				rows = append(rows, eventRow{event: inner, tunnel: e})
+				i++ // consume the inner request too
+				continue
+			}
+		}
+		rows = append(rows, eventRow{event: e})
 	}
-	return r.inv.Plugin
+	return rows
 }
 
-// isSkipRow reports whether r represents a plugin that ran but didn't
-// act on the message (jwt-validation on a bypass path, IBAC on a bypass
-// host, token-exchange with no matching route). Operators usually want
-// these hidden so the timeline shows decisions, not non-events. The
-// `s` keybinding toggles visibility.
-func isSkipRow(r invocationRow) bool {
-	return r.inv != nil && r.inv.Action == pipeline.ActionSkip
+// isTunnelOpen reports whether e is a CONNECT / transparent-redirect
+// tunnel-open. It keys on the explicit Tunnel marker the producer
+// (recordTunnelOpened) sets — NOT on host/extension shape, which an ordinary
+// unparsed outbound request could mimic and get wrongly folded.
+func isTunnelOpen(e *pipeline.SessionEvent) bool {
+	return e.Tunnel
 }
 
-// spanGlyph names which corner / side of a (request, response) pair
-// span a row sits at, for tree-style rendering in the PHASE column.
-// rune (not byte) because the box-drawing characters are multi-byte
-// in UTF-8 and would overflow a byte.
-type spanGlyph rune
-
-const (
-	glyphNone   spanGlyph = 0
-	glyphStart  spanGlyph = '┌' // request row that pairs with a later response
-	glyphMiddle spanGlyph = '│' // row strictly between a paired request and its response
-	glyphEnd    spanGlyph = '└' // response row paired with an earlier request
-)
-
-// spanLevels holds the box-drawing glyphs for up to two levels of
-// nested (req, resp) spans on a single row. outer is the largest span
-// containing the row; inner is the next-largest. Deeper nesting is
-// not surfaced — operators only need the broad shape, not the full
-// nesting tree.
-type spanLevels struct {
-	outer spanGlyph
-	inner spanGlyph
+// isBridgedInner reports whether inner is the decrypted request the TLS bridge
+// produced right after opening tunnel. The two fold into one row: tunnel
+// carries "host:port" + opaque bytes, inner carries the same host (the Host
+// header, usually port-stripped) plus the real method/path. Matching on the
+// host-part (port-insensitive) handles both the standard-443 case (inner host
+// = "example.com") and the non-standard case (inner host = "example.com:8443").
+//
+// Two guards keep unrelated events from folding:
+//   - inner must be another outbound REQUEST, never a response, so a plain
+//     request→response exchange isn't mistaken for a bridged pair.
+//   - inner must NOT itself be a tunnel-open (Tunnel marker). Two back-to-back
+//     passthrough CONNECTs to the same host (common with connection pooling)
+//     would otherwise fold — hiding one real message and mislabeling the other.
+func isBridgedInner(tunnel, inner *pipeline.SessionEvent) bool {
+	host := hostOnly(inner.Host)
+	return inner.Direction == pipeline.Outbound &&
+		inner.Phase == pipeline.SessionRequest &&
+		!isTunnelOpen(inner) &&
+		host != "" &&
+		host == hostOnly(tunnel.Host)
 }
 
-// prefix returns the concatenated rune string for the PHASE-column
-// prefix: e.g. "│┌" when the row is inside an outer span and at the
-// start of an inner span; "└" alone when only an outer endpoint
-// applies; "" when the row is in no pair span.
-func (s spanLevels) prefix() string {
-	switch {
-	case s.outer == glyphNone:
-		return ""
-	case s.inner == glyphNone:
-		return string(rune(s.outer))
-	default:
-		return string([]rune{rune(s.outer), rune(s.inner)})
+// hostOnly returns the host portion of a "host:port" string, or the input
+// unchanged when it carries no port. Handles bracketed IPv6 literals.
+func hostOnly(hostport string) string {
+	if h, _, err := net.SplitHostPort(hostport); err == nil {
+		return h
 	}
+	return hostport
 }
 
-// computeSpanGlyphs assigns each row up to two tree glyphs (outer +
-// inner) drawn from its position relative to all (req, resp) spans in
-// the row list. The two largest spans containing the row are surfaced;
-// deeper nesting is dropped so the PHASE column doesn't blow its
-// 6-char width budget.
-//
-// pairs is the bidirectional map from pairInvocationRows: pairs[i]=j
-// AND pairs[j]=i for any matched pair (i, j). Unpaired rows are absent.
-// n is the total row count.
-func computeSpanGlyphs(pairs map[int]int, n int) []spanLevels {
-	out := make([]spanLevels, n)
-	if len(pairs) == 0 {
-		return out
-	}
-	// Collect each pair (a, b) with a < b once; the resp→req mirror
-	// entries are skipped.
-	type span struct{ a, b int }
-	spans := make([]span, 0, len(pairs)/2)
-	for a, b := range pairs {
-		if a < b {
-			spans = append(spans, span{a, b})
-		}
+// allInvocations returns every plugin invocation on an event, both
+// directions concatenated (inbound first). Returns nil when the event
+// carries no Invocations.
+func allInvocations(e *pipeline.SessionEvent) []pipeline.Invocation {
+	if e == nil || e.Invocations == nil {
+		return nil
 	}
+	out := make([]pipeline.Invocation, 0, len(e.Invocations.Inbound)+len(e.Invocations.Outbound))
+	out = append(out, e.Invocations.Inbound...)
+	out = append(out, e.Invocations.Outbound...)
+	return out
+}
 
-	glyphAt := func(s span, i int) spanGlyph {
-		switch {
-		case i == s.a:
-			return glyphStart
-		case i == s.b:
-			return glyphEnd
-		case s.a < i && i < s.b:
-			return glyphMiddle
-		}
-		return glyphNone
-	}
+// actionRank orders the five invocation verbs for at-a-glance aggregation,
+// highest wins: deny > modify > observe > allow > skip. The pipeline's own
+// outcome only distinguishes deny vs allow (pipeline/outcome.go); the rest of
+// this ordering is an abctl display choice. deny/modify rank top because they
+// changed the message's fate. observe ranks ABOVE allow on purpose: a parser
+// that understood the message (and supplied the METHOD shown on the row) tells
+// the operator more than a gate that merely permitted it — and surfacing the
+// gate while METHOD came from the parser reads as inconsistent. skip is the
+// floor (a plugin ran but didn't apply); 0 is the sentinel for "no plugin
+// acted".
+func actionRank(a pipeline.InvocationAction) int {
+	switch a {
+	case pipeline.ActionDeny:
+		return 5
+	case pipeline.ActionModify:
+		return 4
+	case pipeline.ActionObserve:
+		return 3
+	case pipeline.ActionAllow:
+		return 2
+	case pipeline.ActionSkip:
+		return 1
+	}
+	return 0
+}
 
-	for i := range n {
-		// Find every span this row participates in (endpoint or
-		// strictly inside).
-		var participating []span
-		for _, s := range spans {
-			if s.a <= i && i <= s.b {
-				participating = append(participating, s)
-			}
-		}
-		if len(participating) == 0 {
+// topInvocation returns the highest-ranked invocation satisfying keep and how
+// many tied at that top rank. winner is the zero Invocation (Action "", rank 0)
+// when none satisfy keep.
+func topInvocation(invs []pipeline.Invocation, keep func(pipeline.Invocation) bool) (pipeline.Invocation, int) {
+	best := -1
+	count := 0
+	var winner pipeline.Invocation
+	for _, iv := range invs {
+		if !keep(iv) {
 			continue
 		}
-		// Sort by width descending — outer (widest) first, then inner.
-		// Stable so equal-width spans keep their declaration order
-		// (irrelevant in practice but keeps tests deterministic).
-		sort.SliceStable(participating, func(p, q int) bool {
-			return (participating[p].b - participating[p].a) >
-				(participating[q].b - participating[q].a)
-		})
-		out[i].outer = glyphAt(participating[0], i)
-		if len(participating) > 1 {
-			out[i].inner = glyphAt(participating[1], i)
+		switch r := actionRank(iv.Action); {
+		case r > best:
+			best, count, winner = r, 1, iv
+		case r == best:
+			count++
 		}
 	}
-	return out
+	return winner, count
 }
 
-// flattenInvocations walks the event slice in order and, for each event,
-// emits one invocationRow per Invocation it carries (Inbound then
-// Outbound). Events with no Invocations fall back to a single pseudo-row
-// so parser-only events (a SessionEvent carrying just MCP or A2A with no
-// matching Invocation) remain reachable.
-func flattenInvocations(events []pipeline.SessionEvent) []invocationRow {
-	out := make([]invocationRow, 0, len(events))
-	for i := range events {
-		e := &events[i]
-		if e.Invocations == nil || (len(e.Invocations.Inbound) == 0 && len(e.Invocations.Outbound) == 0) {
-			out = append(out, invocationRow{event: e, direction: e.Direction})
-			continue
-		}
-		for j := range e.Invocations.Inbound {
-			out = append(out, invocationRow{
-				event:     e,
-				inv:       &e.Invocations.Inbound[j],
-				direction: pipeline.Inbound,
-			})
+// shadowFlagged reports whether any invocation is a shadow deny/modify — a
+// plugin that ran under on_error: observe and WOULD have blocked or rewritten
+// the message, but didn't (the framework converted its Reject to a pass and
+// set Shadow). These are the signal a shadow-policy rollout is watching for.
+func shadowFlagged(invs []pipeline.Invocation) bool {
+	for _, iv := range invs {
+		if iv.Shadow && (iv.Action == pipeline.ActionDeny || iv.Action == pipeline.ActionModify) {
+			return true
 		}
-		for j := range e.Invocations.Outbound {
-			out = append(out, invocationRow{
-				event:     e,
-				inv:       &e.Invocations.Outbound[j],
-				direction: pipeline.Outbound,
-			})
+	}
+	return false
+}
+
+// eventAction folds a message's per-plugin invocations into the single ACTION +
+// PLUGIN cell pair shown in the timeline. The headline reflects what actually
+// took effect:
+//
+//   - The winner is the highest-ranked ENFORCED (non-shadow) invocation —
+//     deny > modify > observe > allow > skip (see actionRank). A shadow
+//     deny/modify did NOT take effect (outcome.go enforces deny only when
+//     !Shadow), so it must not headline over the action that really applied.
+//     PLUGIN names that plugin, or "N plugins" when several tie at the top.
+//   - A trailing "*" flags that a shadow policy would have blocked/changed the
+//     message (e.g. "allow*"). Lets a rollout stay scannable without claiming
+//     a block that never happened.
+//   - When nothing enforced acted above a skip, a shadow deny/modify — if
+//     present — becomes the headline ("deny*"), since it's the only signal
+//     worth surfacing (a lone shadow deny on an otherwise-passthrough request).
+//   - Otherwise "—  —": nothing meaningful happened. A skip never credits a
+//     plugin (naming a skipper like "token-exchange" on an unrelated host
+//     reads as if it processed the message). Per-plugin detail is on drill-in.
+func eventAction(invs []pipeline.Invocation) (action, plugin string) {
+	winner, count := topInvocation(invs, func(iv pipeline.Invocation) bool { return !iv.Shadow })
+	shadow := shadowFlagged(invs)
+
+	if actionRank(winner.Action) > actionRank(pipeline.ActionSkip) {
+		action = string(winner.Action)
+		if shadow {
+			action += "*"
+		}
+		if count == 1 {
+			plugin = winner.Plugin
+		} else {
+			plugin = fmt.Sprintf("%d plugins", count)
+		}
+		return action, plugin
+	}
+
+	if shadow {
+		sh, _ := topInvocation(invs, func(iv pipeline.Invocation) bool { return iv.Shadow })
+		return string(sh.Action) + "*", sh.Plugin
+	}
+
+	return "—", "—"
+}
+
+// eventInactive reports whether no plugin took a meaningful action — either no
+// invocations at all (a pure passthrough / unprocessed message) or every
+// invocation was a skip (plugins ran but none matched). A shadow deny/modify
+// counts as activity (it flags a would-have-blocked message worth keeping
+// visible during a rollout). The `s` key hides inactive messages so an
+// operator can focus on plugin activity.
+func eventInactive(invs []pipeline.Invocation) bool {
+	if len(invs) == 0 {
+		return true
+	}
+	for _, iv := range invs {
+		if iv.Action != pipeline.ActionSkip {
+			return false
 		}
 	}
-	return out
+	return true
 }
 
 func shortDirection(d pipeline.Direction) string {
@@ -400,22 +402,28 @@ func shortPhase(p pipeline.SessionPhase) string {
 	return "?"
 }
 
-// (authCell and responsiblePlugin are gone — their roles moved onto
-// invocationRow's actionCell/pluginCell because each row now corresponds
-// to exactly one plugin's invocation rather than a whole event.)
-
-func eventMethod(e pipeline.SessionEvent) string {
+// eventMethodValue is the raw, untruncated method/model for an event — the A2A
+// method, inference model, or MCP method. Used for logic (pairing, filtering)
+// where truncation would conflate distinct names sharing a 22-char prefix or
+// hide searchable suffixes.
+func eventMethodValue(e pipeline.SessionEvent) string {
 	switch {
 	case e.A2A != nil:
-		return truncStr(e.A2A.Method, 22)
+		return e.A2A.Method
 	case e.Inference != nil:
-		return truncStr(e.Inference.Model, 22)
+		return e.Inference.Model
 	case e.MCP != nil:
-		return truncStr(e.MCP.Method, 22)
+		return e.MCP.Method
 	}
 	return ""
 }
 
+// eventMethod is the display form of the method/model — truncated to the
+// METHOD column width. Render-only; never compare or search on it.
+func eventMethod(e pipeline.SessionEvent) string {
+	return truncStr(eventMethodValue(e), 22)
+}
+
 func statusCell(e pipeline.SessionEvent) string {
 	if e.StatusCode == 0 {
 		return ""
@@ -455,90 +463,63 @@ func truncStr(s string, n int) string {
 	return s[:n-1] + "…"
 }
 
-// computeEventPairIDs assigns a small integer to every SessionEvent,
-// sharing one integer across a (request, response) pair and minting a new
-// one for unpaired events. The pairing decision is delegated to the row-
-// level pair map from pairInvocationRows — if any plugin row on event A
-// pairs with a plugin row on event B, then A and B pair at the event
-// level too. This keeps the # column's IDs consistent with the `└resp`
-// glyph the operator already sees (both derive from the same plugin-
-// level (direction, plugin) match).
+// computeEventPairs matches each response row to its request row and returns
+// two views of the result:
 //
-// Deriving from pairInvocationRows rather than recomputing by
-// direction+host+method avoids a class of bugs with "featureless"
-// requests (no parser matched, so method is empty): multiple concurrent
-// passthrough calls to the same host all share the same host+method
-// key and a naive matcher claims the wrong response.
+//   - ids: a small integer per event, shared across a (request, response)
+//     exchange and freshly minted for unpaired rows. Drives the # column.
+//   - partner: a bidirectional row-index map (partner[i]=j and partner[j]=i)
+//     for matched pairs. Drives the PHASE-column span glyphs.
 //
-// IDs are keyed by event pointer so render loops can look up a row's ID
-// without knowing the slice index. IDs start at 1 and increment in
-// first-seen row order so adjacent pairs get adjacent integers.
-func computeEventPairIDs(rowSpecs []invocationRow, pairs map[int]int) map[*pipeline.SessionEvent]int {
-	// Derive event-level pairs from row-level pairs. pairs is symmetric
-	// (pairs[i]=j and pairs[j]=i), so iterating either entry sets the
-	// map symmetrically. Last-write-wins when a single event pairs
-	// through multiple plugins, but in practice all plugin rows on one
-	// request event point at the same response event.
-	eventPair := make(map[*pipeline.SessionEvent]*pipeline.SessionEvent)
-	for i, j := range pairs {
-		ei, ej := rowSpecs[i].event, rowSpecs[j].event
-		if ei != ej {
-			eventPair[ei] = ej
-		}
-	}
-
-	// Event-level fallback for pairs the row-level matcher can't see.
-	// When a response event has no plugin invocations (e.g. a bypass
-	// path like /.well-known/agent.json — jwt-validation skipped on the
-	// request and no parser matched on the response), its pseudo-row
-	// has no (direction, plugin) key and pairInvocationRows leaves it
-	// unpaired. Scan the ordered event list and link each unpaired
-	// response to the closest preceding unpaired request with matching
-	// direction + host so the # column still reflects the pairing.
-	//
-	// Closest-preceding match is sufficient for bypass traffic where
-	// the response event immediately follows its request in the slice.
-	// Multiple concurrent bypass requests on the same host could
-	// theoretically cross-pair, but that's a near-simultaneous
-	// duplicate-path pattern we don't expect in real traffic.
-	orderedEvents := orderedUniqueEvents(rowSpecs)
-	for i, e := range orderedEvents {
-		if e.Phase != pipeline.SessionResponse {
-			continue
-		}
-		if _, paired := eventPair[e]; paired {
+// Each response row is matched to the closest preceding unpaired request row
+// sharing direction + host (port-normalized) + method. The method component
+// keeps a fire-and-forget request (e.g. MCP notifications/initialized, which
+// never gets a response) from stealing a later response that belongs to a
+// different method.
+//
+// Closest-preceding adjacency is sufficient for current traffic, where a
+// response follows its request. Concurrent same-host+method calls could in
+// principle cross-pair, but this is a navigational cue, not a correctness
+// guarantee; a server-side correlation id would be the fix if that ever bites.
+//
+// IDs are keyed by event pointer so the render loop can look one up without
+// knowing the row index. They start at 1 and increment in first-seen row order
+// so adjacent exchanges get adjacent integers.
+func computeEventPairs(rows []eventRow) (map[*pipeline.SessionEvent]int, map[int]int) {
+	partner := make(map[int]int) // row index → matched row index
+	for j := range rows {
+		rj := rows[j].event
+		if rj.Phase != pipeline.SessionResponse {
 			continue
 		}
-		for j := i - 1; j >= 0; j-- {
-			prev := orderedEvents[j]
-			if prev.Phase != pipeline.SessionRequest {
+		for i := j - 1; i >= 0; i-- {
+			if _, taken := partner[i]; taken {
 				continue
 			}
-			if _, already := eventPair[prev]; already {
+			ri := rows[i].event
+			if ri.Phase != pipeline.SessionRequest {
 				continue
 			}
-			if prev.Direction != e.Direction || prev.Host != e.Host {
+			if ri.Direction != rj.Direction ||
+				hostOnly(ri.Host) != hostOnly(rj.Host) ||
+				eventMethodValue(*ri) != eventMethodValue(*rj) {
 				continue
 			}
-			eventPair[e] = prev
-			eventPair[prev] = e
+			partner[i] = j
+			partner[j] = i
 			break
 		}
 	}
 
-	ids := make(map[*pipeline.SessionEvent]int)
-	seen := make(map[*pipeline.SessionEvent]bool)
+	ids := make(map[*pipeline.SessionEvent]int, len(rows))
 	next := 0
-	for _, rs := range rowSpecs {
-		e := rs.event
-		if seen[e] {
+	for i := range rows {
+		e := rows[i].event
+		if _, done := ids[e]; done {
 			continue
 		}
-		seen[e] = true
-		// If this event's paired partner has already been assigned an
-		// ID (partner appeared earlier in row order), reuse it.
-		if partner := eventPair[e]; partner != nil {
-			if pid, ok := ids[partner]; ok {
+		if p, ok := partner[i]; ok {
+			if pid, ok := ids[rows[p].event]; ok {
 				ids[e] = pid
 				continue
 			}
@@ -546,64 +527,182 @@ func computeEventPairIDs(rowSpecs []invocationRow, pairs map[int]int) map[*pipel
 		next++
 		ids[e] = next
 	}
-	return ids
+	return ids, partner
 }
 
-// orderedUniqueEvents returns distinct event pointers in the order they
-// first appear in rowSpecs. Used by computeEventPairIDs' event-level
-// fallback to walk events sequentially while looking backward for
-// unpaired request counterparts.
-func orderedUniqueEvents(rowSpecs []invocationRow) []*pipeline.SessionEvent {
-	seen := make(map[*pipeline.SessionEvent]bool, len(rowSpecs))
-	out := make([]*pipeline.SessionEvent, 0, len(rowSpecs))
-	for _, rs := range rowSpecs {
-		if seen[rs.event] {
+// spanGlyph names which corner / side of a (request, response) exchange a row
+// sits at, for the tree-style bracket in the PHASE column. rune (not byte)
+// because the box-drawing characters are multi-byte in UTF-8.
+type spanGlyph rune
+
+const (
+	glyphNone   spanGlyph = 0
+	glyphStart  spanGlyph = '┌' // request row that pairs with a later response
+	glyphMiddle spanGlyph = '│' // row between a paired request and its response
+	glyphEnd    spanGlyph = '└' // response row paired with an earlier request
+)
+
+// spanLevels holds the box-drawing glyphs for up to two nested exchanges on a
+// single row. outer is the widest exchange containing the row; inner is the
+// next-widest. Deeper nesting is dropped — operators only need the broad
+// shape, and the PHASE column has a finite width budget.
+type spanLevels struct {
+	outer spanGlyph
+	inner spanGlyph
+}
+
+// prefix returns the concatenated rune string for the PHASE-column prefix:
+// e.g. "│┌" when the row is inside an outer exchange and opens an inner one;
+// "└" alone when only an outer endpoint applies; "" when the row is in no
+// exchange span.
+func (s spanLevels) prefix() string {
+	switch {
+	case s.outer == glyphNone:
+		return ""
+	case s.inner == glyphNone:
+		return string(rune(s.outer))
+	default:
+		return string([]rune{rune(s.outer), rune(s.inner)})
+	}
+}
+
+// computeSpanGlyphs assigns each row up to two tree glyphs (outer + inner)
+// from its position relative to all (request, response) exchange spans. The
+// two widest spans containing the row are surfaced; deeper nesting is dropped
+// so the PHASE column doesn't blow its width budget.
+//
+// pairs is the bidirectional map from computeEventPairs: pairs[i]=j AND
+// pairs[j]=i for any matched pair (i, j). Unpaired rows are absent. n is the
+// total row count.
+func computeSpanGlyphs(pairs map[int]int, n int) []spanLevels {
+	out := make([]spanLevels, n)
+	if len(pairs) == 0 {
+		return out
+	}
+	// Collect each pair (a, b) with a < b once; the resp→req mirror entries
+	// are skipped.
+	type span struct{ a, b int }
+	spans := make([]span, 0, len(pairs)/2)
+	for a, b := range pairs {
+		if a < b {
+			spans = append(spans, span{a, b})
+		}
+	}
+
+	glyphAt := func(s span, i int) spanGlyph {
+		switch {
+		case i == s.a:
+			return glyphStart
+		case i == s.b:
+			return glyphEnd
+		case s.a < i && i < s.b:
+			return glyphMiddle
+		}
+		return glyphNone
+	}
+
+	for i := range n {
+		// Find every span this row participates in (endpoint or strictly
+		// inside).
+		var participating []span
+		for _, s := range spans {
+			if s.a <= i && i <= s.b {
+				participating = append(participating, s)
+			}
+		}
+		if len(participating) == 0 {
 			continue
 		}
-		seen[rs.event] = true
-		out = append(out, rs.event)
+		// Sort by width descending — widest first, narrowest last. Stable so
+		// equal-width spans keep declaration order (deterministic tests).
+		sort.SliceStable(participating, func(p, q int) bool {
+			return (participating[p].b - participating[p].a) >
+				(participating[q].b - participating[q].a)
+		})
+		// outer = the widest containing span (the broadest context). inner =
+		// the NARROWEST containing span — the row's own tightest exchange —
+		// NOT the second-widest. A row that is an endpoint of a deeply-nested
+		// pair must still show its ┌/└ corner so its request and response
+		// connect visually; picking the second-widest would let an
+		// intermediate enclosing span's middle bar mask it. Example: a
+		// tools/list pair nested inside both an a2a message/stream span and a
+		// long-lived $transport/stream span would otherwise render "││" on
+		// both rows instead of "│┌" / "│└".
+		out[i].outer = glyphAt(participating[0], i)
+		if len(participating) > 1 {
+			out[i].inner = glyphAt(participating[len(participating)-1], i)
+		}
 	}
 	return out
 }
 
-// matchInvocationRow does a case-insensitive substring match across every
-// string field the operator might reasonably search for — the invocation's
-// own fields plus the containing event's protocol extensions. Two prefix
-// shortcuts:
+// matchEventRow does a case-insensitive substring match across every string
+// field the operator might reasonably search for — the event's host/method,
+// the fields of every plugin invocation on it, and its protocol extensions.
+// A folded tunnel's fields are searched too, so filtering by a bridged
+// origin's host still surfaces the collapsed row. Two prefix shortcuts:
 //
-//   - `deny` alone matches SessionDenied events and any invocation
-//     whose Action == ActionDeny — the one-word "show me failures"
-//     filter.
-//   - `plugin:<name>` matches rows whose escape-hatch Plugins map on
-//     the parent event has <name> as a key.
-func matchInvocationRow(r invocationRow, q string) bool {
+//   - `deny` alone matches a SessionDenied event and any invocation whose
+//     Action == ActionDeny — the one-word "show me failures" filter.
+//   - `plugin:<name>` matches rows whose escape-hatch Plugins map has <name>
+//     as a key.
+func matchEventRow(r eventRow, q string) bool {
 	q = strings.ToLower(q)
 
 	if q == "deny" {
-		if r.event.Phase == pipeline.SessionDenied {
+		return eventMatchesDeny(r.event) || (r.tunnel != nil && eventMatchesDeny(r.tunnel))
+	}
+
+	if after, ok := strings.CutPrefix(q, "plugin:"); ok {
+		if _, present := r.event.Plugins[after]; present {
 			return true
 		}
-		if r.inv != nil && r.inv.Action == pipeline.ActionDeny {
-			return true
+		if r.tunnel != nil {
+			_, present := r.tunnel.Plugins[after]
+			return present
 		}
 		return false
 	}
 
-	if after, ok := strings.CutPrefix(q, "plugin:"); ok {
-		_, present := r.event.Plugins[after]
-		return present
+	hay := eventHaystack(r.event)
+	if r.tunnel != nil {
+		hay = append(hay, eventHaystack(r.tunnel)...)
 	}
+	for _, s := range hay {
+		if strings.Contains(strings.ToLower(s), q) {
+			return true
+		}
+	}
+	return false
+}
 
-	e := r.event
-	hay := []string{e.Host, eventMethod(*e)}
-	if r.inv != nil {
-		hay = append(hay,
-			r.inv.Plugin, string(r.inv.Action), r.inv.Reason, r.inv.Path)
+// eventMatchesDeny reports whether e is a deny — either the terminal
+// SessionDenied phase or any invocation with ActionDeny.
+func eventMatchesDeny(e *pipeline.SessionEvent) bool {
+	if e.Phase == pipeline.SessionDenied {
+		return true
+	}
+	for _, iv := range allInvocations(e) {
+		if iv.Action == pipeline.ActionDeny {
+			return true
+		}
+	}
+	return false
+}
+
+// eventHaystack collects every searchable string on an event: host, method,
+// each invocation's plugin/action/reason/path and detail key=values, the
+// caller identity, and protocol-specific content (A2A parts, MCP error, the
+// inference completion / finish reason).
+func eventHaystack(e *pipeline.SessionEvent) []string {
+	hay := []string{e.Host, eventMethodValue(*e)}
+	for _, iv := range allInvocations(e) {
+		hay = append(hay, iv.Plugin, string(iv.Action), iv.Reason, iv.Path)
 		// Plugin-specific diagnostic context — iterate keys + values so
 		// filter text matches on e.g. "target_audience" / the target
-		// audience value without the UI having to know which keys
-		// each plugin writes.
-		for k, v := range r.inv.Details {
+		// audience value without the UI having to know which keys each
+		// plugin writes.
+		for k, v := range iv.Details {
 			hay = append(hay, k, v)
 		}
 	}
@@ -622,70 +721,7 @@ func matchInvocationRow(r invocationRow, q string) bool {
 	if e.Inference != nil {
 		hay = append(hay, e.Inference.Completion, e.Inference.FinishReason)
 	}
-	for _, s := range hay {
-		if strings.Contains(strings.ToLower(s), q) {
-			return true
-		}
-	}
-	return false
-}
-
-// pairInvocationRows pairs request-phase rows with their response-phase
-// counterparts by (direction, plugin). Each plugin's contribution on the
-// request side connects to its own contribution on the response side,
-// independent of other plugins in the same pipeline — so a jwt-validation
-// request row pairs with a jwt-validation response row even when several
-// other plugins fired on the same event.
-//
-// Sequential pairing is good enough for current traffic: each request
-// row is paired with the NEXT response row that shares (direction, plugin)
-// and hasn't been claimed.
-func pairInvocationRows(rows []invocationRow) map[int]int {
-	pairs := make(map[int]int)
-	// Pair key includes plugin + direction + method (from whichever
-	// parser extension is populated). Without the method component,
-	// a fire-and-forget request like MCP's notifications/initialized
-	// would greedily claim the NEXT mcp-parser response — typically
-	// the response to tools/list — and orphan the actual tools/list
-	// request from its own response. Method discrimination makes the
-	// match specific: mcp-parser/out/tools/list only pairs with
-	// mcp-parser/out/tools/list. Auth plugins have no method; empty
-	// methods still pair with empty methods (same key), preserving
-	// pair behaviour for token-exchange and jwt-validation rows.
-	key := func(r invocationRow) (string, pipeline.Direction, bool) {
-		if r.inv == nil {
-			return "", r.direction, false
-		}
-		return r.inv.Plugin + "|" + eventMethod(*r.event), r.direction, true
-	}
-	for i := range rows {
-		if rows[i].event.Phase != pipeline.SessionRequest {
-			continue
-		}
-		if _, already := pairs[i]; already {
-			continue
-		}
-		k, dir, ok := key(rows[i])
-		if !ok {
-			continue
-		}
-		for j := i + 1; j < len(rows); j++ {
-			if rows[j].event.Phase != pipeline.SessionResponse {
-				continue
-			}
-			if _, taken := pairs[j]; taken {
-				continue
-			}
-			rk, rdir, rok := key(rows[j])
-			if !rok || rk != k || rdir != dir {
-				continue
-			}
-			pairs[i] = j
-			pairs[j] = i
-			break
-		}
-	}
-	return pairs
+	return hay
 }
 
 // identityBannerStyle renders the small bordered box above the events
diff --git a/authbridge/cmd/abctl/tui/events_pane_test.go b/authbridge/cmd/abctl/tui/events_pane_test.go
index 55e2aabe4..02855b35a 100644
--- a/authbridge/cmd/abctl/tui/events_pane_test.go
+++ b/authbridge/cmd/abctl/tui/events_pane_test.go
@@ -5,9 +5,54 @@ import (
 	"testing"
 	"time"
 
+	tea "github.com/charmbracelet/bubbletea"
+
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
 )
 
+// TestPageActivePane_EventsTable verifies PgDn/PgUp page the events table by a
+// near-full screen (one row of overlap), clamped to the row range — the lever
+// for sessions that now hold up to session.max_events (500) rows.
+func TestPageActivePane_EventsTable(t *testing.T) {
+	events := make([]pipeline.SessionEvent, 40)
+	for i := range events {
+		events[i] = pipeline.SessionEvent{
+			Direction: pipeline.Outbound, Phase: pipeline.SessionRequest,
+			Host: "h", Inference: &pipeline.InferenceExtension{Model: "m"},
+		}
+	}
+	m := &model{
+		pane: paneEvents, selectedSess: "s", bodyHeight: 12,
+		events: map[string][]pipeline.SessionEvent{"s": events},
+	}
+	m.eventsTbl = newEventsTable()
+	m.rebuildEventsTable()
+	m.eventsTbl.SetCursor(0)
+
+	h := m.eventsTbl.Height()
+	if h < 2 {
+		t.Fatalf("table height too small to page: %d", h)
+	}
+	m.pageActivePane(tea.KeyMsg{Type: tea.KeyPgDown})
+	if got := m.eventsTbl.Cursor(); got != h-1 {
+		t.Errorf("PgDn from top: cursor=%d, want %d", got, h-1)
+	}
+	m.pageActivePane(tea.KeyMsg{Type: tea.KeyPgUp})
+	if got := m.eventsTbl.Cursor(); got != 0 {
+		t.Errorf("PgUp back to top: cursor=%d, want 0", got)
+	}
+
+	// b/f (the keys shown in the footer, work on any keyboard) page the same way.
+	m.pageActivePane(tea.KeyMsg{Type: tea.KeyRunes, Runes: []rune("f")})
+	if got := m.eventsTbl.Cursor(); got != h-1 {
+		t.Errorf("f (page down) from top: cursor=%d, want %d", got, h-1)
+	}
+	m.pageActivePane(tea.KeyMsg{Type: tea.KeyRunes, Runes: []rune("b")})
+	if got := m.eventsTbl.Cursor(); got != 0 {
+		t.Errorf("b (page up) back to top: cursor=%d, want 0", got)
+	}
+}
+
 // TestShortPhase covers the rendered string for every SessionPhase.
 // SessionDenied renders as "req" (not "deny") because the deny
 // outcome is already on the row in the ACTION + STATUS columns —
@@ -29,406 +74,614 @@ func TestShortPhase(t *testing.T) {
 	}
 }
 
-// TestInvocationRow_Cells exercises the ACTION and PLUGIN column
-// renderers for each shape a row can take: an Invocation with an action,
-// multiple invocations (the row is per-invocation, each carries only
-// its own plugin), and the pseudo-row fallback when an event has no
-// Invocations at all.
-func TestInvocationRow_Cells(t *testing.T) {
-	evWithInv := &pipeline.SessionEvent{
-		Invocations: &pipeline.Invocations{
-			Inbound: []pipeline.Invocation{
-				{Plugin: "jwt-validation", Action: pipeline.ActionAllow},
-				{Plugin: "a2a-parser", Action: pipeline.ActionObserve},
-			},
-		},
+// TestEventAction_Precedence locks the per-event ACTION/PLUGIN aggregation.
+// One row per message: the winning action is the highest-ranked across all
+// invocations (deny > modify > observe > allow > skip), the PLUGIN cell names
+// the single responsible plugin, and a shadow deny gets the "*" suffix.
+func TestEventAction_Precedence(t *testing.T) {
+	ev := func(invs ...pipeline.Invocation) *pipeline.SessionEvent {
+		return &pipeline.SessionEvent{Invocations: &pipeline.Invocations{Inbound: invs}}
 	}
 	cases := []struct {
 		name       string
-		row        invocationRow
+		event      *pipeline.SessionEvent
 		wantAction string
 		wantPlugin string
 	}{
 		{
-			name:       "empty pseudo-row",
-			row:        invocationRow{event: &pipeline.SessionEvent{}},
+			name:       "no invocations → passthrough markers",
+			event:      &pipeline.SessionEvent{},
 			wantAction: "—",
 			wantPlugin: "—",
 		},
 		{
-			name: "inbound allow",
-			row: invocationRow{
-				event:     evWithInv,
-				inv:       &evWithInv.Invocations.Inbound[0],
-				direction: pipeline.Inbound,
-			},
-			wantAction: "allow",
+			name: "deny dominates observes",
+			event: ev(
+				pipeline.Invocation{Plugin: "a2a-parser", Action: pipeline.ActionObserve},
+				pipeline.Invocation{Plugin: "jwt-validation", Action: pipeline.ActionDeny},
+				pipeline.Invocation{Plugin: "mcp-parser", Action: pipeline.ActionObserve},
+			),
+			wantAction: "deny",
 			wantPlugin: "jwt-validation",
 		},
 		{
-			name: "inbound observe (parser)",
-			row: invocationRow{
-				event:     evWithInv,
-				inv:       &evWithInv.Invocations.Inbound[1],
-				direction: pipeline.Inbound,
-			},
+			name: "modify beats allow",
+			event: ev(
+				pipeline.Invocation{Plugin: "jwt-validation", Action: pipeline.ActionAllow},
+				pipeline.Invocation{Plugin: "token-exchange", Action: pipeline.ActionModify},
+			),
+			wantAction: "modify",
+			wantPlugin: "token-exchange",
+		},
+		{
+			// The real #23 case: a gate allows AND a parser observes. The
+			// parser (which supplied METHOD) is the informative headline, so
+			// observe must outrank allow.
+			name: "observe beats allow (parser over gate)",
+			event: ev(
+				pipeline.Invocation{Plugin: "jwt-validation", Action: pipeline.ActionAllow},
+				pipeline.Invocation{Plugin: "a2a-parser", Action: pipeline.ActionObserve},
+			),
 			wantAction: "observe",
 			wantPlugin: "a2a-parser",
 		},
 		{
-			name: "shadow deny (observe mode)",
-			row: invocationRow{
-				event: &pipeline.SessionEvent{},
-				inv: &pipeline.Invocation{
-					Plugin: "pii-scrubber",
-					Action: pipeline.ActionDeny,
-					Shadow: true,
-				},
-				direction: pipeline.Inbound,
-			},
-			// Asterisk suffix flags the would-have-blocked event so
-			// operators can visually separate real denies from shadow
-			// denies in the timeline.
+			// skip-only (plugins ran but none applied) reads as a passthrough:
+			// no plugin is credited, because naming a skipper (e.g.
+			// token-exchange on an unrelated host) implies it processed the
+			// message.
+			name: "all skip → passthrough markers (no plugin credited)",
+			event: ev(
+				pipeline.Invocation{Plugin: "jwt-validation", Action: pipeline.ActionSkip},
+				pipeline.Invocation{Plugin: "token-exchange", Action: pipeline.ActionSkip},
+			),
+			wantAction: "—",
+			wantPlugin: "—",
+		},
+		{
+			// A single skip (the row-30 www.example.org case) is also a
+			// passthrough — token-exchange skipped, nothing was done.
+			name: "single skip → passthrough markers",
+			event: ev(
+				pipeline.Invocation{Plugin: "token-exchange", Action: pipeline.ActionSkip},
+			),
+			wantAction: "—",
+			wantPlugin: "—",
+		},
+		{
+			name: "shadow deny gets asterisk",
+			event: ev(
+				pipeline.Invocation{Plugin: "pii-scrubber", Action: pipeline.ActionDeny, Shadow: true},
+			),
 			wantAction: "deny*",
 			wantPlugin: "pii-scrubber",
 		},
+		{
+			name: "single observe (parser-only)",
+			event: ev(
+				pipeline.Invocation{Plugin: "inference-parser", Action: pipeline.ActionObserve},
+			),
+			wantAction: "observe",
+			wantPlugin: "inference-parser",
+		},
+		{
+			// Shadow deny + a real allow: the deny ran under on_error: observe
+			// and did NOT block, so the headline must reflect the enforced
+			// allow (not "deny*"), with "*" flagging that a shadow policy
+			// would have blocked. The would-have-blocked decision is surfaced
+			// without claiming a block that never happened.
+			name: "shadow deny with enforced allow → allow*",
+			event: ev(
+				pipeline.Invocation{Plugin: "pii-scrubber", Action: pipeline.ActionDeny, Shadow: true},
+				pipeline.Invocation{Plugin: "jwt-validation", Action: pipeline.ActionAllow},
+			),
+			wantAction: "allow*",
+			wantPlugin: "jwt-validation",
+		},
+		{
+			// Shadow deny + a parser observe: enforced winner is observe; the
+			// shadow is flagged with "*".
+			name: "shadow deny with enforced observe → observe*",
+			event: ev(
+				pipeline.Invocation{Plugin: "pii-scrubber", Action: pipeline.ActionDeny, Shadow: true},
+				pipeline.Invocation{Plugin: "a2a-parser", Action: pipeline.ActionObserve},
+			),
+			wantAction: "observe*",
+			wantPlugin: "a2a-parser",
+		},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
-			if got := tc.row.actionCell(); got != tc.wantAction {
-				t.Errorf("actionCell = %q, want %q", got, tc.wantAction)
+			gotAction, gotPlugin := eventAction(allInvocations(tc.event))
+			if gotAction != tc.wantAction {
+				t.Errorf("action = %q, want %q", gotAction, tc.wantAction)
 			}
-			if got := tc.row.pluginCell(); got != tc.wantPlugin {
-				t.Errorf("pluginCell = %q, want %q", got, tc.wantPlugin)
+			if gotPlugin != tc.wantPlugin {
+				t.Errorf("plugin = %q, want %q", gotPlugin, tc.wantPlugin)
+			}
+		})
+	}
+}
+
+// TestEventAction_OutboundInvocations confirms outbound-direction invocations
+// participate in aggregation (forward-proxy events record on the Outbound
+// slot).
+func TestEventAction_OutboundInvocations(t *testing.T) {
+	ev := &pipeline.SessionEvent{
+		Direction: pipeline.Outbound,
+		Invocations: &pipeline.Invocations{
+			Outbound: []pipeline.Invocation{
+				{Plugin: "token-exchange", Action: pipeline.ActionModify},
+			},
+		},
+	}
+	action, plugin := eventAction(allInvocations(ev))
+	if action != "modify" || plugin != "token-exchange" {
+		t.Errorf("eventAction = (%q, %q), want (modify, token-exchange)", action, plugin)
+	}
+}
+
+// TestEventInactive covers the predicate the `s` toggle uses to hide
+// passthrough / skip-only messages. A message with no invocations (a
+// passthrough the operator can now see) and a message where every plugin
+// skipped are both inactive; any non-skip action makes it active.
+func TestEventInactive(t *testing.T) {
+	ev := func(invs ...pipeline.Invocation) *pipeline.SessionEvent {
+		return &pipeline.SessionEvent{Invocations: &pipeline.Invocations{Inbound: invs}}
+	}
+	cases := []struct {
+		name  string
+		event *pipeline.SessionEvent
+		want  bool
+	}{
+		{"no invocations (passthrough)", &pipeline.SessionEvent{}, true},
+		{"all skip", ev(
+			pipeline.Invocation{Plugin: "jwt-validation", Action: pipeline.ActionSkip},
+			pipeline.Invocation{Plugin: "token-exchange", Action: pipeline.ActionSkip},
+		), true},
+		{"one observe among skips", ev(
+			pipeline.Invocation{Plugin: "jwt-validation", Action: pipeline.ActionSkip},
+			pipeline.Invocation{Plugin: "a2a-parser", Action: pipeline.ActionObserve},
+		), false},
+		{"allow", ev(pipeline.Invocation{Plugin: "jwt-validation", Action: pipeline.ActionAllow}), false},
+		{"deny", ev(pipeline.Invocation{Plugin: "jwt-validation", Action: pipeline.ActionDeny}), false},
+		{"shadow deny counts as activity", ev(
+			pipeline.Invocation{Plugin: "pii-scrubber", Action: pipeline.ActionDeny, Shadow: true},
+		), false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := eventInactive(allInvocations(tc.event)); got != tc.want {
+				t.Errorf("eventInactive = %v, want %v", got, tc.want)
 			}
 		})
 	}
 }
 
-// TestFlattenInvocations covers the core expansion: an event with N
-// invocations should produce N rows; an event with zero invocations
-// should still produce one pseudo-row so the event stays reachable.
-func TestFlattenInvocations(t *testing.T) {
+// TestBuildEventRows_MultiPluginIsOneRow locks the core fix: a single message
+// touched by three plugins is exactly ONE display row (not three), and its
+// aggregate headlines the plugin that did the meaningful work (a2a-parser
+// observing) rather than the gate that allowed or the parser that skipped.
+func TestBuildEventRows_MultiPluginIsOneRow(t *testing.T) {
 	events := []pipeline.SessionEvent{
-		// 2 inbound invocations → 2 rows
 		{
 			Direction: pipeline.Inbound,
+			Phase:     pipeline.SessionRequest,
+			Host:      "weather-agent",
 			Invocations: &pipeline.Invocations{
 				Inbound: []pipeline.Invocation{
 					{Plugin: "jwt-validation", Action: pipeline.ActionAllow},
 					{Plugin: "a2a-parser", Action: pipeline.ActionObserve},
+					{Plugin: "mcp-parser", Action: pipeline.ActionSkip},
 				},
 			},
 		},
-		// 1 outbound invocation → 1 row
-		{
-			Direction: pipeline.Outbound,
-			Invocations: &pipeline.Invocations{
-				Outbound: []pipeline.Invocation{
-					{Plugin: "token-exchange", Action: pipeline.ActionSkip},
-				},
-			},
-		},
-		// no invocations → 1 pseudo-row
-		{Direction: pipeline.Inbound},
-	}
-	got := flattenInvocations(events)
-	if len(got) != 4 {
-		t.Fatalf("flattenInvocations returned %d rows, want 4", len(got))
 	}
-	if got[0].inv == nil || got[0].inv.Plugin != "jwt-validation" {
-		t.Errorf("row 0 = %+v, want jwt-validation", got[0])
+	rows := buildEventRows(events)
+	if len(rows) != 1 {
+		t.Fatalf("3-plugin message produced %d rows, want 1", len(rows))
 	}
-	if got[1].inv == nil || got[1].inv.Plugin != "a2a-parser" {
-		t.Errorf("row 1 = %+v, want a2a-parser", got[1])
+	if rows[0].tunnel != nil {
+		t.Errorf("row should have no folded tunnel")
 	}
-	if got[2].inv == nil || got[2].inv.Plugin != "token-exchange" {
-		t.Errorf("row 2 = %+v, want token-exchange", got[2])
+	action, plugin := eventAction(rows[0].invocations())
+	if action != "observe" {
+		t.Errorf("aggregate action = %q, want observe", action)
 	}
-	if got[3].inv != nil {
-		t.Errorf("row 3 should be pseudo-row with nil inv, got %+v", got[3])
+	if plugin != "a2a-parser" {
+		t.Errorf("aggregate plugin = %q, want a2a-parser", plugin)
 	}
 }
 
-// TestPairInvocationRows verifies that each plugin's request row pairs
-// with its own response row independently. A pipeline with
-// jwt-validation + a2a-parser on both request and response phases yields
-// 4 rows (2 req + 2 resp), and pairing should connect them in-plugin:
-// jwt-validation-req ↔ jwt-validation-resp; a2a-parser-req ↔
-// a2a-parser-resp.
-func TestPairInvocationRows(t *testing.T) {
-	inv := func(plugin string, action pipeline.InvocationAction) *pipeline.Invocation {
-		return &pipeline.Invocation{Plugin: plugin, Action: action}
-	}
-	reqEv := &pipeline.SessionEvent{Direction: pipeline.Inbound, Phase: pipeline.SessionRequest}
-	respEv := &pipeline.SessionEvent{Direction: pipeline.Inbound, Phase: pipeline.SessionResponse}
-	rows := []invocationRow{
-		{event: reqEv, inv: inv("jwt-validation", pipeline.ActionAllow), direction: pipeline.Inbound},
-		{event: reqEv, inv: inv("a2a-parser", pipeline.ActionObserve), direction: pipeline.Inbound},
-		{event: respEv, inv: inv("jwt-validation", pipeline.ActionAllow), direction: pipeline.Inbound},
-		{event: respEv, inv: inv("a2a-parser", pipeline.ActionObserve), direction: pipeline.Inbound},
-	}
-	pairs := pairInvocationRows(rows)
-	if pairs[0] != 2 || pairs[2] != 0 {
-		t.Errorf("expected jwt-validation pair 0↔2, got %v", pairs)
-	}
-	if pairs[1] != 3 || pairs[3] != 1 {
-		t.Errorf("expected a2a-parser pair 1↔3, got %v", pairs)
+// TestBuildEventRows_EmptyInvocationsResponse confirms a status-only response
+// that no plugin acted on (now recorded server-side) renders as one row
+// carrying its status.
+func TestBuildEventRows_EmptyInvocationsResponse(t *testing.T) {
+	events := []pipeline.SessionEvent{
+		{Direction: pipeline.Outbound, Phase: pipeline.SessionRequest, Host: "example.com"},
+		{Direction: pipeline.Outbound, Phase: pipeline.SessionResponse, Host: "example.com", StatusCode: 404},
+	}
+	rows := buildEventRows(events)
+	if len(rows) != 2 {
+		t.Fatalf("got %d rows, want 2", len(rows))
+	}
+	resp := rows[1].event
+	if got := statusCell(*resp); got != "404" {
+		t.Errorf("response statusCell = %q, want 404", got)
+	}
+	action, plugin := eventAction(allInvocations(resp))
+	if action != "—" || plugin != "—" {
+		t.Errorf("no-plugin response aggregate = (%q, %q), want (—, —)", action, plugin)
 	}
 }
 
-// TestMatchInvocationRow_DenyShortcut verifies that typing "deny" in the
-// filter box surfaces both the SessionDenied phase AND any invocation
-// whose Action is ActionDeny (jwt-validation or token-exchange
-// denials).
-func TestMatchInvocationRow_DenyShortcut(t *testing.T) {
-	denied := invocationRow{
-		event: &pipeline.SessionEvent{Phase: pipeline.SessionDenied},
+// TestBuildEventRows_CollapsesBridgedConnect locks the CONNECT-fold (Part C):
+// a tunnel-open (host:port, opaque) immediately followed by the decrypted
+// inner request (same host, real method) is ONE row keyed on the inner
+// request, with the tunnel attached. The inner response stays a separate row.
+func TestBuildEventRows_CollapsesBridgedConnect(t *testing.T) {
+	events := []pipeline.SessionEvent{
+		// CONNECT tunnel-open — opaque, host:port, gate invocation only.
+		{
+			Direction: pipeline.Outbound,
+			Phase:     pipeline.SessionRequest,
+			Host:      "api.anthropic.com:443",
+			Tunnel:    true,
+			Invocations: &pipeline.Invocations{
+				Outbound: []pipeline.Invocation{{Plugin: "jwt-validation", Action: pipeline.ActionSkip}},
+			},
+		},
+		// Decrypted inner request — real model, host without port.
+		{
+			Direction: pipeline.Outbound,
+			Phase:     pipeline.SessionRequest,
+			Host:      "api.anthropic.com",
+			Inference: &pipeline.InferenceExtension{Model: "claude-3-5-sonnet"},
+			Invocations: &pipeline.Invocations{
+				Outbound: []pipeline.Invocation{{Plugin: "inference-parser", Action: pipeline.ActionObserve}},
+			},
+		},
+		// Inner response.
+		{
+			Direction:  pipeline.Outbound,
+			Phase:      pipeline.SessionResponse,
+			Host:       "api.anthropic.com",
+			Inference:  &pipeline.InferenceExtension{Model: "claude-3-5-sonnet", TotalTokens: 1200},
+			StatusCode: 200,
+		},
 	}
-	if !matchInvocationRow(denied, "deny") {
-		t.Error("SessionDenied event should match the `deny` shortcut")
+	rows := buildEventRows(events)
+	if len(rows) != 2 {
+		t.Fatalf("bridged call produced %d rows, want 2 (collapsed req + resp)", len(rows))
 	}
-
-	inboundDeny := invocationRow{
-		event: &pipeline.SessionEvent{Phase: pipeline.SessionRequest},
-		inv:   &pipeline.Invocation{Action: pipeline.ActionDeny},
+	// Row 0: the inner request, with the CONNECT folded as tunnel.
+	if rows[0].event.Inference == nil || rows[0].event.Inference.Model != "claude-3-5-sonnet" {
+		t.Errorf("row 0 should be the decrypted inner request")
 	}
-	if !matchInvocationRow(inboundDeny, "deny") {
-		t.Error("inbound-deny invocation should match the `deny` shortcut")
+	if rows[0].tunnel == nil {
+		t.Fatalf("row 0 should carry the folded CONNECT tunnel")
 	}
-
-	clean := invocationRow{
-		event: &pipeline.SessionEvent{Phase: pipeline.SessionRequest},
-		inv:   &pipeline.Invocation{Action: pipeline.ActionAllow},
+	if rows[0].tunnel.Host != "api.anthropic.com:443" {
+		t.Errorf("folded tunnel host = %q, want api.anthropic.com:443", rows[0].tunnel.Host)
 	}
-	if matchInvocationRow(clean, "deny") {
-		t.Error("allow invocation should NOT match the `deny` shortcut")
+	// Row 0 host should be the clean inner host (no :443).
+	if rows[0].event.Host != "api.anthropic.com" {
+		t.Errorf("collapsed row host = %q, want api.anthropic.com", rows[0].event.Host)
+	}
+	// Row 1: the response, no tunnel.
+	if rows[1].event.Phase != pipeline.SessionResponse || rows[1].tunnel != nil {
+		t.Errorf("row 1 should be the standalone inner response")
 	}
 }
 
-// TestMatchInvocationRow_PluginSubstring verifies that filtering by plugin
-// name substring-matches against the Invocation.Plugin field so operators
-// can isolate one plugin's rows.
-func TestMatchInvocationRow_PluginSubstring(t *testing.T) {
-	row := invocationRow{
-		event: &pipeline.SessionEvent{Phase: pipeline.SessionRequest},
-		inv:   &pipeline.Invocation{Plugin: "jwt-validation", Action: pipeline.ActionSkip, Reason: "path_bypass", Path: "/healthz"},
-	}
-	if !matchInvocationRow(row, "jwt-validation") {
-		t.Error("filter jwt-validation should match")
-	}
-	if !matchInvocationRow(row, "path_bypass") {
-		t.Error("filter by reason should match")
+// TestBuildEventRows_PassthroughTunnelStandsAlone covers every shape in which a
+// non-bridged CONNECT tunnel-open must remain its own row rather than fold:
+// a lone trailing CONNECT, two back-to-back CONNECTs to the SAME host
+// (connection pooling — must NOT fold into each other), and a CONNECT followed
+// by a different-host request (host-mismatch guard).
+func TestBuildEventRows_PassthroughTunnelStandsAlone(t *testing.T) {
+	connect := func(host string) pipeline.SessionEvent {
+		return pipeline.SessionEvent{Direction: pipeline.Outbound, Phase: pipeline.SessionRequest, Host: host, Tunnel: true}
 	}
-	if !matchInvocationRow(row, "/healthz") {
-		t.Error("filter by path should match")
+	cases := []struct {
+		name   string
+		events []pipeline.SessionEvent
+	}{
+		{
+			name:   "lone trailing CONNECT",
+			events: []pipeline.SessionEvent{connect("passthrough.example:443")},
+		},
+		{
+			name: "two CONNECTs to the same host (pooling) must not fold",
+			events: []pipeline.SessionEvent{
+				connect("api.example.com:443"),
+				connect("api.example.com:443"),
+			},
+		},
+		{
+			name: "CONNECT then different-host request",
+			events: []pipeline.SessionEvent{
+				connect("passthrough.example:443"),
+				{Direction: pipeline.Outbound, Phase: pipeline.SessionRequest, Host: "other.example",
+					Inference: &pipeline.InferenceExtension{Model: "x"}},
+			},
+		},
 	}
-	if matchInvocationRow(row, "token-exchange") {
-		t.Error("filter token-exchange should NOT match a jwt-validation row")
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			rows := buildEventRows(tc.events)
+			if len(rows) != len(tc.events) {
+				t.Fatalf("got %d rows, want %d (nothing should fold)", len(rows), len(tc.events))
+			}
+			for i, r := range rows {
+				if r.tunnel != nil {
+					t.Errorf("row %d unexpectedly folded a tunnel", i)
+				}
+			}
+		})
 	}
 }
 
-// TestMatchInvocationRow_PluginPrefix tests the `plugin:<name>` escape-
-// hatch filter — matches when the event's Plugins map contains <name>.
-func TestMatchInvocationRow_PluginPrefix(t *testing.T) {
-	row := invocationRow{
-		event: &pipeline.SessionEvent{
-			Plugins: map[string]json.RawMessage{
-				"rate-limiter": json.RawMessage(`{"allowed":true}`),
-			},
+// TestBuildEventRows_TunnelInvocationsFoldIntoRow locks #3: a bridged row's
+// ACTION/inactive view must include the folded CONNECT tunnel's gate
+// invocations, so an egress gate that ALLOWED the CONNECT is reflected even
+// when the decrypted inner request itself saw no plugin activity.
+func TestBuildEventRows_TunnelInvocationsFoldIntoRow(t *testing.T) {
+	events := []pipeline.SessionEvent{
+		// CONNECT tunnel-open — an egress gate explicitly ALLOWED it.
+		{
+			Direction: pipeline.Outbound, Phase: pipeline.SessionRequest, Host: "api.example.com:443", Tunnel: true,
+			Invocations: &pipeline.Invocations{Outbound: []pipeline.Invocation{
+				{Plugin: "egress-policy", Action: pipeline.ActionAllow},
+			}},
 		},
+		// Decrypted inner request — no plugin matched (opaque passthrough body).
+		{Direction: pipeline.Outbound, Phase: pipeline.SessionRequest, Host: "api.example.com"},
 	}
-	if !matchInvocationRow(row, "plugin:rate-limiter") {
-		t.Error("expected match on plugin:rate-limiter")
+	rows := buildEventRows(events)
+	if len(rows) != 1 || rows[0].tunnel == nil {
+		t.Fatalf("expected 1 folded row, got %d (tunnel=%v)", len(rows), rows[0].tunnel != nil)
 	}
-	if matchInvocationRow(row, "plugin:nonexistent") {
-		t.Error("expected no match for a plugin not in the map")
+	// The inner event alone has no invocations; the row must surface the
+	// tunnel's allow.
+	if eventInactive(rows[0].invocations()) {
+		t.Error("row with a tunnel-level allow should not be inactive")
+	}
+	action, plugin := eventAction(rows[0].invocations())
+	if action != "allow" || plugin != "egress-policy" {
+		t.Errorf("folded ACTION = (%q, %q), want (allow, egress-policy)", action, plugin)
 	}
 }
 
-// TestComputeEventPairIDs_BypassResponseWithEmptyInvocations locks the
-// event-level fallback pairing: when a response event has no plugin
-// invocations at all (e.g. jwt-validation bypass response), it should
-// still pair with its preceding request event via direction+host match
-// so the # column shows the same ID on both rows.
-func TestComputeEventPairIDs_BypassResponseWithEmptyInvocations(t *testing.T) {
+// TestRebuildEventsTable_HideInactive is the integration check for #5: the
+// hideInactive toggle (predicate → row build → footer count) suppresses
+// passthrough/skip-only messages while keeping partially-active ones.
+func TestRebuildEventsTable_HideInactive(t *testing.T) {
 	events := []pipeline.SessionEvent{
-		// Event 0: bypass req — jwt-validation skip invocation
-		{
-			Direction: pipeline.Inbound,
-			Phase:     pipeline.SessionRequest,
-			Invocations: &pipeline.Invocations{Inbound: []pipeline.Invocation{{
-				Plugin: "jwt-validation",
-				Phase:  pipeline.InvocationPhaseRequest,
-				Action: pipeline.ActionSkip,
+		// active — a2a observe
+		{Direction: pipeline.Inbound, Phase: pipeline.SessionRequest, Host: "agent",
+			Invocations: &pipeline.Invocations{Inbound: []pipeline.Invocation{
+				{Plugin: "jwt-validation", Action: pipeline.ActionSkip}, // partially-active: skip + observe
+				{Plugin: "a2a-parser", Action: pipeline.ActionObserve},
 			}}},
-		},
-		// Event 1: bypass resp — no invocations (response-phase filter returns empty)
-		{Direction: pipeline.Inbound, Phase: pipeline.SessionResponse, StatusCode: 200},
-		// Event 2: bypass req (different bypass path, same direction+host="")
-		{
-			Direction: pipeline.Inbound,
-			Phase:     pipeline.SessionRequest,
-			Invocations: &pipeline.Invocations{Inbound: []pipeline.Invocation{{
-				Plugin: "jwt-validation",
-				Phase:  pipeline.InvocationPhaseRequest,
-				Action: pipeline.ActionSkip,
+		// inactive — no invocations (passthrough response)
+		{Direction: pipeline.Outbound, Phase: pipeline.SessionResponse, Host: "x", StatusCode: 200},
+		// inactive — skip-only
+		{Direction: pipeline.Outbound, Phase: pipeline.SessionRequest, Host: "y",
+			Invocations: &pipeline.Invocations{Outbound: []pipeline.Invocation{
+				{Plugin: "token-exchange", Action: pipeline.ActionSkip},
 			}}},
-		},
-		// Event 3: bypass resp
-		{Direction: pipeline.Inbound, Phase: pipeline.SessionResponse, StatusCode: 200},
+	}
+	m := &model{selectedSess: "s", events: map[string][]pipeline.SessionEvent{"s": events}}
+	m.eventsTbl = newEventsTable()
+
+	m.hideInactive = false
+	m.rebuildEventsTable()
+	if len(m.visibleRows) != 3 || m.hiddenInactive != 0 {
+		t.Fatalf("show-all: rows=%d hidden=%d, want 3/0", len(m.visibleRows), m.hiddenInactive)
 	}
 
-	rows := flattenInvocations(events)
-	pairs := pairInvocationRows(rows)
-	ids := computeEventPairIDs(rows, pairs)
+	m.hideInactive = true
+	m.rebuildEventsTable()
+	if len(m.visibleRows) != 1 || m.hiddenInactive != 2 {
+		t.Fatalf("hide: rows=%d hidden=%d, want 1/2", len(m.visibleRows), m.hiddenInactive)
+	}
+	// The surviving row is the partially-active a2a message.
+	if action, _ := eventAction(m.visibleRows[0].invocations()); action != "observe" {
+		t.Errorf("surviving row action = %q, want observe", action)
+	}
+}
 
-	id0, id1 := ids[&events[0]], ids[&events[1]]
-	id2, id3 := ids[&events[2]], ids[&events[3]]
+// TestComputeEventPairIDs pairs each response row with its preceding request
+// row by direction + host + method, sharing one # across the exchange and
+// minting fresh integers for unpaired rows.
+func TestComputeEventPairIDs(t *testing.T) {
+	events := []pipeline.SessionEvent{
+		{Direction: pipeline.Inbound, Phase: pipeline.SessionRequest, Host: "weather-agent"},
+		{Direction: pipeline.Inbound, Phase: pipeline.SessionResponse, Host: "weather-agent", StatusCode: 200},
+		{Direction: pipeline.Outbound, Phase: pipeline.SessionRequest, Host: "tool"},
+		{Direction: pipeline.Outbound, Phase: pipeline.SessionResponse, Host: "tool", StatusCode: 200},
+	}
+	rows := buildEventRows(events)
+	ids, _ := computeEventPairs(rows)
 
-	if id0 != id1 {
-		t.Errorf("bypass req/resp #1: got ids (%d,%d), want equal", id0, id1)
+	if ids[&events[0]] != ids[&events[1]] {
+		t.Errorf("inbound req/resp should share id, got %d vs %d", ids[&events[0]], ids[&events[1]])
 	}
-	if id2 != id3 {
-		t.Errorf("bypass req/resp #2: got ids (%d,%d), want equal", id2, id3)
+	if ids[&events[2]] != ids[&events[3]] {
+		t.Errorf("outbound req/resp should share id, got %d vs %d", ids[&events[2]], ids[&events[3]])
 	}
-	if id0 == id2 {
-		t.Errorf("different bypass pairs should have different ids, both got %d", id0)
+	if ids[&events[0]] == ids[&events[2]] {
+		t.Errorf("distinct exchanges should have distinct ids, both got %d", ids[&events[0]])
 	}
 }
 
-// TestPairInvocationRows_MethodDiscrimination locks the method-aware
-// pairing. Fire-and-forget MCP methods (notifications/initialized) have
-// no response; a subsequent tools/list req+resp pair must not be
-// disrupted by the notification's mcp-parser row greedily claiming the
-// tools/list response row.
-func TestPairInvocationRows_MethodDiscrimination(t *testing.T) {
+// TestComputeEventPairIDs_MethodDiscrimination locks method-aware pairing: a
+// fire-and-forget request (MCP notifications/initialized, no response) must
+// not steal the response that belongs to a later tools/list request.
+func TestComputeEventPairIDs_MethodDiscrimination(t *testing.T) {
 	mk := func(phase pipeline.SessionPhase, method string) pipeline.SessionEvent {
 		return pipeline.SessionEvent{
 			Direction: pipeline.Outbound,
 			Phase:     phase,
+			Host:      "tool",
 			MCP:       &pipeline.MCPExtension{Method: method},
-			Invocations: &pipeline.Invocations{Outbound: []pipeline.Invocation{{
-				Plugin: "mcp-parser",
-				Phase:  invocationPhaseFor(phase),
-				Action: pipeline.ActionObserve,
-			}}},
 		}
 	}
 	events := []pipeline.SessionEvent{
-		mk(pipeline.SessionRequest, "notifications/initialized"), // no resp (fire and forget)
+		mk(pipeline.SessionRequest, "notifications/initialized"), // no response (fire and forget)
 		mk(pipeline.SessionRequest, "tools/list"),
 		mk(pipeline.SessionResponse, "tools/list"),
 	}
-	rows := flattenInvocations(events)
-	pairs := pairInvocationRows(rows)
-	ids := computeEventPairIDs(rows, pairs)
+	rows := buildEventRows(events)
+	ids, _ := computeEventPairs(rows)
 
 	if ids[&events[1]] != ids[&events[2]] {
-		t.Errorf("tools/list req and resp must share ID, got %d vs %d",
-			ids[&events[1]], ids[&events[2]])
+		t.Errorf("tools/list req and resp must share id, got %d vs %d", ids[&events[1]], ids[&events[2]])
 	}
 	if ids[&events[0]] == ids[&events[1]] {
-		t.Errorf("notifications/initialized (orphan) must not share ID with tools/list, both got %d",
-			ids[&events[0]])
+		t.Errorf("notifications/initialized must not share id with tools/list, both got %d", ids[&events[0]])
 	}
 }
 
-func invocationPhaseFor(p pipeline.SessionPhase) pipeline.InvocationPhase {
-	if p == pipeline.SessionResponse {
-		return pipeline.InvocationPhaseResponse
+// TestMatchEventRow_DenyShortcut verifies that typing "deny" surfaces both the
+// SessionDenied phase AND any invocation whose Action is ActionDeny.
+func TestMatchEventRow_DenyShortcut(t *testing.T) {
+	denied := eventRow{event: &pipeline.SessionEvent{Phase: pipeline.SessionDenied}}
+	if !matchEventRow(denied, "deny") {
+		t.Error("SessionDenied event should match the `deny` shortcut")
+	}
+
+	inboundDeny := eventRow{event: &pipeline.SessionEvent{
+		Phase:       pipeline.SessionRequest,
+		Invocations: &pipeline.Invocations{Inbound: []pipeline.Invocation{{Action: pipeline.ActionDeny}}},
+	}}
+	if !matchEventRow(inboundDeny, "deny") {
+		t.Error("event with a deny invocation should match the `deny` shortcut")
+	}
+
+	clean := eventRow{event: &pipeline.SessionEvent{
+		Phase:       pipeline.SessionRequest,
+		Invocations: &pipeline.Invocations{Inbound: []pipeline.Invocation{{Action: pipeline.ActionAllow}}},
+	}}
+	if matchEventRow(clean, "deny") {
+		t.Error("allow-only event should NOT match the `deny` shortcut")
 	}
-	return pipeline.InvocationPhaseRequest
 }
 
-// Build a realistic auth-only request/response pair and assert that the
-// flatten → pair pipeline connects them end-to-end. Regression-protects
-// the chart-default case (jwt-validation only, no parsers).
-func TestFlattenPair_AuthOnlyEndToEnd(t *testing.T) {
-	now := time.Date(2026, 5, 8, 14, 22, 5, 0, time.UTC)
-	invs := &pipeline.Invocations{Inbound: []pipeline.Invocation{{Plugin: "jwt-validation", Action: pipeline.ActionAllow}}}
-	events := []pipeline.SessionEvent{
-		{At: now, Direction: pipeline.Inbound, Phase: pipeline.SessionRequest, Invocations: invs, Host: "weather-agent"},
-		{At: now.Add(12 * time.Millisecond), Direction: pipeline.Inbound, Phase: pipeline.SessionResponse, Invocations: invs, Host: "weather-agent", StatusCode: 200, Duration: 12 * time.Millisecond},
+// TestMatchEventRow_PluginSubstring verifies substring matching across an
+// event's invocation fields (plugin name, reason, path).
+func TestMatchEventRow_PluginSubstring(t *testing.T) {
+	row := eventRow{event: &pipeline.SessionEvent{
+		Phase: pipeline.SessionRequest,
+		Invocations: &pipeline.Invocations{Inbound: []pipeline.Invocation{
+			{Plugin: "jwt-validation", Action: pipeline.ActionSkip, Reason: "path_bypass", Path: "/healthz"},
+		}},
+	}}
+	if !matchEventRow(row, "jwt-validation") {
+		t.Error("filter jwt-validation should match")
+	}
+	if !matchEventRow(row, "path_bypass") {
+		t.Error("filter by reason should match")
+	}
+	if !matchEventRow(row, "/healthz") {
+		t.Error("filter by path should match")
 	}
+	if matchEventRow(row, "token-exchange") {
+		t.Error("filter token-exchange should NOT match a jwt-validation-only event")
+	}
+}
 
-	rows := flattenInvocations(events)
-	if len(rows) != 2 {
-		t.Fatalf("expected 2 rows, got %d", len(rows))
+// TestMatchEventRow_TunnelFields confirms a folded tunnel's fields are
+// searchable on the collapsed row — filtering by the bridged origin's
+// host:port still surfaces the row even though the row's own host is
+// port-stripped.
+func TestMatchEventRow_TunnelFields(t *testing.T) {
+	row := eventRow{
+		event:  &pipeline.SessionEvent{Host: "api.anthropic.com"},
+		tunnel: &pipeline.SessionEvent{Host: "api.anthropic.com:443"},
+	}
+	if !matchEventRow(row, "anthropic.com:443") {
+		t.Error("filter on the tunnel host:port should match the collapsed row")
 	}
-	pairs := pairInvocationRows(rows)
-	if pairs[0] != 1 || pairs[1] != 0 {
-		t.Errorf("expected auth-only req/resp to pair: got %v", pairs)
+}
+
+// TestMatchEventRow_PluginPrefix tests the `plugin:<name>` escape-hatch filter
+// against the event's Plugins map.
+func TestMatchEventRow_PluginPrefix(t *testing.T) {
+	row := eventRow{event: &pipeline.SessionEvent{
+		Plugins: map[string]json.RawMessage{
+			"rate-limiter": json.RawMessage(`{"allowed":true}`),
+		},
+	}}
+	if !matchEventRow(row, "plugin:rate-limiter") {
+		t.Error("expected match on plugin:rate-limiter")
 	}
-	if got := rows[0].actionCell(); got != "allow" {
-		t.Errorf("req actionCell = %q, want allow", got)
+	if matchEventRow(row, "plugin:nonexistent") {
+		t.Error("expected no match for a plugin not in the map")
 	}
-	if got := rows[1].pluginCell(); got != "jwt-validation" {
-		t.Errorf("resp pluginCell = %q, want jwt-validation", got)
+}
+
+// TestStatusCell exercises the realistic auth-only request/response shape:
+// a response carrying a 200 renders that status; a request renders blank.
+func TestStatusCell(t *testing.T) {
+	now := time.Date(2026, 5, 8, 14, 22, 5, 0, time.UTC)
+	req := pipeline.SessionEvent{At: now, Phase: pipeline.SessionRequest, Host: "weather-agent"}
+	resp := pipeline.SessionEvent{At: now.Add(12 * time.Millisecond), Phase: pipeline.SessionResponse,
+		Host: "weather-agent", StatusCode: 200, Duration: 12 * time.Millisecond}
+	if got := statusCell(req); got != "" {
+		t.Errorf("request statusCell = %q, want empty", got)
+	}
+	if got := statusCell(resp); got != "200" {
+		t.Errorf("response statusCell = %q, want 200", got)
 	}
-	if got := statusCell(*rows[1].event); got != "200" {
-		t.Errorf("statusCell = %q, want 200", got)
+	if got := durationCell(resp); got != "12ms" {
+		t.Errorf("durationCell = %q, want 12ms", got)
 	}
 }
 
-// TestIsSkipRow checks the predicate the events table uses to decide
-// whether a row should be hidden under the default skip-hiding mode.
-// Only Action=skip rows return true; the pseudo-row with no
-// Invocation, and rows for any other action, are kept.
-func TestIsSkipRow(t *testing.T) {
+// TestHostOnly covers port stripping (used by collapse + pairing), including
+// the no-port and IPv6 cases.
+func TestHostOnly(t *testing.T) {
+	cases := []struct{ in, want string }{
+		{"example.com:443", "example.com"},
+		{"example.com", "example.com"},
+		{"[::1]:8443", "::1"},
+	}
+	for _, tc := range cases {
+		if got := hostOnly(tc.in); got != tc.want {
+			t.Errorf("hostOnly(%q) = %q, want %q", tc.in, got, tc.want)
+		}
+	}
+}
+
+// TestSpanLevels_Prefix locks the PHASE-column prefix: empty levels render as
+// empty string; one level renders one glyph; two levels render two glyphs.
+func TestSpanLevels_Prefix(t *testing.T) {
 	cases := []struct {
 		name string
-		row  invocationRow
-		want bool
+		s    spanLevels
+		want string
 	}{
-		{
-			name: "skip action",
-			row:  invocationRow{inv: &pipeline.Invocation{Action: pipeline.ActionSkip}},
-			want: true,
-		},
-		{
-			name: "allow action",
-			row:  invocationRow{inv: &pipeline.Invocation{Action: pipeline.ActionAllow}},
-			want: false,
-		},
-		{
-			name: "deny action",
-			row:  invocationRow{inv: &pipeline.Invocation{Action: pipeline.ActionDeny}},
-			want: false,
-		},
-		{
-			name: "observe action",
-			row:  invocationRow{inv: &pipeline.Invocation{Action: pipeline.ActionObserve}},
-			want: false,
-		},
-		{
-			name: "modify action",
-			row:  invocationRow{inv: &pipeline.Invocation{Action: pipeline.ActionModify}},
-			want: false,
-		},
-		{
-			name: "pseudo-row with no invocation",
-			// Parser-only events emit a pseudo-row so they remain
-			// reachable. These should never be classified as skips —
-			// hiding them would lose protocol-only events from the
-			// timeline.
-			row:  invocationRow{event: &pipeline.SessionEvent{}, inv: nil},
-			want: false,
-		},
+		{"none", spanLevels{}, ""},
+		{"outer only — start", spanLevels{outer: glyphStart}, "┌"},
+		{"outer only — middle", spanLevels{outer: glyphMiddle}, "│"},
+		{"outer only — end", spanLevels{outer: glyphEnd}, "└"},
+		{"both — outer middle, inner start", spanLevels{outer: glyphMiddle, inner: glyphStart}, "│┌"},
+		{"both — outer middle, inner end", spanLevels{outer: glyphMiddle, inner: glyphEnd}, "│└"},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
-			if got := isSkipRow(tc.row); got != tc.want {
-				t.Errorf("isSkipRow = %v, want %v", got, tc.want)
+			if got := tc.s.prefix(); got != tc.want {
+				t.Errorf("prefix() = %q, want %q", got, tc.want)
 			}
 		})
 	}
 }
 
-// TestComputeSpanGlyphs covers per-row tree-glyph assignment for the
-// PHASE column. Up to two levels of (req, resp) span nesting are
-// surfaced — the largest containing span as outer, the next-largest
-// as inner, deeper levels dropped.
+// TestComputeSpanGlyphs covers per-row tree-glyph assignment for the PHASE
+// column. Up to two levels of (request, response) nesting are surfaced — the
+// widest containing span as outer, the next-widest as inner, deeper dropped.
 func TestComputeSpanGlyphs(t *testing.T) {
 	none := spanLevels{}
 	outer := func(g spanGlyph) spanLevels { return spanLevels{outer: g} }
@@ -440,123 +693,71 @@ func TestComputeSpanGlyphs(t *testing.T) {
 		n     int
 		want  []spanLevels
 	}{
+		{"no pairs", nil, 3, []spanLevels{none, none, none}},
 		{
-			name:  "no pairs",
-			pairs: nil,
-			n:     3,
-			want:  []spanLevels{none, none, none},
-		},
-		{
-			name: "adjacent pair (req at 0, resp at 1)",
-			// Bidirectional, like pairInvocationRows emits.
+			name:  "adjacent pair",
 			pairs: map[int]int{0: 1, 1: 0},
 			n:     2,
 			want:  []spanLevels{outer(glyphStart), outer(glyphEnd)},
 		},
 		{
-			name:  "pair with one row in between",
+			name:  "one row in between",
 			pairs: map[int]int{0: 2, 2: 0},
 			n:     3,
 			want:  []spanLevels{outer(glyphStart), outer(glyphMiddle), outer(glyphEnd)},
 		},
 		{
-			name: "two non-overlapping pairs",
-			pairs: map[int]int{
-				0: 2, 2: 0,
-				3: 5, 5: 3,
-			},
-			n: 6,
-			want: []spanLevels{
-				outer(glyphStart), outer(glyphMiddle), outer(glyphEnd),
-				outer(glyphStart), outer(glyphMiddle), outer(glyphEnd),
-			},
-		},
-		{
-			name: "nested pairs — outer (0,5), inner (2,3)",
-			// Both levels are now visible: the outer's continuation
-			// glyph sits next to the inner's corner so an operator can
-			// see "row 2 is inside an outer span AND opens an inner
-			// span" at a glance.
+			// The real shape: an outer a2a exchange (0,5) bracketing two inner
+			// inference exchanges (1,2) and (3,4).
+			name: "nested exchanges (a2a containing two inference calls)",
 			pairs: map[int]int{
 				0: 5, 5: 0,
-				2: 3, 3: 2,
+				1: 2, 2: 1,
+				3: 4, 4: 3,
 			},
 			n: 6,
-			want: []spanLevels{
-				outer(glyphStart),             // 0: outer starts here
-				outer(glyphMiddle),            // 1: only outer participates
-				both(glyphMiddle, glyphStart), // 2: outer continues, inner starts
-				both(glyphMiddle, glyphEnd),   // 3: outer continues, inner ends
-				outer(glyphMiddle),            // 4: only outer participates
-				outer(glyphEnd),               // 5: outer ends here
-			},
-		},
-		{
-			name: "real-world shape — long outer pair with inner pairs interleaved",
-			// Mirrors the IBAC demo timeline:
-			//   row 0: a2a-parser req     ┌
-			//   row 1: jwt-validation     │            (event-1 continuation)
-			//   row 2: inference-parser req  ┌
-			//   row 3: inference-parser resp └
-			//   row 4: a2a-parser resp    └
-			pairs: map[int]int{
-				0: 4, 4: 0, // outer a2a-parser
-				2: 3, 3: 2, // inner inference-parser
-			},
-			n: 5,
 			want: []spanLevels{
 				outer(glyphStart),
-				outer(glyphMiddle),
+				both(glyphMiddle, glyphStart),
+				both(glyphMiddle, glyphEnd),
 				both(glyphMiddle, glyphStart),
 				both(glyphMiddle, glyphEnd),
 				outer(glyphEnd),
 			},
 		},
 		{
-			name: "deeper nesting collapses to two levels",
-			// outer (0, 7), middle (1, 6), innermost (2, 5). Row 3 sits
-			// inside all three; only the two largest surface so the
-			// PHASE column doesn't blow its width budget.
+			// The #52 case: a pair (2,3) nested THREE deep — inside a middle
+			// span (1,4) inside an outer span (0,5). The innermost pair's
+			// endpoints must still show their ┌/└ corners (so its req/resp
+			// connect) rather than the middle span's bar masking them. inner =
+			// the row's narrowest containing span, not the second-widest.
+			name: "triple-nested innermost pair keeps its corners",
 			pairs: map[int]int{
-				0: 7, 7: 0,
-				1: 6, 6: 1,
-				2: 5, 5: 2,
+				0: 5, 5: 0,
+				1: 4, 4: 1,
+				2: 3, 3: 2,
 			},
-			n: 8,
+			n: 6,
 			want: []spanLevels{
-				outer(glyphStart),
-				both(glyphMiddle, glyphStart),  // outer middle + middle-pair start
-				both(glyphMiddle, glyphMiddle), // middle-pair middle wins over innermost (deeper dropped)
-				both(glyphMiddle, glyphMiddle), // same
-				both(glyphMiddle, glyphMiddle), // same
-				both(glyphMiddle, glyphMiddle), // middle-pair middle (innermost dropped)
-				both(glyphMiddle, glyphEnd),    // middle-pair ends here, outer continues
-				outer(glyphEnd),
+				outer(glyphStart),             // 0: outer starts
+				both(glyphMiddle, glyphStart), // 1: outer mid, middle-span starts
+				both(glyphMiddle, glyphStart), // 2: outer mid, innermost STARTS (was masked to middle)
+				both(glyphMiddle, glyphEnd),   // 3: outer mid, innermost ENDS (was masked to middle)
+				both(glyphMiddle, glyphEnd),   // 4: outer mid, middle-span ends
+				outer(glyphEnd),               // 5: outer ends
 			},
 		},
-		{
-			name: "out-of-bounds endpoint gracefully ignored",
-			// Defensive: pairInvocationRows shouldn't emit pairs with
-			// indices >= n, but the helper must not panic if a future
-			// caller reuses it on a truncated row slice. The span (0,10)
-			// is still treated as containing rows 1 and 2 because they
-			// satisfy a < i < b.
-			pairs: map[int]int{0: 10, 10: 0},
-			n:     3,
-			want:  []spanLevels{outer(glyphStart), outer(glyphMiddle), outer(glyphMiddle)},
-		},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
 			got := computeSpanGlyphs(tc.pairs, tc.n)
 			if len(got) != len(tc.want) {
-				t.Fatalf("len = %d, want %d (got %v)", len(got), len(tc.want), got)
+				t.Fatalf("len = %d, want %d", len(got), len(tc.want))
 			}
 			for i := range tc.want {
 				if got[i] != tc.want[i] {
 					t.Errorf("row %d: got {outer=%q inner=%q}, want {outer=%q inner=%q}",
-						i,
-						string(rune(got[i].outer)), string(rune(got[i].inner)),
+						i, string(rune(got[i].outer)), string(rune(got[i].inner)),
 						string(rune(tc.want[i].outer)), string(rune(tc.want[i].inner)))
 				}
 			}
@@ -564,28 +765,42 @@ func TestComputeSpanGlyphs(t *testing.T) {
 	}
 }
 
-// TestSpanLevels_Prefix locks the rendered prefix the rebuildEventsTable
-// uses to populate the PHASE column. Empty levels render as empty
-// string; one level renders one glyph; two levels render two glyphs.
-func TestSpanLevels_Prefix(t *testing.T) {
-	cases := []struct {
-		name string
-		s    spanLevels
-		want string
-	}{
-		{"none", spanLevels{}, ""},
-		{"outer only — start", spanLevels{outer: glyphStart}, "┌"},
-		{"outer only — middle", spanLevels{outer: glyphMiddle}, "│"},
-		{"outer only — end", spanLevels{outer: glyphEnd}, "└"},
-		{"both levels — outer middle, inner start", spanLevels{outer: glyphMiddle, inner: glyphStart}, "│┌"},
-		{"both levels — outer middle, inner end", spanLevels{outer: glyphMiddle, inner: glyphEnd}, "│└"},
+// TestComputeEventPairs_NestedExchangeGlyphs is the end-to-end #23 shape: an
+// inbound a2a message/stream request, two outbound inference exchanges during
+// processing, then the a2a response. The a2a request/response must pair and
+// bracket (┌ … └) with the inference exchanges nested (│┌ … │└) inside.
+func TestComputeEventPairs_NestedExchangeGlyphs(t *testing.T) {
+	a2aReq := pipeline.SessionEvent{Direction: pipeline.Inbound, Phase: pipeline.SessionRequest,
+		Host: "claude-agent", A2A: &pipeline.A2AExtension{Method: "message/stream"}}
+	infReq1 := pipeline.SessionEvent{Direction: pipeline.Outbound, Phase: pipeline.SessionRequest,
+		Host: "litellm", Inference: &pipeline.InferenceExtension{Model: "claude"}}
+	infResp1 := pipeline.SessionEvent{Direction: pipeline.Outbound, Phase: pipeline.SessionResponse,
+		Host: "litellm", Inference: &pipeline.InferenceExtension{Model: "claude"}, StatusCode: 200}
+	infReq2 := pipeline.SessionEvent{Direction: pipeline.Outbound, Phase: pipeline.SessionRequest,
+		Host: "litellm", Inference: &pipeline.InferenceExtension{Model: "claude"}}
+	infResp2 := pipeline.SessionEvent{Direction: pipeline.Outbound, Phase: pipeline.SessionResponse,
+		Host: "litellm", Inference: &pipeline.InferenceExtension{Model: "claude"}, StatusCode: 200}
+	a2aResp := pipeline.SessionEvent{Direction: pipeline.Inbound, Phase: pipeline.SessionResponse,
+		Host: "claude-agent", A2A: &pipeline.A2AExtension{Method: "message/stream"}, StatusCode: 200}
+	events := []pipeline.SessionEvent{a2aReq, infReq1, infResp1, infReq2, infResp2, a2aResp}
+
+	rows := buildEventRows(events)
+	ids, partner := computeEventPairs(rows)
+
+	// a2a request (row 0) pairs with a2a response (row 5), spanning everything.
+	if partner[0] != 5 || partner[5] != 0 {
+		t.Errorf("a2a req/resp should pair 0↔5, got partner=%v", partner)
 	}
-	for _, tc := range cases {
-		t.Run(tc.name, func(t *testing.T) {
-			if got := tc.s.prefix(); got != tc.want {
-				t.Errorf("prefix() = %q, want %q", got, tc.want)
-			}
-		})
+	if ids[&events[0]] != ids[&events[5]] {
+		t.Errorf("a2a req/resp should share #, got %d vs %d", ids[&events[0]], ids[&events[5]])
+	}
+
+	glyphs := computeSpanGlyphs(partner, len(rows))
+	want := []string{"┌", "│┌", "│└", "│┌", "│└", "└"}
+	for i, w := range want {
+		if got := glyphs[i].prefix(); got != w {
+			t.Errorf("row %d prefix = %q, want %q", i, got, w)
+		}
 	}
 }
 
diff --git a/authbridge/cmd/abctl/tui/keys.go b/authbridge/cmd/abctl/tui/keys.go
index d2bd60e04..8898ef3f3 100644
--- a/authbridge/cmd/abctl/tui/keys.go
+++ b/authbridge/cmd/abctl/tui/keys.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/charmbracelet/bubbles/table"
 	tea "github.com/charmbracelet/bubbletea"
 
 	"github.com/kagenti/kagenti-extensions/authbridge/cmd/abctl/apiclient"
@@ -144,12 +145,14 @@ func (m *model) handleKey(msg tea.KeyMsg) tea.Cmd {
 		return nil
 
 	case "s":
-		// Toggle skip-row visibility. Only meaningful while the events
-		// pane is active, but accepting the key on any pane keeps the
-		// keybinding simple and lets operators "set their preference"
-		// before drilling into a session. rebuildEventsTable is a no-op
-		// when no session is selected.
-		m.showSkips = !m.showSkips
+		// Toggle hiding of passthrough / skip-only messages. Default is
+		// off (show everything); turning it on focuses the timeline on
+		// plugin activity. Only meaningful while the events pane is
+		// active, but accepting the key on any pane keeps the keybinding
+		// simple and lets operators set their preference before drilling
+		// into a session. rebuildEventsTable is a no-op when no session
+		// is selected.
+		m.hideInactive = !m.hideInactive
 		m.rebuildEventsTable()
 		return nil
 
@@ -203,11 +206,11 @@ func (m *model) handleKey(msg tea.KeyMsg) tea.Cmd {
 			// Snapshot in case the stream hasn't yet delivered history.
 			return m.snapshotCmd(id)
 		case paneEvents:
-			ev := m.selectedEvent()
-			if ev == nil {
+			er, ok := m.selectedEventRow()
+			if !ok {
 				return nil
 			}
-			m.showDetail(ev, m.selectedInvocation())
+			m.showDetail(er)
 			m.pane = paneDetail
 			return nil
 		case panePipeline:
@@ -270,6 +273,15 @@ func (m *model) handleKey(msg tea.KeyMsg) tea.Cmd {
 		m.goBottom()
 		return nil
 
+	case "pgup", "pgdown", "pgdn", "b", "f":
+		// Page the active pane. Sessions can hold up to session.max_events
+		// (500) rows, so one-row-at-a-time nav isn't enough. b/f (less/vim
+		// style) work on any keyboard; PgUp/PgDn map to fn+↑/fn+↓ on Mac
+		// laptops. Explicit here (rather than the table component's own
+		// binding) so all of them share a one-row overlap for context and the
+		// detail viewport pages too.
+		return m.pageActivePane(msg)
+
 	case "P":
 		// Open the registered-plugin catalog. Available from any
 		// session-view pane; in --endpoint mode the picker fields
@@ -385,6 +397,42 @@ func (m *model) goBottom() {
 	}
 }
 
+// pageActivePane moves the active pane by one page. Tables move the cursor by
+// (visible height − 1) rows — one row of overlap keeps context across the
+// jump — clamped to the row range by table.MoveUp/MoveDown. The detail/
+// plugin-detail viewport delegates to its built-in page scrolling. Picker
+// panes (namespaces/pods) never reach here; they return early in handleKey
+// and page via their own component's binding.
+func (m *model) pageActivePane(msg tea.KeyMsg) tea.Cmd {
+	up := msg.String() == "pgup" || msg.String() == "b"
+	page := func(t *table.Model) {
+		n := t.Height() - 1
+		if n < 1 {
+			n = 1
+		}
+		if up {
+			t.MoveUp(n)
+		} else {
+			t.MoveDown(n)
+		}
+	}
+	switch m.pane {
+	case paneEvents:
+		page(&m.eventsTbl)
+	case paneSessions:
+		page(&m.sessionsTbl)
+	case panePipeline:
+		page(&m.pipelineTbl)
+	case paneCatalog:
+		page(&m.catalogTbl)
+	case paneDetail, panePluginDetail:
+		var cmd tea.Cmd
+		m.detailVp, cmd = m.detailVp.Update(msg)
+		return cmd
+	}
+	return nil
+}
+
 // setFlash shows a transient message in the footer for flashDuration.
 func (m *model) setFlash(s string) {
 	m.flash = s
@@ -408,13 +456,17 @@ func (m *model) helpView() string {
 		}
 		return "[↑↓] nav  [↵] drill  [tab] pipeline  [/] filter  [p] pause  [q] quit"
 	case paneEvents:
-		base := "[↑↓] nav  [↵] detail  [esc] back  [/] filter  [s] skips  [p] pause  [q] quit"
-		// Surface the hidden-skip count so a sparse timeline doesn't
-		// look like data loss. Only annotate when there's something
-		// to say (skips off AND at least one row was hidden).
-		if !m.showSkips && m.hiddenSkips > 0 {
-			base = fmt.Sprintf("%s  ·  %d skip%s hidden",
-				base, m.hiddenSkips, plural(m.hiddenSkips))
+		skipHint := "[s] hide passthru/skip"
+		if m.hideInactive {
+			skipHint = "[s] show all"
+		}
+		base := "[↑↓] nav  [b/f] page  [↵] detail  [esc] back  [/] filter  " + skipHint + "  [p] pause  [q] quit"
+		// Surface the hidden-message count so a filtered timeline doesn't
+		// look like data loss. Only annotate when hiding is on AND at
+		// least one message was hidden.
+		if m.hideInactive && m.hiddenInactive > 0 {
+			base = fmt.Sprintf("%s  ·  %d hidden",
+				base, m.hiddenInactive)
 		}
 		return base
 	case paneDetail:
@@ -476,7 +528,7 @@ func (m *model) layout() {
 	// Re-wrap the detail viewport to the new width so long JSON values
 	// continue to fit after a terminal resize.
 	if m.detailEvent != nil {
-		m.showDetail(m.detailEvent, m.detailInvocation)
+		m.showDetail(m.detailRow)
 	}
 }
 
diff --git a/authbridge/cmd/authbridge-envoy/main.go b/authbridge/cmd/authbridge-envoy/main.go
index 97722c84b..439b04fb0 100644
--- a/authbridge/cmd/authbridge-envoy/main.go
+++ b/authbridge/cmd/authbridge-envoy/main.go
@@ -201,7 +201,7 @@ func main() {
 				slog.Warn("invalid session.ttl, using default", "value", cfg.Session.TTL, "error", err)
 			}
 		}
-		maxEvents := 100
+		maxEvents := 500 // raised from 100: recording every message (incl. no-plugin-activity) ~doubles volume
 		if cfg.Session.MaxEvents > 0 {
 			maxEvents = cfg.Session.MaxEvents
 		}
diff --git a/authbridge/cmd/authbridge-lite/main.go b/authbridge/cmd/authbridge-lite/main.go
index 26058c957..0f1199d0d 100644
--- a/authbridge/cmd/authbridge-lite/main.go
+++ b/authbridge/cmd/authbridge-lite/main.go
@@ -188,7 +188,7 @@ func main() {
 				slog.Warn("invalid session.ttl, using default", "value", cfg.Session.TTL, "error", err)
 			}
 		}
-		maxEvents := 100
+		maxEvents := 500 // raised from 100: recording every message (incl. no-plugin-activity) ~doubles volume
 		if cfg.Session.MaxEvents > 0 {
 			maxEvents = cfg.Session.MaxEvents
 		}
diff --git a/authbridge/cmd/authbridge-proxy/main.go b/authbridge/cmd/authbridge-proxy/main.go
index e997b0982..4f6306bab 100644
--- a/authbridge/cmd/authbridge-proxy/main.go
+++ b/authbridge/cmd/authbridge-proxy/main.go
@@ -254,7 +254,7 @@ func main() {
 				slog.Warn("invalid session.ttl, using default", "value", cfg.Session.TTL, "error", err)
 			}
 		}
-		maxEvents := 100
+		maxEvents := 500 // raised from 100: recording every message (incl. no-plugin-activity) ~doubles volume
 		if cfg.Session.MaxEvents > 0 {
 			maxEvents = cfg.Session.MaxEvents
 		}