From 4ecfca86836d8a58a1a819d9415b69dd01c54a39 Mon Sep 17 00:00:00 2001 From: yashraj S Date: Fri, 31 Jul 2026 08:35:24 +0530 Subject: [PATCH] fix(go-adk): add per-call session isolation for Agent tools (#2153) ## Summary Fixes #2137 The bug : every call a parent agent makes to a sub-agent (Agent tool) reuses the same A2A **context_id** since it's minted once when the tool is built. worker just uses that **context_id** as its session id directly ( **executor.go: sessionID := reqCtx.ContextID** ) so if a coordinator fires off N parallel calls to the same sub-agent in one turn , they all pile into a single shared session instead of getting their own. Fix is an opt-in **IsolateSessions** flag on Agent-type tools : ```yaml spec: declarative: tools: - type: Agent agent: name: worker isolateSessions: true # each call to worker gets its own session ``` Default behavior (flag unset/false) doesn't change one **context_id** for the tool's whole lifetime so stateful sub-agents keep their session continuity. with the flag on we mint a fresh **context_id** on every call so each invocation is isolated. Doesn't touch the **x-kagent-root-context-id** header stuff. that's still what carries cross-turn continuity and it stays stable either way. ## What changed - `agent_types.go`: added `Tool.IsolateSessions *bool` plus a CEL rule so it can only be set when `type: Agent` - `adk/types.go`: `RemoteAgentConfig.IsolateSessions bool` - `compiler.go`: passes the flag through into `RemoteAgentConfig` - `agent.go`: forwards the flag to `NewKAgentRemoteA2ATool`. also skips adding an entry to `subagentSessionIDs` when isolated since there's no single session id to pre-stamp function_call parts with anymore - `remote_a2a_tool.go`: added `isolateSessions` plus a small `nextContextID()` helper that either returns the stable id or mints a new one. This gets reported back as `subagent_session_id` - Python `types.py`: added `isolate_sessions` for schema parity only, it doesn't do anything on that side yet (matches what the issue scoped out) - Regenerated CRDs/deepcopy with `make controller-manifests` - Added a test for the new context-id logic and a golden fixture (`agent_with_isolated_session_tool`) to check it flows through the whole translation pipeline ## About the UI Isolated tools don't have one fixed session id, so the executor can't pre-populate the stamp map for them at startup. Instead the UI grabs the session id per call from `subagent_session_id` in the function_response. `AgentCallDisplay` already reads that field, so nothing new needed there. ## Not doing in this PR - Not a concurrency limiter, that's separate (`max-concurrency.md`) - HITL resume is unaffected, it still uses the context_id from the confirmation payload - Only the Go runtime respects this flag right now. Python accepts it for config parity but the low-level tool doesn't use it yet ## Tests Ran `go test ./adk/pkg/tools/... ./adk/pkg/agent/... ./api/...` and the golden translator tests. Everything passes. Confirmed `isolate_sessions: true` shows up correctly in the generated `config.json` for the new fixture. ## One small thing I noticed In `handleResume`, `processResult` now sets `subagent_session_id` from the contextID I pass in, but there's older code right after it that sets the same key again with a fallback to `lastContextID`. Not wrong, just a bit redundant now since it writes the same value twice in the normal case. Left it as is since it's harmless but flagging it in case someone wants it cleaned up. --------- Signed-off-by: Yashraj Shukla Co-authored-by: Eitan Yarmush Signed-off-by: Jet Chiang --- go/adk/cmd/main.go | 13 +- go/adk/examples/oneshot/main.go | 2 +- go/adk/pkg/a2a/converter.go | 38 --- go/adk/pkg/a2a/converter_test.go | 53 --- go/adk/pkg/a2a/executor.go | 46 ++- go/adk/pkg/agent/agent.go | 34 +- go/adk/pkg/agent/createllm_test.go | 2 +- go/adk/pkg/runner/adapter.go | 20 +- go/adk/pkg/tools/remote_a2a_tool.go | 176 ++++++---- go/adk/pkg/tools/remote_a2a_tool_test.go | 81 +++++ go/api/adk/types.go | 6 + .../config/crd/bases/kagent.dev_agents.yaml | 22 ++ .../crd/bases/kagent.dev_sandboxagents.yaml | 22 ++ go/api/v1alpha2/agent_types.go | 21 ++ go/api/v1alpha2/zz_generated.deepcopy.go | 5 + .../controller/translator/agent/compiler.go | 9 +- .../agent_with_isolated_session_tool.yaml | 49 +++ .../agent_with_isolated_session_tool.json | 301 ++++++++++++++++++ .../templates/kagent.dev_agents.yaml | 22 ++ .../templates/kagent.dev_sandboxagents.yaml | 22 ++ .../kagent-adk/src/kagent/adk/types.py | 5 + ui/src/lib/messageHandlers.ts | 22 +- 22 files changed, 747 insertions(+), 224 deletions(-) create mode 100644 go/core/internal/controller/translator/agent/testdata/inputs/agent_with_isolated_session_tool.yaml create mode 100644 go/core/internal/controller/translator/agent/testdata/outputs/agent_with_isolated_session_tool.json diff --git a/go/adk/cmd/main.go b/go/adk/cmd/main.go index 10883651ec..9627d9ed58 100644 --- a/go/adk/cmd/main.go +++ b/go/adk/cmd/main.go @@ -186,7 +186,7 @@ func main() { logger.Info("Memory service enabled", "appName", appName) } - runnerConfig, subagentSessionIDs, err := runnerpkg.CreateRunnerConfig(ctx, agentConfig, sessionService, appName, memoryService, kagentURL, httpClient) + runnerConfig, err := runnerpkg.CreateRunnerConfig(ctx, agentConfig, sessionService, appName, memoryService, kagentURL, httpClient) if err != nil { logger.Error(err, "Failed to create Google ADK Runner config") os.Exit(1) @@ -194,12 +194,11 @@ func main() { stream := agentConfig.GetStream() executor := a2a.NewKAgentExecutor(a2a.KAgentExecutorConfig{ - RunnerConfig: runnerConfig, - SubagentSessionIDs: subagentSessionIDs, - SessionService: sessionService, - Stream: stream, - AppName: appName, - Logger: logger, + RunnerConfig: runnerConfig, + SessionService: sessionService, + Stream: stream, + AppName: appName, + Logger: logger, }) // Build the agent card. diff --git a/go/adk/examples/oneshot/main.go b/go/adk/examples/oneshot/main.go index 7dc6642058..d6c4abdb36 100644 --- a/go/adk/examples/oneshot/main.go +++ b/go/adk/examples/oneshot/main.go @@ -66,7 +66,7 @@ func main() { agentConfig.Stream = &t } - adkAgent, err := agent.CreateGoogleADKAgent(ctx, agentConfig, "oneshot") + adkAgent, err := agent.CreateGoogleADKAgent(ctx, agentConfig, "oneshot", nil) if err != nil { fmt.Fprintf(os.Stderr, "error creating agent: %v\n", err) os.Exit(1) diff --git a/go/adk/pkg/a2a/converter.go b/go/adk/pkg/a2a/converter.go index ca45a7094c..fe4aca167c 100644 --- a/go/adk/pkg/a2a/converter.go +++ b/go/adk/pkg/a2a/converter.go @@ -114,44 +114,6 @@ func convertDataPartToGenAI(p *a2atype.DataPart, typeKey string) (*genai.Part, e return adka2a.ToGenAIPart(p) } -// stampSubagentSessionID adds kagent_subagent_session_id to function_call -// DataParts when the tool name is present in subagentSessionIDs. -// Part can be either a *a2atype.DataPart or a2atype.DataPart. -func stampSubagentSessionID(part a2atype.Part, subagentSessionIDs map[string]string) a2atype.Part { - switch p := part.(type) { - case *a2atype.DataPart: - cp := *p - stampSubagentSessionIDOnDataPart(&cp, subagentSessionIDs) - return cp - case a2atype.DataPart: - cp := p - stampSubagentSessionIDOnDataPart(&cp, subagentSessionIDs) - return cp - default: - return part - } -} - -func stampSubagentSessionIDOnDataPart(dp *a2atype.DataPart, subagentSessionIDs map[string]string) { - if dp == nil || len(subagentSessionIDs) == 0 { - return - } - if dp.Metadata == nil { - dp.Metadata = map[string]any{} - } - partType, _ := ReadMetadataValue(dp.Metadata, A2ADataPartMetadataTypeKey) - if partType != A2ADataPartMetadataTypeFunctionCall { - return - } - toolName, _ := dp.Data[PartKeyName].(string) - if toolName == "" { - return - } - if sessionID, ok := subagentSessionIDs[toolName]; ok && sessionID != "" { - dp.Metadata[GetKAgentMetadataKey("subagent_session_id")] = sessionID - } -} - // toA2AMetadataMap converts v to map[string]any via JSON so values placed in A2A func toA2AMetadataMap(v any) (map[string]any, error) { if v == nil { diff --git a/go/adk/pkg/a2a/converter_test.go b/go/adk/pkg/a2a/converter_test.go index c9ec15a908..95fc2e3834 100644 --- a/go/adk/pkg/a2a/converter_test.go +++ b/go/adk/pkg/a2a/converter_test.go @@ -199,59 +199,6 @@ func TestMessageToGenAIContent_NilMessage(t *testing.T) { } } -// --------------------------------------------------------------------------- -// stampSubagentSessionID -// --------------------------------------------------------------------------- - -func TestStampSubagentSessionID_FunctionCallPart(t *testing.T) { - subagentIDs := map[string]string{"k8s_agent": "session-abc"} - - dp := &a2atype.DataPart{ - Data: map[string]any{ - PartKeyName: "k8s_agent", - PartKeyArgs: map[string]any{"request": "list pods"}, - }, - Metadata: map[string]any{ - adka2a.ToA2AMetaKey("type"): A2ADataPartMetadataTypeFunctionCall, - }, - } - updated := stampSubagentSessionID(dp, subagentIDs) - updatedDP, ok := updated.(a2atype.DataPart) - if !ok { - t.Fatalf("updated part type = %T, want a2atype.DataPart", updated) - } - - sessionID, has := updatedDP.Metadata[GetKAgentMetadataKey("subagent_session_id")] - if !has { - t.Fatal("expected kagent_subagent_session_id in metadata, not found") - } - if sessionID != "session-abc" { - t.Errorf("session_id = %q, want session-abc", sessionID) - } -} - -func TestStampSubagentSessionID_UnknownTool(t *testing.T) { - subagentIDs := map[string]string{"k8s_agent": "session-abc"} - - dp := &a2atype.DataPart{ - Data: map[string]any{ - PartKeyName: "unknown_tool", - }, - Metadata: map[string]any{ - adka2a.ToA2AMetaKey("type"): A2ADataPartMetadataTypeFunctionCall, - }, - } - updated := stampSubagentSessionID(dp, subagentIDs) - updatedDP, ok := updated.(a2atype.DataPart) - if !ok { - t.Fatalf("updated part type = %T, want a2atype.DataPart", updated) - } - - if _, ok := updatedDP.Metadata[GetKAgentMetadataKey("subagent_session_id")]; ok { - t.Error("expected no subagent_session_id for unknown tool") - } -} - // --------------------------------------------------------------------------- // toA2AMetadataMap // --------------------------------------------------------------------------- diff --git a/go/adk/pkg/a2a/executor.go b/go/adk/pkg/a2a/executor.go index f4ba0ab5d7..2c091cbc6e 100644 --- a/go/adk/pkg/a2a/executor.go +++ b/go/adk/pkg/a2a/executor.go @@ -30,24 +30,22 @@ const ( // KAgentExecutorConfig holds the configuration for KAgentExecutor type KAgentExecutorConfig struct { - RunnerConfig runner.Config - SubagentSessionIDs map[string]string - SessionService adksession.Service - Stream bool - AppName string - SkillsDirectory string - Logger logr.Logger + RunnerConfig runner.Config + SessionService adksession.Service + Stream bool + AppName string + SkillsDirectory string + Logger logr.Logger } // KAgentExecutor implements a2asrv.AgentExecutor type KAgentExecutor struct { - runnerConfig runner.Config - subagentSessionIDs map[string]string - sessionService adksession.Service - stream bool - appName string - skillsDirectory string - logger logr.Logger + runnerConfig runner.Config + sessionService adksession.Service + stream bool + appName string + skillsDirectory string + logger logr.Logger } var _ a2asrv.AgentExecutor = (*KAgentExecutor)(nil) @@ -62,13 +60,12 @@ func NewKAgentExecutor(cfg KAgentExecutorConfig) *KAgentExecutor { skillsDir = defaultSkillsDirectory } return &KAgentExecutor{ - runnerConfig: cfg.RunnerConfig, - subagentSessionIDs: cfg.SubagentSessionIDs, - sessionService: cfg.SessionService, - stream: cfg.Stream, - appName: cfg.AppName, - skillsDirectory: skillsDir, - logger: cfg.Logger.WithName("kagent-executor"), + runnerConfig: cfg.RunnerConfig, + sessionService: cfg.SessionService, + stream: cfg.Stream, + appName: cfg.AppName, + skillsDirectory: skillsDir, + logger: cfg.Logger.WithName("kagent-executor"), } } @@ -218,9 +215,6 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestCont return fmt.Errorf("inbound message conversion failed: %w", err) } - // 7. Use pre-built subagent session ID map (built by runner bundle). - subagentSessionIDs := e.subagentSessionIDs - // 8. Create runner. r, err := runner.New(e.runnerConfig) if err != nil { @@ -328,10 +322,6 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestCont if isEmptyDataPart(a2aPart) { continue } - // Stamp kagent_subagent_session_id onto function_call DataParts. - if len(subagentSessionIDs) > 0 { - a2aPart = stampSubagentSessionID(a2aPart, subagentSessionIDs) - } a2aParts = append(a2aParts, a2aPart) } diff --git a/go/adk/pkg/agent/agent.go b/go/adk/pkg/agent/agent.go index b956c6026e..c6c78f1e5a 100644 --- a/go/adk/pkg/agent/agent.go +++ b/go/adk/pkg/agent/agent.go @@ -33,21 +33,13 @@ const ( // CreateGoogleADKAgent creates a Google ADK agent from AgentConfig. // agentName is used as the ADK agent identity (appears in event Author field). // extraTools are appended to the agent's tool list (e.g. save_memory). -func CreateGoogleADKAgent(ctx context.Context, agentConfig *adk.AgentConfig, agentName string, extraTools ...tool.Tool) (agent.Agent, error) { - a, _, err := CreateGoogleADKAgentWithSubagentSessionIDs(ctx, agentConfig, agentName, nil, extraTools...) - return a, err -} - -// CreateGoogleADKAgentWithSubagentSessionIDs creates a Google ADK agent and a -// map of remote-subagent tool name → A2A context session ID (for stamping -// outbound A2A events). Callers that only need the agent can use -// CreateGoogleADKAgent. -// Optional stsPlugin can be provided for token propagation to MCP tools. -func CreateGoogleADKAgentWithSubagentSessionIDs(ctx context.Context, agentConfig *adk.AgentConfig, agentName string, stsPlugin *sts.TokenPropagationPlugin, extraTools ...tool.Tool) (agent.Agent, map[string]string, error) { +// Optional stsPlugin can be provided for token propagation to MCP tools; pass +// nil if token propagation is not needed. +func CreateGoogleADKAgent(ctx context.Context, agentConfig *adk.AgentConfig, agentName string, stsPlugin *sts.TokenPropagationPlugin, extraTools ...tool.Tool) (agent.Agent, error) { log := logr.FromContextOrDiscard(ctx) if agentConfig == nil { - return nil, nil, fmt.Errorf("agent config is required") + return nil, fmt.Errorf("agent config is required") } propagateToken := strings.ToLower(os.Getenv("KAGENT_PROPAGATE_TOKEN")) == "true" @@ -57,7 +49,6 @@ func CreateGoogleADKAgentWithSubagentSessionIDs(ctx context.Context, agentConfig } toolsets := mcp.CreateToolsets(ctx, agentConfig.HttpTools, agentConfig.SseTools, propagateToken, dynamicHeaderProvider) mcpAppToolNames := mcp.MCPAppToolNamesFromToolsets(toolsets) - subagentSessionIDs := make(map[string]string) var remoteAgentTools []tool.Tool for _, remoteAgent := range agentConfig.RemoteAgents { @@ -65,12 +56,9 @@ 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, 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) - } - if sessionID != "" { - subagentSessionIDs[remoteAgent.Name] = sessionID + return nil, fmt.Errorf("failed to create remote A2A tool for %s: %w", remoteAgent.Name, err) } remoteAgentTools = append(remoteAgentTools, remoteTool) log.Info("Wired remote A2A agent tool", "name", remoteAgent.Name, "url", remoteAgent.Url) @@ -78,16 +66,16 @@ func CreateGoogleADKAgentWithSubagentSessionIDs(ctx context.Context, agentConfig localTools, err := buildAgentTools(agentConfig, remoteAgentTools, extraTools, log) if err != nil { - return nil, nil, err + return nil, err } if agentConfig.Model == nil { - return nil, nil, fmt.Errorf("model configuration is required") + return nil, fmt.Errorf("model configuration is required") } llmModel, err := CreateLLM(ctx, agentConfig.Model, log) if err != nil { - return nil, nil, fmt.Errorf("failed to create LLM: %w", err) + return nil, fmt.Errorf("failed to create LLM: %w", err) } if agentName == "" { @@ -153,14 +141,14 @@ func CreateGoogleADKAgentWithSubagentSessionIDs(ctx context.Context, agentConfig llmAgent, err := llmagent.New(llmAgentConfig) if err != nil { - return nil, nil, fmt.Errorf("failed to create LLM agent: %w", err) + return nil, fmt.Errorf("failed to create LLM agent: %w", err) } log.Info("Successfully created Google ADK LLM agent", "toolsCount", len(llmAgentConfig.Tools), "toolsetsCount", len(llmAgentConfig.Toolsets)) - return llmAgent, subagentSessionIDs, nil + return llmAgent, nil } func buildAgentTools(agentConfig *adk.AgentConfig, remoteAgentTools, extraTools []tool.Tool, log logr.Logger) ([]tool.Tool, error) { diff --git a/go/adk/pkg/agent/createllm_test.go b/go/adk/pkg/agent/createllm_test.go index 1722b2a9fe..da758e4d98 100644 --- a/go/adk/pkg/agent/createllm_test.go +++ b/go/adk/pkg/agent/createllm_test.go @@ -52,7 +52,7 @@ func runAgent(t *testing.T, agentCfg *adk.AgentConfig, prompt string) string { t.Helper() ctx := logr.NewContext(t.Context(), logr.Discard()) - adkAgent, err := CreateGoogleADKAgent(ctx, agentCfg, "test-agent") + adkAgent, err := CreateGoogleADKAgent(ctx, agentCfg, "test-agent", nil) require.NoError(t, err) sessionService := adksession.InMemoryService() diff --git a/go/adk/pkg/runner/adapter.go b/go/adk/pkg/runner/adapter.go index d40d07b265..c7bd7fc6c2 100644 --- a/go/adk/pkg/runner/adapter.go +++ b/go/adk/pkg/runner/adapter.go @@ -37,14 +37,14 @@ func CreateRunnerConfig( memoryService *kagentmemory.KagentMemoryService, kagentURL string, httpClient *http.Client, -) (runner.Config, map[string]string, error) { +) (runner.Config, error) { log := logr.FromContextOrDiscard(ctx) var extraTools []adktool.Tool if memoryService != nil { saveTool, err := kagentmemory.NewSaveMemoryTool(memoryService) if err != nil { - return runner.Config{}, nil, fmt.Errorf("failed to create save_memory tool: %w", err) + return runner.Config{}, fmt.Errorf("failed to create save_memory tool: %w", err) } extraTools = append(extraTools, saveTool) } @@ -52,15 +52,15 @@ func CreateRunnerConfig( if agentConfig.ShareTools != nil && *agentConfig.ShareTools && kagentURL != "" && httpClient != nil { createTool, err := tools.NewCreateShareLinkTool(httpClient, kagentURL, appName) if err != nil { - return runner.Config{}, nil, fmt.Errorf("failed to create create_share_link tool: %w", err) + return runner.Config{}, fmt.Errorf("failed to create create_share_link tool: %w", err) } listTool, err := tools.NewListShareLinksTool(httpClient, kagentURL, appName) if err != nil { - return runner.Config{}, nil, fmt.Errorf("failed to create list_share_links tool: %w", err) + return runner.Config{}, fmt.Errorf("failed to create list_share_links tool: %w", err) } deleteTool, err := tools.NewDeleteShareLinkTool(httpClient, kagentURL, appName) if err != nil { - return runner.Config{}, nil, fmt.Errorf("failed to create delete_share_link tool: %w", err) + return runner.Config{}, fmt.Errorf("failed to create delete_share_link tool: %w", err) } extraTools = append(extraTools, createTool, listTool, deleteTool) log.Info("Share link tools enabled") @@ -68,12 +68,12 @@ func CreateRunnerConfig( stsPlugin, err := buildTokenPropagationPlugin(ctx, log) if err != nil { - return runner.Config{}, nil, err + return runner.Config{}, err } - adkAgent, subagentSessionIDs, err := agent.CreateGoogleADKAgentWithSubagentSessionIDs(ctx, agentConfig, agentNameFromAppName(appName), stsPlugin, extraTools...) + adkAgent, err := agent.CreateGoogleADKAgent(ctx, agentConfig, agentNameFromAppName(appName), stsPlugin, extraTools...) if err != nil { - return runner.Config{}, nil, fmt.Errorf("failed to create agent: %w", err) + return runner.Config{}, fmt.Errorf("failed to create agent: %w", err) } adkSessionService := sessionService @@ -94,7 +94,7 @@ func CreateRunnerConfig( if stsPlugin != nil { p, err := stsPlugin.ADKPlugin() if err != nil { - return runner.Config{}, nil, fmt.Errorf("failed to create STS ADK plugin: %w", err) + return runner.Config{}, fmt.Errorf("failed to create STS ADK plugin: %w", err) } if p != nil { adkPlugins = append(adkPlugins, p) @@ -111,7 +111,7 @@ func CreateRunnerConfig( }, } - return cfg, subagentSessionIDs, nil + return cfg, nil } func buildTokenPropagationPlugin(ctx context.Context, log logr.Logger) (*sts.TokenPropagationPlugin, error) { diff --git a/go/adk/pkg/tools/remote_a2a_tool.go b/go/adk/pkg/tools/remote_a2a_tool.go index 705c5c9a68..81e1c202a6 100644 --- a/go/adk/pkg/tools/remote_a2a_tool.go +++ b/go/adk/pkg/tools/remote_a2a_tool.go @@ -149,41 +149,89 @@ type remoteA2AState struct { initOnce sync.Once initErr error - lastContextID string + // sharedContextID 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. Unused when isolateSessions is true — each + // call mints its own id instead (see contextIDForCall). + sharedContextID string + + // isolateSessions mints a fresh context_id per call (see contextIDForCall) + // instead of reusing sharedContextID, 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 +// remoteA2AResponse is the typed return value for every remote A2A tool +// invocation. Using one shared struct (instead of ad-hoc map[string]any +// literals per branch) means every response path — success, input_required, +// and failure — carries the same fields, so a field like SubagentSessionID +// can't be silently forgotten in one branch while present in another. +// functiontool.New infers the tool's output schema from this type. +type remoteA2AResponse struct { + Result string `json:"result,omitempty"` + Error string `json:"error,omitempty"` + Status string `json:"status,omitempty"` + WaitingFor string `json:"waiting_for,omitempty"` + Subagent string `json:"subagent,omitempty"` + SubagentSessionID string `json:"subagent_session_id,omitempty"` + KAgentUsageMetadata map[string]any `json:"kagent_usage_metadata,omitempty"` +} + +// NewKAgentRemoteA2ATool creates a function tool that calls a remote A2A agent +// and propagates HITL state. +// +// It intentionally does not return a session id: the constructor-time +// context_id used to be pre-stamped onto outbound function_call events so +// the UI could link the Activity panel before a response arrived, but that +// model only works when a tool has exactly one session for its whole +// lifetime — it breaks down for isolateSessions, where the real session is +// per invocation, not per tool instance. Every call now reports its own +// actual context_id back as SubagentSessionID in the tool's response (see +// remoteA2AResponse), which is the single source of truth the UI reads from +// for both isolated and non-isolated tools alike. // // 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, 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, + sharedContextID: a2atype.NewContextID(), + isolateSessions: isolateSessions, } ft, err := functiontool.New(functiontool.Config{ Name: name, Description: description, - }, func(ctx adkagent.Context, in remoteA2AInput) (map[string]any, error) { + }, func(ctx adkagent.Context, in remoteA2AInput) (remoteA2AResponse, error) { return state.run(ctx, in.Request) }) if err != nil { - return nil, "", fmt.Errorf("failed to create remote A2A function tool for %s: %w", name, err) + return nil, fmt.Errorf("failed to create remote A2A function tool for %s: %w", name, err) + } + return ft, nil +} + +// contextIDForCall 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 sharedContextID otherwise (shared session +// for the lifetime of the tool). +func (s *remoteA2AState) contextIDForCall() string { + if s.isolateSessions { + return a2atype.NewContextID() } - return ft, state.lastContextID, nil + return s.sharedContextID } // ensureClient lazily resolves the agent card and initialises the A2A client. @@ -239,7 +287,7 @@ func (s *remoteA2AState) ensureClient(ctx context.Context) (*a2aclient.Client, e } // run dispatches to handleResume or handleFirstCall based on ToolConfirmation presence. -func (s *remoteA2AState) run(ctx adkagent.Context, requestText string) (map[string]any, error) { +func (s *remoteA2AState) run(ctx adkagent.Context, requestText string) (remoteA2AResponse, error) { if ctx.ToolConfirmation() != nil { return s.handleResume(ctx) } @@ -247,35 +295,36 @@ func (s *remoteA2AState) run(ctx adkagent.Context, requestText string) (map[stri } // handleFirstCall is Phase 1: send the request to the remote agent. -func (s *remoteA2AState) handleFirstCall(ctx adkagent.Context, requestText string) (map[string]any, error) { +func (s *remoteA2AState) handleFirstCall(ctx adkagent.Context, requestText string) (remoteA2AResponse, error) { if requestText == "" { - return map[string]any{"error": "missing or empty 'request' argument"}, nil + return remoteA2AResponse{Error: "missing or empty 'request' argument"}, nil } client, err := s.ensureClient(ctx) if err != nil { - return map[string]any{"error": err.Error()}, nil + return remoteA2AResponse{Error: err.Error()}, nil } + contextID := s.contextIDForCall() 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()) result, err := client.SendMessage(sendCtx, &a2atype.MessageSendParams{Message: message}) if err != nil { slog.Error("Remote agent request failed", "tool", s.name, "error", err) - return map[string]any{"error": fmt.Sprintf("Remote agent '%s' request failed: %v", s.name, err)}, nil + return remoteA2AResponse{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. -func (s *remoteA2AState) handleResume(ctx adkagent.Context) (map[string]any, error) { +func (s *remoteA2AState) handleResume(ctx adkagent.Context) (remoteA2AResponse, error) { confirmation := ctx.ToolConfirmation() payload, _ := confirmation.Payload.(map[string]any) hitlPayload := a2a.ParseHitlConfirmationPayload(payload) @@ -289,7 +338,7 @@ func (s *remoteA2AState) handleResume(ctx adkagent.Context) (map[string]any, err if taskID == "" { slog.Error("Resume for remote agent but no task_id in confirmation payload", "tool", s.name) - return map[string]any{"error": fmt.Sprintf("Cannot resume remote agent '%s': missing task context.", subagentName)}, nil + return remoteA2AResponse{Error: fmt.Sprintf("Cannot resume remote agent '%s': missing task context.", subagentName)}, nil } decisionData := buildDecisionData(confirmation.Confirmed, hitlPayload) @@ -311,7 +360,7 @@ func (s *remoteA2AState) handleResume(ctx adkagent.Context) (map[string]any, err client, err := s.ensureClient(ctx) if err != nil { - return map[string]any{"error": err.Error()}, nil + return remoteA2AResponse{Error: err.Error()}, nil } sendCtx := context.WithValue(ctx, userIDContextKey{}, ctx.UserID()) @@ -319,63 +368,77 @@ func (s *remoteA2AState) handleResume(ctx adkagent.Context) (map[string]any, err result, err := client.SendMessage(sendCtx, &a2atype.MessageSendParams{Message: message}) if err != nil { slog.Error("Remote agent resume failed", "tool", subagentName, "error", err) - return map[string]any{"error": fmt.Sprintf("Remote agent '%s' resume failed: %v", subagentName, err)}, nil + return remoteA2AResponse{Error: fmt.Sprintf("Remote agent '%s' resume failed: %v", subagentName, err)}, nil } - ret, retErr := s.processResult(ctx, 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 - if retErr == nil && ret != nil { - sessionID := contextID - if sessionID == "" { - sessionID = s.lastContextID - } - ret["subagent_session_id"] = sessionID - } - return ret, retErr + // contextID here is whatever the confirmation payload carried (the + // original subagent session from the paused task). It is always the + // correct id to report — unlike before, there is no fallback to a + // construction-time id: for an isolated tool, sharedContextID was never + // actually used in any A2A call, so falling back to it would report a + // session id that doesn't correspond to any real subagent activity. + return s.processResult(ctx, contextID, result) } // processResult converts a SendMessageResult into a tool return value. -func (s *remoteA2AState) processResult(ctx adkagent.Context, result a2atype.SendMessageResult) (map[string]any, error) { +// contextID is the A2A context_id this call was sent under (from +// contextIDForCall, or the confirmation payload on resume) and is reported +// back as SubagentSessionID on every branch — success, input_required, and +// failure alike — so the UI's AgentCallDisplay can always link the card to +// the session that actually ran the call. This is the single source of +// truth the UI reads from; there is no separate constructor-time id to fall +// back on (see NewKAgentRemoteA2ATool). +func (s *remoteA2AState) processResult(ctx adkagent.Context, contextID string, result a2atype.SendMessageResult) (remoteA2AResponse, error) { switch r := result.(type) { case *a2atype.Message: - return map[string]any{"result": extractTextFromMessage(r)}, nil + return remoteA2AResponse{ + Result: extractTextFromMessage(r), + SubagentSessionID: contextID, + }, nil case *a2atype.Task: switch r.Status.State { case a2atype.TaskStateInputRequired: - return s.handleInputRequired(ctx, r), nil + return s.handleInputRequired(ctx, r, contextID), nil case a2atype.TaskStateFailed: text := extractTextFromTask(r) if text == "" { text = fmt.Sprintf("Remote agent '%s' failed.", s.name) } - return map[string]any{"error": text}, nil + return remoteA2AResponse{ + Error: text, + SubagentSessionID: contextID, + }, nil default: // completed — include sub-agent's final LLM usage from task.metadata // so the parent can display it on the AgentCall card in the UI. // Mirrors Python's _extract_usage_from_task(task). - text := extractTextFromTask(r) - ret := map[string]any{ - "result": text, - "subagent_session_id": s.lastContextID, + ret := remoteA2AResponse{ + Result: extractTextFromTask(r), + SubagentSessionID: contextID, } if usage := extractUsageFromTask(r); usage != nil { - ret["kagent_usage_metadata"] = usage + ret.KAgentUsageMetadata = usage } return ret, nil } default: - return map[string]any{"error": fmt.Sprintf("Remote agent '%s' returned no result.", s.name)}, nil + return remoteA2AResponse{ + Error: fmt.Sprintf("Remote agent '%s' returned no result.", s.name), + SubagentSessionID: contextID, + }, nil } } // handleInputRequired pauses parent agent execution via RequestConfirmation. -func (s *remoteA2AState) handleInputRequired(ctx adkagent.Context, task *a2atype.Task) map[string]any { +// contextID is reported back as SubagentSessionID so the UI can link the +// pending Activity panel to the paused subagent session before the human +// decision is forwarded and a final result comes back. +func (s *remoteA2AState) handleInputRequired(ctx adkagent.Context, task *a2atype.Task, contextID string) remoteA2AResponse { if task == nil { slog.Error("Subagent returned input_required without task", "tool", s.name) - return map[string]any{ - "error": fmt.Sprintf("Remote agent '%s' returned input_required without task context.", s.name), + return remoteA2AResponse{ + Error: fmt.Sprintf("Remote agent '%s' returned input_required without task context.", s.name), + SubagentSessionID: contextID, } } @@ -412,10 +475,11 @@ func (s *remoteA2AState) handleInputRequired(ctx adkagent.Context, task *a2atype if err := ctx.RequestConfirmation(hint, confirmPayload.ToMap()); err != nil { slog.Error("Failed to request confirmation", "tool", s.name, "error", err) } - return map[string]any{ - "status": "pending", - "waiting_for": "subagent_approval", - "subagent": s.name, + return remoteA2AResponse{ + Status: "pending", + WaitingFor: "subagent_approval", + Subagent: s.name, + SubagentSessionID: contextID, } } diff --git a/go/adk/pkg/tools/remote_a2a_tool_test.go b/go/adk/pkg/tools/remote_a2a_tool_test.go index 14bcd8f3a7..fa94137f2c 100644 --- a/go/adk/pkg/tools/remote_a2a_tool_test.go +++ b/go/adk/pkg/tools/remote_a2a_tool_test.go @@ -4,6 +4,7 @@ import ( "context" "testing" + a2atype "github.com/a2aproject/a2a-go/a2a" "github.com/a2aproject/a2a-go/a2aclient" "github.com/a2aproject/a2a-go/a2asrv" ) @@ -124,3 +125,83 @@ func assertSingleHeader(t *testing.T, req *a2aclient.Request, key, want string) t.Errorf("%s: got %q, want %q", key, got[0], want) } } + +// TestContextIDForCall_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 TestContextIDForCall_IsolateSessions(t *testing.T) { + t.Run("isolated: each call gets a distinct, non-empty context_id", func(t *testing.T) { + s := &remoteA2AState{isolateSessions: true, sharedContextID: "stable-id"} + + first := s.contextIDForCall() + second := s.contextIDForCall() + + 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 sharedContextID", func(t *testing.T) { + s := &remoteA2AState{isolateSessions: false, sharedContextID: "stable-id"} + + first := s.contextIDForCall() + second := s.contextIDForCall() + + if first != "stable-id" || second != "stable-id" { + t.Errorf("expected both calls to reuse sharedContextID %q, got %q and %q", "stable-id", first, second) + } + }) +} + +// TestProcessResult_SetsSubagentSessionIDOnEveryBranch covers the review +// feedback on #2153: subagent_session_id must be present in the response for +// every result shape (direct Message, completed Task, input_required Task, +// failed Task, and the unrecognised-result fallback) — not just the +// completed-Task branch — since it is the UI's only source of truth for +// linking the AgentCallDisplay Activity panel to the correct subagent +// session, especially when isolateSessions means every call has a distinct id. +func TestProcessResult_SetsSubagentSessionIDOnEveryBranch(t *testing.T) { + const contextID = "call-specific-context-id" + s := &remoteA2AState{name: "worker"} + ctx := context.Background() + + t.Run("direct Message result", func(t *testing.T) { + msg := &a2atype.Message{Parts: a2atype.ContentParts{a2atype.TextPart{Text: "hi"}}} + resp, err := s.processResult(nil, contextID, msg) + _ = ctx + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if resp.SubagentSessionID != contextID { + t.Errorf("SubagentSessionID = %q, want %q", resp.SubagentSessionID, contextID) + } + }) + + t.Run("failed Task result", func(t *testing.T) { + task := &a2atype.Task{Status: a2atype.TaskStatus{State: a2atype.TaskStateFailed}} + resp, err := s.processResult(nil, contextID, task) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if resp.SubagentSessionID != contextID { + t.Errorf("SubagentSessionID = %q, want %q", resp.SubagentSessionID, contextID) + } + if resp.Error == "" { + t.Errorf("expected a non-empty Error for a failed task") + } + }) + + t.Run("unrecognised result type", func(t *testing.T) { + resp, err := s.processResult(nil, contextID, nil) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if resp.SubagentSessionID != contextID { + t.Errorf("SubagentSessionID = %q, want %q", resp.SubagentSessionID, contextID) + } + }) +} diff --git a/go/api/adk/types.go b/go/api/adk/types.go index d2740c5974..e14a52879c 100644 --- a/go/api/adk/types.go +++ b/go/api/adk/types.go @@ -398,6 +398,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. diff --git a/go/api/config/crd/bases/kagent.dev_agents.yaml b/go/api/config/crd/bases/kagent.dev_agents.yaml index df5067aad0..f9d7eefcfc 100644 --- a/go/api/config/crd/bases/kagent.dev_agents.yaml +++ b/go/api/config/crd/bases/kagent.dev_agents.yaml @@ -13344,6 +13344,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: @@ -13414,6 +13434,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 diff --git a/go/api/config/crd/bases/kagent.dev_sandboxagents.yaml b/go/api/config/crd/bases/kagent.dev_sandboxagents.yaml index 5bb6ae641f..caf39a5091 100644 --- a/go/api/config/crd/bases/kagent.dev_sandboxagents.yaml +++ b/go/api/config/crd/bases/kagent.dev_sandboxagents.yaml @@ -11001,6 +11001,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: @@ -11071,6 +11091,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 diff --git a/go/api/v1alpha2/agent_types.go b/go/api/v1alpha2/agent_types.go index 1e45b83ce7..317f75e1e6 100644 --- a/go/api/v1alpha2/agent_types.go +++ b/go/api/v1alpha2/agent_types.go @@ -519,6 +519,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"` @@ -527,6 +528,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 diff --git a/go/api/v1alpha2/zz_generated.deepcopy.go b/go/api/v1alpha2/zz_generated.deepcopy.go index 9a96decf8a..8ebe13ee07 100644 --- a/go/api/v1alpha2/zz_generated.deepcopy.go +++ b/go/api/v1alpha2/zz_generated.deepcopy.go @@ -1993,6 +1993,11 @@ func (in *Tool) DeepCopyInto(out *Tool) { *out = new(TypedReference) **out = **in } + if in.IsolateSessions != nil { + in, out := &in.IsolateSessions, &out.IsolateSessions + *out = new(bool) + **out = **in + } if in.HeadersFrom != nil { in, out := &in.HeadersFrom, &out.HeadersFrom *out = make([]ValueRef, len(*in)) diff --git a/go/core/internal/controller/translator/agent/compiler.go b/go/core/internal/controller/translator/agent/compiler.go index eba707b33f..6354f7fed7 100644 --- a/go/core/internal/controller/translator/agent/compiler.go +++ b/go/core/internal/controller/translator/agent/compiler.go @@ -359,10 +359,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) diff --git a/go/core/internal/controller/translator/agent/testdata/inputs/agent_with_isolated_session_tool.yaml b/go/core/internal/controller/translator/agent/testdata/inputs/agent_with_isolated_session_tool.yaml new file mode 100644 index 0000000000..74aa6c41da --- /dev/null +++ b/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 diff --git a/go/core/internal/controller/translator/agent/testdata/outputs/agent_with_isolated_session_tool.json b/go/core/internal/controller/translator/agent/testdata/outputs/agent_with_isolated_session_tool.json new file mode 100644 index 0000000000..573eb0b829 --- /dev/null +++ b/go/core/internal/controller/translator/agent/testdata/outputs/agent_with_isolated_session_tool.json @@ -0,0 +1,301 @@ +{ + "agentCard": { + "capabilities": { + "streaming": true + }, + "defaultInputModes": [ + "text" + ], + "defaultOutputModes": [ + "text" + ], + "description": "", + "name": "coordinator_agent", + "skills": null, + "supportedInterfaces": [ + { + "protocolBinding": "JSONRPC", + "protocolVersion": "0.3", + "url": "http://coordinator-agent.test:8080" + }, + { + "protocolBinding": "JSONRPC", + "protocolVersion": "1.0", + "url": "http://coordinator-agent.test:8080" + } + ], + "version": "" + }, + "config": { + "description": "", + "instruction": "You are a coordinating agent that delegates tasks to a worker agent, potentially in parallel.", + "model": { + "base_url": "", + "model": "gpt-4o", + "type": "openai" + }, + "remote_agents": [ + { + "isolate_sessions": true, + "name": "test__NS__worker_agent", + "url": "http://worker-agent.test:8080" + } + ], + "stream": false + }, + "manifest": [ + { + "apiVersion": "v1", + "kind": "Secret", + "metadata": { + "labels": { + "app": "kagent", + "app.kubernetes.io/managed-by": "kagent", + "app.kubernetes.io/name": "coordinator-agent", + "app.kubernetes.io/part-of": "kagent", + "kagent": "coordinator-agent" + }, + "name": "coordinator-agent", + "namespace": "test", + "ownerReferences": [ + { + "apiVersion": "kagent.dev/v1alpha2", + "blockOwnerDeletion": true, + "controller": true, + "kind": "Agent", + "name": "coordinator-agent", + "uid": "" + } + ] + }, + "stringData": { + "agent-card.json": "{\n \"defaultInputModes\": [\n \"text\"\n ],\n \"defaultOutputModes\": [\n \"text\"\n ],\n \"description\": \"\",\n \"name\": \"coordinator_agent\",\n \"version\": \"\",\n \"skills\": [],\n \"capabilities\": {\n \"streaming\": true\n },\n \"supportedInterfaces\": [\n {\n \"url\": \"http://coordinator-agent.test:8080\",\n \"protocolBinding\": \"JSONRPC\",\n \"protocolVersion\": \"0.3\"\n },\n {\n \"url\": \"http://coordinator-agent.test:8080\",\n \"protocolBinding\": \"JSONRPC\",\n \"protocolVersion\": \"1.0\"\n }\n ],\n \"url\": \"http://coordinator-agent.test:8080\",\n \"protocolVersion\": \"0.3\",\n \"preferredTransport\": \"JSONRPC\"\n}", + "config.json": "{\"model\":{\"type\":\"openai\",\"model\":\"gpt-4o\",\"base_url\":\"\"},\"description\":\"\",\"instruction\":\"You are a coordinating agent that delegates tasks to a worker agent, potentially in parallel.\",\"remote_agents\":[{\"name\":\"test__NS__worker_agent\",\"url\":\"http://worker-agent.test:8080\",\"isolate_sessions\":true}],\"stream\":false}" + } + }, + { + "apiVersion": "v1", + "kind": "ServiceAccount", + "metadata": { + "labels": { + "app": "kagent", + "app.kubernetes.io/managed-by": "kagent", + "app.kubernetes.io/name": "coordinator-agent", + "app.kubernetes.io/part-of": "kagent", + "kagent": "coordinator-agent" + }, + "name": "coordinator-agent", + "namespace": "test", + "ownerReferences": [ + { + "apiVersion": "kagent.dev/v1alpha2", + "blockOwnerDeletion": true, + "controller": true, + "kind": "Agent", + "name": "coordinator-agent", + "uid": "" + } + ] + } + }, + { + "apiVersion": "apps/v1", + "kind": "Deployment", + "metadata": { + "labels": { + "app": "kagent", + "app.kubernetes.io/managed-by": "kagent", + "app.kubernetes.io/name": "coordinator-agent", + "app.kubernetes.io/part-of": "kagent", + "kagent": "coordinator-agent" + }, + "name": "coordinator-agent", + "namespace": "test", + "ownerReferences": [ + { + "apiVersion": "kagent.dev/v1alpha2", + "blockOwnerDeletion": true, + "controller": true, + "kind": "Agent", + "name": "coordinator-agent", + "uid": "" + } + ] + }, + "spec": { + "selector": { + "matchLabels": { + "app": "kagent", + "kagent": "coordinator-agent" + } + }, + "strategy": { + "rollingUpdate": { + "maxSurge": 1, + "maxUnavailable": 0 + }, + "type": "RollingUpdate" + }, + "template": { + "metadata": { + "annotations": { + "kagent.dev/config-hash": "12596745338766631722" + }, + "labels": { + "app": "kagent", + "app.kubernetes.io/managed-by": "kagent", + "app.kubernetes.io/name": "coordinator-agent", + "app.kubernetes.io/part-of": "kagent", + "kagent": "coordinator-agent" + } + }, + "spec": { + "containers": [ + { + "args": [ + "--host", + "0.0.0.0", + "--port", + "8080", + "--filepath", + "/config" + ], + "env": [ + { + "name": "OPENAI_API_KEY", + "valueFrom": { + "secretKeyRef": { + "key": "api-key", + "name": "openai-secret" + } + } + }, + { + "name": "KAGENT_NAMESPACE", + "valueFrom": { + "fieldRef": { + "fieldPath": "metadata.namespace" + } + } + }, + { + "name": "KAGENT_NAME", + "value": "coordinator-agent" + }, + { + "name": "KAGENT_URL", + "value": "http://kagent-controller.kagent:8083" + } + ], + "image": "ghcr.io/kagent-dev/kagent/app:dev", + "imagePullPolicy": "IfNotPresent", + "name": "kagent", + "ports": [ + { + "containerPort": 8080, + "name": "http" + } + ], + "readinessProbe": { + "httpGet": { + "path": "/.well-known/agent-card.json", + "port": "http" + }, + "initialDelaySeconds": 15, + "periodSeconds": 15, + "timeoutSeconds": 15 + }, + "resources": { + "limits": { + "cpu": "2", + "memory": "1Gi" + }, + "requests": { + "cpu": "100m", + "memory": "384Mi" + } + }, + "volumeMounts": [ + { + "mountPath": "/config", + "name": "config" + }, + { + "mountPath": "/var/run/secrets/tokens", + "name": "kagent-token" + } + ] + } + ], + "serviceAccountName": "coordinator-agent", + "volumes": [ + { + "name": "config", + "secret": { + "secretName": "coordinator-agent" + } + }, + { + "name": "kagent-token", + "projected": { + "sources": [ + { + "serviceAccountToken": { + "audience": "kagent", + "expirationSeconds": 3600, + "path": "kagent-token" + } + } + ] + } + } + ] + } + } + }, + "status": {} + }, + { + "apiVersion": "v1", + "kind": "Service", + "metadata": { + "labels": { + "app": "kagent", + "app.kubernetes.io/managed-by": "kagent", + "app.kubernetes.io/name": "coordinator-agent", + "app.kubernetes.io/part-of": "kagent", + "kagent": "coordinator-agent" + }, + "name": "coordinator-agent", + "namespace": "test", + "ownerReferences": [ + { + "apiVersion": "kagent.dev/v1alpha2", + "blockOwnerDeletion": true, + "controller": true, + "kind": "Agent", + "name": "coordinator-agent", + "uid": "" + } + ] + }, + "spec": { + "ports": [ + { + "name": "http", + "port": 8080, + "targetPort": 8080 + } + ], + "selector": { + "app": "kagent", + "kagent": "coordinator-agent" + }, + "type": "ClusterIP" + }, + "status": { + "loadBalancer": {} + } + } + ] +} \ No newline at end of file diff --git a/helm/kagent-crds/templates/kagent.dev_agents.yaml b/helm/kagent-crds/templates/kagent.dev_agents.yaml index df5067aad0..f9d7eefcfc 100644 --- a/helm/kagent-crds/templates/kagent.dev_agents.yaml +++ b/helm/kagent-crds/templates/kagent.dev_agents.yaml @@ -13344,6 +13344,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: @@ -13414,6 +13434,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 diff --git a/helm/kagent-crds/templates/kagent.dev_sandboxagents.yaml b/helm/kagent-crds/templates/kagent.dev_sandboxagents.yaml index 5bb6ae641f..caf39a5091 100644 --- a/helm/kagent-crds/templates/kagent.dev_sandboxagents.yaml +++ b/helm/kagent-crds/templates/kagent.dev_sandboxagents.yaml @@ -11001,6 +11001,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: @@ -11071,6 +11091,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 diff --git a/python/packages/kagent-adk/src/kagent/adk/types.py b/python/packages/kagent-adk/src/kagent/adk/types.py index 702d1c1fa4..e50937216d 100644 --- a/python/packages/kagent-adk/src/kagent/adk/types.py +++ b/python/packages/kagent-adk/src/kagent/adk/types.py @@ -238,6 +238,11 @@ class RemoteAgentConfig(BaseModel): headers: dict[str, Any] | None = None timeout: float = DEFAULT_TIMEOUT description: str = "" + # isolate_sessions: accepted for schema parity with the Go declarative + # runtime (see go/api/v1alpha2.Tool.IsolateSessions). The Python low-level + # tool (KAgentRemoteA2AToolset / _remote_a2a_tool.py) does not yet honor + # this flag — it only affects agents running on runtime: go. + isolate_sessions: bool = False class BaseLLM(BaseModel): diff --git a/ui/src/lib/messageHandlers.ts b/ui/src/lib/messageHandlers.ts index e8006c27e8..731ef4037f 100644 --- a/ui/src/lib/messageHandlers.ts +++ b/ui/src/lib/messageHandlers.ts @@ -939,13 +939,29 @@ export const createMessageHandlers = (handlers: MessageHandlers) => { } else if (partType === "function_response") { // Skip internal HITL markers: the before_tool_callback stub and - // the ask_user first-invocation pending stub. + // the ask_user first-invocation pending stub. Exception: a + // "pending" response from an Agent-type tool carrying a + // subagent_session_id is a real HITL event surfaced from a + // sub-agent (it needs the user's approval before the sub-agent's + // tool call proceeds) and must not be skipped — without pre- + // stamped session ids (see remoteA2AResponse in + // remote_a2a_tool.go), this is the only place the UI learns the + // session id for that pending sub-agent, and skipping it here + // means the Activity panel never opens for the user to review + // what the sub-agent is about to do before approving it. const responseData = (data as { response?: Record })?.response; const responseStatus = responseData?.status as string | undefined; - if (responseStatus === "confirmation_requested" || responseStatus === "pending") { + const toolData = data as unknown as ToolResponseData; + const isPendingAgentSession = + responseStatus === "pending" && + isAgentToolName(toolData.name) && + typeof responseData?.subagent_session_id === "string"; + if ( + (responseStatus === "confirmation_requested" || responseStatus === "pending") && + !isPendingAgentSession + ) { continue; } - const toolData = data as unknown as ToolResponseData; const source = getSourceFromMetadata(adkMetadata, defaultAgentSource); processFunctionResponsePart(toolData, statusUpdate.contextId, statusUpdate.taskId, source); }