Feat(abctl): one row per network message, show every message

The session timeline was confusing for two reasons: messages no plugin
processed were invisible, and a message processed/skipped by N plugins
showed as N rows (plus a separate row for a TLS-bridged call's CONNECT).

Server (authlib): relax the four-clause append gate to unconditional at
every accept-path emit site (forward-proxy request/response, transparent
tunnel-open, reverse-proxy request/response/streaming) so every message
the pipeline saw is recorded — including passthrough requests and generic
responses (e.g. a 404) no plugin touched. The !skipped guards stay, so
listener.skip_hosts traffic remains suppressed. Raise session.max_events
default 100 -> 500 in all three binaries to absorb the ~2x volume.

abctl: render one table.Row per SessionEvent. eventAction() folds a
message's per-plugin invocations into one ACTION + PLUGIN cell, headlining
the highest-ranked ENFORCED action — deny > modify > observe > allow >
skip. observe outranks allow so a parser (which supplies METHOD) headlines
over a gate that merely allowed; a skip-only or no-plugin message shows
passthrough markers ("—") rather than crediting a plugin that declined to
act. A shadow deny/modify never headlines over the action that really took
effect (the pipeline enforces deny only when !Shadow); it is surfaced with
a trailing "*" instead (e.g. "allow*"), or headlines alone ("deny*") when
nothing enforced acted. The detail pane shows the whole event; a bridged
row also folds the CONNECT tunnel's gate invocations into its ACTION and
inactive-filter view.

A TLS-bridge CONNECT tunnel folds into the decrypted inner request that
follows it, so a bridged call is one request row + one response row, with
a "tunnel:" summary in the detail pane. Two back-to-back passthrough
CONNECTs to the same host are NOT folded (each is its own message).
Event-level span glyphs in PHASE bracket each request/response exchange
and nest outbound calls under the inbound request that caused them. The
`s` key hides passthrough/skip-only messages (default off = show all).

Replaces the per-invocation row model: drops flattenInvocations and the
per-invocation pairing; the span-glyph rendering now operates on events,
so a multi-plugin message no longer duplicates into one row per plugin.

Assisted-By: Claude (Anthropic AI) <noreply@anthropic.com>
Signed-off-by: Hai Huang <huang195@gmail.com>

Author: huang195

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)
 }
 

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

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 {

authbridge/authlib/listener/forwardproxy/transparent.go

@@ -180,9 +180,10 @@ func (s *Server) recordTunnelOpened(pctx *pipeline.Context) {
 		Identity:    pipeline.SnapshotIdentity(pctx),
 		Host:        pctx.Host,
 	}
-	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

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.

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…)"

authbridge/cmd/abctl/tui/detail_pane.go

@@ -9,67 +9,30 @@ 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
 // events show only request-side fields and response events show only
 // 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,13 +44,35 @@ 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
 	}
 	m.detailVp.SetContent(content)
 	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.