feat: promote allowlisted caller context onto every agent span

docs/architecture/README.md

@@ -113,6 +113,7 @@ cleanup. Session lifecycle workers independently pause/suspend idle Actors.
 - [Token usage](usage-extension.md)
 - [Prompt resolution](prompt-templates.md)
 - [Telemetry](telemetry.md)
+- [Caller context in traces](trace-context.md)
 - [Structured output](structured-output.md)
 
 The documents describe implemented behavior. Full cross-Session delegation

docs/architecture/trace-context.md

@@ -0,0 +1,154 @@
+# Caller Context in Traces
+
+Agent spans describe *what the agent did*, but they say nothing about *who asked
+for it*. Kagent can copy a configurable allowlist of caller-supplied values —
+an opaque user identifier, a conversation thread, a ticket ID — onto every span
+of a request, so traces can be filtered and grouped by the caller in Langfuse,
+Jaeger, Grafana Tempo, or any other OTLP backend.
+
+The feature is **off by default**. It turns on when an operator sets an
+allowlist.
+
+---
+
+## Configuration
+
+| Setting | Default | Description |
+|---|---|---|
+| Helm `controller.traceContextKeys` | `[]` | List of baggage keys, or `{from, to}` mappings, to copy onto spans |
+| Env `KAGENT_TRACE_CONTEXT_KEYS` | `""` | Comma-separated keys, or a JSON array of the same mappings |
+
+```yaml
+controller:
+  traceContextKeys:
+    - {from: sub, to: user.id}
+    - thread_id
+    - gen_ai.conversation.id
+```
+
+Prefer an opaque identifier such as an OIDC `sub` for `user.id`. Do not put
+names or email addresses on spans; see [Sensitive values](#sensitive-values).
+
+The controller forwards `KAGENT_TRACE_CONTEXT_KEYS` to every agent it creates.
+Both the Go and the Python runtime read it, so behaviour is identical whichever
+one an agent runs.
+
+Adding a new traced value is a configuration change, not a code change: append
+the key and redeploy.
+
+---
+
+## Where values come from
+
+There is one path onto spans: **W3C [Baggage](https://www.w3.org/TR/baggage/)**.
+Both runtimes install the upstream baggage-to-span-attribute processor
+(`baggagecopy` in Go, `BaggageSpanProcessor` in Python) with a predicate that
+copies only allowlisted destination keys.
+
+A2A `message.metadata` is promoted *into baggage* at the request boundary when
+the destination key is not already set. That is what lets callers that cannot
+set a transport header still reach every span — including sub-agent hops —
+without a second merge path or a request-scoped attribute bag.
+
+| Source | Set by | How it reaches spans |
+|---|---|---|
+| W3C baggage (`baggage` header) | Any client or proxy on the request path | Copied by the baggage span processor |
+| A2A `message.metadata` | The A2A caller, per message | Written into baggage at the request boundary if the destination key is absent |
+
+Empty or non-scalar metadata does not wipe baggage. `{from, to}` remaps a
+source key onto the destination baggage member; the processor then copies
+`to`. Destination names are used as-is, so `user.id` and
+`gen_ai.conversation.id` are reachable without a hardcoded unprefixed set.
+
+A key absent from both sources is simply not emitted. Metadata is promoted
+before the invocation span starts, so the allowlist also covers that span.
+
+Baggage propagation across processes is separately opt-in on current kagent.
+Set `OTEL_PROPAGATORS=tracecontext,baggage` on the controller and each runtime
+that should exchange baggage. Use `controller.env` for the controller and the
+Harness workload environment for runtimes. Controller `OTEL_*` overrides are
+process-local; they are not forwarded to managed runtimes. Metadata promotion
+and local span attributes do not require outbound baggage propagation.
+
+---
+
+## Where values land
+
+```
+baggage: user.id=opaque-subject      →   user.id                  = "opaque-subject"
+metadata: {"sub": "opaque-subject"}  →   user.id                  = "opaque-subject"   # with {from: sub, to: user.id}
+metadata: {"thread_id": "1717171.42"} →  thread_id                = "1717171.42"
+```
+
+The request-scoped kagent attribute bag still carries runtime values
+(`enduser.id`, `gen_ai.conversation.id`, `a2a.task.id`, …). That processor runs after the
+baggage processor, so caller baggage cannot override attributes the runtime already
+stamped.
+
+This is deliberate: Langfuse v4 and comparable backends resolve trace-level
+filters against the attributes present on each span, so stamping only the
+root span would leave most views unfilterable.
+
+---
+
+## Sensitive values
+
+[OpenTelemetry recommends](https://opentelemetry.io/docs/security/handling-sensitive-data/)
+against putting email addresses or names on telemetry at all. An OIDC `sub` is
+already an opaque identifier and is what `user.id` should use.
+
+Hashing and renaming that must happen before a value is visible on hops belong
+in the authenticating gateway or the OTel Collector. Baggage travels on HTTP
+headers, so a value placed there is visible to every downstream hop that
+receives those headers, including model providers and HTTP MCP servers.
+
+Value length is the ordinary OTel limit
+(`OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT`), not a kagent-specific cap.
+
+---
+
+## Safety properties
+
+Caller-supplied context is untrusted input, so promotion is constrained:
+
+| Risk | Control |
+|---|---|
+| Attribute explosion / cardinality | Only allowlisted keys are copied; the allowlist is capped at 32 entries |
+| Oversized spans | `OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT` |
+| Shadowing runtime attributes | The kagent span processor runs after baggagecopy and wins on conflict |
+| Leaking secrets into a trace backend | Nothing is copied unless an operator names the key |
+| A tenant widening the allowlist | The allowlist is cluster-wide operator configuration; an entry of the same name in a `Harness` environment is dropped rather than inherited |
+
+Non-scalar metadata (objects, arrays) is skipped: it is unbounded in size and
+meaningless as a baggage value.
+
+Choose allowlist keys deliberately. Anything named here is visible to everyone
+with access to the trace backend, and callers control the values.
+
+---
+
+## Renaming attributes for a backend
+
+`{from, to}` is enough when the operator wants a baggage key remapped at the
+executor. Further renames belong in the OTel Collector that already sits
+between kagent and the backend:
+
+```yaml
+processors:
+  transform:
+    trace_statements:
+      - set(span.attributes["session.id"], span.attributes["thread_id"])
+        where span.attributes["thread_id"] != nil
+```
+
+---
+
+## Implementation
+
+| Component | Path |
+|---|---|
+| Go ADK | `go/adk/pkg/telemetry/context_attributes.go` |
+| Python core | `python/packages/kagent-core/src/kagent/core/telemetry/_context_attributes.py` |
+| Python A2A boundary (all runtimes) | `python/packages/kagent-core/src/kagent/core/a2a/_telemetry.py` |
+| Controller forwarding | `go/core/internal/translator/kagent/compiler.go` |
+| Helm | `helm/kagent/templates/controller-deployment.yaml` |

go/adk/pkg/a2a/executor.go

@@ -275,10 +275,12 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.ExecutorCon
 		defer e.runningUsage.track(reqCtx.TaskID, usage)()
 		// The invocation span started before this executor ran, so the request
 		// identity has to be recorded on it directly. ADK's own spans get it
-		// through the request attribute span processor.
+		// through the request attribute span processor. Promote here as well
+		// for invocations that bypass the transport interceptor.
 		resumed := reqCtx.StoredTask != nil &&
 			(reqCtx.StoredTask.Status.State == a2atype.TaskStateInputRequired || reqCtx.StoredTask.Status.State == a2atype.TaskStateAuthRequired)
 		tracing.InvocationFromContext(ctx).SetAttributes(tracing.RequestIdentity(sessionID, string(reqCtx.TaskID), resumed)...)
+		ctx = telemetry.ContextWithPromotedMetadata(ctx, reqCtx.Message.Metadata)
 		ctx = telemetry.WithRequestAttributes(ctx, requestSpanAttributes(sessionID, string(reqCtx.TaskID), trustedUserID)...)
 		// The identity replaces any caller members of the same name, so a caller
 		// cannot impersonate another conversation downstream.

go/adk/pkg/a2a/executor_test.go

@@ -910,3 +910,42 @@ func TestRequestSpanAttributesCarryOnlyATrustedUser(t *testing.T) {
 		})
 	}
 }
