feat: Plugin catalog + abctl visibility & validation

authbridge/CLAUDE.md

@@ -406,6 +406,8 @@ When `session.enabled` is true (default) and `listener.session_api_addr` is non-
 | `GET /v1/sessions` | `application/json` | List active sessions: `{sessions: [{id, createdAt, updatedAt, eventCount, active}]}`. |
 | `GET /v1/sessions/{id}` | `application/json` | Full snapshot of one session's events. 404 if unknown/expired. |
 | `GET /v1/events` | `text/event-stream` | SSE stream of new events. Optional `?session=<id>` filters to one session. Heartbeat every 30s. |
+| `GET /v1/pipeline` | `application/json` | Active pipeline composition: `{inbound: [...], outbound: [...]}`. Each plugin entry carries `name`, `direction`, `position`, `bodyAccess`, `writes`, `reads`, plus the static metadata (`requires`, `requiresAny`, `after`, `claims`, `description`) and runtime `config` when present. abctl renders this as the Pipeline pane. |
+| `GET /v1/plugins` | `application/json` | Catalog of every registered plugin (whether or not in the active pipeline): `{plugins: [{name, requires, requiresAny, after, claims, description, ...}]}`. abctl renders this as the Catalog pane (`P` key). 404s when the binary's session API was constructed without `WithCatalog`. |
 | `GET /healthz` | text | Liveness probe. |
 
 ### Quick examples

authbridge/authlib/listener/reverseproxy/server.go

