feat: Add pipeline package with typed extension slots

[huang195]

Summary

Introduces authlib/pipeline/ — the foundation for AuthBridge's plugin architecture. Pure addition, no integration into existing code paths, zero risk to current behavior.

• Plugin interface with context.Context propagation and capability declarations (Reads, Writes, BodyAccess)
• Typed extension slots — MCPExtension (three-variant: tool/resource/prompt), A2AExtension, SecurityExtension (lean: 3 fields), DelegationExtension (structured hops with append-only semantics)
• Pipeline runner — sequential request pass, reverse-order response pass, short-circuit on Reject
• Startup capability validation — rejects pipelines where a plugin reads a slot no earlier plugin writes
• WithSlots option — allows bridge plugins (CPEX, BBR) to register additional slot names without hardcoding
• DelegationExtension.AppendHop() — signals append-only intent for delegation chain

Extension slot shapes deliberately mirror CPEX's CMF where relevant (MCP, Delegation) to keep the future bridge translation layer thin. See PR #342 discussion for design rationale.

Test plan

• 14 unit tests covering: sequential execution, reject short-circuit, reverse response order, capability validation (valid/invalid/unknown slots), WithSlots, AppendHop, context cancellation
• go vet ./pipeline/... clean
• Full go test ./... passes (all existing authlib packages unaffected)

Assisted-By: Claude (Anthropic AI) noreply@anthropic.com

[evaline-ju]

Great to see this starting to put #342 in action!

[evaline-ju]

If multiple writers are allowed to write to the same slot, it seems this Blocked could be changed by a later guardrail or silently overwritten?

[huang195]

Good catch. The intent is tighten-only: Blocked can go false→true but not true→false. Today this is a convention — runtime enforcement (snapshotting before each plugin, rejecting weakening mutations) is planned for Phase 2 alongside the broader extension tier system. Documenting this invariant in the struct comment for now.

[evaline-ju]

Since the PR intends to mirror cpex CMF - I was curious how this is to be resolved with the extension as defined here https://github.com/contextforge-org/contextforge-plugins-framework/blob/main/cpex/framework/extensions/mcp.py#L51-L52 - specifically this metadata only contains the input and output schema and not arguments. Is this a gap that should be resolved with cpex? (I had this question generally for the #342 discussion comment #342 (comment))

[huang195]

Good question. These serve different layers: CPEX's ToolMetadata carries definition-time metadata (input_schema, output_schema — the JSON Schema of what a tool accepts/returns). Our MCPToolMetadata.Args carries runtime arguments (what the caller actually sent in this request).

For policy decisions, we need the runtime args — e.g., "is the caller passing a SQL query to execute_sql?". The schema is useful for validation ("are these args well-formed?") but that's the MCP server's job, not the proxy's.

The bridge translation would map our Args into whatever field CPEX expects for runtime arguments. If CPEX's ToolMetadata truly only has schema and no args field, that might be a gap on their side worth raising — a guardrails plugin needs the actual arguments to inspect, not just the schema.

[evaline-ju]

With aid of agent review - it seems this could be subject to other plugins forging or truncating the chain like:

  // Truncate — hide a delegation hop from downstream audit plugins
  pctx.Extensions.Delegation.Chain = pctx.Extensions.Delegation.Chain[:1]

  // Wipe — make it look like a direct call with no delegation
  pctx.Extensions.Delegation.Chain = nil
  pctx.Extensions.Delegation.Origin = ""
  pctx.Extensions.Delegation.Actor = "attacker"

  // Forge — inject a hop that never happened
  pctx.Extensions.Delegation.Chain = []DelegationHop{
      {SubjectID: "trusted-internal-service", Scopes: []string{"admin"}},
  }

and is advising making chain not public [but I can understand risks about copy being expensive]

type DelegationExtension struct {
      chain  []DelegationHop // unexported — only accessible via methods
      Origin string          // could also be unexported if needed
      Actor  string
  }

  // Chain returns a copy of the delegation chain (not the backing slice).
  func (d *DelegationExtension) Chain() []DelegationHop {
      out := make([]DelegationHop, len(d.chain))
      copy(out, d.chain)
      return out
  }

  // AppendHop is the only way to add to the chain.
  func (d *DelegationExtension) AppendHop(hop DelegationHop) {
      d.chain = append(d.chain, hop)
      if d.Origin == "" {
          d.Origin = hop.SubjectID
      }
      d.Actor = hop.SubjectID
  }

[huang195]

Addressed in 68f0b59. Chain is now unexported, Chain() returns a defensive copy, and AppendHop() is the only mutation path. Thanks for flagging — this was a real security concern for audit trail integrity.

[evaline-ju]

Should the BodyAccess from plugin.go be validated somewhere? Otherwise it seems it could be declared as true but the Body still nil

[huang195]

By design, BodyAccess enforcement is the adapter layer's responsibility, not the pipeline's. The pipeline package doesn't know which deployment mode is active (ext_proc with BUFFERED vs ext_authz which can never provide body). The adapter inspects plugin capabilities at construction time and either buffers the body or fails startup if the mode can't provide it. This is covered in PR 2 when we wire the pipeline into the listener.

[evaline-ju]

What should validate the slot on reading?

If there was something like

plugins := []Plugin{
      mcpParser,   // Capabilities: {Writes: ["mcp"]}
      guardrail,   // Capabilities: {Reads: ["mcp"]}
  }

this should pass validation, but if the MCP parser parses something and doesn't populate the slot on that particular request (say with a notification request instead of a tool call request), are individual (guardrail) plugin contributors going to be responsible for checking that?

