diff --git a/authbridge/authlib/config/config.go b/authbridge/authlib/config/config.go
index effda9173..aa15d5525 100644
--- a/authbridge/authlib/config/config.go
+++ b/authbridge/authlib/config/config.go
@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"os"
 
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
 	"gopkg.in/yaml.v3"
 )
 
@@ -67,19 +68,25 @@ type PipelineStageConfig struct {
 // PluginEntry names a plugin and optionally carries per-instance config.
 //
 // The YAML accepts both the bare-name form ("jwt-validation") and the
-// full form ({name, id, config}). The short form keeps existing pipeline
-// configs parsing unchanged; the long form is what plugins that
-// implement pipeline.Configurable actually need. See
+// full form ({name, id, on_error, config}). The short form keeps
+// existing pipeline configs parsing unchanged; the long form is what
+// plugins that implement pipeline.Configurable actually need. See
 // authbridge/docs/plugin-reference.md for the convention plugins
 // follow when decoding Config.
 //
 // Config is captured as a raw subtree via json.RawMessage so the plugin
 // can do its own DisallowUnknownFields decode against a typed struct —
 // the framework does not interpret it.
+//
+// OnError is the framework-owned wrapper policy (see ErrorPolicy).
+// Plugin authors do not consume it — it lives outside the plugin's
+// own config block so all plugins get the same rollout story without
+// each one re-implementing shadow mode.
 type PluginEntry struct {
-	Name   string          `yaml:"name" json:"name"`
-	ID     string          `yaml:"id,omitempty" json:"id,omitempty"`
-	Config json.RawMessage `yaml:"-" json:"config,omitempty"`
+	Name    string               `yaml:"name" json:"name"`
+	ID      string               `yaml:"id,omitempty" json:"id,omitempty"`
+	OnError pipeline.ErrorPolicy `yaml:"on_error,omitempty" json:"on_error,omitempty"`
+	Config  json.RawMessage      `yaml:"-" json:"config,omitempty"`
 }
 
 // UnmarshalYAML accepts either a bare string or a map. The string form
@@ -108,6 +115,16 @@ func (p *PluginEntry) UnmarshalYAML(node *yaml.Node) error {
 				if err := val.Decode(&p.ID); err != nil {
 					return fmt.Errorf("plugin entry id: %w", err)
 				}
+			case "on_error":
+				var raw string
+				if err := val.Decode(&raw); err != nil {
+					return fmt.Errorf("plugin entry on_error: %w", err)
+				}
+				policy := pipeline.ErrorPolicy(raw)
+				if !policy.Valid() {
+					return fmt.Errorf("plugin entry on_error: %q is not a valid policy (expected: enforce, observe, off)", raw)
+				}
+				p.OnError = policy
 			case "config":
 				// Explicit `config: null` (or `config:` with no value)
 				// decodes to a null-tagged scalar node. Normalize to
diff --git a/authbridge/authlib/config/config_test.go b/authbridge/authlib/config/config_test.go
index 5599aff5c..1a3c4228d 100644
--- a/authbridge/authlib/config/config_test.go
+++ b/authbridge/authlib/config/config_test.go
@@ -8,6 +8,8 @@ import (
 	"strings"
 	"testing"
 	"time"
+
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
 )
 
 // --- Preset Tests ---
@@ -436,3 +438,78 @@ func TestSessionConfig_SessionEnabled(t *testing.T) {
 		})
 	}
 }
