diff --git a/pkg/chipingress/client.go b/pkg/chipingress/client.go
index f51ce93bb7..33aadeba7c 100644
--- a/pkg/chipingress/client.go
+++ b/pkg/chipingress/client.go
@@ -5,13 +5,17 @@ import (
 	"crypto/tls"
 	"fmt"
 	"net"
-	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/google/uuid"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/trace"
+	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
@@ -215,11 +219,51 @@ func WithHeaderProvider(provider HeaderProvider) Opt {
 	return func(c *clientConfig) { c.headerProvider = provider }
 }
 
-// WithResourceAttributeHeaders returns an Opt that attaches the provided resource attributes
-// as sanitized gRPC metadata headers. It combines SanitizeMetadataHeaders with
-// NewStaticHeaderProvider so the safe, validated path is used by default.
+// resourceAttributeLogger receives one warning per dropped resource attribute. It defaults to a
+// no-op so library consumers who don't configure a logger see no output; tests substitute their own
+// via setResourceAttributeLogger.
+var resourceAttributeLogger atomic.Pointer[zap.Logger]
+
+func init() {
+	resourceAttributeLogger.Store(zap.NewNop())
+}
+
+// SetResourceAttributeLogger overrides the logger WithResourceAttributeHeaders warns through when it
+// drops a resource attribute. Intended for host applications that want dropped attributes surfaced
+// in their own logs; the default is a no-op logger.
+func SetResourceAttributeLogger(logger *zap.Logger) {
+	resourceAttributeLogger.Store(logger)
+}
+
+// resourceAttributeDropsCounter counts resource attributes SanitizeMetadataHeaders omitted, by
+// reason, against the global otel MeterProvider. It is created lazily against the global provider
+// (rather than a per-client one from WithMeterProvider) because WithResourceAttributeHeaders runs at
+// Opt-construction time, before any client-level configuration is wired.
+var resourceAttributeDropsCounter = sync.OnceValue(func() metric.Int64Counter {
+	c, _ := otel.Meter("github.com/smartcontractkit/chainlink-common/pkg/chipingress").
+		Int64Counter("chipingress.resource_attribute.dropped",
+			metric.WithDescription("Resource attributes omitted by SanitizeMetadataHeaders, by reason."))
+	return c
+})
+
+// WithResourceAttributeHeaders returns an Opt that attaches the provided resource attributes as
+// gRPC metadata on every request, under ResourceHeaderPrefix. It combines SanitizeMetadataHeaders
+// with NewStaticHeaderProvider so the safe, validated path is used by default, and warns + meters
+// every attribute SanitizeMetadataHeaders had to omit.
+//
+// Attributes are attached once per request rather than to individual events because they describe the
+// producer, not any one event. Chip-ingress fans them out onto every Kafka record the request
+// produces.
 func WithResourceAttributeHeaders(attrs map[string]string) Opt {
-	return WithHeaderProvider(NewStaticHeaderProvider(SanitizeMetadataHeaders(attrs)))
+	sanitized, dropped := SanitizeMetadataHeaders(attrs)
+	for _, d := range dropped {
+		resourceAttributeLogger.Load().Warn("dropping invalid resource attribute",
+			zap.String("key", d.Key), zap.String("reason", d.Reason))
+		if counter := resourceAttributeDropsCounter(); counter != nil {
+			counter.Add(context.Background(), 1, metric.WithAttributes(attribute.String("reason", d.Reason)))
+		}
+	}
+	return WithHeaderProvider(NewStaticHeaderProvider(sanitized))
 }
 
 // WithInsecureConnection configures the client to use an insecure connection (no TLS).
@@ -254,11 +298,15 @@ func WithTracerProvider(provider trace.TracerProvider) Opt {
 	return func(c *clientConfig) { c.tracerProvider = provider }
 }
 
+// nopInfoHeaderKey is the metadata key WithNOPLookup sets, asking chip-ingress to look up NOP info
+// for the authenticated CSA key.
+const nopInfoHeaderKey = "x-include-nop-info"
+
 func WithNOPLookup() Opt {
 	return func(c *clientConfig) {
 		c.nopInfoHeaderProvider = headerProviderFunc(func(ctx context.Context) (map[string]string, error) {
 			return map[string]string{
-				"x-include-nop-info": "true",
+				nopInfoHeaderKey: "true",
 			}, nil
 		})
 	}
@@ -283,42 +331,12 @@ func newHeaderInterceptor(provider HeaderProvider) grpc.UnaryClientInterceptor {
 	}
 }
 
-// EventOpt configures a CloudEvent after its well-known attributes have been set by NewEvent.
-type EventOpt func(*ce.Event)
-
-// sanitizeExtensionName lower-cases name and strips every rune outside [a-z0-9], the character
-// set the CloudEvents spec requires for extension attribute names.
-func sanitizeExtensionName(name string) string {
-	var b strings.Builder
-	for _, r := range strings.ToLower(name) {
-		if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') {
-			b.WriteRune(r)
-		}
-	}
-	return b.String()
-}
-
-// WithResourceAttributeExtensions returns an EventOpt that sets a CloudEvent extension for each
-// entry in attrs, sanitizing keys via sanitizeExtensionName so they satisfy the CloudEvents
-// extension-name character set. Entries that sanitize to an empty string, or that collide with a
-// reserved extension name (see reservedExtensionNames), are skipped. Keys are applied in sorted
-// order so that if two distinct keys sanitize to the same name, the result is deterministic.
-func WithResourceAttributeExtensions(attrs map[string]string) EventOpt {
-	return func(event *ce.Event) {
-		for _, pair := range sanitizeResourceAttributeKeys(attrs, nil) {
-			event.SetExtension(pair.name, attrs[pair.key])
-		}
-	}
-}
-
 // NewEvent creates a new CloudEvent with the specified domain, entity, payload, and optional attributes.
