fix(go-adk): add per-call session isolation for Agent tools

go/adk/pkg/agent/agent.go

@@ -64,10 +64,15 @@ func CreateGoogleADKAgentWithSubagentSessionIDs(ctx context.Context, agentConfig
 			log.Info("Skipping remote agent with empty URL", "name", remoteAgent.Name)
 			continue
 		}
-		remoteTool, sessionID, err := tools.NewKAgentRemoteA2ATool(remoteAgent.Name, remoteAgent.Description, remoteAgent.Url, nil, remoteAgent.Headers, propagateToken)
+		remoteTool, sessionID, err := tools.NewKAgentRemoteA2ATool(remoteAgent.Name, remoteAgent.Description, remoteAgent.Url, nil, remoteAgent.Headers, propagateToken, remoteAgent.IsolateSessions)
 		if err != nil {
 			return nil, nil, fmt.Errorf("failed to create remote A2A tool for %s: %w", remoteAgent.Name, err)
 		}
+		// Isolated tools mint a fresh context_id per call, so there is no
+		// single pre-known session id to stamp function_call parts with;
+		// NewKAgentRemoteA2ATool returns "" in that case and we skip the
+		// map entry. The UI instead links each sub-agent card via the
+		// per-call subagent_session_id in the tool's function_response.
 		if sessionID != "" {
 			subagentSessionIDs[remoteAgent.Name] = sessionID
 		}

go/adk/pkg/tools/remote_a2a_tool.go

@@ -149,30 +149,44 @@ type remoteA2AState struct {
 	initOnce  sync.Once
 	initErr   error
 
+	// lastContextID is the stable A2A context_id used for every call to this
+	// sub-agent when isolateSessions is false (the default): all calls land
+	// in one shared sub-agent session, giving stateful sub-agents session
+	// continuity across calls.
 	lastContextID string
+
+	// isolateSessions mints a fresh context_id per call (see nextContextID)
+	// instead of reusing lastContextID, so each call runs in its own isolated
+	// sub-agent session. Required for parallel fan-out: without it, N
+	// parallel calls in one turn collapse into a single shared sub-agent
+	// session. See go/api/v1alpha2.Tool.IsolateSessions.
+	isolateSessions bool
 }
 
 // NewKAgentRemoteA2ATool creates a function tool that calls a remote A2A agent and
 // propagates HITL state. It returns:
 //   - the tool.Tool to register with the agent config
-//   - the initial A2A context/session ID for subagent session stamping
+//   - the initial A2A context/session ID for subagent session stamping, or ""
+//     when isolateSessions is true (there is no single pre-known session id
+//     to stamp in that case; see remoteA2AState.isolateSessions)
 //
 // The agent card is fetched lazily from baseURL/.well-known/agent.json.
 // If httpClient is nil, a default client is created. The client's transport is
 // wrapped with otelhttp to propagate W3C trace context to subagents.
