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); }