fix(telemetry): count a turn by its segment, not by its invoke_agent spans

docs/architecture/telemetry-contract.md

@@ -353,7 +353,7 @@ Refines `gen_ai.client.inference`, kind `client`. One model request of a harness
 
 ### `kagent.invoke_agent.internal`
 
-Refines `gen_ai.invoke_agent.internal`, kind `internal`. One execution segment of a kagent agent. Named `invoke_agent {gen_ai.agent.name}`. kagent opens it only for a runtime that emits no invoke_agent span of its own. A resumed segment links to the segment that started the native turn, with `kagent.invocation.relationship` set to `resume_origin`. Usage and finish reasons are only on inference spans, so they are opt-in here and never set.
+Refines `gen_ai.invoke_agent.internal`, kind `internal`. One execution segment of a kagent agent. Named `invoke_agent {gen_ai.agent.name}`. kagent opens it only for a runtime that emits no invoke_agent span of its own, or a native harness whose GenAI spans the kagent harness runtime writes. Claude Code still marks its `claude_code.interaction` span invoke_agent, as a child of this one; the reference collector removes that operation from native spans, and a consumer counts turns by `kagent.invocation.segment` either way. A resumed segment links to the segment that started the native turn, with `kagent.invocation.relationship` set to `resume_origin`. Usage and finish reasons are only on inference spans, so they are opt-in here and never set.
 
 | Attribute | Requirement | Note |
 | --- | --- | --- |

docs/architecture/telemetry.md