+
+// TestPluginEntry_OnError covers the three accepted policy values plus
+// the omitted case. An invalid policy must fail loud at YAML parse
+// time so operators catch typos before the pod boots.
+func TestPluginEntry_OnError(t *testing.T) {
+	cases := []struct {
+		name     string
+		yaml     string
+		want     pipeline.ErrorPolicy
+		wantFail bool
+	}{
+		{"enforce explicit", "enforce", pipeline.ErrorPolicyEnforce, false},
+		{"observe", "observe", pipeline.ErrorPolicyObserve, false},
+		{"off", "off", pipeline.ErrorPolicyOff, false},
+		{"typo rejected", "observer", "", true},
+		{"upper rejected", "ENFORCE", "", true},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			dir := t.TempDir()
+			path := filepath.Join(dir, "config.yaml")
+			content := "mode: envoy-sidecar\n" +
+				"pipeline:\n" +
+				"  inbound:\n" +
+				"    plugins:\n" +
+				"      - name: custom-guardrail\n" +
+				"        on_error: " + c.yaml + "\n"
+			if err := os.WriteFile(path, []byte(content), 0600); err != nil {
+				t.Fatal(err)
+			}
+			cfg, err := Load(path)
+			if c.wantFail {
+				if err == nil {
+					t.Fatalf("expected parse error for on_error=%q", c.yaml)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("Load: %v", err)
+			}
+			got := cfg.Pipeline.Inbound.Plugins[0].OnError
+			if got != c.want {
+				t.Errorf("OnError = %q, want %q", got, c.want)
+			}
+		})
+	}
+}
+
+// TestPluginEntry_OnError_Omitted verifies the empty-string default.
+// Resolved() should treat an absent on_error as enforce; the parsed
+// field itself stays empty so round-tripping YAML doesn't invent a
+// value the operator didn't write.
+func TestPluginEntry_OnError_Omitted(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "config.yaml")
+	content := "mode: envoy-sidecar\n" +
+		"pipeline:\n" +
+		"  inbound:\n" +
+		"    plugins:\n" +
+		"      - name: custom-guardrail\n"
+	if err := os.WriteFile(path, []byte(content), 0600); err != nil {
+		t.Fatal(err)
+	}
+	cfg, err := Load(path)
+	if err != nil {
+		t.Fatalf("Load: %v", err)
+	}
+	got := cfg.Pipeline.Inbound.Plugins[0].OnError
+	if got != "" {
+		t.Errorf("omitted OnError parsed as %q, want empty", got)
+	}
+	if got.Resolved() != pipeline.ErrorPolicyEnforce {
+		t.Errorf("Resolved() of empty = %q, want enforce", got.Resolved())
+	}
+}
diff --git a/authbridge/authlib/pipeline/context.go b/authbridge/authlib/pipeline/context.go
index dedf50114..c130cdbbd 100644
--- a/authbridge/authlib/pipeline/context.go
+++ b/authbridge/authlib/pipeline/context.go
@@ -104,14 +104,18 @@ type Context struct {
 
 	Extensions Extensions
 
-	// currentPlugin and currentPhase are framework-owned fields set by
-	// Pipeline.Run / RunResponse around each plugin dispatch. They feed
-	// the Record / Allow / Skip / Observe / Modify / DenyAndRecord helper
-	// methods so plugin code doesn't have to repeat its own Name(),
-	// the phase it's in, or the direction on every Invocation literal.
-	// Unexported so plugins can only set them indirectly (via the framework).
+	// currentPlugin, currentPhase, and currentPolicy are framework-owned
+	// fields set by Pipeline.Run / RunResponse around each plugin
+	// dispatch. They feed the Record / Allow / Skip / Observe / Modify /
+	// DenyAndRecord helper methods so plugin code doesn't have to repeat
+	// its own Name(), the phase it's in, or the direction on every
+	// Invocation literal. currentPolicy is consulted by SetBody /
+	// SetResponseBody to suppress mutation propagation when the current
+	// plugin runs under ErrorPolicyObserve. Unexported so plugins can
+	// only set them indirectly (via the framework).
 	currentPlugin string
 	currentPhase  InvocationPhase
+	currentPolicy ErrorPolicy
 
 	// bodyMutated / responseBodyMutated flag that a plugin called
 	// SetBody / SetResponseBody on this context. Listeners read the flag
@@ -135,16 +139,35 @@ type Context struct {
 // pctx in their own dispatch loops (not strictly via Pipeline.Run) need
 // to set the same fields to get consistent Invocation attribution.
 // Production plugins should never call this directly.
+//
+// Sets the policy to the default (enforce). Call setCurrent (unexported)
+// to stamp a non-default policy — only Pipeline.Run/RunResponse should
+// supply a non-default, and they use the unexported path.
 func (c *Context) SetCurrentPlugin(name string, phase InvocationPhase) {
-	c.currentPlugin = name
-	c.currentPhase = phase
+	c.setCurrent(name, phase, ErrorPolicyEnforce)
 }
 
 // ClearCurrentPlugin resets the framework-owned attribution fields.
 // Paired with SetCurrentPlugin.
 func (c *Context) ClearCurrentPlugin() {
+	c.clearCurrent()
+}
+
+// setCurrent stamps the per-dispatch attribution including the current
+// plugin's on_error policy. Internal to the pipeline package; exported
+// SetCurrentPlugin is the listener-facing entry point and defaults
+// policy to enforce.
+func (c *Context) setCurrent(name string, phase InvocationPhase, policy ErrorPolicy) {
+	c.currentPlugin = name
+	c.currentPhase = phase
+	c.currentPolicy = policy.Resolved()
+}
+
+// clearCurrent zeroes the per-dispatch attribution fields.
+func (c *Context) clearCurrent() {
 	c.currentPlugin = ""
 	c.currentPhase = ""
+	c.currentPolicy = ""
 }
 
 // Record appends an Invocation to pctx under the current pipeline
@@ -224,6 +247,13 @@ func (c *Context) DenyAndRecord(reason, code, message string) Action {
 // SetBody mutate the in-memory Context (readers downstream see the
 // change), but the wire is unchanged.
 //
+// Under ErrorPolicyObserve (shadow mode) SetBody is a NO-OP on bytes:
+// the in-memory body is not replaced, bodyMutated stays false, and
+// downstream plugins continue to see the original. A modify
+// Invocation is still recorded — with Shadow=true — so operators can
+// count "would have redacted" on the rollout dashboard. Plugin code
+// therefore looks identical under enforce and observe.
+//
 // SetBody auto-emits a modify-action Invocation with Reason
 // "body_rewritten" and publishes a plugin-public event under
 // "body-mutation/event" carrying the before/after length and sha256 —
@@ -233,6 +263,10 @@ func (c *Context) DenyAndRecord(reason, code, message string) Action {
 // Callers should NOT assign pctx.Body directly — the listener wouldn't
 // know to propagate the change, and the Invocation wouldn't be emitted.
 func (c *Context) SetBody(newBody []byte) {
+	if c.currentPolicy == ErrorPolicyObserve {
+		c.recordShadowBodyMutation("request", c.Body, newBody)
+		return
+	}
 	old := c.Body
 	c.Body = newBody
 	c.bodyMutated = true
@@ -242,14 +276,44 @@ func (c *Context) SetBody(newBody []byte) {
 // SetResponseBody is the response-side analogue of SetBody. Used by
 // plugins that redact or rewrite the upstream response (prompt-safety
 // guardrails on LLM output, content filters, DLP). Same contract —
-// Invocation + body-mutation/event emitted; never logs the body.
+// Invocation + body-mutation/event emitted; never logs the body —
+// and the same observe-mode suppression: under ErrorPolicyObserve the
+// response body is untouched and the Invocation is marked Shadow=true.
 func (c *Context) SetResponseBody(newBody []byte) {
+	if c.currentPolicy == ErrorPolicyObserve {
+		c.recordShadowBodyMutation("response", c.ResponseBody, newBody)
+		return
+	}
 	old := c.ResponseBody
 	c.ResponseBody = newBody
 	c.responseBodyMutated = true
 	c.emitBodyMutation("response", old, newBody)
 }
 
+// recordShadowBodyMutation emits the would-mutate Invocation for a
+// SetBody / SetResponseBody call that was suppressed under observe
+// mode. Mirrors emitBodyMutation's telemetry (length + sha256 delta)
+// so dashboards get the same shape they see under enforce, just with
+// Shadow=true and no wire-level effect.
+func (c *Context) recordShadowBodyMutation(phase string, oldBody, newBody []byte) {
+	c.Record(Invocation{
+		Action: ActionModify,
+		Reason: "body_rewritten",
+		Shadow: true,
+	})
+	if c.Extensions.Custom == nil {
+		c.Extensions.Custom = map[string]any{}
+	}
+	c.Extensions.Custom["body-mutation"+PluginEventSuffix] = bodyMutationEvent{
+		Phase:        phase,
+		Plugin:       c.currentPlugin,
+		LengthBefore: len(oldBody),
+		LengthAfter:  len(newBody),
+		SHA256Before: hashHex(oldBody),
+		SHA256After:  hashHex(newBody),
+	}
+}
+
 // BodyMutated reports whether a plugin called SetBody during this
 // request. Listeners check this after Run to decide whether to emit a
 // body mutation on the wire. Stream-scoped — a new Context starts with
@@ -353,3 +417,37 @@ type AgentIdentity struct {
 	WorkloadID  string
 	TrustDomain string
 }
+
+// markLastInvocationShadow finds the most recent Invocation authored
+// by pluginName in the given phase (within the current direction
+// bucket) and flips its Shadow flag to true. Returns true when a
+// matching record was found. Used by the pipeline to retroactively
+// tag a plugin's deny record as shadow once Pipeline.Run has observed
+// that the plugin returned Reject under ErrorPolicyObserve.
+//
+// Walking backwards and matching on Plugin+Phase is O(N) in the
+// worst case but typically short-circuits at index -1 — plugins
+// almost always Record right before returning Reject, so the last
+// entry matches.
+func (c *Context) markLastInvocationShadow(pluginName string, phase InvocationPhase) bool {
+	if c.Extensions.Invocations == nil {
+		return false
+	}
+	var list []Invocation
+	switch c.Direction {
+	case Inbound:
+		list = c.Extensions.Invocations.Inbound
+	case Outbound:
+		list = c.Extensions.Invocations.Outbound
+	default:
+		return false
+	}
+	for i := len(list) - 1; i >= 0; i-- {
+		if list[i].Plugin != pluginName || list[i].Phase != phase {
+			continue
+		}
+		list[i].Shadow = true
+		return true
+	}
+	return false
+}
diff --git a/authbridge/authlib/pipeline/extensions.go b/authbridge/authlib/pipeline/extensions.go
index 73d526e0c..9ab73bb6c 100644
--- a/authbridge/authlib/pipeline/extensions.go
+++ b/authbridge/authlib/pipeline/extensions.go
@@ -231,6 +231,46 @@ type Invocations struct {
 	Outbound []Invocation `json:"outbound,omitempty"`
 }
 
+// FilteredByPhase returns a new *Invocations containing only entries
+// whose Phase matches the argument. Strict match — untagged entries
+// (Phase == "") are dropped because the framework always populates
+// Phase via Context.Record; an untagged entry is a plugin bug and
+// including it in the wrong phase would double-report in one event
+// and be missing from the other.
+//
+// The underlying Invocation values are copied shallowly; mutating a
+// returned entry's Details map mutates the source. Acceptable for
+// the per-request flow where the original lives only on pctx and is
+// discarded after recording.
+//
+// Returns nil when no entries match so callers can drop the field
+// from the SessionEvent without a null check.
+//
+// Intended for reject-event recording in listeners and for the
+// accept-path phase split (one SessionEvent per phase). Listeners
+// that need full independence from pctx's lifecycle layer their own
+// snapshot on top.
+func (in *Invocations) FilteredByPhase(phase InvocationPhase) *Invocations {
+	if in == nil {
+		return nil
+	}
+	out := &Invocations{}
+	for _, inv := range in.Inbound {
+		if inv.Phase == phase {
+			out.Inbound = append(out.Inbound, inv)
+		}
+	}
+	for _, inv := range in.Outbound {
+		if inv.Phase == phase {
+			out.Outbound = append(out.Outbound, inv)
+		}
+	}
+	if out.Inbound == nil && out.Outbound == nil {
+		return nil
+	}
+	return out
+}
+
 // Invocation records one plugin's action on one pipeline pass. Plugin is
 // the plugin's Name() for traceability. Action is the universal 5-value
 // verb (see Action). Reason is a stable machine-readable label paired
@@ -295,6 +335,15 @@ type Invocation struct {
 	// The session API has no auth on it — only safe-to-log data
 	// belongs in Invocation.Details.
 	Details map[string]string `json:"details,omitempty"`
+
+	// Shadow reports that the plugin ran under on_error: observe and
+	// its decision (deny or modify) was NOT applied to the request.
+	// An operator reading a would-have-blocked timeline filters on
+	// Shadow=true to count rollout-candidate events; a dashboard that
+	// aggregates "effective denies" filters Shadow=false. The
+	// framework, not the plugin, sets this — plugin code looks
+	// identical under enforce and observe.
+	Shadow bool `json:"shadow,omitempty"`
 }
 
 // DelegationExtension tracks the token delegation chain across hops.
diff --git a/authbridge/authlib/pipeline/pipeline.go b/authbridge/authlib/pipeline/pipeline.go
index 8e89aa36b..0ceec20dc 100644
--- a/authbridge/authlib/pipeline/pipeline.go
+++ b/authbridge/authlib/pipeline/pipeline.go
@@ -7,8 +7,13 @@ import (
 )
 
 // Pipeline holds an ordered list of plugins and runs them sequentially.
+// policies[i] holds the on_error ErrorPolicy that wraps plugins[i]; the
+// slice is always the same length as plugins (guaranteed by New) so
+// policyAt is a bounds-safe lookup. An empty ErrorPolicy resolves to
+// ErrorPolicyEnforce via the Resolved() method.
 type Pipeline struct {
-	plugins []Plugin
+	plugins  []Plugin
+	policies []ErrorPolicy
 }
 
 // defaultSlots lists the built-in extension slot names.
@@ -26,6 +31,7 @@ type Option func(*options)
 
 type options struct {
 	extraSlots []string
+	policies   []ErrorPolicy
 }
 
 // WithSlots registers additional valid extension slot names beyond the built-in set.
@@ -36,6 +42,18 @@ func WithSlots(slots ...string) Option {
 	}
 }
 
+// WithPolicies attaches per-plugin on_error policies in parallel with
+// the plugin slice passed to New. policies[i] belongs to plugins[i];
+// an empty entry defaults to ErrorPolicyEnforce. If fewer policies are
+// supplied than plugins, the remaining plugins use the default
+// (enforce). Supplying more policies than plugins is a programmer
+// error and New returns an error.
+func WithPolicies(policies ...ErrorPolicy) Option {
+	return func(o *options) {
+		o.policies = append(o.policies, policies...)
+	}
+}
+
 // New creates a Pipeline from the given plugins after validating capability wiring.
 // Returns an error if any plugin declares a read on a slot that no earlier plugin writes.
 func New(plugins []Plugin, opts ...Option) (*Pipeline, error) {
@@ -53,30 +71,53 @@ func New(plugins []Plugin, opts ...Option) (*Pipeline, error) {
 	if err := validateCapabilities(plugins, validSlots); err != nil {
 		return nil, err
 	}
-	return &Pipeline{plugins: plugins}, nil
+	if len(o.policies) > len(plugins) {
+		return nil, fmt.Errorf("pipeline: WithPolicies has %d entries but only %d plugins", len(o.policies), len(plugins))
+	}
+	policies := make([]ErrorPolicy, len(plugins))
+	copy(policies, o.policies)
+	return &Pipeline{plugins: plugins, policies: policies}, nil
 }
 
 // Run executes the request phase of the pipeline sequentially.
 // If any plugin returns Reject, the pipeline stops and returns that action
 // with Violation.PluginName populated.
 //
+// Plugins configured with ErrorPolicyOff are skipped entirely — they
+// are not dispatched and contribute no Invocation. Plugins under
+// ErrorPolicyObserve are dispatched normally, but a Reject return is
+// converted into a pass-through: the Violation is recorded as a
+// shadow Invocation and the pipeline continues to the next plugin.
+// Body mutations under observe are also suppressed — see
+// Context.SetBody / SetResponseBody.
+//
 // Before dispatching into each plugin, Run stamps pctx with the plugin's
-// name and the current phase so the plugin's Record / Allow / Skip /
-// Observe / Modify / DenyAndRecord helpers can fill Invocation.Plugin
-// and Invocation.Phase automatically. The stamp is cleared after each
-// plugin returns so a plugin that spawns a goroutine capturing pctx
-// won't mis-attribute a late-arriving Record to itself.
+// name, the current phase, and the current policy so the plugin's
+// Record / Allow / Skip / Observe / Modify / DenyAndRecord helpers can
+// fill Invocation.Plugin and Invocation.Phase automatically. The stamp
+// is cleared after each plugin returns so a plugin that spawns a
+// goroutine capturing pctx won't mis-attribute a late-arriving Record
+// to itself.
 func (p *Pipeline) Run(ctx context.Context, pctx *Context) Action {
-	for _, plugin := range p.plugins {
+	for i, plugin := range p.plugins {
+		policy := p.policyAt(i)
+		if policy == ErrorPolicyOff {
+			slog.Debug("pipeline: plugin disabled (on_error: off)", "plugin", plugin.Name())
+			continue
+		}
 		if ctx.Err() != nil {
 			slog.Info("pipeline: request cancelled", "plugin", plugin.Name())
 			return Deny("pipeline.cancelled", "request cancelled")
 		}
-		pctx.SetCurrentPlugin(plugin.Name(), InvocationPhaseRequest)
+		pctx.setCurrent(plugin.Name(), InvocationPhaseRequest, policy)
 		action := plugin.OnRequest(ctx, pctx)
-		pctx.ClearCurrentPlugin()
+		pctx.clearCurrent()
 		if action.Type == Reject {
 			stampPluginName(&action, plugin.Name())
+			if policy == ErrorPolicyObserve {
+				markShadowAndLog(pctx, plugin.Name(), InvocationPhaseRequest, action, "request")
+				continue
+			}
 			logReject(plugin.Name(), action, "pipeline: plugin rejected request")
 			return action
 		}
@@ -88,19 +129,28 @@ func (p *Pipeline) Run(ctx context.Context, pctx *Context) Action {
 // RunResponse executes the response phase in reverse order.
 // The last plugin in the chain sees the response first.
 //
-// See Run for the pctx attribution stamping. Same pattern, phase set
-// to InvocationPhaseResponse.
+// See Run for the pctx attribution stamping, the off-policy skip, and
+// the observe-policy shadow conversion. Same pattern, phase set to
+// InvocationPhaseResponse.
 func (p *Pipeline) RunResponse(ctx context.Context, pctx *Context) Action {
 	for i := len(p.plugins) - 1; i >= 0; i-- {
+		policy := p.policyAt(i)
+		if policy == ErrorPolicyOff {
+			continue
+		}
 		if ctx.Err() != nil {
 			slog.Info("pipeline: response cancelled", "plugin", p.plugins[i].Name())
 			return Deny("pipeline.cancelled", "request cancelled")
 		}
-		pctx.SetCurrentPlugin(p.plugins[i].Name(), InvocationPhaseResponse)
+		pctx.setCurrent(p.plugins[i].Name(), InvocationPhaseResponse, policy)
 		action := p.plugins[i].OnResponse(ctx, pctx)
-		pctx.ClearCurrentPlugin()
+		pctx.clearCurrent()
 		if action.Type == Reject {
 			stampPluginName(&action, p.plugins[i].Name())
+			if policy == ErrorPolicyObserve {
+				markShadowAndLog(pctx, p.plugins[i].Name(), InvocationPhaseResponse, action, "response")
+				continue
+			}
 			logReject(p.plugins[i].Name(), action, "pipeline: plugin rejected response")
 			return action
 		}
@@ -108,6 +158,63 @@ func (p *Pipeline) RunResponse(ctx context.Context, pctx *Context) Action {
 	return Action{Type: Continue}
 }
 
+// policyAt returns the resolved policy for plugins[i]. The policies
+// slice is always the same length as plugins (New guarantees this),
+// but we check defensively so a zero-value Pipeline (constructed
+// outside New, e.g. in a test) doesn't panic.
+func (p *Pipeline) policyAt(i int) ErrorPolicy {
+	if i < len(p.policies) {
+		return p.policies[i].Resolved()
+	}
+	return ErrorPolicyEnforce
+}
+
+// markShadowAndLog records the would-have-denied Invocation as
+// Shadow=true and emits a WARN log. If the plugin already appended a
+// deny Invocation (typical for gate plugins that call
+// DenyAndRecord / Record before returning Reject), we mark that
+// record instead of appending a duplicate — otherwise dashboards
+// would double-count a single decision. Synthesize a record only
+// when the plugin returned Reject without having recorded its own
+// invocation (rare: plugin bug or non-recording denial helper).
+func markShadowAndLog(pctx *Context, pluginName string, phase InvocationPhase, action Action, phaseLabel string) {
+	status, _, _ := action.Violation.Render()
+	marked := pctx.markLastInvocationShadow(pluginName, phase)
+	if !marked {
+		// Use the Violation's machine-stable code as Reason so
+		// dashboards grouping denials by reason see the plugin's
+		// actual deny code for both recorded and synthesized paths.
+		// The "synthesized" signal lives in Details so operators can
+		// still distinguish "plugin Recorded then Deny'd" from
+		// "plugin Deny'd without Recording" when debugging.
+		reason := "plugin.unspecified"
+		if action.Violation != nil && action.Violation.Code != "" {
+			reason = action.Violation.Code
+		}
+		inv := Invocation{
+			Plugin: pluginName,
+			Phase:  phase,
+			Action: ActionDeny,
+			Reason: reason,
+			Path:   pctx.Path,
+			Shadow: true,
+		}
+		if action.Violation != nil {
+			inv.Details = map[string]string{
+				"synthesized":       "true",
+				"would_deny_reason": action.Violation.Reason,
+			}
+		}
+		pctx.Record(inv)
+	}
+	slog.Warn("pipeline: plugin would have denied (shadow)",
+		"plugin", pluginName,
+		"phase", phaseLabel,
+		"status", status,
+		"code", action.Violation.Code,
+		"reason", action.Violation.Reason)
+}
+
 // stampPluginName annotates a reject action with the plugin that produced
 // it, so listeners and clients can attribute the denial without the
 // plugin remembering to set it.
diff --git a/authbridge/authlib/pipeline/pipeline_test.go b/authbridge/authlib/pipeline/pipeline_test.go
index cacb4e0bd..ab3f72406 100644
--- a/authbridge/authlib/pipeline/pipeline_test.go
+++ b/authbridge/authlib/pipeline/pipeline_test.go
@@ -674,3 +674,231 @@ func TestPipelineStop_NoShutdownersIsNoop(t *testing.T) {
 	}
 	p.Stop(context.Background()) // must not panic
 }
+func TestPipelineRun_ObservePolicyConvertsRejectToContinue(t *testing.T) {
+	var downstreamCalled bool
+	shadow := &stubPlugin{
+		name: "would-deny",
+		onReq: func(_ context.Context, pctx *Context) Action {
+			return pctx.DenyAndRecord("guardrail_triggered", "policy.content-blocked", "blocked")
+		},
+	}
+	downstream := &stubPlugin{
+		name: "downstream",
+		onReq: func(_ context.Context, _ *Context) Action {
+			downstreamCalled = true
+			return Action{Type: Continue}
+		},
+	}
+	p, err := New([]Plugin{shadow, downstream}, WithPolicies(ErrorPolicyObserve, ErrorPolicyEnforce))
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	pctx := &Context{}
+	got := p.Run(context.Background(), pctx)
+	if got.Type != Continue {
+		t.Fatalf("action type = %v, want Continue (observe suppresses Reject)", got.Type)
+	}
+	if !downstreamCalled {
+		t.Error("downstream plugin not called; observe mode must not stop the pipeline")
+	}
+	if pctx.Extensions.Invocations == nil || len(pctx.Extensions.Invocations.Inbound) == 0 {
+		t.Fatal("no invocations recorded")
+	}
+	deny := pctx.Extensions.Invocations.Inbound[0]
+	if deny.Action != ActionDeny {
+		t.Errorf("shadow invocation action = %q, want deny (plugin's own record is preserved)", deny.Action)
+	}
+	if !deny.Shadow {
+		t.Error("shadow=false on the would-deny record; observe mode must mark it")
+	}
+}
+
+// TestPipelineRun_EnforceIsDefault confirms that a plugin without a
+// specified policy behaves exactly as before: Reject stops the
+// pipeline and returns a Reject action upstream. Regression guard —
+// the point of adding WithPolicies was to NOT change existing
+// deployments.
+func TestPipelineRun_EnforceIsDefault(t *testing.T) {
+	plugin := &stubPlugin{
+		name: "denier",
+		onReq: func(_ context.Context, _ *Context) Action {
+			return Deny("policy.forbidden", "no")
+		},
+	}
+	// No WithPolicies — entries default to enforce.
+	p, err := New([]Plugin{plugin})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	got := p.Run(context.Background(), &Context{})
+	if got.Type != Reject {
+		t.Errorf("action type = %v, want Reject (default is enforce)", got.Type)
+	}
+}
+
+// TestPipelineRun_OffPolicySkipsDispatch verifies off-mode plugins
+// don't run at all — not even their OnRequest. A kill-switched
+// plugin must leave no trace in invocations, matching the "plugin
+// absent from config" behavior.
+func TestPipelineRun_OffPolicySkipsDispatch(t *testing.T) {
+	var called bool
+	disabled := &stubPlugin{
+		name: "disabled",
+		onReq: func(_ context.Context, _ *Context) Action {
+			called = true
+			return Action{Type: Continue}
+		},
+	}
+	p, err := New([]Plugin{disabled}, WithPolicies(ErrorPolicyOff))
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	pctx := &Context{}
+	got := p.Run(context.Background(), pctx)
+	if got.Type != Continue {
+		t.Errorf("action = %v, want Continue", got.Type)
+	}
+	if called {
+		t.Error("disabled plugin was dispatched; off must skip it")
+	}
+	if pctx.Extensions.Invocations != nil && len(pctx.Extensions.Invocations.Inbound) > 0 {
+		t.Errorf("off-policy plugin appended invocations: %+v", pctx.Extensions.Invocations.Inbound)
+	}
+}
+
+// TestPipelineRunResponse_ObservePolicySuppressesReject ensures the
+// observe semantics apply symmetrically to the response phase.
+// Response-side guardrails (DLP on tool output, jailbreak detection
+// on model replies) are exactly the class that most needs shadow
+// rollout — they inspect generated content the author can't preview.
+func TestPipelineRunResponse_ObservePolicySuppressesReject(t *testing.T) {
+	shadow := &stubPlugin{
+		name: "would-deny-response",
+		onResp: func(_ context.Context, pctx *Context) Action {
+			return pctx.DenyAndRecord("pii_leak", "policy.content-blocked", "leak")
+		},
+	}
+	p, err := New([]Plugin{shadow}, WithPolicies(ErrorPolicyObserve))
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	pctx := &Context{}
+	got := p.RunResponse(context.Background(), pctx)
+	if got.Type != Continue {
+		t.Errorf("action type = %v, want Continue (observe on response phase)", got.Type)
+	}
+	if pctx.Extensions.Invocations == nil || len(pctx.Extensions.Invocations.Inbound) == 0 {
+		t.Fatal("no response-phase invocation recorded")
+	}
+	inv := pctx.Extensions.Invocations.Inbound[0]
+	if inv.Phase != InvocationPhaseResponse {
+		t.Errorf("phase = %q, want response", inv.Phase)
+	}
+	if !inv.Shadow {
+		t.Error("response-phase shadow flag not set")
+	}
+}
+
+// TestPipelineRun_ObserveSynthesizesRecordWhenPluginSkipsIt covers
+// the defensive path: a plugin that returns Reject without first
+// recording its own Invocation (programmer error, or use of the
+// bare Deny helper instead of DenyAndRecord). The framework still
+// produces a Shadow=true record so the would-have-blocked event is
+// visible on the dashboard.
+func TestPipelineRun_ObserveSynthesizesRecordWhenPluginSkipsIt(t *testing.T) {
+	silent := &stubPlugin{
+		name: "silent-rejecter",
+		onReq: func(_ context.Context, _ *Context) Action {
+			// Note: no pctx.Record / DenyAndRecord.
+			return Deny("policy.forbidden", "no")
+		},
+	}
+	p, err := New([]Plugin{silent}, WithPolicies(ErrorPolicyObserve))
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	pctx := &Context{}
+	got := p.Run(context.Background(), pctx)
+	if got.Type != Continue {
+		t.Errorf("action = %v, want Continue", got.Type)
+	}
+	if pctx.Extensions.Invocations == nil || len(pctx.Extensions.Invocations.Inbound) != 1 {
+		t.Fatalf("want exactly one synthesized invocation, got: %+v", pctx.Extensions.Invocations)
+	}
+	inv := pctx.Extensions.Invocations.Inbound[0]
+	// Reason carries the plugin's machine-stable deny code so
+	// dashboards grouping by Reason see the actual deny across
+	// both recorded and synthesized paths.
+	if inv.Reason != "policy.forbidden" {
+		t.Errorf("reason = %q, want policy.forbidden (the plugin's deny code)", inv.Reason)
+	}
+	if !inv.Shadow {
+		t.Error("Shadow flag not set on synthesized record")
+	}
+	// Details["synthesized"] distinguishes "plugin didn't Record then
+	// returned Deny" from "plugin Recorded then returned Deny" for
+	// debugging, without hiding the deny code behind a synthetic Reason.
+	if inv.Details["synthesized"] != "true" {
+		t.Errorf("Details[synthesized] = %q, want true", inv.Details["synthesized"])
+	}
+	if inv.Details["would_deny_reason"] != "no" {
+		t.Errorf("would_deny_reason = %q, want %q", inv.Details["would_deny_reason"], "no")
+	}
+}
+
+// TestSetBody_ObserveModeIsNoop verifies the body-mutation side of
+// observe mode: SetBody records a shadow Invocation but does not
+// replace the in-memory body or flip bodyMutated. Downstream readers
+// and the wire both continue to see the original bytes, so a
+// redaction plugin running in shadow can be trusted to leave
+// production traffic unchanged.
+func TestSetBody_ObserveModeIsNoop(t *testing.T) {
+	mutator := &stubPlugin{
+		name: "redactor",
+		caps: PluginCapabilities{WritesBody: true},
+		onReq: func(_ context.Context, pctx *Context) Action {
+			pctx.SetBody([]byte("REDACTED"))
+			return Action{Type: Continue}
+		},
+	}
+	p, err := New([]Plugin{mutator}, WithPolicies(ErrorPolicyObserve))
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	pctx := &Context{Body: []byte("secret")}
+	got := p.Run(context.Background(), pctx)
+	if got.Type != Continue {
+		t.Fatalf("action = %v, want Continue", got.Type)
+	}
+	if string(pctx.Body) != "secret" {
+		t.Errorf("body = %q, want untouched (observe mode must not replace bytes)", pctx.Body)
+	}
+	if pctx.BodyMutated() {
+		t.Error("BodyMutated=true under observe; wire must see original body")
+	}
+	if pctx.Extensions.Invocations == nil || len(pctx.Extensions.Invocations.Inbound) != 1 {
+		t.Fatalf("want one invocation, got: %+v", pctx.Extensions.Invocations)
+	}
+	inv := pctx.Extensions.Invocations.Inbound[0]
+	if !inv.Shadow || inv.Reason != "body_rewritten" {
+		t.Errorf("expected shadow modify invocation, got action=%q reason=%q shadow=%v",
+			inv.Action, inv.Reason, inv.Shadow)
+	}
+}
+
+// TestNew_RejectsTooManyPolicies locks the defensive check in New:
+// a caller that supplies more policies than plugins is making a
+// parallel-slice mistake, not a well-formed pipeline. Fail loud at
+// construction rather than silently truncate.
+func TestNew_RejectsTooManyPolicies(t *testing.T) {
+	_, err := New(
+		[]Plugin{&stubPlugin{name: "a"}},
+		WithPolicies(ErrorPolicyEnforce, ErrorPolicyObserve),
+	)
+	if err == nil {
+		t.Fatal("expected error; 2 policies for 1 plugin is a parallel-slice bug")
+	}
+	if !strings.Contains(err.Error(), "WithPolicies") {
+		t.Errorf("error should name WithPolicies: %q", err)
+	}
+}
diff --git a/authbridge/authlib/pipeline/policy.go b/authbridge/authlib/pipeline/policy.go
new file mode 100644
index 000000000..e223bb93a
--- /dev/null
+++ b/authbridge/authlib/pipeline/policy.go
@@ -0,0 +1,63 @@
+package pipeline
+
+// ErrorPolicy controls how the pipeline reacts when a plugin returns a
+// Reject action (i.e., decides to deny a request or response). The
+// policy wraps the plugin — plugin authors do not consume it.
+//
+// Default is ErrorPolicyEnforce: Reject becomes an HTTP error response
+// and pipeline execution stops. Operators roll out a new guardrail in
+// ErrorPolicyObserve first: the plugin still evaluates and may still
+// return Reject, but the framework converts the Reject into a
+// pass-through and records the would-have-denied Invocation as
+// Shadow=true. ErrorPolicyOff is a kill-switch — the plugin is
+// not dispatched at all.
+//
+// Shadowing or disabling auth gates (jwt-validation, token-exchange)
+// is an authentication bypass dressed as a feature — operators SHOULD
+// leave those plugins on ErrorPolicyEnforce (the default).
+//
+// The framework does NOT enforce this at startup: nothing prevents
+// on_error: observe on jwt-validation today. A built-in-registry
+// sealing pass is planned as a follow-up to reject a non-enforce
+// policy on reserved gates at build time.
+type ErrorPolicy string
+
+const (
+	// ErrorPolicyEnforce is the default. A Reject action becomes the
+	// HTTP error the Violation describes, and the pipeline stops.
+	ErrorPolicyEnforce ErrorPolicy = "enforce"
+
+	// ErrorPolicyObserve (shadow mode) evaluates the plugin normally
+	// but turns a Reject into a pass-through. Used to canary a new
+	// guardrail: operators watch the shadow-deny counter before
+	// flipping to enforce. Body mutations (SetBody / SetResponseBody)
+	// also do not propagate to the wire under observe — the plugin's
+	// decision logic runs identically, but its effect is muted.
+	ErrorPolicyObserve ErrorPolicy = "observe"
+
+	// ErrorPolicyOff disables the plugin. The plugin is not dispatched.
+	// Off is an operator kill-switch without a redeploy.
+	ErrorPolicyOff ErrorPolicy = "off"
+)
+
+// Valid reports whether p is a recognized ErrorPolicy. The empty
+// string is valid and means "use the default" (enforce); callers that
+// need the defaulted value use the Resolved method instead.
+func (p ErrorPolicy) Valid() bool {
+	switch p {
+	case "", ErrorPolicyEnforce, ErrorPolicyObserve, ErrorPolicyOff:
+		return true
+	default:
+		return false
+	}
+}
+
+// Resolved returns the policy with "" normalized to ErrorPolicyEnforce.
+// Callers that dispatch on policy value should use Resolved so they
+// don't have to treat "" as a synonym for "enforce" themselves.
+func (p ErrorPolicy) Resolved() ErrorPolicy {
+	if p == "" {
+		return ErrorPolicyEnforce
+	}
+	return p
+}
diff --git a/authbridge/authlib/plugins/registry.go b/authbridge/authlib/plugins/registry.go
index 88b0664de..8ff8c409a 100644
--- a/authbridge/authlib/plugins/registry.go
+++ b/authbridge/authlib/plugins/registry.go
@@ -99,7 +99,15 @@ func factoryFor(name string) (PluginFactory, bool) {
 // currently-registered plugin — typo-catching diagnostic.
 func Build(entries []config.PluginEntry, opts ...pipeline.Option) (*pipeline.Pipeline, error) {
 	ps := make([]pipeline.Plugin, 0, len(entries))
+	policies := make([]pipeline.ErrorPolicy, 0, len(entries))
 	for _, e := range entries {
+		// ErrorPolicyOff removes the plugin from the running pipeline
+		// entirely — no Configure, no Init, no dispatch. Operators use
+		// off as a kill-switch without deleting the entry from YAML,
+		// which makes re-enabling a one-line edit.
+		if e.OnError.Resolved() == pipeline.ErrorPolicyOff {
+			continue
+		}
 		factory, ok := factoryFor(e.Name)
 		if !ok {
 			return nil, fmt.Errorf("unknown plugin %q (registered: %v)", e.Name, RegisteredPlugins())
@@ -113,6 +121,8 @@ func Build(entries []config.PluginEntry, opts ...pipeline.Option) (*pipeline.Pip
 			return nil, fmt.Errorf("plugin %q does not accept configuration", e.Name)
 		}
 		ps = append(ps, p)
+		policies = append(policies, e.OnError.Resolved())
 	}
+	opts = append(opts, pipeline.WithPolicies(policies...))
 	return pipeline.New(ps, opts...)
 }
diff --git a/authbridge/cmd/abctl/tui/events_pane.go b/authbridge/cmd/abctl/tui/events_pane.go
index f97586ef2..6f2e9ec41 100644
--- a/authbridge/cmd/abctl/tui/events_pane.go
+++ b/authbridge/cmd/abctl/tui/events_pane.go
@@ -176,6 +176,15 @@ 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 string(r.inv.Action)
 }
 
@@ -612,4 +621,3 @@ func truncateScopes(scopes []string, n int) string {
 	}
 	return strings.Join(scopes[:n], ", ") + fmt.Sprintf(" +%d more", len(scopes)-n)
 }
-
diff --git a/authbridge/cmd/abctl/tui/events_pane_test.go b/authbridge/cmd/abctl/tui/events_pane_test.go
index 06e053913..80f98d9af 100644
--- a/authbridge/cmd/abctl/tui/events_pane_test.go
+++ b/authbridge/cmd/abctl/tui/events_pane_test.go
@@ -63,6 +63,23 @@ func TestInvocationRow_Cells(t *testing.T) {
 			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.
+			wantAction: "deny*",
+			wantPlugin: "pii-scrubber",
+		},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
diff --git a/authbridge/cmd/authbridge/listener/extproc/server.go b/authbridge/cmd/authbridge/listener/extproc/server.go
index 8ff8ffe59..bf6afc115 100644
--- a/authbridge/cmd/authbridge/listener/extproc/server.go
+++ b/authbridge/cmd/authbridge/listener/extproc/server.go
@@ -251,37 +251,75 @@ func (s *Server) recordInboundReject(pctx *pipeline.Context, action pipeline.Act
 	s.Sessions.Append(inboundSessionID(pctx), ev)
 }
 
+// recordOutboundReject emits a SessionDenied event for outbound requests
+// a pipeline plugin rejected. Symmetric to recordInboundReject on the
+// inbound side. Called BEFORE rejectFromAction returns, so denied
+// outbound calls appear in /v1/sessions and abctl rather than vanishing
+// with only a 4xx/5xx on the agent side — the observability surface
+// that guardrail plugins (rate-limit, policy, intent-based) depend on
+// to show operators what they blocked and why.
+//
+// Uses the same ActiveSession bucketing as recordOutboundSession: an
+// outbound call inherits the most-recently-updated session. When no
+// active session exists the event lands in DefaultSessionID. Matches
+// the correctness envelope of the accept path.
+//
+// Skips recording when no Invocations were appended — the deny came
+// from a plugin that didn't contribute diagnostic context, and a
+// content-free SessionDenied event would be noise without attribution.
+func (s *Server) recordOutboundReject(pctx *pipeline.Context, action pipeline.Action) {
+	if s.Sessions == nil || pctx.Extensions.Invocations == nil {
+		return
+	}
+	sid := s.Sessions.ActiveSession()
+	if sid == "" {
+		sid = session.DefaultSessionID
+	}
+	var status int
+	var code, message string
+	if action.Violation != nil {
+		status = action.Violation.Status
+		if status == 0 {
+			status = pipeline.StatusFromCode(action.Violation.Code)
+		}
+		code = action.Violation.Code
+		message = action.Violation.Reason
+	}
+	ev := pipeline.SessionEvent{
+		At:          time.Now(),
+		Direction:   pipeline.Outbound,
+		Phase:       pipeline.SessionDenied,
+		Invocations: snapshotInvocations(pctx.Extensions.Invocations, pipeline.InvocationPhaseRequest),
+		Plugins:     snapshotPlugins(pctx.Extensions.Custom),
+		Identity:    snapshotIdentity(pctx),
+		Host:        pctx.Host,
+		StatusCode:  status,
+		Error: &pipeline.EventError{
+			Kind:    "policy",
+			Code:    code,
+			Message: message,
+		},
+		Duration: durationSince(pctx.StartedAt),
+	}
+	s.Sessions.Append(sid, ev)
+}
+
 // snapshotInvocations returns a shallow copy of the Invocations extension
 // filtered by phase. Plugins append to pctx.Extensions.Invocations as
 // both OnRequest and OnResponse fire; the full list lives there for
 // cross-phase inspection. At record time each SessionEvent should carry
 // only the invocations from its own phase, so request events don't
 // double-report request-phase entries AFTER the response phase has
-// already added its own. Each Invocation carries its phase tag (set by
-// the producer) — request events pass InvocationPhaseRequest, response
-// events pass InvocationPhaseResponse, denied events pass
+// already added its own. Request events pass InvocationPhaseRequest,
+// response events pass InvocationPhaseResponse, denied events pass
 // InvocationPhaseRequest (denial terminates the pass before response
 // runs). Returns nil when no matching entry exists, so the recording
 // gate can check for "no invocations on this phase" cleanly.
+//
+// Thin wrapper over (*Invocations).FilteredByPhase — kept as a named
+// function so recordXxxSession call sites stay grep-friendly.
 func snapshotInvocations(ext *pipeline.Invocations, phase pipeline.InvocationPhase) *pipeline.Invocations {
-	if ext == nil {
-		return nil
-	}
-	var inbound, outbound []pipeline.Invocation
-	for _, inv := range ext.Inbound {
-		if inv.Phase == phase {
-			inbound = append(inbound, inv)
-		}
-	}
-	for _, inv := range ext.Outbound {
-		if inv.Phase == phase {
-			outbound = append(outbound, inv)
-		}
-	}
-	if len(inbound) == 0 && len(outbound) == 0 {
-		return nil
-	}
-	return &pipeline.Invocations{Inbound: inbound, Outbound: outbound}
+	return ext.FilteredByPhase(phase)
 }
 
 // snapshotPlugins collects plugin-public observability events from
@@ -363,18 +401,18 @@ func (s *Server) recordOutboundResponseSession(pctx *pipeline.Context) {
 	}
 	plugins := snapshotPlugins(pctx.Extensions.Custom)
 	ev := pipeline.SessionEvent{
-		At:             time.Now(),
-		Direction:      pipeline.Outbound,
-		Phase:          pipeline.SessionResponse,
-		MCP:            snapshotMCP(pctx.Extensions.MCP),
-		Inference:      snapshotInference(pctx.Extensions.Inference),
-		Invocations:    snapshotInvocations(pctx.Extensions.Invocations, pipeline.InvocationPhaseResponse),
-		Plugins:        plugins,
-		Identity:       snapshotIdentity(pctx),
-		StatusCode:     pctx.StatusCode,
-		Error:          deriveError(pctx),
-		Host:     pctx.Host,
-		Duration: durationSince(pctx.StartedAt),
+		At:          time.Now(),
+		Direction:   pipeline.Outbound,
+		Phase:       pipeline.SessionResponse,
+		MCP:         snapshotMCP(pctx.Extensions.MCP),
+		Inference:   snapshotInference(pctx.Extensions.Inference),
+		Invocations: snapshotInvocations(pctx.Extensions.Invocations, pipeline.InvocationPhaseResponse),
+		Plugins:     plugins,
+		Identity:    snapshotIdentity(pctx),
+		StatusCode:  pctx.StatusCode,
+		Error:       deriveError(pctx),
+		Host:        pctx.Host,
+		Duration:    durationSince(pctx.StartedAt),
 	}
 	// Auth / Plugins alone qualify for recording; matches the widened
 	// gate in recordInboundSession so outbound denials and plugin-public
@@ -462,15 +500,15 @@ func (s *Server) recordOutboundSession(pctx *pipeline.Context) {
 	}
 	plugins := snapshotPlugins(pctx.Extensions.Custom)
 	ev := pipeline.SessionEvent{
-		At:             time.Now(),
-		Direction:      pipeline.Outbound,
-		Phase:          pipeline.SessionRequest,
-		MCP:            snapshotMCP(pctx.Extensions.MCP),
-		Inference:      snapshotInference(pctx.Extensions.Inference),
-		Invocations:    snapshotInvocations(pctx.Extensions.Invocations, pipeline.InvocationPhaseRequest),
-		Plugins:        plugins,
-		Identity:       snapshotIdentity(pctx),
-		Host: pctx.Host,
+		At:          time.Now(),
+		Direction:   pipeline.Outbound,
+		Phase:       pipeline.SessionRequest,
+		MCP:         snapshotMCP(pctx.Extensions.MCP),
+		Inference:   snapshotInference(pctx.Extensions.Inference),
+		Invocations: snapshotInvocations(pctx.Extensions.Invocations, pipeline.InvocationPhaseRequest),
+		Plugins:     plugins,
+		Identity:    snapshotIdentity(pctx),
+		Host:        pctx.Host,
 	}
 	if ev.MCP != nil || ev.Inference != nil || ev.Invocations != nil || plugins != nil {
 		s.Sessions.Append(sid, ev)
@@ -540,6 +578,7 @@ func (s *Server) handleOutbound(stream extprocv3.ExternalProcessor_ProcessServer
 	originalAuth := pctx.Headers.Get("Authorization")
 	action := s.OutboundPipeline.Run(ctx, pctx)
 	if action.Type == pipeline.Reject {
+		s.recordOutboundReject(pctx, action)
 		return rejectFromAction(action), nil
 	}
 
@@ -575,6 +614,7 @@ func (s *Server) handleOutboundBody(stream extprocv3.ExternalProcessor_ProcessSe
 	originalAuth := pctx.Headers.Get("Authorization")
 	action := s.OutboundPipeline.Run(ctx, pctx)
 	if action.Type == pipeline.Reject {
+		s.recordOutboundReject(pctx, action)
 		return rejectFromAction(action), nil
 	}
 
diff --git a/authbridge/cmd/authbridge/listener/extproc/server_test.go b/authbridge/cmd/authbridge/listener/extproc/server_test.go
index a4b015976..e70ec6f36 100644
--- a/authbridge/cmd/authbridge/listener/extproc/server_test.go
+++ b/authbridge/cmd/authbridge/listener/extproc/server_test.go
@@ -1319,6 +1319,120 @@ func TestRecordInboundReject_SkipsWithoutAuth(t *testing.T) {
 	}
 }
 
+// TestRecordOutboundReject_EmitsDeniedPhase verifies that a rejected
+// outbound request produces a SessionDenied event with the outbound
+// direction tag, the plugin's Invocation context, and the Violation
+// mapped onto StatusCode + EventError. Gap: before this, outbound
+// denies (future IBAC, rate-limit, guardrail plugins) would surface
+// only as a 4xx on the agent and never appear in /v1/sessions or abctl.
+func TestRecordOutboundReject_EmitsDeniedPhase(t *testing.T) {
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+	// Seed an active session so ActiveSession() returns it instead of
+	// empty — mirrors the real request flow (inbound recorded first).
+	store.Append("sess-active", pipeline.SessionEvent{
+		At: time.Now().Add(-50 * time.Millisecond),
+		Direction: pipeline.Inbound, Phase: pipeline.SessionRequest,
+		A2A: &pipeline.A2AExtension{Method: "message/send", SessionID: "sess-active"},
+	})
+	s := &Server{Sessions: store}
+
+	pctx := &pipeline.Context{
+		Direction: pipeline.Outbound,
+		StartedAt: time.Now().Add(-10 * time.Millisecond),
+		Host:      "external.example",
+		Extensions: pipeline.Extensions{
+			Invocations: &pipeline.Invocations{
+				Outbound: []pipeline.Invocation{{
+					Plugin: "ibac",
+					Phase:  pipeline.InvocationPhaseRequest,
+					Action: pipeline.ActionDeny,
+					Reason: "blocked",
+					Details: map[string]string{
+						"intent_preview": "summarize my emails",
+						"action":         "POST http://external.example",
+						"llm_reason":     "action unrelated to user intent",
+					},
+				}},
+			},
+		},
+	}
+	action := pipeline.DenyStatus(403, "ibac.blocked", "action unrelated to user intent")
+	s.recordOutboundReject(pctx, action)
+
+	v := store.View("sess-active")
+	if v == nil || len(v.Events) != 2 {
+		t.Fatalf("expected 2 events under sess-active (seed + denied), got %v", v)
+	}
+	ev := v.Events[1]
+	if ev.Direction != pipeline.Outbound {
+		t.Errorf("Direction = %v, want Outbound", ev.Direction)
+	}
+	if ev.Phase != pipeline.SessionDenied {
+		t.Errorf("Phase = %v, want SessionDenied", ev.Phase)
+	}
+	if ev.StatusCode != 403 {
+		t.Errorf("StatusCode = %d, want 403", ev.StatusCode)
+	}
+	if ev.Error == nil || ev.Error.Code != "ibac.blocked" {
+		t.Errorf("Error = %+v, want code=ibac.blocked", ev.Error)
+	}
+	if ev.Host != "external.example" {
+		t.Errorf("Host = %q, want external.example", ev.Host)
+	}
+	if ev.Invocations == nil || len(ev.Invocations.Outbound) != 1 ||
+		ev.Invocations.Outbound[0].Action != pipeline.ActionDeny {
+		t.Errorf("Invocations context lost on denied event: %+v", ev.Invocations)
+	}
+	if ev.Duration <= 0 {
+		t.Errorf("Duration = %v, want > 0", ev.Duration)
+	}
+}
+
+// TestRecordOutboundReject_NoActiveSession verifies the default-bucket
+// fallback: when no session has been appended yet, the denial lands
+// under DefaultSessionID so operators can still see it in /v1/sessions.
+func TestRecordOutboundReject_NoActiveSession(t *testing.T) {
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+	s := &Server{Sessions: store}
+
+	pctx := &pipeline.Context{
+		Direction: pipeline.Outbound,
+		StartedAt: time.Now(),
+		Extensions: pipeline.Extensions{
+			Invocations: &pipeline.Invocations{
+				Outbound: []pipeline.Invocation{{
+					Plugin: "ibac", Action: pipeline.ActionDeny,
+					Phase: pipeline.InvocationPhaseRequest, Reason: "blocked",
+				}},
+			},
+		},
+	}
+	action := pipeline.DenyStatus(403, "ibac.blocked", "blocked")
+	s.recordOutboundReject(pctx, action)
+
+	if v := store.View(session.DefaultSessionID); v == nil || len(v.Events) != 1 {
+		t.Fatalf("expected 1 event under default session, got %v", v)
+	}
+}
+
+// TestRecordOutboundReject_SkipsWithoutInvocations mirrors the inbound
+// skip rule: a denial with no Invocation context would produce a
+// content-free event, noise for operators without attribution.
+func TestRecordOutboundReject_SkipsWithoutInvocations(t *testing.T) {
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+	s := &Server{Sessions: store}
+
+	action := pipeline.DenyStatus(403, "policy.forbidden", "forbidden")
+	s.recordOutboundReject(&pipeline.Context{Direction: pipeline.Outbound}, action)
+
+	if v := store.View(session.DefaultSessionID); v != nil {
+		t.Errorf("expected no event recorded, got %+v", v)
+	}
+}
+
 // TestSnapshotPlugins_FiltersByEventSuffix verifies the plugin-public
 // observability convention: only Custom entries with keys ending in
 // pipeline.PluginEventSuffix are promoted to SessionEvent.Plugins.
diff --git a/authbridge/cmd/authbridge/listener/forwardproxy/server.go b/authbridge/cmd/authbridge/listener/forwardproxy/server.go
index d7cac81d7..f3c96d43f 100644
--- a/authbridge/cmd/authbridge/listener/forwardproxy/server.go
+++ b/authbridge/cmd/authbridge/listener/forwardproxy/server.go
@@ -84,6 +84,7 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) {
 	action := s.OutboundPipeline.Run(r.Context(), pctx)
 
 	if action.Type == pipeline.Reject {
+		s.recordOutboundReject(pctx, action)
 		writeRejection(w, action)
 		return
 	}
@@ -207,6 +208,50 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// recordOutboundReject emits a SessionDenied event for outbound
+// requests a pipeline plugin rejected. Symmetric to the accept path's
+// session recording (above). Lets guardrail plugins (rate-limit,
+// intent-based, content policy) show operators what was blocked and
+// why via /v1/sessions and abctl, instead of the block appearing only
+// as a 4xx/5xx on the agent side.
+//
+// Skips when no Invocations were appended — the deny came from a
+// plugin that didn't contribute diagnostic context, and a content-free
+// SessionDenied event would be noise without attribution.
+func (s *Server) recordOutboundReject(pctx *pipeline.Context, action pipeline.Action) {
+	if s.Sessions == nil || pctx.Extensions.Invocations == nil {
+		return
+	}
+	sid := s.Sessions.ActiveSession()
+	if sid == "" {
+		sid = session.DefaultSessionID
+	}
+	var status int
+	var code, message string
+	if action.Violation != nil {
+		status = action.Violation.Status
+		if status == 0 {
+			status = pipeline.StatusFromCode(action.Violation.Code)
+		}
+		code = action.Violation.Code
+		message = action.Violation.Reason
+	}
+	ev := pipeline.SessionEvent{
+		At:          time.Now(),
+		Direction:   pipeline.Outbound,
+		Phase:       pipeline.SessionDenied,
+		Invocations: pctx.Extensions.Invocations.FilteredByPhase(pipeline.InvocationPhaseRequest),
+		Host:        pctx.Host,
+		StatusCode:  status,
+		Error: &pipeline.EventError{
+			Kind:    "policy",
+			Code:    code,
+			Message: message,
+		},
+	}
+	s.Sessions.Append(sid, ev)
+}
+
 func extractBearer(authHeader string) string {
 	if len(authHeader) > 7 && strings.EqualFold(authHeader[:7], "bearer ") {
 		return authHeader[7:]
diff --git a/authbridge/cmd/authbridge/listener/forwardproxy/server_test.go b/authbridge/cmd/authbridge/listener/forwardproxy/server_test.go
index 2fe6ec382..77e7d43df 100644
--- a/authbridge/cmd/authbridge/listener/forwardproxy/server_test.go
+++ b/authbridge/cmd/authbridge/listener/forwardproxy/server_test.go
@@ -9,14 +9,16 @@ import (
 	"net/url"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/auth"
-	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/tokenexchange/cache"
-	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/tokenexchange/exchange"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/jwtvalidation/validation"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/plugintesting"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/tokenexchange/cache"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/tokenexchange/exchange"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/routing"
-	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/jwtvalidation/validation"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/session"
 )
 
 type mockVerifier struct {
@@ -335,3 +337,70 @@ func TestForwardProxy_NoBodyBuffering_WhenNotNeeded(t *testing.T) {
 		t.Errorf("status = %d, want 200", resp.StatusCode)
 	}
 }
+
+// TestRecordOutboundReject_EmitsDeniedPhase verifies the forward-proxy
+// listener's reject-event recording: a rejected outbound request
+// produces a SessionDenied event with the plugin Invocation context
+// and the Violation mapped to StatusCode + EventError.
+func TestRecordOutboundReject_EmitsDeniedPhase(t *testing.T) {
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+	store.Append("sess-active", pipeline.SessionEvent{
+		At:        time.Now().Add(-50 * time.Millisecond),
+		Direction: pipeline.Inbound,
+		Phase:     pipeline.SessionRequest,
+		A2A:       &pipeline.A2AExtension{Method: "message/send", SessionID: "sess-active"},
+	})
+	s := &Server{Sessions: store}
+
+	pctx := &pipeline.Context{
+		Direction: pipeline.Outbound,
+		Host:      "external.example",
+		Extensions: pipeline.Extensions{
+			Invocations: &pipeline.Invocations{
+				Outbound: []pipeline.Invocation{{
+					Plugin: "ibac", Action: pipeline.ActionDeny,
+					Phase: pipeline.InvocationPhaseRequest, Reason: "blocked",
+					Details: map[string]string{"llm_reason": "unrelated to user intent"},
+				}},
+			},
+		},
+	}
+	action := pipeline.DenyStatus(403, "ibac.blocked", "unrelated to user intent")
+	s.recordOutboundReject(pctx, action)
+
+	v := store.View("sess-active")
+	if v == nil || len(v.Events) != 2 {
+		t.Fatalf("expected 2 events under sess-active, got %+v", v)
+	}
+	ev := v.Events[1]
+	if ev.Direction != pipeline.Outbound || ev.Phase != pipeline.SessionDenied {
+		t.Errorf("Direction/Phase = %v/%v, want Outbound/SessionDenied", ev.Direction, ev.Phase)
+	}
+	if ev.StatusCode != 403 || ev.Error == nil || ev.Error.Code != "ibac.blocked" {
+		t.Errorf("Status/Error = %d/%+v, want 403/ibac.blocked", ev.StatusCode, ev.Error)
+	}
+	if ev.Host != "external.example" {
+		t.Errorf("Host = %q, want external.example", ev.Host)
+	}
+	if ev.Invocations == nil || len(ev.Invocations.Outbound) != 1 {
+		t.Errorf("Invocations lost on denied event: %+v", ev.Invocations)
+	}
+}
+
+// TestRecordOutboundReject_SkipsWithoutInvocations confirms the skip
+// rule matches extproc's equivalent: denials with no diagnostic
+// context are not recorded, so session stream attribution stays
+// meaningful.
+func TestRecordOutboundReject_SkipsWithoutInvocations(t *testing.T) {
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+	s := &Server{Sessions: store}
+
+	action := pipeline.DenyStatus(403, "policy.forbidden", "forbidden")
+	s.recordOutboundReject(&pipeline.Context{Direction: pipeline.Outbound}, action)
+
+	if v := store.View(session.DefaultSessionID); v != nil {
+		t.Errorf("expected no event, got %+v", v)
+	}
+} 
\ No newline at end of file
diff --git a/authbridge/cmd/authbridge/listener/reverseproxy/server.go b/authbridge/cmd/authbridge/listener/reverseproxy/server.go
index ac40ed98a..66d2409e8 100644
--- a/authbridge/cmd/authbridge/listener/reverseproxy/server.go
+++ b/authbridge/cmd/authbridge/listener/reverseproxy/server.go
@@ -95,6 +95,7 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) {
 
 	action := s.InboundPipeline.Run(r.Context(), pctx)
 	if action.Type == pipeline.Reject {
+		s.recordInboundReject(pctx, action)
 		writeRejection(w, action)
 		return
 	}
@@ -177,6 +178,58 @@ func (s *Server) errorHandler(w http.ResponseWriter, _ *http.Request, err error)
 	http.Error(w, `{"error":"bad gateway"}`, http.StatusBadGateway)
 }
 
+// recordInboundReject emits a SessionDenied event for inbound requests
+// a pipeline plugin rejected. Lets gate plugins (jwt-validation and
+// future inbound guardrails) show operators what was blocked and why
+// via /v1/sessions and abctl, instead of the block appearing only as
+// a 401/403 on the caller side.
+//
+// Skips when no Invocations were appended — the deny came from a
+// plugin that didn't contribute diagnostic context, and a content-free
+// SessionDenied event would be noise without attribution.
+func (s *Server) recordInboundReject(pctx *pipeline.Context, action pipeline.Action) {
+	if s.Sessions == nil || pctx.Extensions.Invocations == nil {
+		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
+	}
+	var status int
+	var code, message string
+	if action.Violation != nil {
+		status = action.Violation.Status
+		if status == 0 {
+			status = pipeline.StatusFromCode(action.Violation.Code)
+		}
+		code = action.Violation.Code
+		message = action.Violation.Reason
+	}
+	ev := pipeline.SessionEvent{
+		At:          time.Now(),
+		Direction:   pipeline.Inbound,
+		Phase:       pipeline.SessionDenied,
+		Invocations: pctx.Extensions.Invocations.FilteredByPhase(pipeline.InvocationPhaseRequest),
+		Host:        pctx.Host,
+		StatusCode:  status,
+		Error: &pipeline.EventError{
+			Kind:    "policy",
+			Code:    code,
+			Message: message,
+		},
+	}
+	s.Sessions.Append(sid, ev)
+}
+
 // writeRejection renders a pipeline Reject to the http.ResponseWriter,
 // preserving the plugin's status, headers, and body.
 func writeRejection(w http.ResponseWriter, action pipeline.Action) {
diff --git a/authbridge/cmd/authbridge/listener/reverseproxy/server_test.go b/authbridge/cmd/authbridge/listener/reverseproxy/server_test.go
index 9447312f3..f16a791ae 100644
--- a/authbridge/cmd/authbridge/listener/reverseproxy/server_test.go
+++ b/authbridge/cmd/authbridge/listener/reverseproxy/server_test.go
@@ -8,12 +8,14 @@ import (
 	"net/http/httptest"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/auth"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/bypass"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline"
-	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/plugintesting"
 	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/jwtvalidation/validation"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins/plugintesting"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/session"
 )
 
 type mockVerifier struct {
@@ -295,3 +297,91 @@ func TestReverseProxy_BodyNotBuffered_WhenNotNeeded(t *testing.T) {
 		t.Errorf("status = %d, want 200", resp.StatusCode)
 	}
 }
+
+// TestRecordInboundReject_EmitsDeniedPhase verifies the reverse-proxy
+// listener's inbound reject-event recording. Gap: before this, an
+// inbound request denied by jwt-validation or any other gate plugin
+// produced a 401/403 on the wire but no SessionDenied event — abctl
+// and /v1/sessions showed nothing, making misconfigurations invisible.
+func TestRecordInboundReject_EmitsDeniedPhase(t *testing.T) {
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+	s := &Server{Sessions: store}
+
+	pctx := &pipeline.Context{
+		Direction: pipeline.Inbound,
+		Host:      "agent.example",
+		Extensions: pipeline.Extensions{
+			A2A: &pipeline.A2AExtension{SessionID: "sess-abc"},
+			Invocations: &pipeline.Invocations{
+				Inbound: []pipeline.Invocation{{
+					Plugin: "jwt-validation", Action: pipeline.ActionDeny,
+					Phase: pipeline.InvocationPhaseRequest, Reason: "jwt_failed",
+					Details: map[string]string{
+						"expected_issuer": "http://issuer.example",
+					},
+				}},
+			},
+		},
+	}
+	action := pipeline.DenyStatus(401, "auth.unauthorized", "token validation failed")
+	s.recordInboundReject(pctx, action)
+
+	v := store.View("sess-abc")
+	if v == nil || len(v.Events) != 1 {
+		t.Fatalf("expected 1 event under sess-abc, got %+v", v)
+	}
+	ev := v.Events[0]
+	if ev.Direction != pipeline.Inbound || ev.Phase != pipeline.SessionDenied {
+		t.Errorf("Direction/Phase = %v/%v, want Inbound/SessionDenied", ev.Direction, ev.Phase)
+	}
+	if ev.StatusCode != 401 || ev.Error == nil || ev.Error.Code != "auth.unauthorized" {
+		t.Errorf("Status/Error = %d/%+v, want 401/auth.unauthorized", ev.StatusCode, ev.Error)
+	}
+	if ev.Invocations == nil || len(ev.Invocations.Inbound) != 1 {
+		t.Errorf("Invocations lost on denied event: %+v", ev.Invocations)
+	}
+}
+
+// TestRecordInboundReject_FallbackBucketing confirms the bucket
+// selection falls through A2A.SessionID → ActiveSession → default.
+// A request with no A2A context (bypass path that denied in a later
+// gate) lands in the default bucket so operators still see it.
+func TestRecordInboundReject_FallbackBucketing(t *testing.T) {
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+	s := &Server{Sessions: store}
+
+	pctx := &pipeline.Context{
+		Direction: pipeline.Inbound,
+		Extensions: pipeline.Extensions{
+			Invocations: &pipeline.Invocations{
+				Inbound: []pipeline.Invocation{{
+					Plugin: "jwt-validation", Action: pipeline.ActionDeny,
+					Phase: pipeline.InvocationPhaseRequest, Reason: "no_header",
+				}},
+			},
+		},
+	}
+	action := pipeline.DenyStatus(401, "auth.unauthorized", "no bearer token")
+	s.recordInboundReject(pctx, action)
+
+	if v := store.View(session.DefaultSessionID); v == nil || len(v.Events) != 1 {
+		t.Fatalf("expected 1 event in default bucket, got %+v", v)
+	}
+}
+
+// TestRecordInboundReject_SkipsWithoutInvocations confirms the skip
+// rule: denials with no diagnostic context are not recorded.
+func TestRecordInboundReject_SkipsWithoutInvocations(t *testing.T) {
+	store := session.New(5*time.Minute, 100, 0)
+	defer store.Close()
+	s := &Server{Sessions: store}
+
+	action := pipeline.DenyStatus(401, "auth.unauthorized", "denied")
+	s.recordInboundReject(&pipeline.Context{Direction: pipeline.Inbound}, action)
+
+	if v := store.View(session.DefaultSessionID); v != nil {
+		t.Errorf("expected no event, got %+v", v)
+	}
+}
diff --git a/authbridge/docs/plugin-reference.md b/authbridge/docs/plugin-reference.md
index 9a773ef86..eef870890 100644
--- a/authbridge/docs/plugin-reference.md
+++ b/authbridge/docs/plugin-reference.md
@@ -40,16 +40,125 @@ pipeline:
           audience_file: "/shared/client-id.txt"
           bypass_paths:
             - "/healthz"
+      - name: pii-scrubber
+        on_error: observe                # canary: log would-blocks, don't block
+        config:
+          patterns: [ssn, credit_card]
 ```
 
 - **`name`** — required. Must match a key in the plugin registry.
 - **`id`** — optional. Defaults to `name`. Lets two instances of the same
   plugin coexist with different config (not yet exercised, but the shape
   is reserved).
+- **`on_error`** — optional. One of `enforce` (default), `observe`, or
+  `off`. See [`on_error` policy](#on_error-policy) below.
 - **`config`** — optional. Arbitrary YAML sub-tree owned by the plugin.
   The framework does not interpret it; it's captured as `json.RawMessage`
   and handed to `Configure`.
 
+## `on_error` policy
+
+> **Naming caveat.** Despite the name, `on_error` controls how the
+> framework handles **intentional `Deny` actions** returned by the
+> plugin — it is not a panic/error handler. A panic or a
+> runtime-error return still surfaces as a 500 regardless of this
+> setting. Reach for `on_error` to stage the rollout of a new
+> guardrail (observe before enforce) or to toggle a plugin off
+> without a redeploy; reach for something else when you want to
+> bound misbehavior.
+
+`on_error` is a **framework-owned** wrapper around the plugin — plugin
+authors do not read it, implement it, or branch on it. Its job is to
+let operators roll out a new guardrail without risking production.
+
+| Policy | Plugin dispatched? | Reject → | Body mutation → | Typical use |
+|---|---|---|---|---|
+| `enforce` (default) | yes | HTTP error, pipeline stops | applied to wire | Production guardrails |
+| `observe` | yes | shadow `Invocation`, request passes | suppressed (no-op) | Canarying a new plugin |
+| `off` | **no** | n/a | n/a | Kill-switch without redeploy |
+
+### Observe is plugin-transparent
+
+Under `observe`, the plugin's `OnRequest` / `OnResponse` runs exactly as
+under `enforce`. If it returns `pipeline.Deny(...)`, the framework
+intercepts: it marks the plugin's `Invocation` with `Shadow: true`, logs
+a `WARN pipeline: plugin would have denied (shadow)` line, and continues
+the pipeline. The request is not blocked. Body-mutation calls
+(`SetBody` / `SetResponseBody`) likewise record a `Shadow: true`
+invocation but do not alter the in-memory body or the wire bytes —
+downstream plugins and the upstream see the original.
+
+The upshot: the same plugin binary, dispatched the same way, is safe to
+ship in `observe` for a week while operators watch shadow metrics,
+then flipped to `enforce` with confidence.
+
+### Shadow timeline query
+
+Operators count would-have-blocked events by filtering Invocations on
+`shadow: true`:
+
+- `count(Invocations where shadow=true)` — rollout candidate volume
+- `count(Invocations where shadow=true and action="deny") by plugin` —
+  per-plugin shadow block rate
+- `count(Invocations where shadow=false and action="deny") by plugin` —
+  enforced denials (unchanged by this feature)
+
+### Don't put matched content in `Violation.Reason`
+
+`Reason` (the free-text explanation on `pipeline.Deny` / `DenyAndRecord`
+and on `Invocation.Reason`) flows to two places: the session store at
+`/v1/sessions` and, under `observe`, a `WARN` log line. Both live
+outside the authorization domain of the request itself — logs typically
+aggregate to a different backend than session events, with different
+retention and access policies.
+
+Plugin authors: **keep `Reason` to a machine-stable short code and/or a
+generic description.** Do not echo matched content, raw user input, or
+credential-shaped substrings into `Reason`.
+
+```go
+// GOOD
+return pctx.DenyAndRecord("ssn_match", "pii.detected", "matched PII pattern")
+
+// BAD — echoes a matched SSN into logs and the session store:
+return pctx.DenyAndRecord("ssn_match", "pii.detected",
+    fmt.Sprintf("matched SSN %s at offset 42", ssn))
+```
+
+If you need to record which pattern matched for debugging, put a
+**hash or stable fingerprint** of the content in `Invocation.Details`
+(`Details["match_sha256"] = "a1b2c3..."`), never the raw match. The
+same rule applies to `Details` values — see the existing "NEVER put
+raw tokens, signatures, or client credentials" note on the Invocation
+type below.
+
+### Off vs. removing the entry
+
+Both achieve "don't run this plugin." `off` exists so a single field
+flip re-enables the plugin without re-adding the whole block to YAML.
+An `off` entry is not `Configure`d and not added to the running
+pipeline; its `config:` subtree is not validated. Remove the entry
+entirely if you don't anticipate re-enabling it.
+
+### What `on_error` does not do
+
+- Not a circuit breaker — a crashing plugin in `observe` still crashes
+  on every request. Bound crash loops with a separate mechanism.
+- Not sampling — `observe` runs the plugin on 100% of traffic.
+  Percentage rollout is a future feature.
+- Not a timeout — a slow plugin still blocks the request. Per-plugin
+  deadlines are a separate knob.
+- Not a panic / runtime-error handler (see the leading caveat at the
+  top of this section).
+
+### Applicability to auth gates
+
+`on_error` is a generic framework knob that applies to every plugin
+including built-in auth gates (`jwt-validation`, `token-exchange`).
+Shadowing auth turns authentication into a suggestion — don't do this
+in production. Shadow-mode is for third-party guardrails being
+canaried; auth gates should stay on `enforce` (the default).
+
 ## The Configurable interface
 
 ```go
@@ -315,6 +424,13 @@ type Invocation struct {
     // Plugin-specific diagnostic context. Opaque to the framework;
     // abctl renders as key=value rows in the detail pane.
     Details map[string]string
+
+    // Shadow is framework-set; plugins never write it. True when the
+    // plugin ran under on_error: observe and its decision (deny or
+    // modify) was NOT applied to the request. Dashboards partition
+    // on Shadow: enforced outcomes (shadow=false) vs rollout
+    // candidates (shadow=true).
+    Shadow bool
 }
 ```
 
diff --git a/authbridge/docs/plugin-tutorial.md b/authbridge/docs/plugin-tutorial.md
index 93dbba76e..394622f8f 100644
--- a/authbridge/docs/plugin-tutorial.md
+++ b/authbridge/docs/plugin-tutorial.md
@@ -129,6 +129,26 @@ When you want to emit an invocation AND reject in one call, use
 return pctx.DenyAndRecord("caller_not_allowed", "policy.forbidden", "caller not permitted")
 ```
 
+### Ship in observe mode first
+
+For any guardrail or policy plugin that can false-positive — PII
+detection, prompt-injection scoring, jailbreak classifiers — roll it
+out under `on_error: observe` before flipping to `enforce`:
+
+```yaml
+- name: your-new-guardrail
+  on_error: observe          # log would-blocks, don't actually block
+  config: { ... }
+```
+
+In `observe`, the framework converts the plugin's `Deny` into a
+pass-through and marks the deny `Invocation` with `Shadow: true`. Your
+plugin code is unchanged — the rollout knob lives outside the plugin.
+After a soak period, compare the shadow-deny rate against your
+acceptable false-positive threshold, then flip to `enforce`. See
+[`plugin-reference.md`](./plugin-reference.md#on_error-policy) for
+full semantics.
+
 ## Step 4 — Add config
 
 If your plugin needs configurable knobs, implement