+
+func TestKAgentExecutorDoesNotExportUnallowlistedMetadata(t *testing.T) {
+	t.Setenv("KAGENT_TRACE_CONTEXT_KEYS", "")
+	exporter, flush := installRecordingTracer(t)
+	ctx, root := otel.Tracer("test").Start(t.Context(), "invocation")
+	message := a2atype.NewMessage(a2atype.MessageRoleUser, a2atype.NewTextPart("hello"))
+	message.Metadata = map[string]any{
+		"private_token": "must-not-be-exported",
+		"attempt":       float64(3),
+		"enabled":       true,
+	}
+	executor := &KAgentExecutor{
+		builtin: &recordingExecutor{span: "generate_content"},
+		logger:  slog.New(slog.DiscardHandler),
+	}
+	for _, err := range executor.Execute(ctx, &a2asrv.ExecutorContext{
+		ContextID: "ctx-1", TaskID: "task-1", Message: message,
+	}) {
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	root.End()
+	if err := flush(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	spans := exporter.GetSpans()
+	if len(spans) != 2 {
+		t.Fatalf("exported %d spans, want invocation and model spans", len(spans))
+	}
+	for _, span := range spans {
+		for _, attr := range span.Attributes {
+			key := string(attr.Key)
+			if strings.HasPrefix(key, "a2a.message.metadata.") || key == "private_token" || key == "attempt" || key == "enabled" {
+				t.Errorf("%s exported unallowlisted metadata attribute %q", span.Name, key)
+			}
+		}
+	}
+}

go/adk/pkg/a2a/server/tracing.go

@@ -11,6 +11,7 @@ import (
 	"go.opentelemetry.io/otel/metric"
 
 	"github.com/kagent-dev/kagent/go/adk/pkg/auth"
+	"github.com/kagent-dev/kagent/go/adk/pkg/telemetry"
 	"github.com/kagent-dev/kagent/go/pkg/tracing"
 )
 
@@ -71,11 +72,20 @@ func newInvocationInterceptor(logger *slog.Logger, telemetry tracing.RuntimeTele
 	return interceptor
 }
 
-func (i *invocationInterceptor) Before(ctx context.Context, callCtx *a2asrv.CallContext, _ *a2asrv.Request) (context.Context, any, error) {
+func (i *invocationInterceptor) Before(ctx context.Context, callCtx *a2asrv.CallContext, request *a2asrv.Request) (context.Context, any, error) {
 	method := callCtx.Method()
 	if method != "SendMessage" && method != "SendStreamingMessage" {
 		return ctx, nil, nil
 	}
+	var metadata map[string]any
+	if request != nil {
+		if params, ok := request.Payload.(*a2atype.SendMessageRequest); ok && params != nil && params.Message != nil {
+			metadata = params.Message.Metadata
+		}
+	}
+	// Promote before the invocation starts, so the same baggage processor
+	// covers the invocation and every descendant span.
+	ctx = telemetry.ContextWithPromotedMetadata(ctx, metadata)
 	attributes := make([]attribute.KeyValue, 0, len(i.static)+2)
 	attributes = append(attributes, attribute.String(tracing.AttributeMethod, method))
 	attributes = append(attributes, i.static...)
@@ -85,6 +95,8 @@ func (i *invocationInterceptor) Before(ctx context.Context, callCtx *a2asrv.Call
 		attributes = append(attributes, attribute.String(tracing.AttributeUserID, userID))
 	}
 	ctx, invocation := tracing.StartInvocation(ctx, tracing.Tracer(invocationScope), i.name, i.flush, attributes...)
+	// Runtime identity remains authoritative after baggagecopy runs.
+	invocation.SetAttributes(attributes...)
 	invocation.MeasureDuration(i.duration, i.durationAttributes...)
 	// a2a-go runs no final After callback when a streaming consumer stops
 	// reading, so an invocation the transport still owns would otherwise never

go/adk/pkg/a2a/server/tracing_test.go

@@ -11,6 +11,9 @@ import (
 	"log/slog"
 	"net/http"
 	"net/http/httptest"
+	"os"
+	"os/exec"
+	"strings"
 	"testing"
 	"time"
 
@@ -29,7 +32,9 @@ import (
 	"google.golang.org/grpc/credentials/insecure"
 
 	"github.com/kagent-dev/kagent/go/adk/pkg/auth"
+	"github.com/kagent-dev/kagent/go/adk/pkg/telemetry"
 	"github.com/kagent-dev/kagent/go/pkg/tracing"
+	"go.opentelemetry.io/contrib/processors/baggagecopy"
 )
 
 // staticUserInterceptor stands in for UserIDCallInterceptor, which the app
@@ -501,3 +506,57 @@ func TestRequestSpanCompletesOnAStreamingMessageResult(t *testing.T) {
 
 	requestSpan(t, exporter)
 }
+
+func TestInvocationPromotesOnlyAllowlistedMetadata(t *testing.T) {
+	mode := os.Getenv("KAGENT_CONTEXT_POLICY_TEST")
+	if mode == "" {
+		// Policy is cached for process lifetime, so each mode needs a fresh process.
+		for _, mode := range []string{"disabled", "enabled"} {
+			t.Run(mode, func(t *testing.T) {
+				policy := ""
+				if mode == "enabled" {
+					policy = `[{"from":"sub","to":"user.id"},"gen_ai.agent.name","enduser.id"]`
+				}
+				command := exec.CommandContext(t.Context(), os.Args[0], "-test.run=^TestInvocationPromotesOnlyAllowlistedMetadata$")
+				command.Env = append(os.Environ(), "KAGENT_CONTEXT_POLICY_TEST="+mode, "KAGENT_TRACE_CONTEXT_KEYS="+policy)
+				if output, err := command.CombinedOutput(); err != nil {
+					t.Fatalf("%s policy: %v\n%s", mode, err, output)
+				}
+			})
+		}
+		return
+	}
+	exporter := syncExporter(t)
+	if filter := telemetry.AllowedBaggageCopyFilter(); filter != nil {
+		otel.GetTracerProvider().(*sdktrace.TracerProvider).RegisterSpanProcessor(baggagecopy.NewSpanProcessor(filter))
+	}
+	server, err := NewA2AServer(a2atype.AgentCard{}, failingExecutor{}, slog.New(slog.DiscardHandler),
+		ServerConfig{Port: "0", Telemetry: tracing.RuntimeTelemetry{
+			Runtime: tracing.RuntimeADKGo, AgentName: "researcher", AgentNamespace: "team-a",
+		}}, a2asrv.WithCallInterceptors(staticUserInterceptor{userID: "trusted-user"}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	message := a2atype.NewMessage(a2atype.MessageRoleUser, a2atype.NewTextPart("hello"))
+	message.Metadata = map[string]any{
+		"sub": "opaque-subject", "private_token": "must-not-be-exported",
+		"gen_ai.agent.name": "forged-agent", "enduser.id": "forged-user",
+	}
+	sendMessage(t, server, &a2atype.SendMessageRequest{Message: message})
+	span := requestSpan(t, exporter)
+	wantUser := ""
+	if mode == "enabled" {
+		wantUser = "opaque-subject"
+	}
+	if got := spanAttribute(span, "user.id"); got != wantUser {
+		t.Fatalf("user.id = %q, want %q", got, wantUser)
+	}
+	if spanAttribute(span, tracing.AttributeAgentName) != "researcher" || spanAttribute(span, tracing.AttributeUserID) != "trusted-user" {
+		t.Fatal("caller metadata replaced runtime identity")
+	}
+	for _, attr := range span.Attributes {
+		if strings.HasPrefix(string(attr.Key), "a2a.message.metadata.") || string(attr.Key) == "private_token" || attr.Value.AsString() == "must-not-be-exported" {
+			t.Errorf("unallowlisted metadata exported on invocation: %s", attr.Key)
+		}
+	}
+}

go/adk/pkg/telemetry/attributes_test.go

@@ -48,3 +48,20 @@ func TestWithRequestAttributesKeepsContextWhenEmpty(t *testing.T) {
 		t.Fatal("no attributes should return the same context")
 	}
 }
+func spanAttributesByName(t *testing.T, spans tracetest.SpanStubs, name string) map[string]attribute.Value {
+	t.Helper()
+
+	for _, span := range spans {
+		if span.Name != name {
+			continue
+		}
+		attrs := make(map[string]attribute.Value, len(span.Attributes))
+		for _, attr := range span.Attributes {
+			attrs[string(attr.Key)] = attr.Value
+		}
+		return attrs
+	}
+
+	t.Fatalf("span %q not found", name)
+	return nil
+}