feat(telemetry): Go runtime tracing and go-adk 1.0.0 (#1618)

- Adds OTLP exporter setup similar to python runtime
- Adds span processor to add various span attributes for python adk
parity
- Bumps go-adk to latest v1.0.0
- Aligns various A2A messaging behaviour between Python and Go so
clients can switch between the two without having to rewrite A2A client
logic, this includes streaming and persisted tasks and events

An example trace with go runtime agent and subagent:

<img width="1697" height="732" alt="Screenshot 2026-04-02 at 3 49 22 PM"
src="https://github.com/user-attachments/assets/42068a10-d9ec-4a40-af12-d99981fda1db"
/>

---------

Signed-off-by: Jet Chiang <pokyuen.jetchiang-ext@solo.io>

Author: supreme-gg-gg

go/adk/cmd/main.go

@@ -18,6 +18,7 @@ import (
 	kagentmemory "github.com/kagent-dev/kagent/go/adk/pkg/memory"
 	runnerpkg "github.com/kagent-dev/kagent/go/adk/pkg/runner"
 	"github.com/kagent-dev/kagent/go/adk/pkg/session"
+	"github.com/kagent-dev/kagent/go/adk/pkg/telemetry"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 )
@@ -93,8 +94,38 @@ func main() {
 		"sseTools", len(agentConfig.SseTools),
 		"remoteAgents", len(agentConfig.RemoteAgents))
 
+	kagentName := os.Getenv("KAGENT_NAME")
+	kagentNamespace := os.Getenv("KAGENT_NAMESPACE")
+
 	// Derive app name from env or agent card.
-	appName := deriveAppName(agentCard, logger)
+	appName := deriveAppName(kagentName, kagentNamespace, agentCard, logger)
+
+	// Fall back to appName / "default" so traces always have a non-empty service identity.
+	serviceNameSource := kagentName
+	if serviceNameSource == "" {
+		serviceNameSource = appName
+	}
+	serviceNamespaceSource := kagentNamespace
+	if serviceNamespaceSource == "" {
+		serviceNamespaceSource = "default"
+	}
+	serviceName := strings.ReplaceAll(serviceNameSource, "-", "_")
+	serviceNamespace := strings.ReplaceAll(serviceNamespaceSource, "-", "_")
+	shutdownTelemetry, telemetryEnabled, telErr := telemetry.Init(context.Background(), serviceName, serviceNamespace)
+	if telErr != nil {
+		logger.Error(telErr, "Failed to initialize ADK telemetry providers; continuing without telemetry export")
+	} else if telemetryEnabled {
+		defer func() {
+			shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+			defer cancel()
+			if err := shutdownTelemetry(shutdownCtx); err != nil {
+				logger.Error(err, "Failed to shutdown telemetry providers cleanly")
+			}
+		}()
+		logger.Info("ADK telemetry initialized")
+	} else {
+		logger.Info("ADK telemetry disabled (set OTEL_TRACING_ENABLED or OTEL_LOGGING_ENABLED to true)")
+	}
 
 	// Create authenticated HTTP client when kagent persistence is enabled.
 	// This client is shared between the executor's session service and
@@ -195,10 +226,7 @@ func main() {
 	}
 }
 
