diff --git a/authbridge/authlib/go.mod b/authbridge/authlib/go.mod
index a2ca3e191..90c3d0a2a 100644
--- a/authbridge/authlib/go.mod
+++ b/authbridge/authlib/go.mod
@@ -12,6 +12,10 @@ require (
 	github.com/open-policy-agent/opa v1.18.2
 	github.com/rossoctl/context-guru v0.0.0-20260720181432-8fc7c7b36563
 	github.com/spiffe/go-spiffe/v2 v2.8.1
+	go.opentelemetry.io/otel v1.44.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0
+	go.opentelemetry.io/otel/sdk v1.44.0
+	go.opentelemetry.io/otel/trace v1.44.0
 	golang.org/x/net v0.57.0
 	golang.org/x/sys v0.47.0
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa
@@ -45,6 +49,7 @@ require (
 	github.com/bytedance/gopkg v0.1.3 // indirect
 	github.com/bytedance/sonic v1.15.1 // indirect
 	github.com/bytedance/sonic/loader v0.5.1 // indirect
+	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudwego/base64x v0.1.6 // indirect
 	github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2 // indirect
@@ -56,6 +61,7 @@ require (
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/goccy/go-json v0.10.6 // indirect
 	github.com/google/uuid v1.6.0 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect
 	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/huandu/go-clone v1.7.3 // indirect
 	github.com/huandu/go-sqlbuilder v1.41.0 // indirect
@@ -108,10 +114,9 @@ require (
 	github.com/yashtewari/glob-intersection v0.2.0 // indirect
 	github.com/yosida95/uritemplate/v3 v3.0.2 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
-	go.opentelemetry.io/otel v1.44.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 // indirect
 	go.opentelemetry.io/otel/metric v1.44.0 // indirect
-	go.opentelemetry.io/otel/sdk v1.44.0 // indirect
-	go.opentelemetry.io/otel/trace v1.44.0 // indirect
+	go.opentelemetry.io/proto/otlp v1.10.0 // indirect
 	go.starlark.net v0.0.0-20260708150628-5395d018f003 // indirect
 	go.yaml.in/yaml/v2 v2.4.4 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
@@ -120,6 +125,7 @@ require (
 	golang.org/x/sync v0.22.0 // indirect
 	golang.org/x/text v0.40.0 // indirect
 	golang.org/x/time v0.15.0 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect
 	google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af // indirect
 	gopkg.in/ini.v1 v1.67.3 // indirect
 	oras.land/oras-go/v2 v2.6.2 // indirect
diff --git a/authbridge/authlib/plugins/lineage/config.go b/authbridge/authlib/plugins/lineage/config.go
new file mode 100644
index 000000000..c030c7e05
--- /dev/null
+++ b/authbridge/authlib/plugins/lineage/config.go
@@ -0,0 +1,79 @@
+package lineage
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"strings"
+)
+
+// Config holds the per-plugin configuration decoded from the pipeline YAML.
+type Config struct {
+	// OTelEndpoint is the OTLP gRPC endpoint (host:port or http://host:port).
+	// Default: "localhost:4317"
+	OTelEndpoint string `json:"otel_endpoint"`
+
+	// CaptureIO when true attaches parsed request/response content as
+	// input.value (request span) and output.value (response span)
+	// attributes, enabling Phoenix to display message content inline.
+	//
+	// For A2A (inbound agent calls): input = user message parts, output = artifact.
+	// For MCP tools/call: input = tool params JSON, output = tool result JSON.
+	// For Inference (LLM): input = messages array JSON, output = completion text.
+	//
+	// Off by default — enable only if traces do not contain PII or the
+	// OTel backend enforces appropriate access controls.
+	CaptureIO bool `json:"capture_io"`
+
+	// BypassPaths lists URL path prefixes that should not generate lineage
+	// hops. Useful for suppressing infrastructure polling (agent-card
+	// discovery, health checks) that would otherwise flood the lineage graph.
+	// Default: ["/.well-known/", "/healthz", "/readyz", "/health"]
+	BypassPaths []string `json:"bypass_paths"`
+
+	// BypassHosts lists target host substrings (matched against pctx.Host)
+	// that should not generate lineage hops. Useful for suppressing
+	// infrastructure outbound calls such as OTel trace exports.
+	// Default: ["otel-collector", "jaeger", "zipkin", "prometheus"]
+	BypassHosts []string `json:"bypass_hosts"`
+
+	// SelfID is the agent's own stable identifier, emitted as the
+	// lineage.self.id fact on every span. Typically the Keycloak client ID
+	// of this workload. If empty, SelfIDFile is consulted instead.
+	SelfID string `json:"self_id"`
+
+	// SelfIDFile is the path to a file containing the agent's own client ID.
+	// Defaults to /shared/client-id.txt (the operator-mounted credential).
+	// Ignored when SelfID is set.
+	SelfIDFile string `json:"self_id_file"`
+}
+
+func defaultConfig() Config {
+	return Config{
+		OTelEndpoint: "localhost:4317",
+		BypassPaths:  []string{"/.well-known/", "/healthz", "/readyz", "/health"},
+		BypassHosts:  []string{"otel-collector", "jaeger", "zipkin", "prometheus"},
+		SelfIDFile:   "/shared/client-id.txt",
+	}
+}
+
+func decodeConfig(raw json.RawMessage) (Config, error) {
+	cfg := defaultConfig()
+	if len(raw) == 0 {
+		return cfg, nil
+	}
+	// Unknown keys are a boot error: a typo'd knob (capture-io, selfid_file)
+	// must not silently run with defaults.
+	dec := json.NewDecoder(bytes.NewReader(raw))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&cfg); err != nil {
+		return Config{}, fmt.Errorf("lineage-telemetry config: %w", err)
+	}
+	if cfg.OTelEndpoint == "" {
+		cfg.OTelEndpoint = "localhost:4317"
+	}
+	// Strip http:// or https:// prefix — gRPC NewClient expects host:port only.
+	cfg.OTelEndpoint = strings.TrimPrefix(cfg.OTelEndpoint, "https://")
+	cfg.OTelEndpoint = strings.TrimPrefix(cfg.OTelEndpoint, "http://")
+	return cfg, nil
+}
diff --git a/authbridge/authlib/plugins/lineage/plugin.go b/authbridge/authlib/plugins/lineage/plugin.go
new file mode 100644
index 000000000..d31169f54
--- /dev/null
+++ b/authbridge/authlib/plugins/lineage/plugin.go
@@ -0,0 +1,746 @@
+// Package lineage provides the lineage-telemetry authbridge plugin.
+//
+// Two-span model (see docs/sidecar-wire-contract.md in the lab-data-governance
+// repo, the consumer side — the law this file implements). Each HTTP exchange through the sidecar produces TWO OTLP spans:
+//
+//   - a request span, emitted as soon as the request has been seen and
+//     forwarded, carrying caller-side facts + input.value; and
+//   - a response span, emitted at stream end (even when no response was
+//     produced), carrying status/outcome facts + output.value.
+//
+// Both spans are ended immediately at emission — no span is held open across
+// the wait. lineage.exchange.id (= the request span's own span id) is echoed
+// on both so the consumer pairs them. The plugin emits FACTS ONLY (direction,
+// protocol, endpoints, parsed payloads); all vocabulary — hop kinds, entity
+// kinds, caller/callee — lives in the consumer's classify(). See the "removed
+// vs today" migration map in the contract for the attrs this no longer emits.
+//
+// The plugin implements Finisher so the response span is emitted at stream
+// end whatever the outcome — including denials that happen AFTER the request
+// span was recorded (a response-phase deny, or a request-phase deny by a
+// plugin ordered after this one); pctx.Outcome() is available at that point
+// and maps to lineage.outcome=denied + lineage.denied_by. LIMITATION: the
+// pipeline YAML places this plugin after the gate plugins (ordering is not
+// soft-declared under this capabilities model — position in the list is the
+// contract), and the pipeline short-circuits on a request-phase Reject,
+// so an exchange denied by a gate BEFORE OnRequest ran emits NO spans at all —
+// it is invisible to lineage. Moving lineage ahead of the gates (spans for
+// denied traffic too) is a named follow-up, not current behavior.
+package lineage
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"os"
+	"strings"
+	"sync/atomic"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
+	"go.opentelemetry.io/otel/propagation"
+	"go.opentelemetry.io/otel/sdk/resource"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
+	"go.opentelemetry.io/otel/trace"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/rossoctl/cortex/authbridge/authlib/pipeline"
+	"github.com/rossoctl/cortex/authbridge/authlib/plugins"
+)
+
+const pluginName = "lineage-telemetry"
+
+// tracestateStampKey is the W3C tracestate member that carries the sidecar
+// parent chain — the single channel every lineage element reads its parent
+// from and writes its own request span id into (wire contract v1.5). Inbound
+// stamps the request it forwards to its own app; the app's propagate-only shim
+// carries tracestate through its per-request causal chain (contextvars), so
+// the member surfaces on exactly the outbound calls that inbound caused.
+// Outbound re-stamps the request it forwards to the peer, whose inbound
+// sidecar reads it as its parent. Parent precedence is stamp > wire parent in
+// BOTH directions, and the chosen source is recorded as the
+// lineage.parent.source fact. The forwarded traceparent is never modified:
+// the sidecar chain lives entirely in this member, so an app that emits its
+// own spans keeps an intact traceparent chain toward its own backend while
+// the sidecar chain stays self-consistent in ours. (Until v1.4 the outbound
+// instead rewrote the forwarded traceparent — the splice; v1.5 removed it.)
+//
+// The key names the consuming data-governance system (W3C convention: the key
+// identifies the owner of the entry) and is deliberately platform-neutral —
+// it was `kglin` until 2026-08-04; the name never lands in stored data, so
+// renaming is wire-only.
+//
+// A trace-keyed map (one entry per trace, "the last inbound seen") used to sit
+// between the two. It was removed: its answer is correct only while exactly one
+// inbound of that trace is in flight — a precondition it never checked and could
+// not verify — and when it was wrong it produced a real, exported, walkable
+// parent that was simply untrue. Un-stamped traffic falls to the wire
+// parent, which is an app-internal span this pipeline never exported: the
+// interaction still derives in full, but as a trace entry rather than a child.
+// A visibly missing edge is recoverable; a silently wrong one is not.
+const tracestateStampKey = "dg-parent"
+
+func init() {
+	plugins.RegisterPlugin(pluginName, func() pipeline.Plugin { return NewLineageTelemetry() })
+}
+
+// exchangeState carries what OnFinish needs to emit the response span as the
+// twin of the request span emitted in OnRequest.
+type exchangeState struct {
+	// reqCtx is the (already-ended) request span's context — the parent of
+	// the response span. An ended span's SpanContext is a valid parent.
+	reqCtx trace.SpanContext
+	// common holds the attributes shared by both spans (lineage.direction,
+	// self.id, peer.*, protocol, exchange.id) — NOT lineage.role, which
+	// differs per span. Computed once so both spans agree byte-for-byte.
+	common   []attribute.KeyValue
+	spanKind trace.SpanKind
+	// spanName is the request span's name; the response span appends " response".
+	spanName string
+	// protocol is the request span's lineage.protocol fact; the response
+	// span's output.value must be read through the SAME protocol's parser
+	// (parsers are precedence-ordered, not mutually exclusive — mcp-parser
+	// also matches any JSON-RPC body, including every a2a exchange).
+	protocol string
+}
+
+// LineageTelemetry emits OTel spans for each request hop observed by authbridge.
+type LineageTelemetry struct {
+	cfg        Config
+	tp         *sdktrace.TracerProvider
+	tracer     trace.Tracer
+	ready      atomic.Bool
+	propagator propagation.TextMapPropagator
+	selfID     string // agent's own client ID for the lineage.self.id fact
+}
+
+// NewLineageTelemetry constructs an unconfigured plugin. Configure + Init must
+// run before it serves traffic (guarded by Ready()).
+func NewLineageTelemetry() *LineageTelemetry {
+	return &LineageTelemetry{
+		propagator: propagation.TraceContext{},
+	}
+}
+
+func (p *LineageTelemetry) Name() string { return pluginName }
+
+func (p *LineageTelemetry) Capabilities() pipeline.PluginCapabilities {
+	return pipeline.PluginCapabilities{
+		// At least one protocol parser must be present and earlier in the
+		// chain: the protocol fact and both payload reductions read the
+		// parsers' Extensions. A chain that misorders lineage before its
+		// parsers (or has none) fails at startup instead of silently
+		// emitting lineage.protocol="http" on everything. jwt-validation
+		// ordering (for the principal facts) cannot be soft-declared under
+		// this capabilities model — list it before lineage in the YAML.
+		RequiresAny: []string{"a2a-parser", "mcp-parser", "inference-parser"},
+		// The contract is cited major.minor only, deliberately: patch
+		// revisions (v1.5.x) clarify prose and never change span semantics,
+		// so a patch bump must not imply a producer change.
+		Description: "Emits two facts-only lineage spans per HTTP exchange (wire contract v1.5).",
+	}
+}
+
+func (p *LineageTelemetry) Configure(raw json.RawMessage) error {
+	cfg, err := decodeConfig(raw)
+	if err != nil {
+		return err
+	}
+	p.cfg = cfg
+	return nil
+}
+
+func (p *LineageTelemetry) Init(ctx context.Context) error {
+	endpoint := p.cfg.OTelEndpoint
+	conn, err := grpc.NewClient(endpoint,
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	if err != nil {
+		return fmt.Errorf("lineage-telemetry: gRPC dial %s: %w", endpoint, err)
+	}
+
+	exporter, err := otlptracegrpc.New(ctx,
+		otlptracegrpc.WithGRPCConn(conn),
+	)
+	if err != nil {
+		return fmt.Errorf("lineage-telemetry: OTLP exporter: %w", err)
+	}
+
+	res, err := resource.New(ctx,
+		resource.WithAttributes(
+			semconv.ServiceNameKey.String("authbridge"),
+			attribute.String("authbridge.component", pluginName),
+		),
+	)
+	if err != nil {
+		slog.Warn("lineage-telemetry: resource detection failed, using default", "error", err)
+		res = resource.Default()
+	}
+
+	p.tp = sdktrace.NewTracerProvider(
+		sdktrace.WithBatcher(exporter),
+		sdktrace.WithResource(res),
+	)
+	p.tracer = p.tp.Tracer("authbridge/" + pluginName)
+
+	// Resolve self identity for the lineage.self.id fact. Every span this
+	// plugin emits is a claim of the form "X did Y"; with no X there is no
+	// claim to make, so an unresolvable identity refuses to start rather
+	// than serving traffic under a plausible-but-wrong label ("no mechanism
+	// may guess", contract v1.3). Note the asymmetry with this file's other
+	// unknowns: a missing status, payload or parent anchor is a missing PART
+	// of a fact and degrades honestly (abandoned / NULL / parent.source=wire).
+	// Identity is the fact's subject — it has no degraded form, and a shared
+	// placeholder would collapse every unidentified pod onto one entity row
+	// (entity id = uuid5("{kind}:{self.id}"), and entities is upsert-only).
+	if p.cfg.SelfID != "" {
+		p.selfID = p.cfg.SelfID
+	} else if p.cfg.SelfIDFile != "" {
+		raw, err := os.ReadFile(p.cfg.SelfIDFile)
+		if err != nil {
+			return fmt.Errorf("lineage-telemetry: no inline self_id and self_id_file unreadable: %w", err)
+		}
+		p.selfID = strings.TrimSpace(string(raw))
+	}
+	if p.selfID == "" {
+		return fmt.Errorf("lineage-telemetry: self identity unresolved (empty self_id and self_id_file %q)", p.cfg.SelfIDFile)
+	}
+
+	p.ready.Store(true)
+	slog.Info("lineage-telemetry: initialized", "endpoint", endpoint, "self_id", p.selfID)
+	return nil
+}
+
+func (p *LineageTelemetry) Shutdown(ctx context.Context) error {
+	if p.tp == nil {
+		return nil
+	}
+	return p.tp.Shutdown(ctx)
+}
+
+func (p *LineageTelemetry) Ready() bool { return p.ready.Load() }
+
+func (p *LineageTelemetry) OnRequest(ctx context.Context, pctx *pipeline.Context) pipeline.Action {
+	if !p.ready.Load() {
+		pctx.Skip("not_ready")
+		return pipeline.Action{Type: pipeline.Continue}
+	}
+
+	// Skip infrastructure paths (health checks, agent-card discovery, etc.)
+	for _, prefix := range p.cfg.BypassPaths {
+		if strings.HasPrefix(pctx.Path, prefix) {
+			pctx.Skip("bypass_path")
+			return pipeline.Action{Type: pipeline.Continue}
+		}
+	}
+
+	// Skip infrastructure outbound targets (OTel exporters, metrics scrapers, etc.)
+	for _, substr := range p.cfg.BypassHosts {
+		if strings.Contains(pctx.Host, substr) {
+			pctx.Skip("bypass_host")
+			return pipeline.Action{Type: pipeline.Continue}
+		}
+	}
+
+	// Extract remote trace context from the incoming W3C traceparent header.
+	// HeaderCarrier wraps http.Header and uses case-insensitive Get/Keys so
+	// canonical-form keys ("Traceparent") match the propagator's lowercase
+	// lookups.
+	remoteCtx := p.propagator.Extract(ctx, propagation.HeaderCarrier(pctx.Headers))
+
+	protocol := protocolOf(pctx)
+	self := serviceLabel(p.selfID)
+	spanKind := spanKindFor(pctx.Direction)
+	spanName := requestSpanName(self, protocol, spanOp(pctx, protocol))
+
+	// Facts shared by both spans (exchange.id is appended once the request
+	// span exists, since it IS the request span id).
+	base := baseAttrs(pctx, self, protocol)
+
+	// Request-span attributes: role + shared facts + request-only facts.
+	reqAttrs := make([]attribute.KeyValue, 0, len(base)+8)
+	reqAttrs = append(reqAttrs, attribute.String("lineage.role", "request"))
+	reqAttrs = append(reqAttrs, base...)
+	reqAttrs = p.appendRequestFacts(reqAttrs, pctx, protocol)
+
+	// (3) parent · (4) emit · (5) re-stamp — wire contract v1.5. The emit is
+	// unconditional; the two calls around it are the stamp machinery.
+	//
+	// >>> OPTION-4 DELETION POINT <<<
+	// A pure read-only sidecar deletes exactly the selectParent and
+	// restampTracestate calls below (and the parent.source fact), parenting on
+	// remoteCtx alone — wire-parent-only propagation. The emit stays. The
+	// trade-off to weigh first: without the stamp, two sidecarred pods can
+	// only be joined through the app's own propagation, so cross-pod
+	// parenting degrades to whatever the wire parent happens to carry.
+	parent, parentSource := selectParent(ctx, remoteCtx)
+	reqAttrs = append(reqAttrs, attribute.String("lineage.parent.source", parentSource))
+	reqCtx := p.emitRequestSpan(parent, spanName, spanKind, reqAttrs)
+	exchangeID := reqCtx.SpanID().String()
+	restampTracestate(pctx, remoteCtx, exchangeID)
+
+	common := make([]attribute.KeyValue, 0, len(base)+1)
+	common = append(common, base...)
+	common = append(common, attribute.String("lineage.exchange.id", exchangeID))
+
+	pipeline.SetState(pctx, pluginName, &exchangeState{
+		reqCtx:   reqCtx,
+		common:   common,
+		spanKind: spanKind,
+		spanName: spanName,
+		protocol: protocol,
+	})
+	pctx.Observe("recorded_request")
+	return pipeline.Action{Type: pipeline.Continue}
+}
+
+// selectParent is step (3) of the single-channel parenting mechanism (wire
+// contract v1.5): the parent is the tracestate stamp — the previous sidecar
+// element in the chain (the caller's outbound for an inbound, this pod's
+// inbound for an outbound) — else the wire parent. Same precedence in both
+// directions. There is deliberately no third option: guessing an attribution
+// is worse than declining to give one. Returns the parent context and the
+// source label the caller emits as the lineage.parent.source fact.
+func selectParent(ctx, remoteCtx context.Context) (context.Context, string) {
+	rsc := trace.SpanContextFromContext(remoteCtx)
+	if rsc.IsValid() {
+		if psc, ok := stampedParent(rsc); ok {
+			return trace.ContextWithRemoteSpanContext(ctx, psc), "tracestate"
+		}
+	}
+	return remoteCtx, "wire"
+}
+
+// emitRequestSpan is step (4): emit the request span under parent and end it
+// immediately — no span is held open across the exchange. lineage.exchange.id
+// is the span's OWN id, so it can only be set after Start. Returns the span's
+// context; an ended span's SpanContext remains a valid parent for the response
+// span, and its span id is the exchange id.
+func (p *LineageTelemetry) emitRequestSpan(
+	parent context.Context,
+	spanName string,
+	spanKind trace.SpanKind,
+	reqAttrs []attribute.KeyValue,
+) trace.SpanContext {
+	_, span := p.tracer.Start(parent, spanName,
+		trace.WithSpanKind(spanKind),
+		trace.WithAttributes(reqAttrs...),
+	)
+	sc := span.SpanContext()
+	span.SetAttributes(attribute.String("lineage.exchange.id", sc.SpanID().String()))
+	span.End()
+	return sc
+}
+
+// restampTracestate is step (5): rewrite the forwarded request's tracestate
+// member with this exchange id — both directions. Inbound: the app's
+// propagate-only shim couriers it to exactly the outbound calls this inbound
+// caused. Outbound: the peer sidecar's inbound reads it as its parent. The
+// forwarded traceparent is never modified (see tracestateStampKey). A valid
+// wire traceparent is required — without one the app's shim starts a fresh
+// root trace and drops the tracestate anyway, so there is nothing to stamp.
+// The listener is responsible for propagating this header mutation (ext_proc
+// emits a SetHeaders diff).
+func restampTracestate(pctx *pipeline.Context, remoteCtx context.Context, exchangeID string) {
+	rsc := trace.SpanContextFromContext(remoteCtx)
+	if !rsc.IsValid() {
+		return
+	}
+	ts, err := rsc.TraceState().Insert(tracestateStampKey, exchangeID)
+	if err != nil {
+		// Stamp attempted and refused (tracestate full or a member malformed,
+		// W3C caps at 32 members / 512 bytes). Without this line the outcome
+		// is indistinguishable from "app has no shim".
+		slog.Warn("lineage-telemetry: tracestate stamp rejected; the next element will attribute as wire",
+			"exchange_id", exchangeID, "error", err)
+		return
+	}
+	pctx.Headers.Set("tracestate", ts.String())
+}
+
+// stampedParent resolves the tracestate stamp on an outbound wire context:
+// the inbound exchange id this pod's sidecar wrote into tracestate on the
+// forwarded request, carried back by the app's shim. Returns ok=false when
+// the member is absent or malformed (caller falls back to the wire parent).
+func stampedParent(rsc trace.SpanContext) (trace.SpanContext, bool) {
+	raw := rsc.TraceState().Get(tracestateStampKey)
+	if raw == "" {
+		return trace.SpanContext{}, false
+	}
+	sid, err := trace.SpanIDFromHex(raw)
+	if err != nil {
+		return trace.SpanContext{}, false
+	}
+	psc := trace.NewSpanContext(trace.SpanContextConfig{
+		TraceID:    rsc.TraceID(),
+		SpanID:     sid,
+		TraceFlags: rsc.TraceFlags(),
+		Remote:     true,
+	})
+	return psc, psc.IsValid()
+}
+
+// OnResponse is a no-op. The response span is emitted in OnFinish (which fires
+// on every finished exchange, including denials and abandonments), not here.
+// The method exists only to satisfy the base pipeline.Plugin interface, which
+// mandates OnResponse; it carries no logic in the two-span model.
+func (p *LineageTelemetry) OnResponse(_ context.Context, _ *pipeline.Context) pipeline.Action {
+	return pipeline.Action{Type: pipeline.Continue}
+}
+
+// OnFinish emits the response span — the twin of the request span, parented
+// under it and echoing the same exchange.id — carrying outcome/status/output.
+// Always fires at stream end, so a bodyless or failed exchange still completes
+// as a first-class pair. Runs under a recover so an unexpected state never
+// crashes the pipeline.
+func (p *LineageTelemetry) OnFinish(ctx context.Context, pctx *pipeline.Context) {
+	defer func() {
+		if r := recover(); r != nil {
+			slog.Warn("lineage-telemetry: OnFinish panic recovered", "recover", r)
+		}
+	}()
+
+	state := pipeline.GetState[exchangeState](pctx, pluginName)
+	if state == nil || !state.reqCtx.IsValid() {
+		return
+	}
+
+	outcome, status, hasStatus, deniedBy := lineageOutcome(pctx.Outcome())
+
+	attrs := make([]attribute.KeyValue, 0, len(state.common)+5)
+	attrs = append(attrs, attribute.String("lineage.role", "response"))
+	attrs = append(attrs, state.common...)
+	attrs = append(attrs, attribute.String("lineage.outcome", outcome))
+	if hasStatus {
+		attrs = append(attrs, attribute.Int("http.status_code", status))
+	}
+	if deniedBy != "" {
+		attrs = append(attrs, attribute.String("lineage.denied_by", deniedBy))
+	}
+	if p.cfg.CaptureIO {
+		if v := ioOutputValue(pctx, state.protocol); v != "" {
+			attrs = append(attrs, attribute.String("output.value", v))
+		}
+	}
+
+	parent := trace.ContextWithRemoteSpanContext(ctx, state.reqCtx)
+	_, span := p.tracer.Start(parent, state.spanName+" response",
+		trace.WithSpanKind(state.spanKind),
+		trace.WithAttributes(attrs...),
+	)
+	span.End()
+}
+
+// lineageOutcome maps the pipeline's 3-value Outcome (allow/deny/error, nil
+// outside OnFinish) onto the contract's lineage.outcome vocabulary
+// (ok|denied|error|abandoned) plus the http.status_code fact. A terminal state
+// with no status written (upstream reset, client disconnect, listener death)
+// is "abandoned" — the row completes as in-flight-turned-failed rather than
+// dangling. hasStatus is false when no status code was produced.
+func lineageOutcome(o *pipeline.Outcome) (outcome string, status int, hasStatus bool, deniedBy string) {
+	if o == nil {
+		return "abandoned", 0, false, ""
+	}
+	switch o.FinalAction {
+	case pipeline.OutcomeAllow:
+		return "ok", o.StatusCode, o.StatusCode > 0, ""
+	case pipeline.OutcomeDeny:
+		return "denied", o.StatusCode, o.StatusCode > 0, o.DenyingPlugin
+	case pipeline.OutcomeError:
+		if o.StatusCode > 0 {
+			return "error", o.StatusCode, true, ""
+		}
+		return "abandoned", 0, false, ""
+	default:
+		return "error", o.StatusCode, o.StatusCode > 0, ""
+	}
+}
+
+// protocolOf reports which parser populated Extensions — the lineage.protocol
+// fact. "http" means no parser matched.
+func protocolOf(pctx *pipeline.Context) string {
+	switch {
+	case pctx.Extensions.A2A != nil:
+		return "a2a"
+	case pctx.Extensions.MCP != nil:
+		return "mcp"
+	case pctx.Extensions.Inference != nil:
+		return "inference"
+	default:
+		return "http"
+	}
+}
+
+// spanKindFor maps direction to OTel SpanKind: inbound is SERVER, outbound is
+// CLIENT. The response span reuses its request span's kind.
+func spanKindFor(dir pipeline.Direction) trace.SpanKind {
+	if dir == pipeline.Inbound {
+		return trace.SpanKindServer
+	}
+	return trace.SpanKindClient
+}
+
+// baseAttrs returns the facts carried on BOTH spans except exchange.id (added
+// once the request span id is known) and role (differs per span).
+func baseAttrs(pctx *pipeline.Context, self, protocol string) []attribute.KeyValue {
+	attrs := []attribute.KeyValue{
+		attribute.String("lineage.direction", pctx.Direction.String()),
+		attribute.String("lineage.self.id", self),
+		attribute.String("lineage.protocol", protocol),
+	}
+	if pctx.Host != "" {
+		attrs = append(attrs, attribute.String("lineage.peer.host", pctx.Host))
+	}
+	return attrs
+}
+
+// appendRequestFacts adds the request-only facts: HTTP method/path/scheme, the
+// protocol-specific parsed facts, validated-JWT principal (inbound only), and
+// input.value when capture_io is on. protocolOf guarantees the matching
+// extension pointer is non-nil.
+func (p *LineageTelemetry) appendRequestFacts(attrs []attribute.KeyValue, pctx *pipeline.Context, protocol string) []attribute.KeyValue {
+	if pctx.Method != "" {
+		attrs = append(attrs, attribute.String("http.method", pctx.Method))
+	}
+	if pctx.Path != "" {
+		attrs = append(attrs, attribute.String("url.path", pctx.Path))
+	}
+	if pctx.Scheme != "" {
+		attrs = append(attrs, attribute.String("url.scheme", pctx.Scheme))
+	}
+	switch protocol {
+	case "a2a":
+		a := pctx.Extensions.A2A
+		if a.Method != "" {
+			attrs = append(attrs, attribute.String("a2a.method", a.Method))
+		}
+		if a.SessionID != "" {
+			attrs = append(attrs, attribute.String("a2a.session_id", a.SessionID))
+		}
+	case "mcp":
+		m := pctx.Extensions.MCP
+		if m.Method != "" {
+			attrs = append(attrs, attribute.String("mcp.method", m.Method))
+		}
+		if t := mcpTool(pctx); t != "" {
+			attrs = append(attrs, attribute.String("mcp.tool", t))
+		}
+	case "inference":
+		if model := pctx.Extensions.Inference.Model; model != "" {
+			attrs = append(attrs, attribute.String("inference.model", model))
+		}
+	}
+	// Principal facts: request span, inbound only, and only from a validated
+	// JWT (pctx.Identity non-nil).
+	if pctx.Direction == pipeline.Inbound && pctx.Identity != nil {
+		if s := pctx.Identity.Subject(); s != "" {
+			attrs = append(attrs, attribute.String("lineage.principal.sub", s))
+		}
+		if c := pctx.Identity.ClientID(); c != "" {
+			attrs = append(attrs, attribute.String("lineage.principal.client", c))
+		}
+	}
+	if p.cfg.CaptureIO {
+		if v := ioInputValue(pctx, protocol); v != "" {
+			attrs = append(attrs, attribute.String("input.value", v))
+		}
+	}
+	return attrs
+}
+
+// requestSpanName builds "{self} {protocol} {op}", dropping the trailing op
+// when it is empty. The response span appends " response".
+func requestSpanName(self, protocol, op string) string {
+	if op == "" {
+		return self + " " + protocol
+	}
+	return self + " " + protocol + " " + op
+}
+
+// spanOp picks the operation label for the span name per protocol:
+// mcp.tool / a2a.method / inference.model, falling back to url.path.
+func spanOp(pctx *pipeline.Context, protocol string) string {
+	var op string
+	switch protocol {
+	case "a2a":
+		if pctx.Extensions.A2A != nil {
+			op = pctx.Extensions.A2A.Method
+		}
+	case "mcp":
+		op = mcpTool(pctx)
+		if op == "" && pctx.Extensions.MCP != nil {
+			op = pctx.Extensions.MCP.Method
+		}
+	case "inference":
+		if pctx.Extensions.Inference != nil {
+			op = pctx.Extensions.Inference.Model
+		}
+	}
+	if op == "" {
+		op = pctx.Path
+	}
+	return op
+}
+
+// mcpTool returns the tool name for an MCP tools/call, or "" otherwise.
+func mcpTool(pctx *pipeline.Context) string {
+	m := pctx.Extensions.MCP
+	if m == nil || m.Method != "tools/call" || m.Params == nil {
+		return ""
+	}
+	if name, ok := m.Params["name"].(string); ok {
+		return name
+	}
+	return ""
+}
+
+// serviceLabel reduces a SPIFFE ID to its last path segment, or returns
+// selfID as-is if it is not a SPIFFE URI. Used for the lineage.self.id fact
+// and span names.
+//
+//	"spiffe://trust-domain/ns/team1/sa/weather-service" → "weather-service"
+//	"weather-service" → "weather-service"
+//
+// selfID is never empty at the only call site: Init refuses to start without
+// a resolved identity (v1.3). There is deliberately no empty-string fallback —
+// inventing a label is the guess that rule exists to forbid.
+func serviceLabel(selfID string) string {
+	parts := strings.Split(selfID, "/")
+	for i := len(parts) - 1; i >= 0; i-- {
+		if parts[i] != "" {
+			return parts[i]
+		}
+	}
+	return selfID
+}
+
+// ioInputValue returns the input.value for a request span: the parsed request
+// content for *protocol* — the hop's lineage.protocol fact — or "" if that
+// parser produced nothing meaningful. Only that protocol's extension is read:
+// parsers are precedence-ordered, not mutually exclusive (mcp-parser matches
+// any JSON-RPC body, including every a2a exchange), so falling through to
+// another parser's output would attach a mislabeled protocol envelope. A hop
+// whose own parser yields nothing keeps a NULL payload — the contract's
+// "interactions are independent of payloads".
+func ioInputValue(pctx *pipeline.Context, protocol string) string {
+	ext := pctx.Extensions
+	switch {
+	case protocol == "a2a" && ext.A2A != nil && len(ext.A2A.Parts) > 0:
+		// Collect all text parts; fall back to JSON if non-text parts present.
+		var texts []string
+		for _, p := range ext.A2A.Parts {
+			if p.Content != "" {
+				texts = append(texts, p.Content)
+			}
+		}
+		if len(texts) > 0 {
+			return strings.Join(texts, "\n")
+		}
+		if b, err := json.Marshal(ext.A2A.Parts); err == nil {
+			return string(b)
+		}
+	case protocol == "inference" && ext.Inference != nil && len(ext.Inference.Messages) > 0:
+		if b, err := json.Marshal(ext.Inference.Messages); err == nil {
+			return string(b)
+		}
+	case protocol == "mcp" && ext.MCP != nil && ext.MCP.Params != nil:
+		// For tools/call, surface just the arguments (the semantically
+		// meaningful part) rather than the full {"name":…,"arguments":…} wrapper.
+		if ext.MCP.Method == "tools/call" {
+			if args, ok := ext.MCP.Params["arguments"]; ok {
+				if b, err := json.Marshal(args); err == nil {
+					return string(b)
+				}
+			}
+		}
+		if b, err := json.Marshal(ext.MCP.Params); err == nil {
+			return string(b)
+		}
+	}
+	return ""
+}
+
+// isA2AProtocolEvent returns true when s is a JSON object carrying an A2A
+// transport-level "kind" field (status-update, task-status-update, etc.)
+// rather than actual content. Used to avoid surfacing protocol metadata
+// as output.value when the a2a-parser captures a protocol event as the
+// artifact instead of the real agent response text.
+func isA2AProtocolEvent(s string) bool {
+	var obj map[string]json.RawMessage
+	if json.Unmarshal([]byte(s), &obj) != nil {
+		return false
+	}
+	var kind string
+	if raw, ok := obj["kind"]; ok {
+		_ = json.Unmarshal(raw, &kind)
+	}
+	return strings.Contains(kind, "status") || strings.Contains(kind, "artifact-update") ||
+		strings.Contains(kind, "Status") || kind == "working" || kind == "canceled"
+}
+
+// ioOutputValue returns the output.value for a response span: the parsed
+// response content for *protocol* — the REQUEST span's lineage.protocol fact —
+// or "" if that parser produced nothing. Only that protocol's extension is
+// read, for the same reason as ioInputValue: mcp-parser also parses every a2a
+// response (any JSON-RPC body), and falling through to it would emit the raw
+// JSON-RPC envelope — including the protocol events isA2AProtocolEvent exists
+// to suppress — as an a2a hop's payload.
+func ioOutputValue(pctx *pipeline.Context, protocol string) string {
+	ext := pctx.Extensions
+	switch {
+	case protocol == "a2a" && ext.A2A != nil && ext.A2A.Artifact != "" && !isA2AProtocolEvent(ext.A2A.Artifact):
+		return ext.A2A.Artifact
+	case protocol == "a2a" && ext.A2A != nil && ext.A2A.ErrorMessage != "":
+		return ext.A2A.ErrorMessage
+	case protocol == "inference" && ext.Inference != nil && ext.Inference.Completion != "":
+		return ext.Inference.Completion
+	case protocol == "inference" && ext.Inference != nil && len(ext.Inference.ToolCalls) > 0:
+		if b, err := json.Marshal(ext.Inference.ToolCalls); err == nil {
+			return string(b)
+		}
+	case protocol == "mcp" && ext.MCP != nil && ext.MCP.Result != nil:
+		// For tools/call results, extract the text content from the MCP
+		// content array rather than returning the full {"content":[…],"_meta":…}
+		// envelope, so the output matches what Phoenix shows for the tool span.
+		if ext.MCP.Method == "tools/call" {
+			if content, ok := ext.MCP.Result["content"]; ok {
+				if items, ok := content.([]any); ok {
+					var texts []string
+					for _, item := range items {
+						if m, ok := item.(map[string]any); ok {
+							if m["type"] == "text" {
+								if t, ok := m["text"].(string); ok && t != "" {
+									texts = append(texts, t)
+								}
+							}
+						}
+					}
+					if len(texts) > 0 {
+						return strings.Join(texts, "\n")
+					}
+				}
+			}
+		}
+		if b, err := json.Marshal(ext.MCP.Result); err == nil {
+			return string(b)
+		}
+	case protocol == "mcp" && ext.MCP != nil && ext.MCP.Err != nil:
+		if b, err := json.Marshal(ext.MCP.Err); err == nil {
+			return string(b)
+		}
+	}
+	return ""
+}
+
+// Compile-time interface assertions.
+var (
+	_ pipeline.Plugin       = (*LineageTelemetry)(nil)
+	_ pipeline.Configurable = (*LineageTelemetry)(nil)
+	_ pipeline.Initializer  = (*LineageTelemetry)(nil)
+	_ pipeline.Shutdowner   = (*LineageTelemetry)(nil)
+	_ pipeline.Finisher     = (*LineageTelemetry)(nil)
+	_ pipeline.Readier      = (*LineageTelemetry)(nil)
+)
diff --git a/authbridge/authlib/plugins/lineage/plugin_test.go b/authbridge/authlib/plugins/lineage/plugin_test.go
new file mode 100644
index 000000000..59de55f05
--- /dev/null
+++ b/authbridge/authlib/plugins/lineage/plugin_test.go
@@ -0,0 +1,912 @@
+package lineage
+
+import (
+	"context"
+	"encoding/json"
+	"maps"
+	"net/http"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/propagation"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	"go.opentelemetry.io/otel/sdk/trace/tracetest"
+	"go.opentelemetry.io/otel/trace"
+
+	"github.com/rossoctl/cortex/authbridge/authlib/pipeline"
+)
+
+// newTestPlugin creates a LineageTelemetry wired to an in-memory span exporter
+// (synchronous, so a span appears the instant it is ended) and marks it ready
+// so Init is not needed.
+func newTestPlugin(t *testing.T) (*LineageTelemetry, *tracetest.InMemoryExporter) {
+	t.Helper()
+	exp := tracetest.NewInMemoryExporter()
+	tp := sdktrace.NewTracerProvider(sdktrace.WithSyncer(exp))
+	p := NewLineageTelemetry()
+	p.tp = tp
+	p.tracer = tp.Tracer("test")
+	p.selfID = "weather-service"
+	p.ready.Store(true)
+	return p, exp
+}
+
+// run drives a full exchange (request pass + finish) through a single-plugin
+// pipeline. Spans are read from the caller's exporter.
+func run(t *testing.T, p *LineageTelemetry, pctx *pipeline.Context, outcome pipeline.Outcome) {
+	t.Helper()
+	pl, err := pipeline.New([]pipeline.Plugin{p})
+	if err != nil {
+		t.Fatalf("pipeline.New: %v", err)
+	}
+	pl.Run(context.Background(), pctx)
+	pl.RunFinish(context.Background(), pctx, outcome)
+}
+
+// allow is the ordinary success outcome.
+func allow(status int) pipeline.Outcome {
+	return pipeline.Outcome{FinalAction: pipeline.OutcomeAllow, StatusCode: status}
+}
+
+// fakeContext mirrors what the real listeners supply. Method is populated
+// because every listener now supplies it (reverseproxy/forwardproxy from
+// r.Method, ext_proc from the :method pseudo-header) — if a listener stops,
+// the fixture must change with it rather than keep asserting a fiction.
+func fakeContext(dir pipeline.Direction, headers http.Header) *pipeline.Context {
+	return &pipeline.Context{
+		Direction: dir,
+		Method:    "POST",
+		Host:      "test-service:8000",
+		Path:      "/test",
+		Headers:   headers,
+	}
+}
+
+// traceparent builds a header carrier naming traceID/spanID as the wire parent.
+func traceparent(traceID, spanID string) http.Header {
+	h := http.Header{}
+	h.Set("traceparent", "00-"+traceID+"-"+spanID+"-01")
+	return h
+}
+
+// extractParent decodes the span context named by the headers' traceparent.
+func extractParent(h http.Header) trace.SpanContext {
+	ctx := propagation.TraceContext{}.Extract(context.Background(), propagation.HeaderCarrier(h))
+	return trace.SpanContextFromContext(ctx)
+}
+
+// roleSplit returns the request and response spans from an exported set,
+// asserting exactly one of each.
+func roleSplit(t *testing.T, spans tracetest.SpanStubs) (req, resp tracetest.SpanStub) {
+	t.Helper()
+	var gotReq, gotResp bool
+	for _, s := range spans {
+		switch attrStr(s, "lineage.role") {
+		case "request":
+			if gotReq {
+				t.Fatal("more than one request span")
+			}
+			req, gotReq = s, true
+		case "response":
+			if gotResp {
+				t.Fatal("more than one response span")
+			}
+			resp, gotResp = s, true
+		default:
+			t.Fatalf("span %q has no lineage.role", s.Name)
+		}
+	}
+	if !gotReq || !gotResp {
+		t.Fatalf("want one request + one response span, got %d spans (req=%v resp=%v)", len(spans), gotReq, gotResp)
+	}
+	return req, resp
+}
+
+// ---- identifiers, pairing, parenting ----
+
+func TestExchange_TwoSpansPairedAndParented(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	pctx := fakeContext(pipeline.Inbound, http.Header{})
+
+	pl, err := pipeline.New([]pipeline.Plugin{p})
+	if err != nil {
+		t.Fatalf("pipeline.New: %v", err)
+	}
+
+	// Emit on sight: the request span exists after the request pass, before finish.
+	pl.Run(context.Background(), pctx)
+	if got := len(exp.GetSpans()); got != 1 {
+		t.Fatalf("after request pass: want 1 span (request), got %d", got)
+	}
+
+	pl.RunFinish(context.Background(), pctx, allow(200))
+	spans := exp.GetSpans()
+	if len(spans) != 2 {
+		t.Fatalf("after finish: want 2 spans, got %d", len(spans))
+	}
+	req, resp := roleSplit(t, spans)
+
+	// exchange.id == request span id, echoed on both.
+	wantID := req.SpanContext.SpanID().String()
+	if got := attrStr(req, "lineage.exchange.id"); got != wantID {
+		t.Errorf("request exchange.id = %q, want %q", got, wantID)
+	}
+	if got := attrStr(resp, "lineage.exchange.id"); got != wantID {
+		t.Errorf("response exchange.id = %q, want %q", got, wantID)
+	}
+
+	// Response span's parent is the request span (same trace).
+	if resp.Parent.SpanID() != req.SpanContext.SpanID() {
+		t.Errorf("response parent span = %s, want request span %s", resp.Parent.SpanID(), req.SpanContext.SpanID())
+	}
+	if resp.SpanContext.TraceID() != req.SpanContext.TraceID() {
+		t.Errorf("response trace = %s, want request trace %s", resp.SpanContext.TraceID(), req.SpanContext.TraceID())
+	}
+
+	// Both spans share the same SpanKind (SERVER for inbound).
+	if req.SpanKind != trace.SpanKindServer || resp.SpanKind != trace.SpanKindServer {
+		t.Errorf("span kinds = %v/%v, want server/server", req.SpanKind, resp.SpanKind)
+	}
+}
+
+// ---- the stamp (single-channel parenting, wire contract v1.5) ----
+
+// TestStamp_OutboundRewritesStampNotTraceparent: the outbound reads its
+// parent from the inbound's stamp, then re-stamps the forwarded tracestate
+// with its OWN request span id for the peer sidecar's inbound to read. The
+// forwarded traceparent is NOT modified (v1.5 removed the splice) — an app
+// chain riding traceparent toward its own backend stays intact.
+func TestStamp_OutboundRewritesStampNotTraceparent(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	const traceID, wireParent = "4bf92f3577b34da6a3ce929d0e0e4736", "00f067aa0ba902b7"
+	const inboundID = "1111111111111111"
+	h := traceparent(traceID, wireParent)
+	h.Set("tracestate", tracestateStampKey+"="+inboundID)
+	pctx := fakeContext(pipeline.Outbound, h)
+	pctx.Extensions.MCP = &pipeline.MCPExtension{Method: "tools/call", Params: map[string]any{"name": "get_weather"}}
+
+	run(t, p, pctx, allow(200))
+
+	req, _ := roleSplit(t, exp.GetSpans())
+	// Parent comes from the inbound's stamp.
+	if got := req.Parent.SpanID().String(); got != inboundID {
+		t.Errorf("parent = %s, want stamped inbound %s", got, inboundID)
+	}
+	// The forwarded traceparent is untouched — still the wire parent.
+	forwarded := extractParent(pctx.Headers)
+	if got := forwarded.SpanID().String(); got != wireParent {
+		t.Errorf("forwarded traceparent parent = %s, want untouched wire parent %s", got, wireParent)
+	}
+	if got := forwarded.TraceID().String(); got != traceID {
+		t.Errorf("forwarded trace = %s, want %s", got, traceID)
+	}
+	// The forwarded tracestate now stamps THIS outbound request span,
+	// replacing the inbound's stamp it consumed.
+	want := tracestateStampKey + "=" + req.SpanContext.SpanID().String()
+	if got := pctx.Headers.Get("tracestate"); got != want {
+		t.Errorf("tracestate = %q, want re-stamp %q", got, want)
+	}
+}
+
+// TestStamp_InboundParentsOnPeerStamp is the cross-pod link: the caller
+// sidecar's outbound stamped tracestate with its request span id, and this
+// inbound must parent on that stamp — not on the wire traceparent, whose
+// span id may belong to an app chain this pipeline never exports.
+func TestStamp_InboundParentsOnPeerStamp(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	const traceID, wireParent = "4bf92f3577b34da6a3ce929d0e0e4736", "00f067aa0ba902b7"
+	const peerOutbound = "2222222222222222"
+	h := traceparent(traceID, wireParent)
+	h.Set("tracestate", tracestateStampKey+"="+peerOutbound)
+	pctx := fakeContext(pipeline.Inbound, h)
+
+	run(t, p, pctx, allow(200))
+
+	req, _ := roleSplit(t, exp.GetSpans())
+	if got := req.Parent.SpanID().String(); got != peerOutbound {
+		t.Errorf("parent = %s, want peer outbound stamp %s", got, peerOutbound)
+	}
+	if got := attrStr(req, "lineage.parent.source"); got != "tracestate" {
+		t.Errorf("lineage.parent.source = %q, want tracestate", got)
+	}
+	// The forwarded stamp now names THIS inbound request span — the app's
+	// shim couriers it to exactly the outbound calls this inbound causes.
+	want := tracestateStampKey + "=" + req.SpanContext.SpanID().String()
+	if got := pctx.Headers.Get("tracestate"); got != want {
+		t.Errorf("tracestate = %q, want re-stamp %q", got, want)
+	}
+}
+
+func TestStamp_InboundHeadersUntouchedExceptStamp(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	h := traceparent("4bf92f3577b34da6a3ce929d0e0e4736", "00f067aa0ba902b7")
+	before := http.Header{}
+	maps.Copy(before, h)
+	pctx := fakeContext(pipeline.Inbound, h)
+
+	run(t, p, pctx, allow(200))
+
+	// The ONLY inbound mutation is the tracestate stamp; traceparent and
+	// everything else are forwarded as they arrived.
+	req, _ := roleSplit(t, exp.GetSpans())
+	want := tracestateStampKey + "=" + req.SpanContext.SpanID().String()
+	if got := pctx.Headers.Get("tracestate"); got != want {
+		t.Errorf("tracestate = %q, want stamp %q", got, want)
+	}
+	after := http.Header{}
+	maps.Copy(after, pctx.Headers)
+	after.Del("tracestate")
+	if !headersEqual(before, after) {
+		t.Errorf("inbound headers beyond tracestate mutated: before=%v after=%v", before, after)
+	}
+	// No stamp arrived, so the parent is the wire traceparent — recorded as such.
+	if got := attrStr(req, "lineage.parent.source"); got != "wire" {
+		t.Errorf("lineage.parent.source = %q, want wire", got)
+	}
+}
+
+func TestStamp_PreservesForeignTracestateMembers(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	h := traceparent("4bf92f3577b34da6a3ce929d0e0e4736", "00f067aa0ba902b7")
+	h.Set("tracestate", "vendor=abc")
+	pctx := fakeContext(pipeline.Inbound, h)
+
+	run(t, p, pctx, allow(200))
+
+	req, _ := roleSplit(t, exp.GetSpans())
+	got := pctx.Headers.Get("tracestate")
+	wantStamp := tracestateStampKey + "=" + req.SpanContext.SpanID().String()
+	if !strings.Contains(got, wantStamp) || !strings.Contains(got, "vendor=abc") {
+		t.Errorf("tracestate = %q, want both %q and vendor=abc", got, wantStamp)
+	}
+}
+
+func TestStamp_NoWireTraceparentNoStamp(t *testing.T) {
+	p, _ := newTestPlugin(t)
+	pctx := fakeContext(pipeline.Inbound, http.Header{})
+
+	run(t, p, pctx, allow(200))
+
+	if got := pctx.Headers.Get("tracestate"); got != "" {
+		t.Errorf("tracestate stamped without a wire traceparent: %q", got)
+	}
+}
+
+// TestStamp_OutboundPrefersStampOverMap is the same-trace fan-in case in
+// miniature: two concurrent inbound exchanges on ONE trace (the trace-keyed
+// map can only hold the later one), then an outbound whose tracestate stamp
+// names the EARLIER inbound. Without the stamp this outbound would collapse
+// onto the map entry — the 1/N misattribution the fanin-test.sh e2e proves.
+func TestStamp_OutboundUsesTheStampedInbound(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	const traceID = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+
+	// Two concurrent inbounds on the SAME trace — the case no trace-keyed
+	// structure can disambiguate, and the reason the stamp exists.
+	run(t, p, fakeContext(pipeline.Inbound, traceparent(traceID, "1111111111111111")), allow(200))
+	in1, _ := roleSplit(t, exp.GetSpans())
+	exp.Reset()
+	run(t, p, fakeContext(pipeline.Inbound, traceparent(traceID, "2222222222222222")), allow(200))
+	in2, _ := roleSplit(t, exp.GetSpans())
+
+	// Outbound couriered in1's stamp back through the app. It must parent
+	// under in1 specifically — not in2, not the wire parent.
+	exp.Reset()
+	h := traceparent(traceID, "3333333333333333")
+	h.Set("tracestate", tracestateStampKey+"="+in1.SpanContext.SpanID().String())
+	run(t, p, fakeContext(pipeline.Outbound, h), allow(200))
+	outReq, _ := roleSplit(t, exp.GetSpans())
+
+	if outReq.Parent.SpanID() != in1.SpanContext.SpanID() {
+		t.Errorf("parent = %s, want stamped inbound %s (the other in-flight inbound was %s)",
+			outReq.Parent.SpanID(), in1.SpanContext.SpanID(), in2.SpanContext.SpanID())
+	}
+	if got := attrStr(outReq, "lineage.parent.source"); got != "tracestate" {
+		t.Errorf("lineage.parent.source = %q, want tracestate", got)
+	}
+}
+
+func TestStamp_MalformedFallsBackToWire(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	const traceID = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+	const wireParent = "3333333333333333"
+
+	// An inbound on this trace exists — and must NOT be used, because a
+	// malformed stamp means "unknown", not "guess for me".
+	run(t, p, fakeContext(pipeline.Inbound, traceparent(traceID, "1111111111111111")), allow(200))
+	in1, _ := roleSplit(t, exp.GetSpans())
+
+	exp.Reset()
+	h := traceparent(traceID, wireParent)
+	h.Set("tracestate", tracestateStampKey+"=nothex")
+	run(t, p, fakeContext(pipeline.Outbound, h), allow(200))
+	outReq, _ := roleSplit(t, exp.GetSpans())
+
+	if got := outReq.Parent.SpanID().String(); got != wireParent {
+		t.Errorf("parent = %s, want wire parent %s", got, wireParent)
+	}
+	if outReq.Parent.SpanID() == in1.SpanContext.SpanID() {
+		t.Error("malformed stamp silently inherited this pod's inbound span")
+	}
+	if got := attrStr(outReq, "lineage.parent.source"); got != "wire" {
+		t.Errorf("lineage.parent.source = %q, want wire", got)
+	}
+}
+
+func TestStamp_ParentSourceWireWhenUnstamped(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	out := fakeContext(pipeline.Outbound, traceparent("cccccccccccccccccccccccccccccccc", "1111111111111111"))
+	run(t, p, out, allow(200))
+	outReq, _ := roleSplit(t, exp.GetSpans())
+	if got := attrStr(outReq, "lineage.parent.source"); got != "wire" {
+		t.Errorf("lineage.parent.source = %q, want wire", got)
+	}
+}
+
+// TestStamp_UnstampedOutboundNeverInheritsInbound is the regression guard for
+// the removal of the trace-keyed map. An outbound with no stamp must fall to the
+// wire parent EVEN WHEN this pod has an inbound span for the same trace. The old
+// map answered such cases from "the last inbound seen", which is correct only
+// while exactly one inbound is in flight — a precondition it never checked. A
+// missing edge is recoverable; a confidently wrong one is not.
+func TestStamp_UnstampedOutboundNeverInheritsInbound(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	const traceID = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+	const wireParent = "3333333333333333"
+
+	run(t, p, fakeContext(pipeline.Inbound, traceparent(traceID, "2222222222222222")), allow(200))
+	inReq, _ := roleSplit(t, exp.GetSpans())
+
+	exp.Reset()
+	run(t, p, fakeContext(pipeline.Outbound, traceparent(traceID, wireParent)), allow(200))
+	outReq, _ := roleSplit(t, exp.GetSpans())
+
+	if outReq.Parent.SpanID() == inReq.SpanContext.SpanID() {
+		t.Fatal("un-stamped outbound inherited this pod's inbound span — the map is back")
+	}
+	if got := outReq.Parent.SpanID().String(); got != wireParent {
+		t.Errorf("parent = %s, want wire parent %s", got, wireParent)
+	}
+	if got := attrStr(outReq, "lineage.parent.source"); got != "wire" {
+		t.Errorf("lineage.parent.source = %q, want wire", got)
+	}
+}
+
+func TestStamp_ConcurrentTracesNeverCross(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	const traceA = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+	const traceB = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+
+	// One inbound on each trace.
+	run(t, p, fakeContext(pipeline.Inbound, traceparent(traceA, "1111111111111111")), allow(200))
+	inA, _ := roleSplit(t, exp.GetSpans())
+	exp.Reset()
+	run(t, p, fakeContext(pipeline.Inbound, traceparent(traceB, "2222222222222222")), allow(200))
+	inB, _ := roleSplit(t, exp.GetSpans())
+
+	// Each outbound couriers its own trace's stamp back.
+	exp.Reset()
+	hA := traceparent(traceA, "3333333333333333")
+	hA.Set("tracestate", tracestateStampKey+"="+inA.SpanContext.SpanID().String())
+	run(t, p, fakeContext(pipeline.Outbound, hA), allow(200))
+	outA, _ := roleSplit(t, exp.GetSpans())
+	if outA.Parent.SpanID() != inA.SpanContext.SpanID() {
+		t.Errorf("outbound A parent = %s, want inbound A %s", outA.Parent.SpanID(), inA.SpanContext.SpanID())
+	}
+	if outA.Parent.SpanID() == inB.SpanContext.SpanID() {
+		t.Error("outbound A crossed into inbound B's span")
+	}
+	if outA.SpanContext.TraceID().String() != traceA {
+		t.Errorf("outbound A trace = %s, want %s", outA.SpanContext.TraceID(), traceA)
+	}
+
+	exp.Reset()
+	hB := traceparent(traceB, "4444444444444444")
+	hB.Set("tracestate", tracestateStampKey+"="+inB.SpanContext.SpanID().String())
+	run(t, p, fakeContext(pipeline.Outbound, hB), allow(200))
+	outB, _ := roleSplit(t, exp.GetSpans())
+	if outB.Parent.SpanID() != inB.SpanContext.SpanID() {
+		t.Errorf("outbound B parent = %s, want inbound B %s", outB.Parent.SpanID(), inB.SpanContext.SpanID())
+	}
+}
+
+// ---- bodyless / unparsed completeness ----
+
+func TestBodyless_UnparsedNoCaptureStillEmitsBothSpans(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	// capture_io defaults false; no parser extensions → protocol http.
+	pctx := fakeContext(pipeline.Outbound, http.Header{})
+
+	run(t, p, pctx, allow(200))
+
+	req, resp := roleSplit(t, exp.GetSpans())
+	if got := attrStr(req, "lineage.protocol"); got != "http" {
+		t.Errorf("protocol = %q, want http", got)
+	}
+	// Complete: both carry the shared facts and the exchange is paired.
+	if attrStr(req, "lineage.exchange.id") == "" || attrStr(resp, "lineage.exchange.id") == "" {
+		t.Error("exchange.id missing on a bodyless span")
+	}
+	if got := attrStr(resp, "lineage.outcome"); got != "ok" {
+		t.Errorf("outcome = %q, want ok", got)
+	}
+	// No payloads captured.
+	if _, ok := findAttr(req, "input.value"); ok {
+		t.Error("input.value present with capture_io off")
+	}
+	if _, ok := findAttr(resp, "output.value"); ok {
+		t.Error("output.value present with capture_io off")
+	}
+}
+
+// ---- outcomes ----
+
+func TestOutcome_Denied(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	pctx := fakeContext(pipeline.Inbound, http.Header{})
+
+	run(t, p, pctx, pipeline.Outcome{
+		FinalAction:   pipeline.OutcomeDeny,
+		StatusCode:    401,
+		DenyingPlugin: "jwt-validation",
+	})
+
+	_, resp := roleSplit(t, exp.GetSpans())
+	if got := attrStr(resp, "lineage.outcome"); got != "denied" {
+		t.Errorf("outcome = %q, want denied", got)
+	}
+	if got := attrStr(resp, "lineage.denied_by"); got != "jwt-validation" {
+		t.Errorf("denied_by = %q, want jwt-validation", got)
+	}
+	if got, ok := intAttr(resp, "http.status_code"); !ok || got != 401 {
+		t.Errorf("http.status_code = %d (ok=%v), want 401", got, ok)
+	}
+}
+
+func TestOutcome_AbandonedHasNoStatus(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	pctx := fakeContext(pipeline.Outbound, http.Header{})
+
+	// Terminal error with no response written (upstream reset / disconnect).
+	run(t, p, pctx, pipeline.Outcome{FinalAction: pipeline.OutcomeError, StatusCode: 0})
+
+	_, resp := roleSplit(t, exp.GetSpans())
+	if got := attrStr(resp, "lineage.outcome"); got != "abandoned" {
+		t.Errorf("outcome = %q, want abandoned", got)
+	}
+	if _, ok := findAttr(resp, "http.status_code"); ok {
+		t.Error("http.status_code present on an abandoned exchange (none was produced)")
+	}
+}
+
+// ---- request facts + capture_io + span names ----
+
+func TestRequestFacts_MCPWithCapture(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	p.cfg.CaptureIO = true
+	pctx := fakeContext(pipeline.Outbound, http.Header{})
+	pctx.Host = "weather-tool-mcp.team1.svc:8000"
+	pctx.Path = "/mcp"
+	pctx.Scheme = "http"
+	pctx.Extensions.MCP = &pipeline.MCPExtension{
+		Method: "tools/call",
+		Params: map[string]any{"name": "get_weather", "arguments": map[string]any{"city": "Tokyo"}},
+		Result: map[string]any{"content": []any{map[string]any{"type": "text", "text": "sunny"}}},
+	}
+
+	run(t, p, pctx, allow(200))
+	req, resp := roleSplit(t, exp.GetSpans())
+
+	checkAttr(t, req, "lineage.protocol", "mcp")
+	checkAttr(t, req, "mcp.method", "tools/call")
+	checkAttr(t, req, "mcp.tool", "get_weather")
+	checkAttr(t, req, "http.method", "POST")
+	checkAttr(t, req, "url.path", "/mcp")
+	checkAttr(t, req, "url.scheme", "http")
+	checkAttr(t, req, "lineage.self.id", "weather-service")
+	checkAttr(t, req, "lineage.peer.host", "weather-tool-mcp.team1.svc:8000")
+	checkAttr(t, req, "lineage.direction", "outbound")
+	checkAttr(t, req, "input.value", `{"city":"Tokyo"}`)
+	checkAttr(t, resp, "output.value", "sunny")
+
+	if req.Name != "weather-service mcp get_weather" {
+		t.Errorf("request span name = %q", req.Name)
+	}
+	if resp.Name != "weather-service mcp get_weather response" {
+		t.Errorf("response span name = %q", resp.Name)
+	}
+	if req.SpanKind != trace.SpanKindClient {
+		t.Errorf("outbound request kind = %v, want client", req.SpanKind)
+	}
+}
+
+func TestRequestFacts_A2AAndInference(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	// A2A.
+	a := fakeContext(pipeline.Outbound, http.Header{})
+	a.Extensions.A2A = &pipeline.A2AExtension{Method: "message/send", SessionID: "sess-123"}
+	run(t, p, a, allow(200))
+	areq, _ := roleSplit(t, exp.GetSpans())
+	checkAttr(t, areq, "lineage.protocol", "a2a")
+	checkAttr(t, areq, "a2a.method", "message/send")
+	checkAttr(t, areq, "a2a.session_id", "sess-123")
+	if _, ok := findAttr(areq, "url.scheme"); ok {
+		t.Error("url.scheme present although the context carried no scheme")
+	}
+	if areq.Name != "weather-service a2a message/send" {
+		t.Errorf("a2a span name = %q", areq.Name)
+	}
+
+	// Inference.
+	exp.Reset()
+	i := fakeContext(pipeline.Outbound, http.Header{})
+	i.Extensions.Inference = &pipeline.InferenceExtension{Model: "qwen2.5:7b"}
+	run(t, p, i, allow(200))
+	ireq, _ := roleSplit(t, exp.GetSpans())
+	checkAttr(t, ireq, "lineage.protocol", "inference")
+	checkAttr(t, ireq, "inference.model", "qwen2.5:7b")
+	if ireq.Name != "weather-service inference qwen2.5:7b" {
+		t.Errorf("inference span name = %q", ireq.Name)
+	}
+}
+
+// mcp-parser attaches to ANY JSON-RPC body — including every a2a exchange —
+// so on an a2a hop both extensions are populated. The payload read is keyed by
+// the protocol fact: when the a2a parser yields nothing (no text parts, a
+// protocol-event artifact), the payload stays ABSENT rather than falling
+// through to the co-populated MCP parse of the same bytes (which would emit
+// the raw JSON-RPC envelope on an lineage.protocol=a2a span).
+func TestCaptureIO_A2ANeverFallsThroughToCoPopulatedMCP(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	p.cfg.CaptureIO = true
+	pctx := fakeContext(pipeline.Outbound, http.Header{})
+	pctx.Extensions.A2A = &pipeline.A2AExtension{
+		Method: "message/send",
+		// A status-update captured as the artifact — a protocol event, filtered.
+		Artifact: `{"kind":"status-update","taskId":"t-1"}`,
+	}
+	pctx.Extensions.MCP = &pipeline.MCPExtension{
+		Method: "message/send",
+		Params: map[string]any{"message": map[string]any{"role": "user"}},
+		Result: map[string]any{"artifacts": []any{map[string]any{"artifactId": "a-1"}}},
+	}
+
+	run(t, p, pctx, allow(200))
+	req, resp := roleSplit(t, exp.GetSpans())
+
+	checkAttr(t, req, "lineage.protocol", "a2a")
+	if v, ok := findAttr(req, "input.value"); ok {
+		t.Errorf("input.value = %q on an a2a hop with no a2a parts — leaked from the co-populated MCP parse", v.Emit())
+	}
+	if v, ok := findAttr(resp, "output.value"); ok {
+		t.Errorf("output.value = %q on an a2a hop whose artifact is a protocol event — leaked from the co-populated MCP parse", v.Emit())
+	}
+	// mcp.* facts belong to mcp hops only; the a2a label must keep them off.
+	if v, ok := findAttr(req, "mcp.method"); ok {
+		t.Errorf("mcp.method = %q emitted on an a2a hop", v.Emit())
+	}
+}
+
+func TestPrincipalFacts_InboundRequestOnly(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	pctx := fakeContext(pipeline.Inbound, http.Header{})
+	pctx.Identity = fakeIdentity{sub: "alice", client: "weather-ui"}
+
+	run(t, p, pctx, allow(200))
+	req, resp := roleSplit(t, exp.GetSpans())
+
+	checkAttr(t, req, "lineage.principal.sub", "alice")
+	checkAttr(t, req, "lineage.principal.client", "weather-ui")
+	// Principal facts are request-only.
+	if _, ok := findAttr(resp, "lineage.principal.sub"); ok {
+		t.Error("lineage.principal.sub leaked onto the response span")
+	}
+}
+
+func TestPrincipalFacts_OutboundNeverEmitsPrincipal(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	pctx := fakeContext(pipeline.Outbound, http.Header{})
+	pctx.Identity = fakeIdentity{sub: "alice", client: "weather-ui"}
+
+	run(t, p, pctx, allow(200))
+	req, _ := roleSplit(t, exp.GetSpans())
+	if _, ok := findAttr(req, "lineage.principal.sub"); ok {
+		t.Error("outbound span carried a principal fact (inbound-only)")
+	}
+}
+
+// ---- the forbidden-keys guard ----
+
+// TestForbiddenKeysNeverEmitted scans every attribute of every span emitted
+// across a spread of exchange shapes and asserts none carries a key from a
+// removed vocabulary. The contract deleted these; this test is the tripwire
+// that keeps them deleted.
+func TestForbiddenKeysNeverEmitted(t *testing.T) {
+	forbidden := []string{"trust.", "lineage.hop.kind", "enduser.id", "openinference.", "source", "authbridge.proxy"}
+
+	shapes := []func() *pipeline.Context{
+		func() *pipeline.Context {
+			c := fakeContext(pipeline.Inbound, http.Header{})
+			c.Identity = fakeIdentity{sub: "alice", client: "weather-ui"}
+			return c
+		},
+		func() *pipeline.Context {
+			c := fakeContext(pipeline.Outbound, http.Header{})
+			c.Extensions.MCP = &pipeline.MCPExtension{Method: "tools/call", Params: map[string]any{"name": "get_weather"}}
+			return c
+		},
+		func() *pipeline.Context {
+			c := fakeContext(pipeline.Outbound, http.Header{})
+			c.Extensions.A2A = &pipeline.A2AExtension{Method: "message/send"}
+			return c
+		},
+		func() *pipeline.Context {
+			c := fakeContext(pipeline.Outbound, http.Header{})
+			c.Extensions.Inference = &pipeline.InferenceExtension{Model: "qwen2.5:7b"}
+			return c
+		},
+	}
+
+	for _, mk := range shapes {
+		p, exp := newTestPlugin(t)
+		p.cfg.CaptureIO = true
+		run(t, p, mk(), allow(200))
+		for _, s := range exp.GetSpans() {
+			for _, kv := range s.Attributes {
+				key := string(kv.Key)
+				for _, bad := range forbidden {
+					if key == bad || strings.HasPrefix(key, bad) {
+						t.Errorf("span %q emitted forbidden attribute %q", s.Name, key)
+					}
+				}
+			}
+		}
+	}
+}
+
+// ---- robustness ----
+
+func TestOnFinish_NoStateDoesNotPanicOrEmit(t *testing.T) {
+	p, exp := newTestPlugin(t)
+	pctx := fakeContext(pipeline.Inbound, http.Header{})
+	// OnFinish without OnRequest having run — no exchangeState stored.
+	p.OnFinish(context.Background(), pctx)
+	if got := len(exp.GetSpans()); got != 0 {
+		t.Errorf("OnFinish with no state emitted %d spans, want 0", got)
+	}
+}
+
+func TestNotReady_SkipsSpan(t *testing.T) {
+	p := NewLineageTelemetry()
+	// Do NOT set ready — Init never called.
+	pctx := fakeContext(pipeline.Inbound, http.Header{})
+	action := p.OnRequest(context.Background(), pctx)
+	if action.Type != pipeline.Continue {
+		t.Fatalf("expected Continue, got %v", action.Type)
+	}
+	if pipeline.GetState[exchangeState](pctx, pluginName) != nil {
+		t.Error("exchangeState should not be set when plugin is not ready")
+	}
+}
+
+// ---- config ----
+
+func TestConfigure_Defaults(t *testing.T) {
+	p := NewLineageTelemetry()
+	if err := p.Configure(nil); err != nil {
+		t.Fatalf("Configure(nil): %v", err)
+	}
+	if p.cfg.OTelEndpoint != "localhost:4317" {
+		t.Errorf("default endpoint = %q, want localhost:4317", p.cfg.OTelEndpoint)
+	}
+	if p.cfg.SelfIDFile != "/shared/client-id.txt" {
+		t.Errorf("default self_id_file = %q", p.cfg.SelfIDFile)
+	}
+}
+
+func TestConfigure_DecodesKeptKeys(t *testing.T) {
+	p := NewLineageTelemetry()
+	raw := json.RawMessage(`{"otel_endpoint":"http://collector:4317","capture_io":true,"self_id":"weather-service"}`)
+	if err := p.Configure(raw); err != nil {
+		t.Fatalf("Configure: %v", err)
+	}
+	if p.cfg.OTelEndpoint != "collector:4317" {
+		t.Errorf("endpoint = %q, want collector:4317 (scheme stripped)", p.cfg.OTelEndpoint)
+	}
+	if !p.cfg.CaptureIO {
+		t.Error("capture_io should be true")
+	}
+	if p.cfg.SelfID != "weather-service" {
+		t.Errorf("self_id = %q", p.cfg.SelfID)
+	}
+}
+
+// ---- helpers ----
+
+type fakeIdentity struct {
+	sub, client string
+	scopes      []string
+}
+
+func (f fakeIdentity) Subject() string  { return f.sub }
+func (f fakeIdentity) ClientID() string { return f.client }
+func (f fakeIdentity) Scopes() []string { return f.scopes }
+
+func findAttr(span tracetest.SpanStub, key string) (attribute.Value, bool) {
+	for _, kv := range span.Attributes {
+		if string(kv.Key) == key {
+			return kv.Value, true
+		}
+	}
+	return attribute.Value{}, false
+}
+
+func attrStr(span tracetest.SpanStub, key string) string {
+	if v, ok := findAttr(span, key); ok {
+		return v.AsString()
+	}
+	return ""
+}
+
+func intAttr(span tracetest.SpanStub, key string) (int64, bool) {
+	if v, ok := findAttr(span, key); ok {
+		return v.AsInt64(), true
+	}
+	return 0, false
+}
+
+// checkAttr asserts a span contains attribute key with the given string value.
+func checkAttr(t *testing.T, span tracetest.SpanStub, key, want string) {
+	t.Helper()
+	got, ok := findAttr(span, key)
+	if !ok {
+		t.Errorf("attribute %q not found in span %q", key, span.Name)
+		return
+	}
+	if got.AsString() != want {
+		t.Errorf("attr %q = %q, want %q", key, got.AsString(), want)
+	}
+}
+
+func headersEqual(a, b http.Header) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for k, av := range a {
+		bv, ok := b[k]
+		if !ok || len(av) != len(bv) {
+			return false
+		}
+		for i := range av {
+			if av[i] != bv[i] {
+				return false
+			}
+		}
+	}
+	return true
+}
+
+// TestInit_RefusesToStartWithoutIdentity locks the v1.3 rule at the identity
+// boundary: a pod whose self identity cannot be resolved must fail at boot,
+// never serve traffic under a plausible-but-wrong label (the old behavior
+// emitted lineage.self.id="agent" from the empty-string serviceLabel).
+func TestInit_RefusesToStartWithoutIdentity(t *testing.T) {
+	cases := []struct {
+		name    string
+		cfg     Config
+		wantErr bool
+	}{
+		{"inline self_id starts", Config{OTelEndpoint: "localhost:4317", SelfID: "weather-service"}, false},
+		{"missing self_id_file refuses", Config{OTelEndpoint: "localhost:4317", SelfIDFile: t.TempDir() + "/absent.txt"}, true},
+		{"no identity source refuses", Config{OTelEndpoint: "localhost:4317"}, true},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			p := NewLineageTelemetry()
+			p.cfg = tc.cfg
+			err := p.Init(context.Background())
+			if p.tp != nil {
+				ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+				_ = p.tp.Shutdown(ctx)
+				cancel()
+			}
+			if tc.wantErr && err == nil {
+				t.Fatal("Init succeeded without a resolvable identity")
+			}
+			if !tc.wantErr && err != nil {
+				t.Fatalf("Init failed with a valid inline self_id: %v", err)
+			}
+			if tc.wantErr && p.Ready() {
+				t.Error("plugin reports Ready after a refused Init")
+			}
+		})
+	}
+}
+
+// TestInit_ReadsSelfIDFile covers the operator-injected path (file, not inline).
+func TestInit_ReadsSelfIDFile(t *testing.T) {
+	dir := t.TempDir()
+	path := dir + "/client-id.txt"
+	if err := os.WriteFile(path, []byte("weather-service\n"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	p := NewLineageTelemetry()
+	p.cfg = Config{OTelEndpoint: "localhost:4317", SelfIDFile: path}
+	if err := p.Init(context.Background()); err != nil {
+		t.Fatalf("Init: %v", err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	_ = p.tp.Shutdown(ctx)
+	cancel()
+	if p.selfID != "weather-service" {
+		t.Errorf("selfID = %q, want trimmed file content", p.selfID)
+	}
+}
+
+// TestConfig_UnknownKeysRefused: a typo'd knob must be a boot error, not a
+// silent run-with-defaults.
+func TestConfig_UnknownKeysRefused(t *testing.T) {
+	if _, err := decodeConfig([]byte(`{"capture-io": true}`)); err == nil {
+		t.Fatal("unknown config key accepted silently")
+	}
+	if _, err := decodeConfig([]byte(`{"capture_io": true, "self_id": "x"}`)); err != nil {
+		t.Fatalf("valid config rejected: %v", err)
+	}
+}
+
+// ---- bypass config ----
+// The one failure mode of bypass_paths / bypass_hosts produces NO signal
+// anywhere: a matched hop is simply absent from the graph. So both directions
+// are pinned — a match emits nothing, a near-miss emits the full pair.
+
+func TestBypassPaths_PrefixMatchEmitsNothing(t *testing.T) {
+	cases := []struct {
+		name  string
+		path  string
+		spans int // spans expected from the exchange
+	}{
+		{"prefix match skipped", "/health/live", 0},
+		{"exact prefix skipped", "/health", 0},
+		{"non-matching path emits", "/api/health-report", 2},
+		{"prefix is anchored, not substring", "/v1/health", 2},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			p, exp := newTestPlugin(t)
+			p.cfg.BypassPaths = []string{"/health"}
+			pctx := fakeContext(pipeline.Inbound, http.Header{})
+			pctx.Path = tc.path
+			run(t, p, pctx, allow(200))
+			if got := len(exp.GetSpans()); got != tc.spans {
+				t.Fatalf("path %q: got %d spans, want %d", tc.path, got, tc.spans)
+			}
+		})
+	}
+}
+
+func TestBypassHosts_SubstringMatchEmitsNothing(t *testing.T) {
+	cases := []struct {
+		name  string
+		host  string
+		spans int
+	}{
+		{"substring match skipped", "otel-collector.rossoctl-system:4317", 0},
+		{"bare name match skipped", "otel-collector:4317", 0},
+		{"unrelated host emits", "weather-tool:8000", 2},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			p, exp := newTestPlugin(t)
+			p.cfg.BypassHosts = []string{"otel-collector"}
+			pctx := fakeContext(pipeline.Outbound, http.Header{})
+			pctx.Host = tc.host
+			run(t, p, pctx, allow(200))
+			if got := len(exp.GetSpans()); got != tc.spans {
+				t.Fatalf("host %q: got %d spans, want %d", tc.host, got, tc.spans)
+			}
+		})
+	}
+}
diff --git a/authbridge/cmd/authbridge-envoy/go.mod b/authbridge/cmd/authbridge-envoy/go.mod
index 066430ba1..823d7f6e8 100644
--- a/authbridge/cmd/authbridge-envoy/go.mod
+++ b/authbridge/cmd/authbridge-envoy/go.mod
@@ -40,6 +40,7 @@ require (
 	github.com/bytedance/gopkg v0.1.3 // indirect
 	github.com/bytedance/sonic v1.15.1 // indirect
 	github.com/bytedance/sonic/loader v0.5.1 // indirect
+	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudwego/base64x v0.1.6 // indirect
 	github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2 // indirect
@@ -54,6 +55,7 @@ require (
 	github.com/gobwas/glob v0.2.3 // indirect
 	github.com/goccy/go-json v0.10.6 // indirect
 	github.com/google/uuid v1.6.0 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect
 	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/huandu/go-clone v1.7.3 // indirect
 	github.com/huandu/go-sqlbuilder v1.41.0 // indirect
@@ -111,9 +113,12 @@ require (
 	github.com/yosida95/uritemplate/v3 v3.0.2 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/otel v1.44.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0 // indirect
 	go.opentelemetry.io/otel/metric v1.44.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.44.0 // indirect
 	go.opentelemetry.io/otel/trace v1.44.0 // indirect
+	go.opentelemetry.io/proto/otlp v1.10.0 // indirect
 	go.starlark.net v0.0.0-20260708150628-5395d018f003 // indirect
 	go.yaml.in/yaml/v2 v2.4.4 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
@@ -124,6 +129,7 @@ require (
 	golang.org/x/sys v0.47.0 // indirect
 	golang.org/x/text v0.40.0 // indirect
 	golang.org/x/time v0.15.0 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect
 	google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af // indirect
 	gopkg.in/ini.v1 v1.67.3 // indirect
diff --git a/authbridge/cmd/authbridge-envoy/plugins_lineage.go b/authbridge/cmd/authbridge-envoy/plugins_lineage.go
new file mode 100644
index 000000000..76ea1e644
--- /dev/null
+++ b/authbridge/cmd/authbridge-envoy/plugins_lineage.go
@@ -0,0 +1,5 @@
+//go:build !exclude_plugin_lineage
+
+package main
+
+import _ "github.com/rossoctl/cortex/authbridge/authlib/plugins/lineage"
diff --git a/authbridge/cmd/authbridge-proxy/go.mod b/authbridge/cmd/authbridge-proxy/go.mod
index 1d0a6f8f4..4a665a863 100644
--- a/authbridge/cmd/authbridge-proxy/go.mod
+++ b/authbridge/cmd/authbridge-proxy/go.mod
@@ -33,6 +33,7 @@ require (
 	github.com/bytedance/gopkg v0.1.3 // indirect
 	github.com/bytedance/sonic v1.15.1 // indirect
 	github.com/bytedance/sonic/loader v0.5.1 // indirect
+	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudwego/base64x v0.1.6 // indirect
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.1 // indirect
@@ -45,6 +46,7 @@ require (
 	github.com/gobwas/glob v0.2.3 // indirect
 	github.com/goccy/go-json v0.10.6 // indirect
 	github.com/google/uuid v1.6.0 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect
 	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/huandu/go-clone v1.7.3 // indirect
 	github.com/huandu/go-sqlbuilder v1.41.0 // indirect
@@ -101,9 +103,12 @@ require (
 	github.com/yosida95/uritemplate/v3 v3.0.2 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/otel v1.44.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0 // indirect
 	go.opentelemetry.io/otel/metric v1.44.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.44.0 // indirect
 	go.opentelemetry.io/otel/trace v1.44.0 // indirect
+	go.opentelemetry.io/proto/otlp v1.10.0 // indirect
 	go.starlark.net v0.0.0-20260708150628-5395d018f003 // indirect
 	go.yaml.in/yaml/v2 v2.4.4 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
@@ -114,6 +119,7 @@ require (
 	golang.org/x/sys v0.47.0 // indirect
 	golang.org/x/text v0.40.0 // indirect
 	golang.org/x/time v0.15.0 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect
 	google.golang.org/grpc v1.82.0 // indirect
 	google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af // indirect
diff --git a/authbridge/cmd/authbridge-proxy/plugins_lineage.go b/authbridge/cmd/authbridge-proxy/plugins_lineage.go
new file mode 100644
index 000000000..76ea1e644
--- /dev/null
+++ b/authbridge/cmd/authbridge-proxy/plugins_lineage.go
@@ -0,0 +1,5 @@
+//go:build !exclude_plugin_lineage
+
+package main
+
+import _ "github.com/rossoctl/cortex/authbridge/authlib/plugins/lineage"