@@ -13,7 +13,7 @@ The contract is a Weaver registry in [`telemetry/registry`](../../telemetry/regi
 
 | Concept | Attributes | Where |
 | --- | --- | --- |
-| Operation | `gen_ai.operation.name` | A span is a GenAI operation when it carries this. kagent writes `invoke_agent` only where the runtime emits none of its own |
+| Operation | `gen_ai.operation.name` | A span is a GenAI operation when it carries this. kagent writes `invoke_agent` only where the runtime emits none of its own or the kagent harness runtime writes its GenAI spans. Count turns by `kagent.invocation.segment` |
 | Runtime | `kagent.runtime` | Resource |
 | Agent | `gen_ai.agent.name`, `gen_ai.agent.id` | invoke_agent and request spans. The resource carries the same identity as `gen_ai.main_agent.name` and `gen_ai.main_agent.id` |
 | Provider and model | `gen_ai.provider.name`, `gen_ai.request.model` | Resource and invoke_agent span, for harnesses compiled against one model |
@@ -33,7 +33,7 @@ The registry describes the contract the runtimes are moving to. The next section
 | Runtime | `kagent.runtime` | Request span | `invoke_agent` | Inference spans | Keys outside the contract |
 | --- | --- | --- | --- | --- | --- |
 | Go ADK | `adk-go` | `otelgrpc` or `otelhttp` SERVER span and kagent's `a2a.request` | The ADK's own | The ADK's own | `gcp.vertex.agent.*`, written by the ADK itself |
-| Claude Code, Codex | `claude`, `codex` | `otelgrpc` or `otelhttp` SERVER span | kagent's wrapper, `invoke_agent <agent>` | kagent's harness runtime, `chat <model>` and `execute_tool <tool>` | None |
+| Claude Code, Codex | `claude`, `codex` | `otelgrpc` or `otelhttp` SERVER span | kagent's wrapper, `invoke_agent <agent>`; Claude Code's `claude_code.interaction`, which also marks itself `invoke_agent`, is its child | kagent's harness runtime, `chat <model>` and `execute_tool <tool>` | None |
 | Python ADK | `adk-python` | gRPC or FastAPI SERVER span and kagent's `a2a.request` | The ADK's own | The ADK's `generate_content` | `gcp.vertex.agent.*`, and usage repeated on `call_llm`, both written by the ADK itself |
 | LangGraph | `langgraph` | gRPC or FastAPI SERVER span | kagent's wrapper, `invoke_agent <agent>` | `opentelemetry-instrumentation-genai-langchain` | None |
 | CrewAI | `crewai` | gRPC or FastAPI SERVER span | kagent's wrapper, `invoke_agent <agent>` | `opentelemetry-instrumentation-genai-openai` and `-genai-anthropic` | None |
@@ -181,7 +181,7 @@ The registry depends on the core semantic conventions at v1.44.0 and on the GenA
 
 Each A2A `SendMessage` or `SendStreamingMessage` request opens one span in the instrumentation scope `github.com/kagent-dev/kagent/go/adk/pkg/a2a/server`. It represents one execution segment, and what it is called depends on whether anything beneath it describes the agent invocation.
 
-For a native harness, Claude Code or Codex, nothing does: the native runtime emits no agent invocation, and kagent's harness runtime writes the model and tool spans. The request span is therefore the GenAI conventions' `invoke_agent` operation, named `invoke_agent <gen_ai.agent.name>`, and it is the anchor consumers should read.
+For a native harness, Claude Code or Codex, kagent's harness runtime writes the GenAI spans in the native runtime's place. The request span is therefore the GenAI conventions' `invoke_agent` operation, named `invoke_agent <gen_ai.agent.name>`, and it is the anchor consumers should read. Claude Code still marks its `claude_code.interaction` span `invoke_agent`, as a child of the request span; the reference collector removes that operation with the native runtime's other GenAI attributes.
 
 For the ADK runtime the ADK emits `invoke_agent` spans of its own, carrying `gen_ai.agent.name` and `gen_ai.conversation.id`: one for the root agent and one for each sub-agent it transfers to within the turn. The root one is the invocation. The request span stays a transport span named `a2a.request` with the same identity attributes and no `gen_ai.operation.name`, so the wrapper never adds an invocation of its own to what the ADK reports. A consumer that counts turns rather than agent invocations counts `kagent.invocation.segment`, which only the request span carries, whichever runtime produced the trace.
 
@@ -268,7 +268,7 @@ CI runs it in the e2e job and uploads the report with the e2e logs. It reports o
 
 - associates Kubernetes metadata by `k8s.pod.uid` only, which only the controller sets. A runtime on Substrate runs inside a worker pod, so a connection IP would name the worker;
 - deletes `gen_ai.usage.*` from the Python ADK's `call_llm` spans, so usage is counted once per model call;
-- deletes `gen_ai.operation.name`, `gen_ai.system`, `gen_ai.request.model`, `gen_ai.response.finish_reasons` and `gen_ai.usage.*` from the native Claude Code and Codex spans, which carry `kagent.genai.producer=runtime`. The harness runtime writes their model and tool spans, so each model call is counted once. A standalone Claude request keeps them, since the runtime writes no span for it;
+- deletes `gen_ai.operation.name`, `gen_ai.system`, `gen_ai.request.model`, `gen_ai.response.finish_reasons` and `gen_ai.usage.*` from the native Claude Code and Codex spans, which carry `kagent.genai.producer=runtime`. The harness runtime writes their model and tool spans, so each model call is counted once, and Claude Code's `claude_code.interaction` span does not count as a second invocation. A standalone Claude request keeps them, since the runtime writes no span for it;
 - deletes captured content and the Go ADK's legacy payload keys, as a backstop for an installation with capture off. Remove `transform/genai_content` from the pipelines when capture is on;
 - derives `gen_ai.invoke_agent.duration` from invoke_agent spans that kagent did not open, before any sampling;
 - exports over gzip OTLP with a `file_storage` sending queue, so a collector restart keeps what it has accepted.

go/core/test/e2e/tracing_test.go

@@ -249,36 +249,46 @@ func (r *otlpTraceReceiver) clear() {
 }
 
 // diagnostic summarizes only the spans useful for explaining a failed
-// assertion: A2A request spans and spans correlated by the injected trace ID.
-// Codex can emit hundreds of native spans, so cap the detail included in logs.
+// assertion: the spans correlated by the injected trace ID, then the A2A
+// request spans of other traces. Codex can emit hundreds of native spans, so
+// cap the detail included in logs; the expected trace comes first, so the
+// parallel cases' request spans never crowd it out.
 func (r *otlpTraceReceiver) diagnostic(expectedTraceID []byte) string {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
 	const detailLimit = 12
 	totalSpans := 0
-	details := make([]string, 0, detailLimit)
+	var expected, others []string
 	for _, resourceSpans := range r.records {
 		serviceName := stringAttribute(resourceSpans.GetResource().GetAttributes(), "service.name")
 		for _, scopeSpans := range resourceSpans.GetScopeSpans() {
 			for _, span := range scopeSpans.GetSpans() {
 				totalSpans++
-				requestSpan := stringAttribute(span.GetAttributes(), tracing.AttributeMethod) != ""
-				if len(details) == detailLimit || !bytes.Equal(span.GetTraceId(), expectedTraceID) && !requestSpan {
+				inTrace := bytes.Equal(span.GetTraceId(), expectedTraceID)
+				if !inTrace && stringAttribute(span.GetAttributes(), tracing.AttributeMethod) == "" {
 					continue
 				}
-				details = append(details, fmt.Sprintf(
-					"service=%q scope=%q name=%q trace_id=%x %s=%q %s=%q",
+				detail := fmt.Sprintf(
+					"service=%q scope=%q name=%q trace_id=%x %s=%q %s=%q %s=%q",
 					serviceName,
 					scopeSpans.GetScope().GetName(),
 					span.GetName(),
 					span.GetTraceId(),
+					tracing.AttributeOperationName, stringAttribute(span.GetAttributes(), tracing.AttributeOperationName),
 					tracing.AttributeMethod, stringAttribute(span.GetAttributes(), tracing.AttributeMethod),
 					tracing.AttributeTaskState, stringAttribute(span.GetAttributes(), tracing.AttributeTaskState),
-				))
+				)
+				if inTrace {
+					expected = append(expected, detail)
+				} else {
+					others = append(others, detail)
+				}
 			}
 		}
 	}
+	details := append(expected, others...)
+	details = details[:min(len(details), detailLimit)]
 	return fmt.Sprintf("exports=%d resource_spans=%d total_spans=%d relevant_spans=%v", r.exports, len(r.records), totalSpans, details)
 }
 
@@ -356,7 +366,7 @@ func TestE2ECompletedChatFlushesTraces(t *testing.T) {
 		modelURL      func(*testing.T) string
 		createModel   func(*testing.T, string) *v1alpha3.ModelConfig
 		prompt        string
-		assertNative  func(*testing.T, *otlpTraceReceiver, []byte)
+		assertNative  func(*testing.T, *otlpTraceReceiver, []byte, capturedSpan)
 	}{
 		{
 			name: "claude", harness: claudeTracingE2EHarness, templateLabel: "claude-tracing",
@@ -368,14 +378,17 @@ func TestE2ECompletedChatFlushesTraces(t *testing.T) {
 				return createClaudeMockModel(t, interactionKubeClient(t), baseURL)
 			},
 			prompt: "What is 2+2?",
-			assertNative: func(t *testing.T, receiver *otlpTraceReceiver, traceID []byte) {
+			// Claude Code's interaction span is its own record of the turn and
+			// marks itself invoke_agent; it runs under kagent's invocation,
+			// which it reaches through TRACEPARENT.
+			assertNative: func(t *testing.T, receiver *otlpTraceReceiver, traceID []byte, invocation capturedSpan) {
 				t.Helper()
-				for _, captured := range receiver.selectSpans(traceID, "", "", "", nil) {
-					if strings.HasPrefix(captured.span.GetName(), "claude_code.") {
+				for _, captured := range receiver.selectSpans(traceID, "", "", "claude_code.interaction", nil) {
+					if bytes.Equal(captured.span.GetParentSpanId(), invocation.span.GetSpanId()) {
 						return
 					}
 				}
-				t.Fatal("no claude_code.* span was exported before suspension")
+				t.Fatalf("no claude_code.interaction span under the invocation was exported before suspension: %s", receiver.diagnostic(traceID))
 			},
 		},
 		{
@@ -388,7 +401,7 @@ func TestE2ECompletedChatFlushesTraces(t *testing.T) {
 				return createCodexMockModel(t, interactionKubeClient(t), baseURL)
 			},
 			prompt: "What is 2+2?",
-			assertNative: func(t *testing.T, receiver *otlpTraceReceiver, traceID []byte) {
+			assertNative: func(t *testing.T, receiver *otlpTraceReceiver, traceID []byte, _ capturedSpan) {
 				t.Helper()
 				for _, captured := range receiver.selectSpans(traceID, "", "", "", nil) {
 					if captured.serviceName == "codex-app-server" || strings.Contains(strings.ToLower(captured.scopeName), "codex") {
@@ -413,15 +426,23 @@ func TestE2ECompletedChatFlushesTraces(t *testing.T) {
 			}
 			// Wait for the Actor to suspend then assert that the completed invocation span was exported.
 			assertActorSuspended(t, fixture)
-			// Invocations are selected by operation alone, the way a consumer
-			// counts them, so a second invoke_agent span in the trace or one
-			// missing an attribute fails the test rather than going unseen. The
+			// Invocations are counted the way the reference collector leaves
+			// them: it removes the GenAI operation from the spans of a native
+			// harness process, such as Claude Code's interaction span, which
+			// carry kagent.genai.producer=runtime on their resource. Any other
+			// invoke_agent span in the trace, or one missing its segment or
+			// another attribute, fails the test rather than going unseen. The
 			// compiler owns this identity, so the assertion holds without any
 			// user-supplied resource marker on the Harness.
 			agentName := template
-			spans := receiver.selectSpans(traceID, "", "", "", map[string]string{tracing.AttributeOperationName: tracing.OperationInvokeAgent})
+			var spans []capturedSpan
+			for _, captured := range receiver.selectSpans(traceID, "", "", "", map[string]string{tracing.AttributeOperationName: tracing.OperationInvokeAgent}) {
+				if stringAttribute(captured.resource.GetAttributes(), tracing.AttributeGenAIProducer) != conv.KagentGenaiProducerRuntime {
+					spans = append(spans, captured)
+				}
+			}
 			if len(spans) != 1 {
-				t.Fatalf("invoke_agent spans = %d, want exactly one before suspension: %s", len(spans), receiver.diagnostic(traceID))
+				t.Fatalf("invoke_agent spans outside the native process = %d, want exactly one before suspension: %s", len(spans), receiver.diagnostic(traceID))
 			}
 			invocation := spans[0]
 			if got, want := invocation.span.GetName(), "invoke_agent "+agentName; got != want {
@@ -510,7 +531,7 @@ func TestE2ECompletedChatFlushesTraces(t *testing.T) {
 					t.Errorf("chat %s = %d (present %v), want %d", key, got, ok, want)
 				}
 			}
-			test.assertNative(t, receiver, traceID)
+			test.assertNative(t, receiver, traceID, invocation)
 		})
 	}
 }

telemetry/registry/signals.yaml

@@ -253,8 +253,12 @@ span_refinements:
     brief: One execution segment of a kagent agent.
     note: >
       Named `invoke_agent {gen_ai.agent.name}`. kagent opens it only for a
-      runtime that emits no invoke_agent span of its own. A resumed segment links
-      to the segment that started the native turn, with
+      runtime that emits no invoke_agent span of its own, or a native harness
+      whose GenAI spans the kagent harness runtime writes. Claude Code still
+      marks its `claude_code.interaction` span invoke_agent, as a child of this
+      one; the reference collector removes that operation from native spans, and
+      a consumer counts turns by `kagent.invocation.segment` either way. A
+      resumed segment links to the segment that started the native turn, with
       `kagent.invocation.relationship` set to `resume_origin`. Usage and finish
       reasons are only on inference spans, so they are opt-in here and never set.
     stability: development

telemetry/resolved.yaml

@@ -4769,7 +4769,7 @@ refinements:
         When `gen_ai.agent.name` is not available, it SHOULD be `invoke_agent`.
         Semantic conventions for individual GenAI systems and frameworks MAY specify different span name format.
     note: |
-      Named `invoke_agent {gen_ai.agent.name}`. kagent opens it only for a runtime that emits no invoke_agent span of its own. A resumed segment links to the segment that started the native turn, with `kagent.invocation.relationship` set to `resume_origin`. Usage and finish reasons are only on inference spans, so they are opt-in here and never set.
+      Named `invoke_agent {gen_ai.agent.name}`. kagent opens it only for a runtime that emits no invoke_agent span of its own, or a native harness whose GenAI spans the kagent harness runtime writes. Claude Code still marks its `claude_code.interaction` span invoke_agent, as a child of this one; the reference collector removes that operation from native spans, and a consumer counts turns by `kagent.invocation.segment` either way. A resumed segment links to the segment that started the native turn, with `kagent.invocation.relationship` set to `resume_origin`. Usage and finish reasons are only on inference spans, so they are opt-in here and never set.
     provenance:
       path: telemetry/registry/signals.yaml
     requirement_level: recommended