feat(pipeline): pctx.Record / Allow / Skip / Observe / Modify helpers for plugin DX

authbridge/CLAUDE.md

@@ -333,6 +333,34 @@ curl -N http://localhost:9094/v1/events
 curl -N "http://localhost:9094/v1/events?session=$SID"
 ```
 
+### Event schema
+
+Every event on `/v1/sessions/{id}` and `/v1/events` carries:
+
+- `at`, `direction`, `phase` — when, which side, what stage. `phase` is one of `"request"`, `"response"`, or `"denied"` (terminal denial from a pipeline plugin — typically a jwt-validation failure).
+- `a2a` / `mcp` / `inference` — protocol parser payloads (one at most).
+- `invocations` — per-plugin invocation records for every plugin that ran on the pipeline pass. Structured as `{inbound: [...], outbound: [...]}`; each entry carries `plugin`, `action` (one of 5 values — see below), `reason` (machine-stable code), and optional plugin-specific context (expected issuer, target audience, cache-hit flag, path, etc.). abctl renders one row per invocation, so operators see an explicit per-plugin timeline.
+- `plugins` — escape-hatch map for plugin-specific observability. Keys are plugin names; values are the raw JSON each plugin emitted. Unknown plugins render as opaque JSON in abctl. See `authlib/plugins/CONVENTIONS.md` "Emitting session events" for the producer contract.
+- `identity`, `host`, `statusCode`, `error`, `durationMs` — request-level context.
+
+### Invocation action vocabulary
+
+Every plugin emits one of these 5 action values per invocation, so operators can scan a timeline without memorizing plugin-specific verbs:
+
+| `action` | Meaning | Example |
+|---|---|---|
+| `allow` | Gate plugin permitted the request | jwt-validation on valid token |
+| `deny` | Gate plugin rejected the request; pipeline stops | jwt-validation on bad token, token-exchange on IdP failure |
+| `skip` | Plugin ran but didn't act on this message | jwt-validation on a bypass path; parser whose body didn't match |
+| `modify` | Plugin mutated the message | token-exchange replaced the Authorization header |
+| `observe` | Plugin attached diagnostic data without changing flow | a2a-parser, mcp-parser, inference-parser when they match |
+
+Use `reason` to discriminate within an action — e.g. `skip/path_bypass` vs `skip/no_matching_route` tell different stories at the detail-pane level but both scan as "skip" in the at-a-glance timeline.
+
+### Gotcha: denied requests
+
+Rejected requests (401 / 503) land as `phase: "denied"` events in `/v1/sessions` when at least one pipeline plugin appended an Invocation before rejecting. If you're debugging an unauthorized-access pattern, the default-session bucket (`GET /v1/sessions/default`) is where denial events aggregate.
+
 ### Disabling
 
 Set `session.enabled: false` in the runtime config to turn off the store (and implicitly the API). Setting `listener.session_api_addr: ""` alone is not currently supported as a selective disable — the preset refills it; if you need store-on-API-off, raise an issue.

authbridge/authlib/auth/auth.go

@@ -282,39 +282,43 @@ func (a *Auth) HandleInbound(ctx context.Context, authHeader, path, audience str
 		a.IncInboundDeny(DENY_NO_HEADER)
 		a.log.Debug("inbound denied: no Authorization header", "path", path)
 		return &InboundResult{
-			Action:     ActionDeny,
-			DenyStatus: http.StatusUnauthorized,
-			DenyReason: "missing Authorization header",
+			Action:         ActionDeny,
+			DenyStatus:     http.StatusUnauthorized,
+			DenyReason:     "missing Authorization header",
+			DenyReasonCode: DENY_NO_HEADER,
 		}
 	}
 	token := extractBearer(authHeader)
 	if token == "" {
 		a.IncInboundDeny(DENY_MALFORMED_HEADER)
 		a.log.Debug("inbound denied: malformed Authorization header", "path", path)
 		return &InboundResult{
-			Action:     ActionDeny,
-			DenyStatus: http.StatusUnauthorized,
-			DenyReason: "invalid Authorization header format",
+			Action:         ActionDeny,
+			DenyStatus:     http.StatusUnauthorized,
+			DenyReason:     "invalid Authorization header format",
+			DenyReasonCode: DENY_MALFORMED_HEADER,
 		}
 	}
 
 	// 3. Validate JWT
 	if a.verifier == nil {
 		a.IncInboundDeny(DENY_VALIDATOR_MISSING)
 		return &InboundResult{
-			Action:     ActionDeny,
-			DenyStatus: http.StatusUnauthorized,
-			DenyReason: "inbound validation not configured",
+			Action:         ActionDeny,
+			DenyStatus:     http.StatusUnauthorized,
+			DenyReason:     "inbound validation not configured",
+			DenyReasonCode: DENY_VALIDATOR_MISSING,
 		}
 	}
 	if audience == "" {
 		audience = a.identity.Load().Audience
 	}
 	if audience == "" {
 		return &InboundResult{
-			Action:     ActionDeny,
-			DenyStatus: http.StatusServiceUnavailable,
-			DenyReason: "identity not yet configured (credentials pending)",
+			Action:         ActionDeny,
+			DenyStatus:     http.StatusServiceUnavailable,
+			DenyReason:     "identity not yet configured (credentials pending)",
+			DenyReasonCode: DENY_VALIDATOR_MISSING,
 		}
 	}
 	a.log.Debug("validating inbound JWT", "path", path, "expectedAudience", audience)
@@ -330,9 +334,10 @@ func (a *Auth) HandleInbound(ctx context.Context, authHeader, path, audience str
 			"expectedIssuer", a.identity.Load().Audience,
 			"error", err)
 		return &InboundResult{
-			Action:     ActionDeny,
-			DenyStatus: http.StatusUnauthorized,
-			DenyReason: "token validation failed",
+			Action:         ActionDeny,
+			DenyStatus:     http.StatusUnauthorized,
+			DenyReason:     "token validation failed",
+			DenyReasonCode: DENY_JWT_FAILED,
 		}
 	}
 
@@ -396,7 +401,14 @@ func (a *Auth) HandleOutbound(ctx context.Context, authHeader, host string) *Out
 		if cached, ok := a.cache.Get(subjectToken, audience); ok {
 			a.IncOutboundReplaceToken(OUTBOUND_ACTION_CACHE_HIT)
 			a.log.Debug("outbound cache hit", "host", host, "audience", audience)
-			return &OutboundResult{Action: ActionReplaceToken, Token: cached}
+			return &OutboundResult{
+				Action:          ActionReplaceToken,
+				Token:           cached,
+				CacheHit:        true,
+				RouteMatched:    true,
+				TargetAudience:  audience,
+				RequestedScopes: scopes,
+			}
 		}
 	}
 
@@ -437,9 +449,13 @@ func (a *Auth) HandleOutbound(ctx context.Context, authHeader, host string) *Out
 			"tokenEndpoint", resolved.TokenEndpoint,
 			"error", err)
 		return &OutboundResult{
-			Action:     ActionDeny,
-			DenyStatus: http.StatusServiceUnavailable,
-			DenyReason: "token exchange failed",
+			Action:          ActionDeny,
+			DenyStatus:      http.StatusServiceUnavailable,
+			DenyReason:      "token exchange failed",
+			DenyReasonCode:  OUTBOUND_TOKEN_EXCHANGE_FAILED,
+			RouteMatched:    true,
+			TargetAudience:  audience,
+			RequestedScopes: scopes,
 		}
 	}
 
@@ -453,7 +469,13 @@ func (a *Auth) HandleOutbound(ctx context.Context, authHeader, host string) *Out
 	a.log.Info("outbound token exchanged", "host", host, "audience", audience)
 	a.log.Debug("outbound exchange details",
 		"host", host, "audience", audience, "expiresIn", resp.ExpiresIn)
-	return &OutboundResult{Action: ActionReplaceToken, Token: resp.AccessToken}
+	return &OutboundResult{
+		Action:          ActionReplaceToken,
+		Token:           resp.AccessToken,
+		RouteMatched:    true,
+		TargetAudience:  audience,
+		RequestedScopes: scopes,
+	}
 }
 
 func (a *Auth) handleNoToken(ctx context.Context, audience, scopes string) *OutboundResult {
@@ -469,9 +491,10 @@ func (a *Auth) handleNoToken(ctx context.Context, audience, scopes string) *Outb
 			a.log.Debug("no token, client_credentials requested but exchanger not configured",
 				"audience", audience)
 			return &OutboundResult{
-				Action:     ActionDeny,
-				DenyStatus: http.StatusServiceUnavailable,
-				DenyReason: "exchanger not configured for client credentials",
+				Action:         ActionDeny,
+				DenyStatus:     http.StatusServiceUnavailable,
+				DenyReason:     "exchanger not configured for client credentials",
+				DenyReasonCode: OUTBOUND_CREDS_REQUESTED_NO_EXCHANGER,
 			}
 		}
 		a.log.Debug("no token, falling back to client_credentials",
@@ -483,9 +506,10 @@ func (a *Auth) handleNoToken(ctx context.Context, audience, scopes string) *Outb
 			a.log.Debug("client credentials failure details",
 				"audience", audience, "scopes", scopes, "error", err)
 			return &OutboundResult{
-				Action:     ActionDeny,
-				DenyStatus: http.StatusServiceUnavailable,
-				DenyReason: "client credentials token acquisition failed",
+				Action:         ActionDeny,
+				DenyStatus:     http.StatusServiceUnavailable,
+				DenyReason:     "client credentials token acquisition failed",
+				DenyReasonCode: OUTBOUND_CREDENTIALS_GRANT_FAILURE,
 			}
 		}
 		a.IncOutboundReplaceToken(OUTBOUND_ACTION_REPLACE_TOKEN)
@@ -496,9 +520,10 @@ func (a *Auth) handleNoToken(ctx context.Context, audience, scopes string) *Outb
 		a.log.Debug("no token, policy denies request",
 			"policy", a.noTokenPolicy, "audience", audience)
 		return &OutboundResult{
-			Action:     ActionDeny,
-			DenyStatus: http.StatusUnauthorized,
-			DenyReason: "missing Authorization header",
+			Action:         ActionDeny,
+			DenyStatus:     http.StatusUnauthorized,
+			DenyReason:     "missing Authorization header",
+			DenyReasonCode: OUTBOUND_NO_TOKEN,
 		}
 	}
 }

authbridge/authlib/auth/auth_test.go

@@ -113,6 +113,64 @@ func TestHandleInbound_NoVerifier_Denies(t *testing.T) {
 	}
 }
 
+// DenyReasonCode must be populated on every ActionDeny InboundResult. The
+// listener/plugin layer uses the code (not the free-form DenyReason string)
+// to populate SessionEvent.Auth.Inbound[].Reason for machine-stable
+// filtering. A drift here would silently leave the event field empty on
+// denied requests.
+func TestHandleInbound_PopulatesDenyReasonCode(t *testing.T) {
+	cases := []struct {
+		name       string
+		cfg        Config
+		authHeader string
+		audience   string
+		want       InboundDenialReason
+	}{
+		{
+			name:       "no header",
+			cfg:        Config{Verifier: &mockVerifier{}},
+			authHeader: "",
+			want:       DENY_NO_HEADER,
+		},
+		{
+			name:       "malformed header",
+			cfg:        Config{Verifier: &mockVerifier{}},
+			authHeader: "Basic abc",
+			want:       DENY_MALFORMED_HEADER,
+		},
+		{
+			name:       "validator missing",
+			cfg:        Config{}, // nil Verifier
+			authHeader: "Bearer tok",
+			want:       DENY_VALIDATOR_MISSING,
+		},
+		{
+			name: "jwt failed",
+			cfg: Config{
+				Verifier: &mockVerifier{err: fmt.Errorf("bad signature")},
+				Identity: IdentityConfig{Audience: "aud"},
+			},
+			authHeader: "Bearer tok",
+			want:       DENY_JWT_FAILED,
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			a := New(tc.cfg)
+			res := a.HandleInbound(context.Background(), tc.authHeader, "/api", tc.audience)
+			if res.Action != ActionDeny {
+				t.Fatalf("expected deny, got %q", res.Action)
+			}
+			if res.DenyReasonCode != tc.want {
+				t.Errorf("DenyReasonCode = %v, want %v", res.DenyReasonCode, tc.want)
+			}
+			if res.DenyReason == "" {
+				t.Error("DenyReason (human string) should still be populated alongside code")
+			}
+		})
+	}
+}
+
 func TestHandleInbound_AudienceOverride(t *testing.T) {
 	mv := &mockVerifier{claims: validClaims()}
 	a := New(Config{

authbridge/authlib/auth/result.go

@@ -27,16 +27,28 @@ const (
 
 // InboundResult is the outcome of inbound JWT validation.
 type InboundResult struct {
-	Action     string            // ActionAllow or ActionDeny
-	Claims     *validation.Claims // non-nil when a valid JWT was present
-	DenyStatus int               // HTTP status code (e.g., 401)
-	DenyReason string            // human-readable error
+	Action         string             // ActionAllow or ActionDeny
+	Claims         *validation.Claims // non-nil when a valid JWT was present
+	DenyStatus     int                // HTTP status code (e.g., 401)
+	DenyReason     string             // human-readable error — safe for logs, response bodies
+	DenyReasonCode InboundDenialReason // machine-stable enum paired with the /stats counter; use for filtering / indexing session events
 }
 
 // OutboundResult is the outcome of outbound token exchange.
 type OutboundResult struct {
-	Action     string // ActionAllow, ActionReplaceToken, or ActionDeny
-	Token      string // replacement token (when Action == ActionReplaceToken)
-	DenyStatus int    // HTTP status code (e.g., 503)
-	DenyReason string // human-readable error
+	Action         string               // ActionAllow, ActionReplaceToken, or ActionDeny
+	Token          string               // replacement token (when Action == ActionReplaceToken)
+	DenyStatus     int                  // HTTP status code (e.g., 503)
+	DenyReason     string               // human-readable error
+	DenyReasonCode OutboundDenialReason // machine-stable enum paired with the /stats counter
+	CacheHit       bool                 // true when Token was served from the exchange cache; safe to read on any Action
+
+	// Route context populated when the router matched the request. Gives
+	// observability callers (token-exchange plugin writing a SessionEvent)
+	// a way to surface which route was chosen without reaching into
+	// routing.ResolvedRoute directly. RouteMatched=false means the
+	// request passed through because no route fired.
+	RouteMatched    bool   // true when Router.Resolve returned a non-nil route (regardless of action)
+	TargetAudience  string // resolved route's audience (or deriver's output); empty on passthrough
+	RequestedScopes string // resolved route's scopes, raw space-separated; empty on passthrough
 }