feat: Add pipeline package with typed extension slots

authbridge/authlib/pipeline/action.go

@@ -0,0 +1,17 @@
+package pipeline
+
+// ActionType represents the result of a plugin's processing.
+type ActionType int
+
+const (
+	Continue ActionType = iota
+	Reject
+)
+
+// Action is returned by a plugin to indicate whether processing should continue
+// or the request should be rejected.
+type Action struct {
+	Type   ActionType
+	Status int    // HTTP status code (only for Reject)
+	Reason string // human-readable reason (only for Reject)
+}

authbridge/authlib/pipeline/context.go

@@ -0,0 +1,41 @@
+package pipeline
+
+import (
+	"net/http"
+
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/routing"
+	"github.com/kagenti/kagenti-extensions/authbridge/authlib/validation"
+)
+
+// Direction indicates whether a request is inbound (caller → this agent) or
+// outbound (this agent → target service).
+type Direction int
+
+const (
+	Inbound Direction = iota
+	Outbound
+)
+
+// Context is the shared state passed through the plugin pipeline.
+// Plugins read and mutate fields directly — there is no separate mutation API.
+type Context struct {
+	Direction Direction
+	Method    string
+	Host      string
+	Path      string
+	Headers   http.Header
+	Body      []byte // nil unless at least one plugin declares BodyAccess: true
+
+	Agent  *AgentIdentity
+	Claims *validation.Claims    // nil before jwt-validation runs
+	Route  *routing.ResolvedRoute
+
+	Extensions Extensions
+}
+
+// AgentIdentity carries the agent's own workload identity.
+type AgentIdentity struct {
+	ClientID    string
+	WorkloadID  string
+	TrustDomain string
+}

authbridge/authlib/pipeline/extensions.go

@@ -0,0 +1,104 @@
+package pipeline
+
+import "time"
+
+// Extensions holds typed extension slots for plugin-to-plugin communication.
+// Each slot is populated by a specific plugin and consumed by downstream plugins.
+type Extensions struct {
+	MCP        *MCPExtension
+	A2A        *A2AExtension
+	Security   *SecurityExtension
+	Delegation *DelegationExtension
+	Custom     map[string]any
+}
+
+// MCPExtension carries parsed MCP JSON-RPC metadata.
+// Exactly one of Tool, Resource, or Prompt is populated per request.
+type MCPExtension struct {
+	Method string // JSON-RPC method: "tools/call", "resources/read", "prompts/get"
+	RPCID  any    // JSON-RPC id for request-response correlation
+
+	Tool     *MCPToolMetadata
+	Resource *MCPResourceMetadata
+	Prompt   *MCPPromptMetadata
+}
+
+// MCPToolMetadata is populated for tools/call requests.
+type MCPToolMetadata struct {
+	Name string
+	Args map[string]any
+}
+
+// MCPResourceMetadata is populated for resources/read requests.
+type MCPResourceMetadata struct {
+	URI string
+}
+
+// MCPPromptMetadata is populated for prompts/get requests.
+type MCPPromptMetadata struct {
+	Name string
+	Args map[string]string
+}
+
+// A2AExtension carries parsed A2A protocol metadata.
+type A2AExtension struct {
+	TaskID string
+	Method string // "tasks/send", "tasks/get", etc.
+	Parts  []A2APart
+}
+
+// A2APart represents a message part in an A2A request.
+type A2APart struct {
+	Type    string // "text", "file", "data"
+	Content string
+}
+
+// SecurityExtension carries guardrail output.
+// Caller identity is already in ctx.Agent and ctx.Claims — this slot is only
+// for downstream signals from content-inspection plugins.
+type SecurityExtension struct {
+	Labels      []string
+	Blocked     bool
+	BlockReason string
+}
+
+// DelegationExtension tracks the token delegation chain across hops.
+// The chain is append-only and unexported to prevent forgery or truncation.
+type DelegationExtension struct {
+	chain  []DelegationHop
+	Origin string // original caller's subject ID
+	Actor  string // current actor's subject ID
+}
+
+// Chain returns a copy of the delegation chain. The copy prevents callers from
+// mutating the backing slice (truncation, reordering, forgery).
+func (d *DelegationExtension) Chain() []DelegationHop {
+	out := make([]DelegationHop, len(d.chain))
+	copy(out, d.chain)
+	return out
+}
+
+// Depth returns the number of hops in the delegation chain.
+func (d *DelegationExtension) Depth() int {
+	return len(d.chain)
+}
+
+// DelegationHop represents one hop in the delegation chain.
+type DelegationHop struct {
+	SubjectID string
+	Scopes    []string
+	Timestamp time.Time
+}
+
+// AppendHop adds a hop to the delegation chain. This is the only way to extend
+// the chain — direct mutation is prevented by the unexported slice.
+//
+// AppendHop is not safe for concurrent use. The pipeline guarantees sequential
+// invocation.
+func (d *DelegationExtension) AppendHop(hop DelegationHop) {
+	d.chain = append(d.chain, hop)
+	if d.Origin == "" {
+		d.Origin = hop.SubjectID
+	}
+	d.Actor = hop.SubjectID
+}