[huang195]

Yes, individual plugins are responsible for nil-checking the extension slot they read. Startup validation catches wiring errors ("you declared reads: [mcp] but no earlier plugin writes mcp"), but a parser plugin might legitimately not populate the slot on every request (e.g., a non-JSON-RPC request passes through mcp-parser without populating Extensions.MCP).

This is the same pattern as ctx.Claims being nil before jwt-validation runs — downstream plugins that read Claims must handle nil. The alternative (always populating a zero-value struct) hides the semantic difference between "not parsed yet" and "parsed but empty".

[evaline-ju]

a direct agent suggestion:

Claims lives here on Context. Go plugins read it directly. But the cpex-bridge plugin must translate pctx into CPEX Extensions before calling mgr.Invoke(). CPEX plugins receive Extensions, not Context — they read ext.Security.Subject.Scopes for authorization decisions.

SecurityExtension in this PR has no Subject field — only Labels, Blocked, BlockReason. So the bridge must know to copy Claims into CPEX's Security.Subject manually. If it doesn't, CPEX authorization plugins silently see no identity.

Suggestion: either add a note here that the bridge is responsible for this translation, or provide a helper method on Context that produces the CPEX-side SecurityExtension with Subject populated from Claims

[huang195]

Correct — the bridge is explicitly responsible for this translation. Claims and Agent are top-level Context fields (AuthBridge's identity model), not extension slots. The bridge plugin's OnRequest will map ctx.Claims → CPEX SecurityExtension.Subject and ctx.Agent → CPEX SecurityExtension.Agent before calling mgr.Invoke().

A helper could live in the bridge package itself (e.g., func toCPEXExtensions(pctx *Context) *cpex.Extensions). I'd rather not add CPEX-aware helpers to the pipeline package — it shouldn't know about CPEX. Will add a doc note on Context.Claims pointing to bridge responsibility.

[evaline-ju]

another agent suggestion taking into account the #342 comment #342 (comment)

When cpex-bridge is implemented as a Plugin in this pipeline, its OnRequest will call mgr.Invoke() and receive a ContextTable back from CPEX. Its OnResponse will need to pass that same ContextTable into a second mgr.Invoke() call.

The bridge plugin needs somewhere to store the ContextTable between the pipeline calling bridge.OnRequest() and later calling bridge.OnResponse(). The Plugin interface provides no mechanism for a plugin to carry per-request state across those two calls. The only available storage is pctx.Extensions.Custom — map[string]any.

Not a blocker for this PR, but worth noting so the bridge doesn't default to the untyped map when it's implemented.

[huang195]

Good callout. The bridge will need to carry CPEX's ContextTable between OnRequest and OnResponse. Using Custom for this would work but loses type safety.

Better approach for the bridge PR: the bridge plugin holds a sync.Map keyed by request ID (from context or header), storing the ContextTable between phases. The pipeline already provides context.Context which can carry a request-scoped ID. This keeps the typed extensions clean and doesn't require a new pipeline-level mechanism.

Will address concretely in the bridge PR.

[mrsabath]

Solid foundation for the plugin architecture proposed in PR #342. Clean interface design with Plugin / PluginCapabilities, typed extension slots that mirror CPEX shapes, and proper startup capability validation. The pipeline is a pure addition with no integration into existing code paths — zero regression risk.

14 unit tests cover the core paths well: sequential execution, reject short-circuit, reverse response order, capability validation, WithSlots extensibility, AppendHop semantics, and context cancellation.

Two minor suggestions below — neither blocks merge.

Areas reviewed: Go code quality, interface design, concurrency safety, test coverage, security
Commits: 2 commits, all signed-off ✓
CI: All checks passing ✓

[mrsabath]

suggestion: AppendHop() mutates Chain, Origin, and Actor without synchronization. The pipeline is currently sequential so this is safe in practice, but if the pipeline ever supports concurrent plugin execution (mentioned in PR #342 open questions), this becomes a data race.

Consider adding a doc comment like // AppendHop is not safe for concurrent use. The pipeline guarantees sequential invocation. — this makes the contract explicit for future maintainers and bridge plugin authors (CPEX, etc.).

[huang195]

Added in 68f0b59 — AppendHop now has an explicit doc comment stating it's not safe for concurrent use and that the pipeline guarantees sequential invocation.

[mrsabath]

suggestion: validateCapabilities validates Reads and Writes against valid slot names, but doesn't validate BodyAccess. If a plugin declares BodyAccess: true, there's no startup check that the deployment mode can actually provide body data (e.g., ext_authz never can, per the PR #342 design doc).

This could be a follow-up — the pipeline package may not have enough deployment-mode context at construction time. Worth a TODO or noting in the design that body-access validation happens at the adapter layer.

[huang195]

Agreed — body-access validation is a deployment-mode concern. The pipeline package doesn't have mode context at construction time. The adapter layer (ext_proc, reverse proxy, etc.) will inspect PluginCapabilities.BodyAccess and either configure body buffering or fail startup if the mode can't provide it. This lands in PR 2.

[mrsabath]

praise: ctx.Err() check before each plugin invocation is a good defensive pattern — prevents wasted work on cancelled requests and gives a clean 499 status. Nice that this was added in the second commit as a refinement.

[mrsabath]

praise: Clean minimal interface — four methods, capability declarations separate from execution, and context.Context propagation throughout. The Reads/Writes slot model with startup validation is a strong foundation that catches wiring bugs before traffic flows.

[huang195]

/do-no-merge yet pls