@@ -230,13 +230,7 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) {
 	// not A2A-only. A non-A2A inbound, or an A2A request that fails to
 	// parse, is intentionally not recorded here.
 	if s.Sessions != nil && pctx.Extensions.A2A != nil {
-		sid := pctx.Extensions.A2A.SessionID
-		if sid == "" {
-			sid = s.Sessions.ActiveSession()
-		}
-		if sid == "" {
-			sid = session.DefaultSessionID
-		}
+		sid := inboundSessionID(pctx)
 		// Snapshot-copy the protocol extension and use the shared helpers
 		// for plugin invocations / observability / identity. Mirrors what
 		// extproc does so request events don't pick up response-phase
@@ -320,13 +314,7 @@ 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.
 	if s.Sessions != nil && pctx.Extensions.A2A != nil {
-		sid := pctx.Extensions.A2A.SessionID
-		if sid == "" {
-			sid = s.Sessions.ActiveSession()
-		}
-		if sid == "" {
-			sid = session.DefaultSessionID
-		}
+		sid := inboundSessionID(pctx)
 		s.Sessions.Append(sid, pipeline.SessionEvent{
 			At:          time.Now(),
 			Direction:   pipeline.Inbound,
@@ -367,18 +355,10 @@ func (s *Server) recordInboundReject(pctx *pipeline.Context, action pipeline.Act
 		return
 	}
 	// Inbound uses the A2A-stated contextId when available; otherwise
-	// falls through to the default bucket. Matches the accept path's
-	// bucketing rule (A2A request event at line 112-125).
-	sid := ""
-	if pctx.Extensions.A2A != nil {
-		sid = pctx.Extensions.A2A.SessionID
-	}
-	if sid == "" {
-		sid = s.Sessions.ActiveSession()
-	}
-	if sid == "" {
-		sid = session.DefaultSessionID
-	}
+	// the default bucket. Same rule as the accept path's
+	// inboundSessionID helper, kept consistent so denial events land
+	// in the same bucket the accepted request would have.
+	sid := inboundSessionID(pctx)
 	var status int
 	var code, message string
 	if action.Violation != nil {
@@ -430,3 +410,19 @@ func requestScheme(r *http.Request) string {
 	}
 	return "http"
 }
+
+// inboundSessionID returns the bucket ID for an inbound event. Mirrors
+// extproc's inboundSessionID: trusts the A2A-stated contextId when
+// non-empty, otherwise routes to DefaultSessionID. Does NOT fall back
+// to ActiveSession() — that fallback was a cross-conversation
+// contamination vector (a new conversation's first turn would inherit
+// the previous conversation's rekeyed bucket, stranding the current
+// turn's events in the prior bucket and creating an orphan one-event
+// session for the response). Rekey on response migrates the Default
+// bucket into the contextId once the agent reveals it.
+func inboundSessionID(pctx *pipeline.Context) string {
+	if pctx.Extensions.A2A != nil && pctx.Extensions.A2A.SessionID != "" {
+		return pctx.Extensions.A2A.SessionID
+	}
+	return session.DefaultSessionID
+}

authbridge/authlib/listener/reverseproxy/server_test.go

@@ -562,3 +562,82 @@ func TestReverseProxy_ModifyResponse_RekeyAndResponseEvent(t *testing.T) {
 		t.Errorf("response event has no Invocations; expected one a2a-stamp observe entry")
 	}
 }
+
+// TestInboundSessionID_NoActiveSessionFallback locks in the
+// reverseproxy's bucketing rule: when the A2A request has no
+// contextId (first turn of a conversation), inboundSessionID returns
+// DefaultSessionID — never falling back to ActiveSession(). The
+// previous fallback was a cross-conversation contamination vector:
+// a fresh chat's first turn would silently inherit a previous
+// conversation's still-alive session bucket, stranding all events
+// in the prior bucket and creating an orphan one-event session for
+// the rekeyed response.
+func TestInboundSessionID_NoActiveSessionFallback(t *testing.T) {
+	cases := []struct {
+		name string
+		ctx  string // pctx.Extensions.A2A.SessionID
+		want string
+	}{
+		{"first turn — empty contextId", "", session.DefaultSessionID},
+		{"subsequent turn — explicit contextId", "ctx-abc", "ctx-abc"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			pctx := &pipeline.Context{
+				Direction: pipeline.Inbound,
+				Extensions: pipeline.Extensions{
+					A2A: &pipeline.A2AExtension{SessionID: tc.ctx},
+				},
+			}
+			if got := inboundSessionID(pctx); got != tc.want {
+				t.Errorf("inboundSessionID = %q, want %q", got, tc.want)
+			}
+		})
+	}
+}
+
+// TestInboundSessionID_NoA2AExtension exercises the auth-only path:
+// non-A2A inbound (e.g. a denied request that never reached the
+// parser) routes to DefaultSessionID, where operators expect to find
+// unauthorized-access events.
+func TestInboundSessionID_NoA2AExtension(t *testing.T) {
+	pctx := &pipeline.Context{Direction: pipeline.Inbound}
+	if got := inboundSessionID(pctx); got != session.DefaultSessionID {
+		t.Errorf("inboundSessionID = %q, want %q", got, session.DefaultSessionID)
+	}
+}
+
+// TestRecordInbound_NewChatDoesntLeakIntoPriorBucket is the
+// integration regression for the original bug: with a previous
+// conversation's session still alive (so ActiveSession() returns it),
+// a new chat's first-turn inbound request lands in DefaultSessionID
+// rather than the previous bucket. The rekey-on-response path then
+// migrates Default → server-assigned contextId, keeping the new
+// conversation in its own bucket.
+func TestRecordInbound_NewChatDoesntLeakIntoPriorBucket(t *testing.T) {
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+
+	// Seed a "previous conversation" bucket and make it the active session.
+	store.Append("prev-conversation", pipeline.SessionEvent{
+		At: time.Now(), Direction: pipeline.Inbound, Phase: pipeline.SessionRequest,
+	})
+	if got := store.ActiveSession(); got != "prev-conversation" {
+		t.Fatalf("setup: ActiveSession = %q, want prev-conversation", got)
+	}
+
+	// New chat's first turn arrives — empty contextId.
+	pctx := &pipeline.Context{
+		Direction: pipeline.Inbound,
+		Extensions: pipeline.Extensions{
+			A2A: &pipeline.A2AExtension{Method: "message/stream"},
+		},
+	}
+	sid := inboundSessionID(pctx)
+	if sid == "prev-conversation" {
+		t.Fatal("regression: new chat's first turn leaked into previous bucket")
+	}
+	if sid != session.DefaultSessionID {
+		t.Errorf("first-turn sid = %q, want %q", sid, session.DefaultSessionID)
+	}
+}

authbridge/authlib/pipeline/plugin.go

@@ -95,6 +95,19 @@ type PluginCapabilities struct {
 	// list." See authlib/contracts/claims.go for the canonical
 	// vocabulary.
 	Claims []string
+
+	// Description is operator-facing prose, one line, ≤80 chars,
+	// describing what this plugin does. Surfaces in `abctl`'s
+	// plugin-detail and catalog panes, and in /v1/plugins. Empty for
+	// plugins that haven't opted in; new plugins should populate it.
+	//
+	// Capabilities are static type-level metadata: Capabilities() must
+	// return the same value for any instance produced by a given
+	// factory. The catalog endpoint relies on this — varying capabilities
+	// by instance state silently produces wrong catalog entries. If a
+	// plugin's behavior varies enough that its capabilities differ,
+	// register it under multiple names.
+	Description string
 }
 
 // Normalize applies compatibility rules to a PluginCapabilities:

authbridge/authlib/plugins/a2aparser/plugin.go

@@ -26,8 +26,9 @@ func (p *A2AParser) Name() string { return "a2a-parser" }
 
 func (p *A2AParser) Capabilities() pipeline.PluginCapabilities {
 	return pipeline.PluginCapabilities{
-		Writes:    []string{"a2a"},
-		ReadsBody: true,
+		Writes:      []string{"a2a"},
+		ReadsBody:   true,
+		Description: "Parses A2A messages into pctx.Extensions.A2A for downstream plugins.",
 	}
 }
 
@@ -109,10 +110,19 @@ func (p *A2AParser) OnRequest(_ context.Context, pctx *pipeline.Context) pipelin
 //
 // Handles both JSON-RPC responses (message/send) and SSE event streams (message/stream).
 func (p *A2AParser) OnResponse(_ context.Context, pctx *pipeline.Context) pipeline.Action {
-	// No Invocation on response when the parser doesn't apply: either
-	// there's no response body or the matching request wasn't an A2A
-	// JSON-RPC call. Keeps the response event clean for non-A2A traffic.
-	if len(pctx.ResponseBody) == 0 || pctx.Extensions.A2A == nil {
+	// Stay silent when the request side never participated — the parser
+	// recorded nothing on request, so recording on response would orphan
+	// the row.
+	if pctx.Extensions.A2A == nil {
+		return pipeline.Action{Type: pipeline.Continue}
+	}
+	// We DID process the request but the response has no body — record
+	// a Skip so abctl can pair the response row with the request row.
+	// Without this, the request invocation orphans and the events table
+	// shows req+resp as unpaired (no ┌/└ glyphs, plugin/action columns
+	// blank on the response side).
+	if len(pctx.ResponseBody) == 0 {
+		pctx.Skip("no_response_body")
 		return pipeline.Action{Type: pipeline.Continue}
 	}
 

authbridge/authlib/plugins/a2aparser/plugin_test.go

@@ -429,9 +429,13 @@ func TestA2AParser_OnResponse_NoRequestContext(t *testing.T) {
 	}
 }
 
+// TestA2AParser_OnResponse_EmptyBody locks the regression: when the
+// request side parsed (Extensions.A2A populated) but response body is
+// empty, the parser MUST record a Skip so abctl pairs the timeline rows.
 func TestA2AParser_OnResponse_EmptyBody(t *testing.T) {
 	p := NewA2AParser()
 	pctx := &pipeline.Context{
+		Direction:  pipeline.Inbound,
 		Extensions: pipeline.Extensions{A2A: &pipeline.A2AExtension{Method: "message/send"}},
 	}
 	action := p.OnResponse(context.Background(), pctx)
@@ -441,6 +445,13 @@ func TestA2AParser_OnResponse_EmptyBody(t *testing.T) {
 	if pctx.Extensions.A2A.SessionID != "" {
 		t.Errorf("SessionID should remain empty, got %q", pctx.Extensions.A2A.SessionID)
 	}
+	if pctx.Extensions.Invocations == nil {
+		t.Fatal("expected a Skip Invocation, got none")
+	}
+	invs := pctx.Extensions.Invocations.Inbound
+	if len(invs) != 1 || invs[0].Action != pipeline.ActionSkip || invs[0].Reason != "no_response_body" {
+		t.Fatalf("expected single Skip/no_response_body, got %+v", invs)
+	}
 }
 
 func TestA2AParser_OnResponse_JSONRPC_ContextID(t *testing.T) {

authbridge/authlib/plugins/ibac/plugin.go

@@ -184,8 +184,9 @@ func (p *IBAC) Capabilities() pipeline.PluginCapabilities {
 		// IBAC must come after so it can read the parsed tool name
 		// and args. If it's absent, IBAC still functions on raw
 		// HTTP — the judge sees method+host+path+body excerpt.
-		After:     []string{"mcp-parser"},
-		ReadsBody: true,
+		After:       []string{"mcp-parser"},
+		ReadsBody:   true,
+		Description: "LLM-judge intent-based access control for outbound tool calls.",
 	}
 }
 

authbridge/authlib/plugins/inferenceparser/plugin.go

@@ -26,8 +26,9 @@ func (p *InferenceParser) Name() string { return "inference-parser" }
 
 func (p *InferenceParser) Capabilities() pipeline.PluginCapabilities {
 	return pipeline.PluginCapabilities{
-		Writes:    []string{"inference"},
-		ReadsBody: true,
+		Writes:      []string{"inference"},
+		ReadsBody:   true,
+		Description: "Parses LLM completions into pctx.Extensions.Inference.",
 	}
 }
 
@@ -95,9 +96,16 @@ func (p *InferenceParser) OnRequest(_ context.Context, pctx *pipeline.Context) p
 // token counts) on pctx.Extensions.Inference. Handles both non-streaming
 // JSON responses and SSE streams from OpenAI-compatible servers.
 func (p *InferenceParser) OnResponse(_ context.Context, pctx *pipeline.Context) pipeline.Action {
-	// No Invocation when the parser doesn't apply — request wasn't
-	// inference or no response body to parse.
-	if len(pctx.ResponseBody) == 0 || pctx.Extensions.Inference == nil {
+	// Stay silent when the request side never participated — the parser
+	// recorded nothing on request, so recording on response would orphan
+	// the row.
+	if pctx.Extensions.Inference == nil {
+		return pipeline.Action{Type: pipeline.Continue}
+	}
+	// We DID process the request but the response has no body — record
+	// a Skip so abctl can pair the response row with the request row.
+	if len(pctx.ResponseBody) == 0 {
+		pctx.Skip("no_response_body")
 		return pipeline.Action{Type: pipeline.Continue}
 	}
 

authbridge/authlib/plugins/inferenceparser/plugin_test.go

@@ -395,9 +395,14 @@ func TestInferenceParser_OnResponse_NoRequestContext(t *testing.T) {
 	}
 }
 
+// TestInferenceParser_OnResponse_EmptyBody locks the regression: when
+// the request side parsed (Extensions.Inference populated) but the
+// response body is empty, the parser MUST record a Skip so abctl pairs
+// the timeline rows.
 func TestInferenceParser_OnResponse_EmptyBody(t *testing.T) {
 	p := NewInferenceParser()
 	pctx := &pipeline.Context{
+		Direction:  pipeline.Outbound,
 		Extensions: pipeline.Extensions{Inference: &pipeline.InferenceExtension{Model: "gpt-4"}},
 	}
 	action := p.OnResponse(context.Background(), pctx)
@@ -407,6 +412,13 @@ func TestInferenceParser_OnResponse_EmptyBody(t *testing.T) {
 	if pctx.Extensions.Inference.Completion != "" {
 		t.Errorf("Completion = %q, want empty", pctx.Extensions.Inference.Completion)
 	}
+	if pctx.Extensions.Invocations == nil {
+		t.Fatal("expected a Skip Invocation, got none")
+	}
+	invs := pctx.Extensions.Invocations.Outbound
+	if len(invs) != 1 || invs[0].Action != pipeline.ActionSkip || invs[0].Reason != "no_response_body" {
+		t.Fatalf("expected single Skip/no_response_body, got %+v", invs)
+	}
 }
 
 func TestInferenceParser_OnResponse_NonStreaming(t *testing.T) {

authbridge/authlib/plugins/jwtvalidation/plugin.go

@@ -209,7 +209,10 @@ func init() {
 func (p *JWTValidation) Name() string { return "jwt-validation" }
 
 func (p *JWTValidation) Capabilities() pipeline.PluginCapabilities {
-	return pipeline.PluginCapabilities{Writes: []string{"security"}}
+	return pipeline.PluginCapabilities{
+		Writes:      []string{"security"},
+		Description: "Inbound JWT validation (signature, issuer, audience) against JWKS.",
+	}
 }
 
 // Configure decodes the plugin's config subtree, applies defaults,

authbridge/authlib/plugins/mcpparser/plugin.go

@@ -26,8 +26,9 @@ func (p *MCPParser) Name() string { return "mcp-parser" }
 
 func (p *MCPParser) Capabilities() pipeline.PluginCapabilities {
 	return pipeline.PluginCapabilities{
-		Writes:    []string{"mcp"},
-		ReadsBody: true,
+		Writes:      []string{"mcp"},
+		ReadsBody:   true,
+		Description: "Parses MCP tool calls/results into pctx.Extensions.MCP.",
 	}
 }
 
@@ -70,12 +71,19 @@ func (p *MCPParser) OnRequest(_ context.Context, pctx *pipeline.Context) pipelin
 }
 
 func (p *MCPParser) OnResponse(_ context.Context, pctx *pipeline.Context) pipeline.Action {
-	// No Invocation when the parser doesn't apply — request wasn't MCP
-	// JSON-RPC or no response body to parse. The unparseable_response
-	// case below IS recorded because it's diagnostic: the request WAS
-	// MCP but the response couldn't be decoded, which usually signals
-	// an upstream protocol bug worth surfacing.
-	if len(pctx.ResponseBody) == 0 || pctx.Extensions.MCP == nil {
+	// Stay silent when the request side never participated — the parser
+	// recorded nothing on request, so recording on response would orphan
+	// the row.
+	if pctx.Extensions.MCP == nil {
+		return pipeline.Action{Type: pipeline.Continue}
+	}
+	// We DID process the request but the response has no body — typical
+	// for JSON-RPC notifications that ack with HTTP 202. Record a Skip
+	// so abctl can pair the response row with the request row in the
+	// timeline (pairing keys on plugin+method+direction; an empty
+	// invocation slot orphans both ends).
+	if len(pctx.ResponseBody) == 0 {
+		pctx.Skip("no_response_body")
 		return pipeline.Action{Type: pipeline.Continue}
 	}