authbridge/authlib/pipeline/pipeline.go

@@ -0,0 +1,109 @@
+package pipeline
+
+import (
+	"context"
+	"fmt"
+)
+
+// Pipeline holds an ordered list of plugins and runs them sequentially.
+type Pipeline struct {
+	plugins []Plugin
+}
+
+// defaultSlots lists the built-in extension slot names.
+var defaultSlots = map[string]bool{
+	"mcp":        true,
+	"a2a":        true,
+	"security":   true,
+	"delegation": true,
+	"custom":     true,
+}
+
+// Option configures pipeline construction.
+type Option func(*options)
+
+type options struct {
+	extraSlots []string
+}
+
+// WithSlots registers additional valid extension slot names beyond the built-in set.
+// Use this when a bridge plugin (e.g., CPEX) produces extensions not in the default set.
+func WithSlots(slots ...string) Option {
+	return func(o *options) {
+		o.extraSlots = append(o.extraSlots, slots...)
+	}
+}
+
+// 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) {
+	var o options
+	for _, opt := range opts {
+		opt(&o)
+	}
+	validSlots := make(map[string]bool, len(defaultSlots)+len(o.extraSlots))
+	for k, v := range defaultSlots {
+		validSlots[k] = v
+	}
+	for _, s := range o.extraSlots {
+		validSlots[s] = true
+	}
+	if err := validateCapabilities(plugins, validSlots); err != nil {
+		return nil, err
+	}
+	return &Pipeline{plugins: plugins}, nil
+}
+
+// Run executes the request phase of the pipeline sequentially.
+// If any plugin returns Reject, the pipeline stops and returns that action.
+func (p *Pipeline) Run(ctx context.Context, pctx *Context) Action {
+	for _, plugin := range p.plugins {
+		if ctx.Err() != nil {
+			return Action{Type: Reject, Status: 499, Reason: "request cancelled"}
+		}
+		action := plugin.OnRequest(ctx, pctx)
+		if action.Type == Reject {
+			return action
+		}
+	}
+	return Action{Type: Continue}
+}
+
+// RunResponse executes the response phase in reverse order.
+// The last plugin in the chain sees the response first.
+func (p *Pipeline) RunResponse(ctx context.Context, pctx *Context) Action {
+	for i := len(p.plugins) - 1; i >= 0; i-- {
+		if ctx.Err() != nil {
+			return Action{Type: Reject, Status: 499, Reason: "request cancelled"}
+		}
+		action := p.plugins[i].OnResponse(ctx, pctx)
+		if action.Type == Reject {
+			return action
+		}
+	}
+	return Action{Type: Continue}
+}
+
+// validateCapabilities checks that every slot a plugin reads has been written
+// by an earlier plugin in the chain.
+func validateCapabilities(plugins []Plugin, validSlots map[string]bool) error {
+	written := make(map[string]bool)
+	for _, plugin := range plugins {
+		caps := plugin.Capabilities()
+		for _, slot := range caps.Reads {
+			if !validSlots[slot] {
+				return fmt.Errorf("plugin %q declares read on unknown slot %q", plugin.Name(), slot)
+			}
+			if !written[slot] {
+				return fmt.Errorf("plugin %q reads slot %q but no earlier plugin writes it", plugin.Name(), slot)
+			}
+		}
+		for _, slot := range caps.Writes {
+			if !validSlots[slot] {
+				return fmt.Errorf("plugin %q declares write on unknown slot %q", plugin.Name(), slot)
+			}
+			written[slot] = true
+		}
+	}
+	return nil
+}