-func deriveAppName(agentCard *a2atype.AgentCard, logger logr.Logger) string {
-	kagentName := os.Getenv("KAGENT_NAME")
-	kagentNamespace := os.Getenv("KAGENT_NAMESPACE")
-
+func deriveAppName(kagentName, kagentNamespace string, agentCard *a2atype.AgentCard, logger logr.Logger) string {
 	if kagentNamespace != "" && kagentName != "" {
 		namespace := strings.ReplaceAll(kagentNamespace, "-", "_")
 		name := strings.ReplaceAll(kagentName, "-", "_")

go/adk/pkg/a2a/executor.go

@@ -13,6 +13,7 @@ import (
 	"github.com/kagent-dev/kagent/go/adk/pkg/session"
 	"github.com/kagent-dev/kagent/go/adk/pkg/skills"
 	"github.com/kagent-dev/kagent/go/adk/pkg/telemetry"
+	"go.opentelemetry.io/otel/attribute"
 	adkagent "google.golang.org/adk/agent"
 	"google.golang.org/adk/runner"
 	"google.golang.org/adk/server/adka2a"
@@ -130,6 +131,8 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestCont
 		spanAttributes["kagent.app_name"] = e.appName
 	}
 	ctx = telemetry.SetKAgentSpanAttributes(ctx, spanAttributes)
+	ctx, invocationSpan := telemetry.StartInvocationSpan(ctx)
+	defer invocationSpan.End()
 
 	// 3. Initialize skills session path.
 	if e.skillsDirectory != "" && sessionID != "" {
@@ -189,11 +192,20 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestCont
 
 	// 9. Emit initial events.
 	if reqCtx.StoredTask == nil {
-		// New task — emit submitted.
-		submitted := a2atype.NewStatusUpdateEvent(reqCtx, a2atype.TaskStateSubmitted, nil)
+		// New task — emit submitted with the user's message
+		submitted := a2atype.NewStatusUpdateEvent(reqCtx, a2atype.TaskStateSubmitted, reqCtx.Message)
 		if err := queue.Write(ctx, submitted); err != nil {
 			return fmt.Errorf("failed to write submitted event: %w", err)
 		}
+	} else if ExtractDecisionFromMessage(reqCtx.Message) != "" {
+		// a2a-go appends incoming message to task history before executor runs.
+		// See https://github.com/a2aproject/a2a-go/blob/v0.3.13/a2asrv/agentexec.go#L188
+		// Remove the pre-appended copy and emit one decision status event.
+		dropPreAppendedDecisionFromHistory(reqCtx.StoredTask, reqCtx.Message)
+		decision := a2atype.NewStatusUpdateEvent(reqCtx, a2atype.TaskStateWorking, reqCtx.Message)
+		if err := queue.Write(ctx, decision); err != nil {
+			return fmt.Errorf("failed to write HITL decision status event: %w", err)
+		}
 	}
 
 	// Base metadata carried on every event (app_name, user_id, session_id).
@@ -217,13 +229,10 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestCont
 
 	// State tracked across the event loop.
 	var (
-		invocationID      string
-		artifactID        a2atype.ArtifactID
-		partialArtifactID a2atype.ArtifactID
-		lastTextParts     a2atype.ContentParts
-		lastTextMeta      map[string]any
-		hitlParts         a2atype.ContentParts
-		runErr            error
+		invocationID        string
+		lastNonPartialParts a2atype.ContentParts
+		hitlParts           a2atype.ContentParts
+		runErr              error
 	)
 
 	for adkEvent, adkErr := range r.Run(ctx, userID, sessionID, content, runConfig) {
@@ -238,6 +247,7 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestCont
 		// Track invocation ID from the first event that has one.
 		if adkEvent.InvocationID != "" && invocationID == "" {
 			invocationID = adkEvent.InvocationID
+			invocationSpan.SetAttributes(attribute.String("gcp.vertex.agent.invocation_id", invocationID))
 		}
 
 		// Build per-event metadata (inherits baseMeta + adds invocation_id, usage etc.).
@@ -299,8 +309,11 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestCont
 		}
 
 		if adkEvent.Partial {
-			// Partial event: emit as working status (text-only) for UI streaming,
-			// and as partial artifact update.
+			// Partial event: emit as working status (text-only) for UI streaming.
+			// Note: Go ADK executor uses TaskArtifactUpdateEvent for partial events,
+			// so we don't need to emit a separate partial artifact update.
+			// However, this is done here in order to match the Python executor's behavior.
+			// Go ADK executor also uses different A2A response formats than Python ADK.
 			textOnly := filterTextParts(a2aParts)
 			if len(textOnly) > 0 {
 				mirrorMeta := maps.Clone(eventMeta)
@@ -313,28 +326,7 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestCont
 					return fmt.Errorf("failed to write partial status event: %w", err)
 				}
 			}
-
-			// Emit partial artifact update.
-			var partialEv *a2atype.TaskArtifactUpdateEvent
-			if partialArtifactID == "" {
-				partialEv = a2atype.NewArtifactEvent(reqCtx, a2aParts...)
-				partialArtifactID = partialEv.Artifact.ID
-			} else {
-				partialEv = a2atype.NewArtifactUpdateEvent(reqCtx, partialArtifactID, a2aParts...)
-			}
-			if partialEv.Artifact.Metadata == nil {
-				partialEv.Artifact.Metadata = map[string]any{}
-			}
-			partialEv.Artifact.Metadata[adka2a.ToA2AMetaKey("partial")] = true
-			partialEv.Metadata = maps.Clone(eventMeta)
-			partialEv.Metadata[adka2a.ToA2AMetaKey("partial")] = true
-			partialEv.Append = false // discard partial events
-			if err := queue.Write(ctx, partialEv); err != nil {
-				return fmt.Errorf("failed to write partial artifact event: %w", err)
-			}
 		} else {
-			// Non-partial event: emit as artifact + mirror as working status.
-			// Mirror: emit working status with non-empty parts.
 			mirrorParts := a2aParts
 			if len(hitlParts) == 0 {
 				// Only mirror when not accumulating HITL parts (those go into input_required).
@@ -345,27 +337,7 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestCont
 				if err := queue.Write(ctx, statusEv); err != nil {
 					return fmt.Errorf("failed to write mirror status event: %w", err)
 				}
-			}
-
-			// Track last text parts for final status injection.
-			if tp := filterTextParts(a2aParts); len(tp) > 0 {
-				lastTextParts = tp
-				lastTextMeta = eventMeta
-			}
-
-			// Emit artifact.
-			var artifactEv *a2atype.TaskArtifactUpdateEvent
-			if artifactID == "" {
-				artifactEv = a2atype.NewArtifactEvent(reqCtx, a2aParts...)
-				artifactID = artifactEv.Artifact.ID
-			} else {
-				artifactEv = a2atype.NewArtifactUpdateEvent(reqCtx, artifactID, a2aParts...)
-			}
-			// Mark as non-partial so downstream consumers skip their own aggregation.
-			artifactEv.Metadata = maps.Clone(eventMeta)
-			artifactEv.Metadata[adka2a.ToA2AMetaKey("partial")] = false
-			if err := queue.Write(ctx, artifactEv); err != nil {
-				return fmt.Errorf("failed to write artifact event: %w", err)
+				lastNonPartialParts = mirrorParts
 			}
 		}
 
@@ -398,15 +370,16 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestCont
 		return queue.Write(ctx, inputRequired)
 	}
 
-	// completed: inject last text into final status if no message present.
-	var finalMsg *a2atype.Message
-	if len(lastTextParts) > 0 {
-		finalMsg = a2atype.NewMessage(a2atype.MessageRoleAgent, lastTextParts...)
-		if len(lastTextMeta) > 0 {
-			finalMsg.Metadata = maps.Clone(lastTextMeta)
+	// Final artifact update with lastChunk=true (if we have parts) and final completed status update (no message payload).
+	if len(lastNonPartialParts) > 0 {
+		finalArtifact := a2atype.NewArtifactEvent(reqCtx, lastNonPartialParts...)
+		finalArtifact.LastChunk = true
+		if err := queue.Write(ctx, finalArtifact); err != nil {
+			return fmt.Errorf("failed to write final artifact event: %w", err)
 		}
 	}
-	completed := a2atype.NewStatusUpdateEvent(reqCtx, a2atype.TaskStateCompleted, finalMsg)
+
+	completed := a2atype.NewStatusUpdateEvent(reqCtx, a2atype.TaskStateCompleted, nil)
 	completed.Final = true
 	completed.Metadata = finalMeta
 	return queue.Write(ctx, completed)
@@ -434,3 +407,19 @@ func extractSessionName(message *a2atype.Message) string {
 	}
 	return ""
 }
+
+// dropPreAppendedDecisionFromHistory removes a pre-appended HITL decision
+// message inserted by a2a-go before executor invocation.
+func dropPreAppendedDecisionFromHistory(task *a2atype.Task, incoming *a2atype.Message) {
+	if task == nil || incoming == nil || len(task.History) == 0 {
+		return
+	}
+	last := task.History[len(task.History)-1]
+	if last == nil || last.ID != incoming.ID {
+		return
+	}
+	if ExtractDecisionFromMessage(last) == "" {
+		return
+	}
+	task.History = task.History[:len(task.History)-1]
+}

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

@@ -12,6 +12,7 @@ import (
 	a2atype "github.com/a2aproject/a2a-go/a2a"
 	"github.com/a2aproject/a2a-go/a2asrv"
 	"github.com/go-logr/logr"
+	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 )
 
 // ServerConfig holds configuration for the A2A server.
@@ -38,6 +39,23 @@ func NewA2AServer(agentCard a2atype.AgentCard, executor a2asrv.AgentExecutor, lo
 	RegisterHealthEndpoints(mux)
 	mux.Handle(a2asrv.WellKnownAgentCardPath, a2asrv.NewStaticAgentCardHandler(&agentCard))
 	mux.Handle("/", jsonrpcHandler)
+	// Wrap the whole server mux to enable trace context extraction and an inbound
+	// HTTP server span for each request.
+	instrumentedHandler := otelhttp.NewHandler(
+		mux,
+		"a2a-server",
+		otelhttp.WithSpanNameFormatter(func(_ string, r *http.Request) string {
+			return r.Method + " " + r.URL.Path
+		}),
+		otelhttp.WithFilter(func(r *http.Request) bool {
+			switch r.URL.Path {
+			case "/health", "/healthz", a2asrv.WellKnownAgentCardPath:
+				return false
+			default:
+				return true
+			}
+		}),
+	)
 
 	addr := ":" + config.Port
 	if config.Host != "" {
@@ -47,7 +65,7 @@ func NewA2AServer(agentCard a2atype.AgentCard, executor a2asrv.AgentExecutor, lo
 	return &A2AServer{
 		httpServer: &http.Server{
 			Addr:    addr,
-			Handler: mux,
+			Handler: instrumentedHandler,
 		},
 		logger: logger,
 		config: config,

go/adk/pkg/memory/kagent_service.go

@@ -80,10 +80,10 @@ func New(cfg Config) (*KagentMemoryService, error) {
 	}, nil
 }
 
-// AddSession implements memory.Service.AddSession.
+// AddSessionToMemory implements memory.Service.AddSessionToMemory.
 // It extracts content from the session, optionally summarizes it, generates embeddings,
 // and stores it via the Kagent API.
-func (s *KagentMemoryService) AddSession(ctx context.Context, session adksession.Session) error {
+func (s *KagentMemoryService) AddSessionToMemory(ctx context.Context, session adksession.Session) error {
 	log := logr.FromContextOrDiscard(ctx)
 	log.V(1).Info("Adding session to memory", "sessionID", session.ID(), "userID", session.UserID())
 
@@ -162,9 +162,9 @@ func (s *KagentMemoryService) storeMemory(ctx context.Context, userID, content s
 	return nil
 }
 
-// Search implements memory.Service.Search.
+// SearchMemory implements memory.Service.SearchMemory.
 // It searches for relevant memories using vector similarity.
-func (s *KagentMemoryService) Search(ctx context.Context, req *memory.SearchRequest) (*memory.SearchResponse, error) {
+func (s *KagentMemoryService) SearchMemory(ctx context.Context, req *memory.SearchRequest) (*memory.SearchResponse, error) {
 	log := logr.FromContextOrDiscard(ctx)
 	log.V(1).Info("Searching memory", "query", req.Query, "userID", req.UserID)
 

go/adk/pkg/memory/kagent_service_test.go

@@ -6,6 +6,7 @@ import (
 	"iter"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"testing"
 	"time"
 
@@ -146,7 +147,7 @@ func TestKagentMemoryService_AddSession(t *testing.T) {
 				model:           nil, // No summarization
 			}
 
-			err := svc.AddSession(context.Background(), tt.session)
+			err := svc.AddSessionToMemory(context.Background(), tt.session)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("AddSession() error = %v, wantErr %v", err, tt.wantErr)
 			}
@@ -251,7 +252,7 @@ func TestKagentMemoryService_Search(t *testing.T) {
 				model:           nil,
 			}
 
-			resp, err := svc.Search(context.Background(), &memory.SearchRequest{
+			resp, err := svc.SearchMemory(context.Background(), &memory.SearchRequest{
 				Query:  tt.query,
 				UserID: tt.userID,
 			})
@@ -407,7 +408,7 @@ func TestKagentMemoryService_ExtractSessionContent(t *testing.T) {
 				t.Error("Expected non-empty content, got empty")
 			}
 
-			if tt.wantContain != "" && !contains(content, tt.wantContain) {
+			if tt.wantContain != "" && !strings.Contains(content, tt.wantContain) {
 				t.Errorf("Expected content to contain %q, got: %s", tt.wantContain, content)
 			}
 		})
@@ -566,16 +567,3 @@ func newMockEventWithFunctionCall(author, functionName string) *adksession.Event
 	}
 	return evt
 }
-
-func contains(s, substr string) bool {
-	return len(s) >= len(substr) && (s == substr || len(s) > 0 && containsHelper(s, substr))
-}
-
-func containsHelper(s, substr string) bool {
-	for i := 0; i <= len(s)-len(substr); i++ {
-		if s[i:i+len(substr)] == substr {
-			return true
-		}
-	}
-	return false
-}