-func NewKAgentRemoteA2ATool(name, description, baseURL string, httpClient *http.Client, extraHeaders map[string]string, propagateToken bool) (tool.Tool, string, error) {
+func NewKAgentRemoteA2ATool(name, description, baseURL string, httpClient *http.Client, extraHeaders map[string]string, propagateToken, isolateSessions bool) (tool.Tool, string, error) {
 	if httpClient == nil {
 		httpClient = &http.Client{}
 	}
 	httpClient = withOTelTransport(httpClient)
 	state := &remoteA2AState{
-		name:           name,
-		description:    description,
-		baseURL:        baseURL,
-		httpClient:     httpClient,
-		extraHeaders:   extraHeaders,
-		propagateToken: propagateToken,
-		lastContextID:  a2atype.NewContextID(),
+		name:            name,
+		description:     description,
+		baseURL:         baseURL,
+		httpClient:      httpClient,
+		extraHeaders:    extraHeaders,
+		propagateToken:  propagateToken,
+		lastContextID:   a2atype.NewContextID(),
+		isolateSessions: isolateSessions,
 	}
 	ft, err := functiontool.New(functiontool.Config{
 		Name:        name,
@@ -183,9 +197,23 @@ func NewKAgentRemoteA2ATool(name, description, baseURL string, httpClient *http.
 	if err != nil {
 		return nil, "", fmt.Errorf("failed to create remote A2A function tool for %s: %w", name, err)
 	}
+	if state.isolateSessions {
+		return ft, "", nil
+	}
 	return ft, state.lastContextID, nil
 }
 
+// nextContextID returns the A2A context_id to stamp on the next outbound
+// call: a fresh id when isolateSessions is enabled (isolated per-call
+// session), or the tool's stable lastContextID otherwise (shared session for
+// the lifetime of the tool).
+func (s *remoteA2AState) nextContextID() string {
+	if s.isolateSessions {
+		return a2atype.NewContextID()
+	}
+	return s.lastContextID
+}
+
 // ensureClient lazily resolves the agent card and initialises the A2A client.
 // Initialization is protected by sync.Once to avoid races under concurrent use.
 func (s *remoteA2AState) ensureClient(ctx context.Context) (*a2aclient.Client, error) {
@@ -257,11 +285,12 @@ func (s *remoteA2AState) handleFirstCall(ctx adkagent.ToolContext, requestText s
 		return map[string]any{"error": err.Error()}, nil
 	}
 
+	contextID := s.nextContextID()
 	message := a2atype.NewMessage(
 		a2atype.MessageRoleUser,
 		a2atype.TextPart{Text: requestText},
 	)
-	message.ContextID = s.lastContextID
+	message.ContextID = contextID
 
 	sendCtx := context.WithValue(ctx, userIDContextKey{}, ctx.UserID())
 	sendCtx = context.WithValue(sendCtx, parentContextIDContextKey{}, ctx.SessionID())
@@ -271,7 +300,7 @@ func (s *remoteA2AState) handleFirstCall(ctx adkagent.ToolContext, requestText s
 		return map[string]any{"error": fmt.Sprintf("Remote agent '%s' request failed: %v", s.name, err)}, nil
 	}
 
-	return s.processResult(ctx, result)
+	return s.processResult(ctx, contextID, result)
 }
 
 // handleResume is Phase 2: forward the user's decision to the remote agent's pending task.
@@ -322,7 +351,7 @@ func (s *remoteA2AState) handleResume(ctx adkagent.ToolContext) (map[string]any,
 		return map[string]any{"error": fmt.Sprintf("Remote agent '%s' resume failed: %v", subagentName, err)}, nil
 	}
 
-	ret, retErr := s.processResult(ctx, result)
+	ret, retErr := s.processResult(ctx, contextID, result)
 	// Prefer the context_id from the confirmation payload (the original subagent
 	// session) over the pre-generated one. Mirrors Python's:
 	//   "subagent_session_id": context_id or self._last_context_id
@@ -337,7 +366,12 @@ func (s *remoteA2AState) handleResume(ctx adkagent.ToolContext) (map[string]any,
 }
 
 // processResult converts a SendMessageResult into a tool return value.
-func (s *remoteA2AState) processResult(ctx adkagent.ToolContext, result a2atype.SendMessageResult) (map[string]any, error) {
+// contextID is the A2A context_id this call was sent under (from
+// nextContextID, or the confirmation payload on resume) and is reported back
+// as subagent_session_id so the UI's AgentCallDisplay links the card to the
+// session that actually ran the call — critical when isolateSessions is true
+// and every call has a different id.
+func (s *remoteA2AState) processResult(ctx adkagent.ToolContext, contextID string, result a2atype.SendMessageResult) (map[string]any, error) {
 	switch r := result.(type) {
 	case *a2atype.Message:
 		return map[string]any{"result": extractTextFromMessage(r)}, nil
@@ -358,7 +392,7 @@ func (s *remoteA2AState) processResult(ctx adkagent.ToolContext, result a2atype.
 			text := extractTextFromTask(r)
 			ret := map[string]any{
 				"result":              text,
-				"subagent_session_id": s.lastContextID,
+				"subagent_session_id": contextID,
 			}
 			if usage := extractUsageFromTask(r); usage != nil {
 				ret["kagent_usage_metadata"] = usage

go/adk/pkg/tools/remote_a2a_tool_test.go

@@ -124,3 +124,34 @@ func assertSingleHeader(t *testing.T, req *a2aclient.Request, key, want string)
 		t.Errorf("%s: got %q, want %q", key, got[0], want)
 	}
 }
+
+// TestNextContextID_IsolateSessions covers the EP#2137 fix: isolated tools
+// mint a fresh context_id per call so parallel/serial calls to the same
+// sub-agent land in independent sessions, while non-isolated tools keep
+// reusing one context_id for session continuity.
+func TestNextContextID_IsolateSessions(t *testing.T) {
+	t.Run("isolated: each call gets a distinct, non-empty context_id", func(t *testing.T) {
+		s := &remoteA2AState{isolateSessions: true, lastContextID: "stable-id"}
+
+		first := s.nextContextID()
+		second := s.nextContextID()
+
+		if first == "" || second == "" {
+			t.Fatalf("expected non-empty context ids, got %q and %q", first, second)
+		}
+		if first == second {
+			t.Errorf("expected distinct context ids for isolated calls, got the same id %q twice", first)
+		}
+	})
+
+	t.Run("not isolated: every call reuses the stable lastContextID", func(t *testing.T) {
+		s := &remoteA2AState{isolateSessions: false, lastContextID: "stable-id"}
+
+		first := s.nextContextID()
+		second := s.nextContextID()
+
+		if first != "stable-id" || second != "stable-id" {
+			t.Errorf("expected both calls to reuse lastContextID %q, got %q and %q", "stable-id", first, second)
+		}
+	})
+}

go/api/adk/types.go

@@ -379,6 +379,12 @@ type RemoteAgentConfig struct {
 	Url         string            `json:"url"`
 	Headers     map[string]string `json:"headers,omitempty"`
 	Description string            `json:"description,omitempty"`
+	// IsolateSessions requests a fresh A2A context_id (and therefore a fresh
+	// sub-agent session) on every call to this remote agent, instead of the
+	// default single shared session per tool lifetime. Honored by the Go
+	// declarative runtime (go/adk/pkg/tools/remote_a2a_tool.go); accepted by
+	// the Python config model for schema parity only (python/packages/kagent-adk).
+	IsolateSessions bool `json:"isolate_sessions,omitempty"`
 }
 
 // EmbeddingConfig is the embedding model config for memory tools.

go/api/config/crd/bases/kagent.dev_agents.yaml

@@ -13222,6 +13222,26 @@ spec:
                               rule: (has(self.value) && !has(self.valueFrom)) || (!has(self.value)
                                 && has(self.valueFrom))
                           type: array
+                        isolateSessions:
+                          description: |-
+                            IsolateSessions controls per-call session isolation for Agent-type tools.
+                            Only valid when Type is Agent.
+
+                            When unset or false (default), every call this agent makes to the
+                            referenced sub-agent reuses the same A2A context_id, so all calls land
+                            in one shared sub-agent session (session continuity for stateful
+                            sub-agents).
+
+                            When true, each call mints a fresh context_id, so every invocation runs
+                            in its own isolated sub-agent session. This is required for parallel
+                            fan-out to a sub-agent: without it, N parallel calls in one turn
+                            collapse into a single shared sub-agent session instead of N
+                            independent ones.
+
+                            Cross-turn/conversation continuity for stateful sub-agents does not
+                            depend on this flag; it rides the x-kagent-root-context-id header,
+                            which stays stable regardless of IsolateSessions.
+                          type: boolean
                         mcpServer:
                           properties:
                             allowedHeaders:
@@ -13292,6 +13312,8 @@ spec:
                         rule: '!(has(self.agent) && self.type != ''Agent'')'
                       - message: type.agent must be specified for Agent filter.type
                         rule: '!(!has(self.agent) && self.type == ''Agent'')'
+                      - message: isolateSessions can only be set when type is Agent
+                        rule: '!(has(self.isolateSessions) && self.type != ''Agent'')'
                     maxItems: 20
                     type: array
                 type: object

go/api/config/crd/bases/kagent.dev_sandboxagents.yaml

@@ -10879,6 +10879,26 @@ spec:
                               rule: (has(self.value) && !has(self.valueFrom)) || (!has(self.value)
                                 && has(self.valueFrom))
                           type: array
+                        isolateSessions:
+                          description: |-
+                            IsolateSessions controls per-call session isolation for Agent-type tools.
+                            Only valid when Type is Agent.
+
+                            When unset or false (default), every call this agent makes to the
+                            referenced sub-agent reuses the same A2A context_id, so all calls land
+                            in one shared sub-agent session (session continuity for stateful
+                            sub-agents).
+
+                            When true, each call mints a fresh context_id, so every invocation runs
+                            in its own isolated sub-agent session. This is required for parallel
+                            fan-out to a sub-agent: without it, N parallel calls in one turn
+                            collapse into a single shared sub-agent session instead of N
+                            independent ones.
+
+                            Cross-turn/conversation continuity for stateful sub-agents does not
+                            depend on this flag; it rides the x-kagent-root-context-id header,
+                            which stays stable regardless of IsolateSessions.
+                          type: boolean
                         mcpServer:
                           properties:
                             allowedHeaders:
@@ -10949,6 +10969,8 @@ spec:
                         rule: '!(has(self.agent) && self.type != ''Agent'')'
                       - message: type.agent must be specified for Agent filter.type
                         rule: '!(!has(self.agent) && self.type == ''Agent'')'
+                      - message: isolateSessions can only be set when type is Agent
+                        rule: '!(has(self.isolateSessions) && self.type != ''Agent'')'
                     maxItems: 20
                     type: array
                 type: object

go/api/v1alpha2/agent_types.go

@@ -474,6 +474,7 @@ const (
 // +kubebuilder:validation:XValidation:message="type.mcpServer must be specified for McpServer filter.type",rule="!(!has(self.mcpServer) && self.type == 'McpServer')"
 // +kubebuilder:validation:XValidation:message="type.agent must be nil if the type is not Agent",rule="!(has(self.agent) && self.type != 'Agent')"
 // +kubebuilder:validation:XValidation:message="type.agent must be specified for Agent filter.type",rule="!(!has(self.agent) && self.type == 'Agent')"
+// +kubebuilder:validation:XValidation:message="isolateSessions can only be set when type is Agent",rule="!(has(self.isolateSessions) && self.type != 'Agent')"
 type Tool struct {
 	// +optional
 	Type ToolProviderType `json:"type,omitempty"`
@@ -482,6 +483,26 @@ type Tool struct {
 	// +optional
 	Agent *TypedReference `json:"agent,omitempty"`
 
+	// IsolateSessions controls per-call session isolation for Agent-type tools.
+	// Only valid when Type is Agent.
+	//
+	// When unset or false (default), every call this agent makes to the
+	// referenced sub-agent reuses the same A2A context_id, so all calls land
+	// in one shared sub-agent session (session continuity for stateful
+	// sub-agents).
+	//
+	// When true, each call mints a fresh context_id, so every invocation runs
+	// in its own isolated sub-agent session. This is required for parallel
+	// fan-out to a sub-agent: without it, N parallel calls in one turn
+	// collapse into a single shared sub-agent session instead of N
+	// independent ones.
+	//
+	// Cross-turn/conversation continuity for stateful sub-agents does not
+	// depend on this flag; it rides the x-kagent-root-context-id header,
+	// which stays stable regardless of IsolateSessions.
+	// +optional
+	IsolateSessions *bool `json:"isolateSessions,omitempty"`
+
 	// HeadersFrom specifies a list of configuration values to be added as
 	// headers to requests sent to the Tool from this agent. The value of
 	// each header is resolved from either a Secret or ConfigMap in the same

go/core/internal/controller/translator/agent/compiler.go

@@ -350,10 +350,11 @@ func (a *adkApiTranslator) translateInlineAgent(ctx context.Context, agent v1alp
 				}
 
 				cfg.RemoteAgents = append(cfg.RemoteAgents, adk.RemoteAgentConfig{
-					Name:        utils.ConvertToPythonIdentifier(utils.GetObjectRef(toolAgent)),
-					Url:         targetURL,
-					Headers:     headers,
-					Description: toolSpec.Description,
+					Name:            utils.ConvertToPythonIdentifier(utils.GetObjectRef(toolAgent)),
+					Url:             targetURL,
+					Headers:         headers,
+					Description:     toolSpec.Description,
+					IsolateSessions: tool.IsolateSessions != nil && *tool.IsolateSessions,
 				})
 			default:
 				return nil, nil, nil, fmt.Errorf("unknown agent type: %s", toolSpec.Type)

go/core/internal/controller/translator/agent/testdata/inputs/agent_with_isolated_session_tool.yaml

@@ -0,0 +1,49 @@
+operation: translateAgent
+targetObject: coordinator-agent
+namespace: test
+objects:
+  - apiVersion: v1
+    kind: Secret
+    metadata:
+      name: openai-secret
+      namespace: test
+    data:
+      api-key: c2stdGVzdC1hcGkta2V5  # base64 encoded "sk-test-api-key"
+  - apiVersion: kagent.dev/v1alpha2
+    kind: ModelConfig
+    metadata:
+      name: nested-model
+      namespace: test
+    spec:
+      provider: OpenAI
+      model: gpt-4o
+      apiKeySecret: openai-secret
+      apiKeySecretKey: api-key
+  - apiVersion: kagent.dev/v1alpha2
+    kind: Agent
+    metadata:
+      name: worker-agent
+      namespace: test
+    spec:
+      type: Declarative
+      declarative:
+        description: A worker agent that can be called in parallel by the coordinator
+        systemMessage: You are a worker agent. Complete the assigned task and report back.
+        modelConfig: nested-model
+        tools: []
+  - apiVersion: kagent.dev/v1alpha2
+    kind: Agent
+    metadata:
+      name: coordinator-agent
+      namespace: test
+    spec:
+      type: Declarative
+      declarative:
+        description: A coordinator agent that fans out isolated parallel calls to a worker
+        systemMessage: You are a coordinating agent that delegates tasks to a worker agent, potentially in parallel.
+        modelConfig: nested-model
+        tools:
+          - type: Agent
+            agent:
+              name: worker-agent
+            isolateSessions: true