+//
+// Resource attributes are deliberately not stamped here. They describe the producer rather than any
+// individual event, so they travel once per request as gRPC metadata (see
+// WithResourceAttributeHeaders) instead of being repeated on every event in a batch.
 func NewEvent(domain, entity string, payload []byte, attributes map[string]any) (CloudEvent, error) {
-	return NewEventWithOpts(domain, entity, payload, attributes)
-}
-
-// NewEventWithOpts creates a new CloudEvent like NewEvent, additionally applying opts (e.g.
-// WithResourceAttributeExtensions) to the event before its data is set.
-func NewEventWithOpts(domain, entity string, payload []byte, attributes map[string]any, opts ...EventOpt) (CloudEvent, error) {
 	event := ce.NewEvent()
 	event.SetSource(domain)
 	event.SetType(entity)
@@ -352,10 +370,6 @@ func NewEventWithOpts(domain, entity string, payload []byte, attributes map[stri
 		event.SetExtension(IdempotencyKeyAttr, val)
 	}
 
-	for _, opt := range opts {
-		opt(&event)
-	}
-
 	err := event.SetData(ceformat.ContentTypeProtobuf, payload)
 	if err != nil {
 		return ce.Event{}, fmt.Errorf("could not set data on event: %w", err)
diff --git a/pkg/chipingress/client_test.go b/pkg/chipingress/client_test.go
index c224e1bb42..16cd5e9657 100644
--- a/pkg/chipingress/client_test.go
+++ b/pkg/chipingress/client_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"strings"
 	"testing"
 	"time"
 
@@ -168,89 +169,6 @@ func TestNewEvent_IdempotencyKey(t *testing.T) {
 	})
 }
 
-func Test_sanitizeExtensionName(t *testing.T) {
-	tests := []struct {
-		name string
-		in   string
-		want string
-	}{
-		{name: "snake_case", in: "chain_id", want: "chainid"},
-		{name: "dotted", in: "k8s.pod.name", want: "k8spodname"},
-		{name: "already valid", in: "chainid", want: "chainid"},
-		{name: "upper case is lowered", in: "ChainID", want: "chainid"},
-		{name: "empty", in: "", want: ""},
-		{name: "all invalid characters", in: "---...", want: ""},
-		{name: "mixed valid and invalid", in: "Service-Name.1", want: "servicename1"},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			assert.Equal(t, tt.want, sanitizeExtensionName(tt.in))
-		})
-	}
-}
-
-func TestNewEventWithOpts_WithResourceAttributeExtensions(t *testing.T) {
-	payload := []byte("body")
-
-	t.Run("sanitized keys/values land on the event", func(t *testing.T) {
-		attrs := map[string]string{"chain_id": "1", "k8s.pod.name": "pod-abc"}
-		event, err := NewEventWithOpts("domain", "entity", payload, nil, WithResourceAttributeExtensions(attrs))
-		require.NoError(t, err)
-		ext := event.Extensions()
-		assert.Equal(t, "1", ext["chainid"])
-		assert.Equal(t, "pod-abc", ext["k8spodname"])
-	})
-
-	t.Run("empty sanitized name is dropped", func(t *testing.T) {
-		attrs := map[string]string{"---": "value"}
-		event, err := NewEventWithOpts("domain", "entity", payload, nil, WithResourceAttributeExtensions(attrs))
-		require.NoError(t, err)
-		assert.Len(t, event.Extensions(), 1) // only the always-set recordedtime extension
-	})
-
-	t.Run("reserved name is skipped", func(t *testing.T) {
-		attrs := map[string]string{IdempotencyKeyAttr: "should-not-override", "subject": "should-not-override"}
-		event, err := NewEventWithOpts("domain", "entity", payload, map[string]any{IdempotencyKeyAttr: "real-key"}, WithResourceAttributeExtensions(attrs))
-		require.NoError(t, err)
-		ext := event.Extensions()
-		assert.Equal(t, "real-key", ext[IdempotencyKeyAttr])
-		assert.Empty(t, event.Subject())
-	})
-
-	t.Run("duplicate sanitized names resolve deterministically to sorted-first key", func(t *testing.T) {
-		attrs := map[string]string{"service.name": "from-dotted", "service_name": "from-snake"}
-		event, err := NewEventWithOpts("domain", "entity", payload, nil, WithResourceAttributeExtensions(attrs))
-		require.NoError(t, err)
-		// sorted order: "service.name" < "service_name" ('.' < '_' in ASCII), so the dotted key wins.
-		assert.Equal(t, "from-dotted", event.Extensions()["servicename"])
-	})
-
-	t.Run("omitting all opts is a no-op", func(t *testing.T) {
-		event, err := NewEventWithOpts("domain", "entity", payload, nil)
-		require.NoError(t, err)
-		assert.Len(t, event.Extensions(), 1) // only the always-set recordedtime extension
-	})
-}
-
-// TestNewEvent_UnchangedSignature is a backward-compatibility guard: NewEvent's exported
-// signature must stay exactly as it was before EventOpt/NewEventWithOpts were introduced, and
-// must remain equivalent to calling NewEventWithOpts with no opts.
-func TestNewEvent_UnchangedSignature(t *testing.T) {
-	payload := []byte("body")
-	attributes := map[string]any{"subject": "example-subject"}
-
-	viaNewEvent, err := NewEvent("domain", "entity", payload, attributes)
-	require.NoError(t, err)
-
-	viaNewEventWithOpts, err := NewEventWithOpts("domain", "entity", payload, attributes)
-	require.NoError(t, err)
-
-	assert.Equal(t, viaNewEventWithOpts.Subject(), viaNewEvent.Subject())
-	assert.Equal(t, viaNewEventWithOpts.Extensions()["recordedtime"].(ce.Timestamp).Truncate(time.Second),
-		viaNewEvent.Extensions()["recordedtime"].(ce.Timestamp).Truncate(time.Second))
-	assert.Equal(t, viaNewEventWithOpts.Data(), viaNewEvent.Data())
-}
-
 func TestEventToProto(t *testing.T) {
 	// Create a test protobuf message
 	testProto := pb.PingResponse{Message: "test message"}
@@ -684,14 +602,21 @@ func TestOptions(t *testing.T) {
 	t.Run("WithResourceAttributeHeaders", func(t *testing.T) {
 		config := defaultCfg
 		WithResourceAttributeHeaders(map[string]string{
-			"Chain-ID": "1",
-			"id":       "skipped", // reserved extension name
-			"chain_id": "2",       // duplicate sanitized key, first wins
+			"Chain-ID":       "1",   // lower-cased, separator preserved
+			"csa_public_key": "abc", // preserved verbatim
+			// Namespaced rather than dropped: prefixing puts them out of reach of the real keys.
+			"te":          "harmless",
+			authHeaderKey: "harmless",
 		})(&config)
 		assert.NotNil(t, config.headerProvider)
 		headers, err := config.headerProvider.Headers(t.Context())
 		require.NoError(t, err)
-		assert.Equal(t, map[string]string{"chainid": "1"}, headers)
+		assert.Equal(t, map[string]string{
+			ResourceHeaderPrefix + "chain-id":                     "1",
+			ResourceHeaderPrefix + "csa_public_key":               "abc",
+			ResourceHeaderPrefix + "te":                           "harmless",
+			ResourceHeaderPrefix + strings.ToLower(authHeaderKey): "harmless",
+		}, headers)
 	})
 
 	t.Run("WithBasicAuth", func(t *testing.T) {
@@ -808,6 +733,56 @@ func TestClient_ChainedHeaderProviders(t *testing.T) {
 	assert.Equal(t, []string{"true"}, capture.lastMD.Get("x-include-nop-info"))
 }
 
+// TestClient_AuthHeaderCoexistsWithResourceAttributes pins down the property the resource-attribute
+// work must never break: the CSA node auth token and the resource-attribute headers travel by two
+// different mechanisms — per-RPC credentials (WithTokenAuth) and a unary interceptor
+// (WithResourceAttributeHeaders) — and both must arrive intact, exactly once, on the same request.
+//
+// It also pins the property that lets the client carry attributes without a reserved-key deny-list:
+// an attribute named after the auth header is namespaced under ResourceHeaderPrefix, so it cannot
+// append a second value under the auth header's own key.
+func TestClient_AuthHeaderCoexistsWithResourceAttributes(t *testing.T) {
+	lis, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	defer lis.Close()
+
+	srv := gp.NewServer()
+	capture := &capturingServer{}
+	pb.RegisterChipIngressServer(srv, capture)
+	go func() { _ = srv.Serve(lis) }()
+	defer srv.Stop()
+
+	const authToken = "2:deadbeef:1:cafe"
+
+	client, err := NewClient(lis.Addr().String(),
+		WithInsecureConnection(),
+		WithTokenAuth(&mockHeaderProvider{headers: map[string]string{authHeaderKey: authToken}}),
+		WithResourceAttributeHeaders(map[string]string{
+			"csa_public_key": "abc123",
+			"service.name":   "chainlink",
+			// Namespaced away from the auth key rather than appended to it.
+			authHeaderKey: "forged",
+		}),
+		WithNOPLookup(),
+	)
+	require.NoError(t, err)
+	defer client.Close() //nolint:errcheck
+
+	_, err = client.Ping(t.Context(), &EmptyRequest{})
+	require.NoError(t, err)
+
+	require.NotNil(t, capture.lastMD)
+	// grpc lower-cases metadata keys on the wire.
+	assert.Equal(t, []string{authToken}, capture.lastMD.Get(authHeaderKey),
+		"the auth token must arrive exactly once, unmodified")
+	assert.Equal(t, []string{"abc123"}, capture.lastMD.Get(ResourceHeaderPrefix+"csa_public_key"))
+	assert.Equal(t, []string{"chainlink"}, capture.lastMD.Get(ResourceHeaderPrefix+"service.name"))
+	assert.Equal(t, []string{"true"}, capture.lastMD.Get("x-include-nop-info"))
+	// The forged attribute landed in the resource namespace, harmlessly.
+	assert.Equal(t, []string{"forged"},
+		capture.lastMD.Get(ResourceHeaderPrefix+strings.ToLower(authHeaderKey)))
+}
+
 func TestWithTLS(t *testing.T) {
 	serverName := "example.com"
 	config := defaultCfg
diff --git a/pkg/chipingress/header_provider.go b/pkg/chipingress/header_provider.go
index 47e2dfcd9b..f69f721d3d 100644
--- a/pkg/chipingress/header_provider.go
+++ b/pkg/chipingress/header_provider.go
@@ -8,6 +8,8 @@ import (
 	"errors"
 	"fmt"
 	"maps"
+	"sort"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -111,50 +113,154 @@ func newStaticHeaderProvider(headers map[string]string, requireTLS bool) HeaderP
 // NewStaticHeaderProvider returns a HeaderProvider that always returns the given headers,
 // for use with WithHeaderProvider to attach fixed, non-auth gRPC metadata (e.g. resource
 // attributes) to every request.
+//
+// This is for the non-auth interceptor path only. It reports RequireTransportSecurity() == false,
+// which WithHeaderProvider never consults — the HeaderProvider interface declares only Headers,
+// and grpc asks only credentials.PerRPCCredentials about transport security. Do not pass the
+// result to WithTokenAuth: that path takes its TLS requirement from the client config
+// (!c.insecureConnection), not from the provider, so the false here would be silently ignored
+// rather than honoured. Use NewHeaderProvider for auth headers.
 func NewStaticHeaderProvider(headers map[string]string) HeaderProvider {
 	return newStaticHeaderProvider(headers, false)
 }
 
-// SanitizeMetadataValue replaces any byte outside the printable ASCII range [0x20-0x7E]
-// with '?'. grpc-go hard-fails the entire RPC when an outgoing metadata value fails this
-// check (unlike the CE-extension path, where an invalid entry is simply dropped), so
-// values headed for gRPC metadata must be normalized before being sent.
-func SanitizeMetadataValue(val string) string {
-	b := []byte(val)
-	out := make([]byte, len(b))
-	for i, c := range b {
-		if c >= 0x20 && c <= 0x7E {
-			out[i] = c
-		} else {
-			out[i] = '?'
+// Limits on resource attributes accepted by SanitizeMetadataHeaders. They reserve headroom in the
+// gRPC HEADERS frame for authentication and normal gRPC metadata, and bound how much of every Kafka
+// record's header space a producer's resource attributes can consume.
+const (
+	maxResourceAttributes          = 32
+	maxResourceAttributeKeyBytes   = 128
+	maxResourceAttributeValueBytes = 512
+	maxResourceAttributeTotalBytes = 4096 // sum of accepted key + value bytes, prefix excluded
+)
+
+// isPrintableASCII reports whether every byte of val is in the printable ASCII range [0x20, 0x7E].
+// grpc-go hard-fails the entire RPC — auth header included — when an outgoing metadata value fails
+// this check, so a value that does not pass is omitted rather than rewritten: a byte-mangled value
+// is a worse outcome than a dropped attribute for an operator-facing routing/observability field.
+func isPrintableASCII(val string) bool {
+	for i := 0; i < len(val); i++ {
+		if c := val[i]; c < 0x20 || c > 0x7E {
+			return false
+		}
+	}
+	return true
+}
+
+// DroppedAttribute records a resource attribute SanitizeMetadataHeaders omitted, and why.
+type DroppedAttribute struct {
+	Key    string
+	Reason string
+}
+
+// Reasons a resource attribute can be omitted by SanitizeMetadataHeaders. Exposed as strings (not
+// an enum type) so callers can attach them to a log field or a metric attribute directly.
+const (
+	reasonInvalidKey    = "invalid_key"
+	reasonInvalidValue  = "invalid_value"
+	reasonDuplicateKey  = "duplicate_key"
+	reasonLimitExceeded = "limit_exceeded"
+)
+
+// isValidMetadataKeyChar reports whether r is allowed in an outgoing gRPC metadata key. grpc-go
+// accepts [0-9a-z-_.] (see internal/metadata.ValidateKey); upper-case is handled by lower-casing
+// before this is called.
+func isValidMetadataKeyChar(r rune) bool {
+	return (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') || r == '.' || r == '-' || r == '_'
+}
+
+// sanitizeMetadataKey validates a resource-attribute key as a valid outgoing gRPC metadata key,
+// without the ResourceHeaderPrefix that SanitizeMetadataHeaders adds, and reports whether it is
+// valid. It never rewrites: a key that fails validation is omitted by the caller rather than
+// mutated, so two distinct configured keys can never collapse into one gRPC metadata key.
+//
+// Valid keys, once lower-cased, match [0-9a-z-_.]+ (grpc's own key charset — see
+// internal/metadata.ValidateKey) and do not end in "-bin", which grpc treats as declaring a
+// base64-encoded binary value. A valid key's structure survives untouched: "csa_public_key" stays
+// "csa_public_key" and "service.name" stays "service.name", which is what lets chip-ingress emit the
+// forwarded header verbatim.
+func sanitizeMetadataKey(key string) (string, bool) {
+	if key == "" {
+		return "", false
+	}
+	lower := strings.ToLower(key)
+	for _, r := range lower {
+		if !isValidMetadataKeyChar(r) {
+			return "", false
 		}
 	}
-	return string(out)
+	if strings.HasSuffix(lower, "-bin") {
+		return "", false
+	}
+	return lower, true
 }
 
-// SanitizeMetadataHeaders sanitizes a map of resource-attribute headers for use as outgoing
-// gRPC metadata (e.g. via NewStaticHeaderProvider). Keys are sanitized with
-// sanitizeExtensionName — the same strict [a-z0-9] charset used for CloudEvent extensions —
-// which is a subset of grpc's allowed metadata-key charset, so a sanitized key can never trip
-// grpc's key validation or the reserved "-bin" suffix, and produces the same key stem as the
-// corresponding CE extension (differing only by the CloudEvents Kafka binding's "ce_" prefix
-// once on the wire). Values are sanitized via SanitizeMetadataValue, since grpc-go fails the
-// whole RPC on a non-printable value. Entries that sanitize to an empty key, or that collide
-// with a reserved extension name (see reservedExtensionNames) or a gRPC-reserved header name
-// (see reservedMetadataKeys), are skipped. Keys are applied in sorted order so duplicate
-// sanitized keys resolve deterministically (first in sorted order wins), matching
-// WithResourceAttributeExtensions' collision handling.
+// SanitizeMetadataHeaders validates a map of resource attributes for use as outgoing gRPC metadata
+// (e.g. via NewStaticHeaderProvider). Every emitted key is ResourceHeaderPrefix followed by the
+// validated key, unchanged, so service.name becomes resource_service.name and csa_public_key becomes
+// resource_csa_public_key. Chip-ingress forwards keys carrying that prefix onto every Kafka record a
+// request produces, emitting the key unchanged.
 //
-// Note: unlike the CloudEvents Kafka binding, gRPC metadata keys are NOT prefixed with "ce_" —
-// that prefix is a CloudEvents-binding concept, not a metadata one, and reusing it here would
-// collide with the CE binding's own "ce_<name>" Kafka header if the server ever forwards gRPC
-// metadata verbatim onto Kafka.
-func SanitizeMetadataHeaders(in map[string]string) map[string]string {
+// The prefix is what makes this safe without a deny-list. The header interceptor appends to outgoing
+// metadata rather than replacing it, so an attribute landing on an existing header name would send
+// two values under one key — an attribute named X-Beholder-Node-Auth-Token would have broken
+// authentication that way. Because every emitted key is prefixed, no attribute can reach a reserved
+// gRPC key: that one becomes resource_x-beholder-node-auth-token, which collides with nothing, and
+// the same holds for authorization, te, content-type, the grpc- prefix and pseudo-headers.
+//
+// An attribute is omitted, rather than rewritten, when: its key is empty, exceeds
+// maxResourceAttributeKeyBytes, fails sanitizeMetadataKey's charset/[-bin] validation, or duplicates
+// an already-accepted key (first in sorted order of the original keys wins); its value exceeds
+// maxResourceAttributeValueBytes or is not printable ASCII (isPrintableASCII); or accepting it would
+// push the accepted count past maxResourceAttributes or the accepted key+value byte total past
+// maxResourceAttributeTotalBytes. Keys are processed in sorted order so every omission is
+// deterministic. dropped records each omission and why, for the caller to warn and meter.
+func SanitizeMetadataHeaders(in map[string]string) (map[string]string, []DroppedAttribute) {
+	keys := make([]string, 0, len(in))
+	for k := range in {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys) // deterministic: first in sorted order wins, and excess entries drop from the tail
+
 	out := make(map[string]string, len(in))
-	for _, pair := range sanitizeResourceAttributeKeys(in, reservedMetadataKeys) {
-		out[pair.name] = SanitizeMetadataValue(in[pair.key])
+	var dropped []DroppedAttribute
+	totalBytes := 0
+	for _, k := range keys {
+		if len(out) >= maxResourceAttributes {
+			dropped = append(dropped, DroppedAttribute{Key: k, Reason: reasonLimitExceeded})
+			continue
+		}
+		if len(k) > maxResourceAttributeKeyBytes {
+			dropped = append(dropped, DroppedAttribute{Key: k, Reason: reasonInvalidKey})
+			continue
+		}
+		name, ok := sanitizeMetadataKey(k)
+		if !ok {
+			dropped = append(dropped, DroppedAttribute{Key: k, Reason: reasonInvalidKey})
+			continue
+		}
+		name = ResourceHeaderPrefix + name
+		if _, dup := out[name]; dup {
+			dropped = append(dropped, DroppedAttribute{Key: k, Reason: reasonDuplicateKey})
+			continue
+		}
+		val := in[k]
+		if len(val) > maxResourceAttributeValueBytes {
+			dropped = append(dropped, DroppedAttribute{Key: k, Reason: reasonInvalidValue})
+			continue
+		}
+		if !isPrintableASCII(val) {
+			dropped = append(dropped, DroppedAttribute{Key: k, Reason: reasonInvalidValue})
+			continue
+		}
+		if totalBytes+len(name)-len(ResourceHeaderPrefix)+len(val) > maxResourceAttributeTotalBytes {
+			dropped = append(dropped, DroppedAttribute{Key: k, Reason: reasonLimitExceeded})
+			continue
+		}
+		totalBytes += len(name) - len(ResourceHeaderPrefix) + len(val)
+		out[name] = val
 	}
-	return out
+	return out, dropped
 }
 
 // newRotatingHeaderProvider returns a HeaderProvider that refreshes its
diff --git a/pkg/chipingress/header_provider_test.go b/pkg/chipingress/header_provider_test.go
index 8069420fd2..ea705d7b75 100644
--- a/pkg/chipingress/header_provider_test.go
+++ b/pkg/chipingress/header_provider_test.go
@@ -4,7 +4,9 @@ import (
 	"context"
 	"crypto/ed25519"
 	"encoding/hex"
+	"fmt"
 	"net"
+	"strings"
 	"testing"
 	"time"
 
@@ -283,63 +285,161 @@ func TestNewStaticHeaderProvider(t *testing.T) {
 	assert.False(t, tlsReq.RequireTransportSecurity())
 }
 
-func TestSanitizeMetadataValue(t *testing.T) {
-	tests := []struct {
-		name string
-		in   string
-		want string
-	}{
-		{name: "printable ASCII is unchanged", in: "chain-1_prod.v2", want: "chain-1_prod.v2"},
-		{name: "empty", in: "", want: ""},
-		{name: "control character replaced", in: "value\nwith\tcontrol", want: "value?with?control"},
-		{name: "non-ASCII UTF-8 replaced byte-wise", in: "café", want: "caf??"},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			assert.Equal(t, tt.want, chipingress.SanitizeMetadataValue(tt.in))
-		})
-	}
-}
+const rp = chipingress.ResourceHeaderPrefix
 
 func TestSanitizeMetadataHeaders(t *testing.T) {
-	t.Run("standard OTel-style keys are sanitized to the same stem as CE extensions", func(t *testing.T) {
-		in := map[string]string{
-			"service.name":  "beholder",
-			"chain_id":      "1",
-			"node-operator": "acme",
-		}
-		got := chipingress.SanitizeMetadataHeaders(in)
+	t.Run("valid keys are prefixed and kept verbatim", func(t *testing.T) {
+		got, dropped := chipingress.SanitizeMetadataHeaders(map[string]string{
+			"service.name":   "beholder",
+			"csa_public_key": "abc123",
+			"node-operator":  "acme",
+			"donid":          "don-1",
+		})
+		assert.Empty(t, dropped)
 		assert.Equal(t, map[string]string{
-			"servicename":  "beholder",
-			"chainid":      "1",
-			"nodeoperator": "acme",
+			rp + "service.name":   "beholder",
+			rp + "csa_public_key": "abc123",
+			rp + "node-operator":  "acme",
+			rp + "donid":          "don-1",
 		}, got)
 	})
 
-	t.Run("empty-after-sanitize keys are dropped", func(t *testing.T) {
-		got := chipingress.SanitizeMetadataHeaders(map[string]string{"---": "value"})
+	t.Run("validate, don't rewrite", func(t *testing.T) {
+		tests := []struct {
+			name    string
+			in      string
+			want    string // "" if the key must be omitted
+			omitted bool
+		}{
+			// grpc accepts [0-9a-z-_.], so a valid key's structure survives untouched and
+			// chip-ingress can emit the forwarded header verbatim.
+			{name: "snake case preserved", in: "csa_public_key", want: rp + "csa_public_key"},
+			{name: "dotted preserved", in: "service.name", want: rp + "service.name"},
+			{name: "upper-cased is lowered", in: "DonID", want: rp + "donid"},
+			{name: "mixed separators preserved", in: "k8s.pod-name_1", want: rp + "k8s.pod-name_1"},
+			// Invalid keys are OMITTED, never rewritten: silently collapsing two distinct
+			// configured keys into one gRPC metadata key is worse than dropping one.
+			{name: "illegal characters omit the attribute", in: "chain id/2:x", omitted: true},
+			{name: "non-ascii omits the attribute", in: "héllo", omitted: true},
+			{name: "empty key omits the attribute", in: "", omitted: true},
+			// A "-bin" suffix tells grpc the value is base64-encoded binary; omit rather than
+			// rewrite so a plain-text resource attribute can't silently start being decoded.
+			{name: "bin suffix omits the attribute", in: "payload-bin", omitted: true},
+			{name: "bin substring is untouched", in: "payload-binary", want: rp + "payload-binary"},
+		}
+		for _, tt := range tests {
+			t.Run(tt.name, func(t *testing.T) {
+				got, dropped := chipingress.SanitizeMetadataHeaders(map[string]string{tt.in: "v"})
+				if tt.omitted {
+					assert.Empty(t, got)
+					require.Len(t, dropped, 1)
+					assert.Equal(t, tt.in, dropped[0].Key)
+					return
+				}
+				assert.Empty(t, dropped)
+				assert.Equal(t, map[string]string{tt.want: "v"}, got)
+			})
+		}
+	})
+
+	// This is the property that replaces the reserved-key set the prefix made redundant. The header
+	// interceptor appends to outgoing metadata rather than replacing, so an attribute landing on an
+	// existing header name would send two values under one key — for the CSA auth token that breaks
+	// authentication. Prefixing puts every valid attribute out of reach of every reserved gRPC key.
+	t.Run("no attribute can collide with a reserved gRPC metadata key", func(t *testing.T) {
+		for _, key := range []string{
+			"x-include-nop-info", // WithNOPLookup
+			"authorization",      // WithBasicAuth
+			"te", "content-type", "cookie", "host", "user-agent",
+			"grpc-timeout", "grpc-encoding",
+		} {
+			got, dropped := chipingress.SanitizeMetadataHeaders(map[string]string{key: "forged"})
+			assert.Empty(t, dropped, "key %q should be valid and namespaced, not dropped", key)
+			require.Len(t, got, 1, "key %q should still be sent, just namespaced", key)
+			for name := range got {
+				assert.True(t, strings.HasPrefix(name, rp), "key %q must be prefixed, got %q", key, name)
+				assert.NotEqual(t, strings.ToLower(key), name, "key %q must not reach the reserved name", key)
+			}
+		}
+		// The CSA auth token header contains uppercase letters and hyphens; hyphens are a valid
+		// gRPC metadata char, so this key is namespaced rather than omitted, same as the others.
+		got, dropped := chipingress.SanitizeMetadataHeaders(map[string]string{"X-Beholder-Node-Auth-Token": "forged"})
+		assert.Empty(t, dropped)
+		assert.Equal(t, map[string]string{rp + "x-beholder-node-auth-token": "forged"}, got)
+	})
+
+	t.Run("CloudEvents context attribute names are kept, they mean nothing as gRPC metadata", func(t *testing.T) {
+		got, dropped := chipingress.SanitizeMetadataHeaders(map[string]string{"subject": "keep-me", "source": "keep-me-too"})
+		assert.Empty(t, dropped)
+		assert.Equal(t, map[string]string{rp + "subject": "keep-me", rp + "source": "keep-me-too"}, got)
+	})
+
+	t.Run("non-printable values omit the whole attribute", func(t *testing.T) {
+		got, dropped := chipingress.SanitizeMetadataHeaders(map[string]string{"chain_id": "1\n2"})
 		assert.Empty(t, got)
+		require.Len(t, dropped, 1)
+		assert.Equal(t, "chain_id", dropped[0].Key)
+		assert.Equal(t, "invalid_value", dropped[0].Reason)
+	})
+
+	t.Run("duplicate keys resolve deterministically to sorted-first key", func(t *testing.T) {
+		// "DonID" and "donid" both validate to "donid"; sorted order of the ORIGINAL keys is
+		// "DonID" < "donid" (upper-case sorts first in ASCII), so "DonID" wins.
+		got, dropped := chipingress.SanitizeMetadataHeaders(map[string]string{"DonID": "upper", "donid": "lower"})
+		assert.Equal(t, map[string]string{rp + "donid": "upper"}, got)
+		require.Len(t, dropped, 1)
+		assert.Equal(t, "donid", dropped[0].Key)
+		assert.Equal(t, "duplicate_key", dropped[0].Reason)
 	})
 
-	t.Run("reserved names are dropped", func(t *testing.T) {
-		got := chipingress.SanitizeMetadataHeaders(map[string]string{chipingress.IdempotencyKeyAttr: "should-not-appear", "subject": "should-not-appear"})
+	t.Run("oversized key is omitted, not truncated", func(t *testing.T) {
+		longKey := strings.Repeat("a", 129)
+		got, dropped := chipingress.SanitizeMetadataHeaders(map[string]string{longKey: "v"})
 		assert.Empty(t, got)
+		require.Len(t, dropped, 1)
+		assert.Equal(t, "invalid_key", dropped[0].Reason)
 	})
 
-	t.Run("gRPC-reserved header 'te' is dropped", func(t *testing.T) {
-		got := chipingress.SanitizeMetadataHeaders(map[string]string{"te": "trailers"})
+	t.Run("oversized value is omitted, not truncated", func(t *testing.T) {
+		longVal := strings.Repeat("v", 513)
+		got, dropped := chipingress.SanitizeMetadataHeaders(map[string]string{"chain_id": longVal})
 		assert.Empty(t, got)
+		require.Len(t, dropped, 1)
+		assert.Equal(t, "invalid_value", dropped[0].Reason)
 	})
 
-	t.Run("non-printable values are sanitized", func(t *testing.T) {
-		got := chipingress.SanitizeMetadataHeaders(map[string]string{"chain_id": "1\n2"})
-		assert.Equal(t, "1?2", got["chainid"])
+	t.Run("attribute count is capped at 32, excess dropped deterministically", func(t *testing.T) {
+		in := make(map[string]string, 33)
+		for i := 0; i < 33; i++ {
+			in[fmt.Sprintf("attr_%02d", i)] = "v"
+		}
+		got, dropped := chipingress.SanitizeMetadataHeaders(in)
+		assert.Len(t, got, 32)
+		require.Len(t, dropped, 1)
+		// Sorted order: "attr_32" sorts last among "attr_00".."attr_32".
+		assert.Equal(t, "attr_32", dropped[0].Key)
+		assert.Equal(t, "limit_exceeded", dropped[0].Reason)
 	})
 
-	t.Run("duplicate sanitized keys resolve deterministically to sorted-first key", func(t *testing.T) {
-		got := chipingress.SanitizeMetadataHeaders(map[string]string{"service.name": "from-dotted", "service_name": "from-snake"})
-		// sorted order: "service.name" < "service_name" ('.' < '_' in ASCII), so the dotted key wins.
-		assert.Equal(t, "from-dotted", got["servicename"])
+	t.Run("total key+value bytes are capped at 4096, tail dropped deterministically", func(t *testing.T) {
+		// Each accepted attribute contributes len(key)+len(value) bytes (prefix excluded). 9
+		// attributes of 500 bytes each would total 4500, over the 4096 cap, so the last one or
+		// two (in sorted order) must be dropped.
+		in := make(map[string]string, 9)
+		val := strings.Repeat("v", 490)
+		for i := 0; i < 9; i++ {
+			in[fmt.Sprintf("attr_%d", i)] = val // key is 6 bytes, so each entry is 496 bytes
+		}
+		got, dropped := chipingress.SanitizeMetadataHeaders(in)
+		assert.NotEmpty(t, dropped)
+		for _, d := range dropped {
+			assert.Equal(t, "limit_exceeded", d.Reason)
+		}
+		total := 0
+		for name, v := range got {
+			total += len(name) - len(rp) + len(v)
+		}
+		assert.LessOrEqual(t, total, 4096)
 	})
 }
 
@@ -383,9 +483,11 @@ func TestSanitizeMetadataHeaders_AvoidsRPCFailure(t *testing.T) {
 	})
 
 	t.Run("sanitized headers succeed", func(t *testing.T) {
+		sanitized, dropped := chipingress.SanitizeMetadataHeaders(dirty)
+		require.Len(t, dropped, 1, "the non-printable value must be omitted, not rewritten")
 		client, err := chipingress.NewClient(lis.Addr().String(),
 			chipingress.WithInsecureConnection(),
-			chipingress.WithHeaderProvider(chipingress.NewStaticHeaderProvider(chipingress.SanitizeMetadataHeaders(dirty))),
+			chipingress.WithHeaderProvider(chipingress.NewStaticHeaderProvider(sanitized)),
 		)
 		require.NoError(t, err)
 		defer client.Close() //nolint:errcheck
diff --git a/pkg/chipingress/resource_attributes.go b/pkg/chipingress/resource_attributes.go
deleted file mode 100644
index 2d5f974e8f..0000000000
--- a/pkg/chipingress/resource_attributes.go
+++ /dev/null
@@ -1,46 +0,0 @@
-package chipingress
-
-import "sort"
-
-// resourceAttrKey pairs a sanitized extension/metadata key name with the original
-// resource-attribute key it was derived from.
-type resourceAttrKey struct {
-	name string
-	key  string
-}
-
-// sanitizeResourceAttributeKeys returns the deduplicated, sorted list of resource-attribute
-// keys that survive sanitization and reservation checks. The returned pairs contain the
-// sanitized name and the original map key, so callers can apply their own value handling.
-//
-// Ordering is deterministic: original keys are sorted lexicographically, and if two keys
-// sanitize to the same name the first one in sorted order wins. extraReserved, if non-nil,
-// is consulted in addition to reservedExtensionNames.
-func sanitizeResourceAttributeKeys(attrs map[string]string, extraReserved map[string]struct{}) []resourceAttrKey {
-	keys := make([]string, 0, len(attrs))
-	for k := range attrs {
-		keys = append(keys, k)
-	}
-	sort.Strings(keys)
-
-	seen := make(map[string]struct{}, len(attrs))
-	result := make([]resourceAttrKey, 0, len(attrs))
-	for _, k := range keys {
-		name := sanitizeExtensionName(k)
-		if name == "" {
-			continue
-		}
-		if _, reserved := reservedExtensionNames[name]; reserved {
-			continue
-		}
-		if _, reserved := extraReserved[name]; reserved {
-			continue
-		}
-		if _, already := seen[name]; already {
-			continue
-		}
-		seen[name] = struct{}{}
-		result = append(result, resourceAttrKey{name: name, key: k})
-	}
-	return result
-}
diff --git a/pkg/chipingress/types.go b/pkg/chipingress/types.go
index 25eede3edb..34551bc93d 100644
--- a/pkg/chipingress/types.go
+++ b/pkg/chipingress/types.go
@@ -13,34 +13,20 @@ import (
 // Kafka headers named "ce_<name>" (e.g., ce_idempotencykey), enabling downstream deduplication.
 const IdempotencyKeyAttr = "idempotencykey"
 
-// reservedExtensionNames holds every CloudEvent extension name that NewEvent sets internally,
-// plus the CloudEvents core context attribute names (id, source, type, specversion, time,
-// subject, dataschema, datacontenttype) and the spec-forbidden "data" name. WithResourceAttributeExtensions
-// consults this set so that a resource attribute can never silently overwrite event-lifecycle
-// metadata or collide with a CloudEvents core attribute.
-var reservedExtensionNames = map[string]struct{}{
-	IdempotencyKeyAttr: {},
-	"recordedtime":     {},
-	"id":               {},
-	"source":           {},
-	"type":             {},
-	"specversion":      {},
-	"time":             {},
-	"subject":          {},
-	"dataschema":       {},
-	"datacontenttype":  {},
-	"data":             {},
-}
-
-// reservedMetadataKeys holds gRPC-reserved header names that could otherwise be reached by
-// sanitizeExtensionName's [a-z0-9] sanitization. Verified against grpc-go v1.79.1's
-// isReservedHeader: every other reserved header (pseudo-headers, "content-type", "grpc-*")
-// contains a ':' or '-' that sanitization strips, so "te" is the only one actually reachable.
-// SanitizeMetadataHeaders consults this set so that edge case is handled deterministically
-// rather than relying on grpc's own (silent) handling of a reserved header.
-var reservedMetadataKeys = map[string]struct{}{
-	"te": {},
-}
+// ResourceHeaderPrefix namespaces producer resource attributes sent as outgoing gRPC metadata.
+// SanitizeMetadataHeaders applies it to every key it emits.
+//
+// It is the wire contract with chip-ingress, which forwards metadata carrying this prefix onto every
+// Kafka record a request produces and emits the key unchanged. Requiring the prefix inbound and
+// preserving it outbound keeps the namespace closed, which is what makes the forwarding safe: a
+// client can only cause a header beginning with this prefix to be written, so a resource attribute
+// cannot shadow a "ce_" header, an identity header the server derives from the verified auth token,
+// or — on this side of the wire — a reserved gRPC metadata key such as the CSA auth token's.
+//
+// The same constant exists in chip-ingress as constants.ResourceHeaderPrefix. Duplicating it across
+// repositories is deliberate, matching how authHeaderKey is already spelled in both pkg/beholder and
+// pkg/chipingress; the two must stay byte-identical or forwarding silently stops.
+const ResourceHeaderPrefix = "resource_"
 
 type (
 	// Cloudevents types