diff --git a/design/EP-2153-a2a-session-isolation.md b/design/EP-2153-a2a-session-isolation.md new file mode 100644 index 0000000000..32afe18eb7 --- /dev/null +++ b/design/EP-2153-a2a-session-isolation.md @@ -0,0 +1,230 @@ +# EP-2153: Per-call session isolation + robust sub-agent session reporting + +* Issue: [#2153](https://github.com/kagent-dev/kagent/issues/2153) +* Upstream PR: https://github.com/kagent-dev/kagent/pull/2153 + +Status: implemented + +## Goal + +Let a declarative agent fan out **parallel, isolated** sub-agent calls, and make +each of those calls **observable** in the UI by always reporting the sub-agent +session it ran in. + +Two capabilities, delivered together: + +1. **Per-call session isolation** (`isolateSessions`) — mint a fresh A2A + `context_id` per sub-agent call so parallel fan-out no longer collapses into + one shared sub-agent session. +2. **A single typed response contract** — every remote A2A tool outcome + (success, `input_required`, failure) returns the per-call `context_id` as + `subagent_session_id`, so the UI can always link a sub-agent card to the + session that produced it. + +## Problem + +### Shared sub-agent session (isolation) + +`go/adk/pkg/tools/remote_a2a_tool.go` generated one `context_id` **once**, at +tool construction (pod startup), and stamped it on **every** outbound A2A +message: + +```go +state := &remoteA2AState{ /* ... */ lastContextID: a2atype.NewContextID() } +// ... +message.ContextID = s.lastContextID // same id for every call, forever +``` + +The worker derives its session id straight from that context id +(`go/adk/pkg/a2a/executor.go`: `sessionID := reqCtx.ContextID`). So N parallel +calls → 1 shared worker session, serialized/interleaved rather than N parallel +sessions. + +Observed: a coordinator emitted 11 sub-agent calls in one model turn; all 11 +collapsed into a single worker session (11 A2A tasks in one context) instead of +11 isolated sessions. + +### Session id silently dropped on some branches (reporting) + +The tool returned each outcome as an ad-hoc `map[string]any` literal, built +independently per branch. Nothing forced a branch to include +`subagent_session_id`, and several omitted it (direct-message result, +`input_required`, and the no-result fallback). With isolation on there is no +stable, pre-known session id to fall back to, so any branch that drops the +per-call id leaves the UI unable to link that sub-agent card to a session. + +## Design + +### `isolateSessions` flag + +A per-Agent-tool boolean. + +- **`false` (default):** one stable `context_id` per tool for its lifetime → + session continuity for stateful sub-agents. Unchanged behavior. +- **`true`:** `context_id` is minted per call via `a2atype.NewContextID()` → + each call is an isolated sub-agent session; parallel fan-out no longer shares + state/history. + +```go +// contextIDForCall returns the A2A context_id (== sub-agent session id) to stamp +// on this call. With isolation enabled every call gets a fresh id; otherwise the +// stable per-tool id is reused. +func (s *remoteA2AState) contextIDForCall() string { + if s.isolateSessions { + return a2atype.NewContextID() + } + return s.lastContextID +} +``` + +Cross-conversation/turn continuity that stateful workers need does **not** +depend on the message `context_id`: it rides the `x-kagent-root-context-id` +header (`lineageHeadersInterceptor`), which stays stable per conversation +regardless of this flag. So isolation only changes which sub-agent *session* a +turn is recorded in. + +### UI linking + +The executor can pre-stamp outgoing `function_call` parts with a pre-known +`subagent_session_id` from a startup map (`subagentSessionIDs`, keyed by tool +name). That model only works when a tool has exactly one session for its whole +lifetime — it breaks down under `isolateSessions`, where the real session is +per call. So an isolated tool contributes **no** map entry (its constructor +returns an empty stamp id), and the UI links each sub-agent card via the +per-call `subagent_session_id` returned in the tool's `function_response` +(consumed by `AgentCallDisplay`). + +For this to hold, that response field must be present on **every** branch — see +below. + +### Typed response contract (the reporting fix) + +Every outcome returns one shared struct instead of a per-branch map literal, so +`SubagentSessionID` (the per-call `context_id`) is a field every path sets and +cannot be forgotten in one branch while present in another: + +```go +// remoteA2AResponse is the single typed return value for every remote A2A tool +// outcome — success, input_required, and failure alike. functiontool.New infers +// the tool's output schema from this type. +type remoteA2AResponse struct { + Result string `json:"result,omitempty"` + Error string `json:"error,omitempty"` + ErrorType string `json:"error_type,omitempty"` + ErrorStep string `json:"error_step,omitempty"` + ErrorMessage string `json:"error_message,omitempty"` + RemediationHint string `json:"remediation_hint,omitempty"` + Status string `json:"status,omitempty"` + WaitingFor string `json:"waiting_for,omitempty"` + Subagent string `json:"subagent,omitempty"` + SubagentSessionID string `json:"subagent_session_id,omitempty"` + SubagentTaskID string `json:"subagent_task_id,omitempty"` + SubagentTaskState string `json:"subagent_task_state,omitempty"` + KAgentUsageMetadata map[string]any `json:"kagent_usage_metadata,omitempty"` +} +``` + +`subagent_session_id` reporting per branch: + +| Outcome | Reported session id | +|---|---| +| completed | per-call `context_id` | +| direct `*Message` result | per-call `context_id` | +| `input_required` (incl. nil-task guard) | paused task's `context_id`, else per-call | +| failed / canceled / rejected | task's `context_id`, else per-call | +| non-terminal after call end | per-call `context_id` | +| request error | per-call `context_id` | +| no result returned | per-call `context_id` | + +`processResult` switches **explicitly** on the terminal task state: only +`completed` is a success; `failed`/`canceled`/`rejected` become typed errors +with sub-agent linkage (`subagent_task_id`, `subagent_task_state`); a task left +in a non-terminal state (`working`/`submitted`/…) after the call ends is never +treated as success. Child-provided structured failure fields (`error_type`, +`error_step`, `error_message`, `remediation_hint`) are overlaid onto the struct +— read from the failed task's status-message data parts, or from a JSON object +embedded in the failure text (optionally fenced) — with the child's value +winning over the generic `subagent_` default. + +## Relationship to PR #2153 + +PR #2153 is the upstream origin of both ideas. This branch implements them as a +clean, self-contained change on top of `main`: + +| Aspect | Upstream PR #2153 | This branch | +|---|---|---| +| `isolateSessions` per-call `context_id` | ✅ | ✅ | +| Typed `remoteA2AResponse`, always report `subagent_session_id` | ✅ | ✅ (extended: task-id/state + structured failure passthrough) | +| Explicit terminal-state switch in `processResult` | partial | ✅ (non-terminal is never success) | +| Tool constructor | positional params | `NewKAgentRemoteA2ATool(RemoteA2AToolConfig)` — config-struct, booleans can't be transposed | +| `subagentSessionIDs` stamp-map plumbing | removed | retained (empty entry for isolated tools; still serves non-isolated tools) | + +## Changes + +| File | Change | +|------|--------| +| `go/api/v1alpha2/agent_types.go` | `IsolateSessions *bool` on `Tool` (CEL: `type=Agent` only) | +| `go/api/adk/types.go` | `IsolateSessions bool` (`isolate_sessions`) on `RemoteAgentConfig` | +| `go/core/internal/controller/translator/agent/compiler.go` | Copy `tool.IsolateSessions` → `RemoteAgentConfig.IsolateSessions` | +| `go/adk/pkg/agent/agent.go` | Pass flag via `RemoteA2AToolConfig`; skip stamp-map entry when isolated | +| `go/adk/pkg/tools/remote_a2a_tool.go` | `isolateSessions` field + `contextIDForCall`; **typed `remoteA2AResponse` on every branch, always setting `subagent_session_id`** | +| `python/packages/kagent-adk/src/kagent/adk/types.py` | `isolate_sessions` field for config-schema parity (Go runtime honors it) | +| generated | CRD YAML (`api/config/crd/bases` + `helm/kagent-crds/templates`) + DeepCopy via `controller-gen` | + +## Deployment / configuration + +```yaml +apiVersion: kagent.dev/v1alpha2 +kind: Agent +metadata: + name: coordinator +spec: + declarative: + runtime: go + tools: + - type: Agent + agent: + name: worker + isolateSessions: true # each worker call = its own isolated session +``` + +- Unset / `false` → current behavior (one shared sub-agent session). +- `true` → fresh sub-agent session per call; enables parallel isolated fan-out. + +## Scope / non-goals + +- **Runtime:** honored by the **Go** declarative runtime. The field is accepted + by the Python config model (schema parity) but the Python low-level tool is + unchanged. +- **Not** a concurrency limiter — capping parallel invocations per pod is a + separate, complementary mechanism. +- **Not** a streaming relay or per-call timeout — this change is limited to + session isolation and the response contract. Those are possible follow-ups. +- No change to HITL resume: resume still targets the original sub-agent session + via the `context_id` stored in the confirmation payload. + +## Behavior notes & caveats + +- Stateful sub-agents that rely on **reused** session history across calls must + keep `isolateSessions: false` (or key state on `x-kagent-root-context-id`). +- With isolation on, a sub-agent's per-call sessions have `source='agent'` and + are hidden from the worker's sidebar by design; they render inline in the + parent chat. +- Parallelism is bounded by how many tool calls the parent LLM emits in a turn + and by any `KAGENT_MAX_CONCURRENCY` on the worker pod. +- The per-call `subagent_session_id` in the `function_response` is now the + single source of truth for UI linkage on both isolated and non-isolated tools. + +## Testing + +- Unit (`go/adk/pkg/tools`): `contextIDForCall` — shared reuses one id, isolated + mints a fresh non-empty id per call; constructor stamp-id is empty for + isolated tools and non-empty otherwise. +- Unit: every `processResult` branch (message, completed, input_required incl. + nil-task, failed/canceled/rejected, non-terminal, no-result) returns a + non-empty `subagent_session_id`. +- Unit: structured failure passthrough (data part, fenced JSON text, plain-text + default) and `parseJSONObject` edge cases. +- Unit (`go/core/.../translator/agent`): `Tool.IsolateSessions` is carried into + `RemoteAgentConfig.IsolateSessions` (true / false / unset→false). +- Regression: existing lineage-header propagation tests stay green. diff --git a/go/adk/pkg/agent/agent.go b/go/adk/pkg/agent/agent.go index 9b92aad74b..cff9295953 100644 --- a/go/adk/pkg/agent/agent.go +++ b/go/adk/pkg/agent/agent.go @@ -65,10 +65,19 @@ func CreateGoogleADKAgentWithSubagentSessionIDs(ctx context.Context, agentConfig log.Info("Skipping remote agent with empty URL", "name", remoteAgent.Name) continue } - remoteTool, sessionID, err := tools.NewKAgentRemoteA2ATool(remoteAgent.Name, remoteAgent.Description, remoteAgent.Url, nil, remoteAgent.Headers, propagateToken) + remoteTool, sessionID, err := tools.NewKAgentRemoteA2ATool(tools.RemoteA2AToolConfig{ + Name: remoteAgent.Name, + Description: remoteAgent.Description, + BaseURL: remoteAgent.Url, + ExtraHeaders: remoteAgent.Headers, + PropagateToken: propagateToken, + IsolateSessions: remoteAgent.IsolateSessions, + }) if err != nil { return nil, nil, fmt.Errorf("failed to create remote A2A tool for %s: %w", remoteAgent.Name, err) } + // Isolated tools return an empty stamp id (each call reports its own + // session id in the function_response instead), so they add no map entry. if sessionID != "" { subagentSessionIDs[remoteAgent.Name] = sessionID } diff --git a/go/adk/pkg/tools/remote_a2a_tool.go b/go/adk/pkg/tools/remote_a2a_tool.go index 705c5c9a68..933569c581 100644 --- a/go/adk/pkg/tools/remote_a2a_tool.go +++ b/go/adk/pkg/tools/remote_a2a_tool.go @@ -2,6 +2,7 @@ package tools import ( "context" + "encoding/json" "fmt" "log/slog" "net/http" @@ -134,6 +135,31 @@ type remoteA2AInput struct { Request string `json:"request"` } +// remoteA2AResponse is the single typed return value for every remote A2A tool +// outcome — success, input_required, and failure alike. Using one shared struct +// instead of ad-hoc map[string]any literals per branch makes SubagentSessionID +// (the per-call context_id == sub-agent session id) a field every path sets, so +// it cannot be present in one branch and silently forgotten in another. This +// matters most with isolateSessions=true: there is no stable, pre-known session +// id to stamp on the outbound function_call, so this per-call id in the response +// is the single source of truth the UI (AgentCallDisplay) links each sub-agent +// card from. functiontool.New infers the tool's output schema from this type. +type remoteA2AResponse struct { + Result string `json:"result,omitempty"` + Error string `json:"error,omitempty"` + ErrorType string `json:"error_type,omitempty"` + ErrorStep string `json:"error_step,omitempty"` + ErrorMessage string `json:"error_message,omitempty"` + RemediationHint string `json:"remediation_hint,omitempty"` + Status string `json:"status,omitempty"` + WaitingFor string `json:"waiting_for,omitempty"` + Subagent string `json:"subagent,omitempty"` + SubagentSessionID string `json:"subagent_session_id,omitempty"` + SubagentTaskID string `json:"subagent_task_id,omitempty"` + SubagentTaskState string `json:"subagent_task_state,omitempty"` + KAgentUsageMetadata map[string]any `json:"kagent_usage_metadata,omitempty"` +} + // remoteA2AState holds the mutable state for one remote A2A agent connection. // All external interaction goes through the tool.Tool returned by NewKAgentRemoteA2ATool. type remoteA2AState struct { @@ -150,38 +176,77 @@ type remoteA2AState struct { initErr error lastContextID string + // isolateSessions mints a fresh context_id per call so each invocation runs + // in its own isolated sub-agent session (parallel fan-out). When false, all + // calls reuse lastContextID for session continuity. + isolateSessions bool +} + +// contextIDForCall returns the A2A context_id (== sub-agent session id) to stamp +// on this call. With isolation enabled every call gets a fresh id; otherwise the +// stable per-tool id is reused so calls share one sub-agent session. +func (s *remoteA2AState) contextIDForCall() string { + if s.isolateSessions { + return a2atype.NewContextID() + } + return s.lastContextID +} + +// RemoteA2AToolConfig configures a remote A2A agent tool. A config struct is +// used instead of positional parameters so booleans like IsolateSessions cannot +// be transposed at the call site. +type RemoteA2AToolConfig struct { + Name string + Description string + BaseURL string + HTTPClient *http.Client // optional; wrapped with otelhttp + ExtraHeaders map[string]string + PropagateToken bool + // IsolateSessions mints a fresh context_id per call so each invocation + // runs in its own isolated sub-agent session (enables parallel fan-out). + IsolateSessions bool } // NewKAgentRemoteA2ATool creates a function tool that calls a remote A2A agent and // propagates HITL state. It returns: // - the tool.Tool to register with the agent config -// - the initial A2A context/session ID for subagent session stamping +// - the initial A2A context/session ID for subagent session stamping (empty when +// IsolateSessions is true, since each call mints its own id and reports it in +// the tool response as subagent_session_id instead) // -// 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 +// 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(cfg RemoteA2AToolConfig) (tool.Tool, string, error) { + httpClient := cfg.HTTPClient 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: cfg.Name, + description: cfg.Description, + baseURL: cfg.BaseURL, + httpClient: httpClient, + extraHeaders: cfg.ExtraHeaders, + propagateToken: cfg.PropagateToken, + lastContextID: a2atype.NewContextID(), + isolateSessions: cfg.IsolateSessions, } ft, err := functiontool.New(functiontool.Config{ - Name: name, - Description: description, - }, func(ctx adkagent.Context, in remoteA2AInput) (map[string]any, error) { + Name: cfg.Name, + Description: cfg.Description, + }, 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", cfg.Name, err) + } + // Isolated tools have no stable, pre-known session id to stamp on the + // outgoing function_call: each call's id is returned in the function_response + // (subagent_session_id) instead. + if cfg.IsolateSessions { + return ft, "", nil } return ft, state.lastContextID, nil } @@ -239,7 +304,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 +312,39 @@ 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), + SubagentSessionID: contextID, + }, nil } - return s.processResult(ctx, result) + return s.processResult(ctx, result, contextID) } // 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 +358,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 +380,15 @@ 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 + } + + // 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 + sessionID := contextID + if sessionID == "" { + sessionID = s.lastContextID } sendCtx := context.WithValue(ctx, userIDContextKey{}, ctx.UserID()) @@ -319,63 +396,175 @@ 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), + SubagentSessionID: sessionID, + }, 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 + return s.processResult(ctx, result, sessionID) } -// processResult converts a SendMessageResult into a tool return value. -func (s *remoteA2AState) processResult(ctx adkagent.Context, result a2atype.SendMessageResult) (map[string]any, error) { +// processResult converts a SendMessageResult into a tool return value with an +// explicit switch on the task state: only `completed` is a success; +// failed/canceled/rejected become typed errors with sub-agent linkage; +// non-terminal states are never treated as success. contextID is the A2A +// context_id used for this call and is reported back as subagent_session_id on +// every branch so the UI can link the sub-agent's session. +func (s *remoteA2AState) processResult(ctx adkagent.Context, result a2atype.SendMessageResult, contextID string) (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 - case a2atype.TaskStateFailed: - text := extractTextFromTask(r) - if text == "" { - text = fmt.Sprintf("Remote agent '%s' failed.", s.name) - } - return map[string]any{"error": text}, 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. + case a2atype.TaskStateCompleted: + // 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, + resp := remoteA2AResponse{ + Result: extractTextFromTask(r), + SubagentSessionID: contextID, + SubagentTaskID: string(r.ID), } if usage := extractUsageFromTask(r); usage != nil { - ret["kagent_usage_metadata"] = usage + resp.KAgentUsageMetadata = usage } - return ret, nil + return resp, nil + case a2atype.TaskStateInputRequired: + return s.handleInputRequired(ctx, r, contextID), nil + case a2atype.TaskStateFailed, a2atype.TaskStateCanceled, a2atype.TaskStateRejected: + return s.failureResult(r, contextID), nil + default: + // working / submitted / auth-required / unknown after the call + // ended: never treat a non-terminal child as success. + return remoteA2AResponse{ + Error: fmt.Sprintf("Remote agent '%s' ended in non-terminal state %q.", s.name, r.Status.State), + ErrorType: "subagent_non_terminal", + SubagentSessionID: contextID, + SubagentTaskID: string(r.ID), + SubagentTaskState: string(r.Status.State), + }, 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 } } +// structuredFailureKeys are child-provided failure fields passed through to the +// parent so it can act deterministically without re-parsing tool text. +var structuredFailureKeys = []string{"error_type", "error_step", "error_message", "remediation_hint"} + +// failureResult builds a typed error for a terminal failed/canceled/rejected +// child task, preserving sub-agent linkage and passing through any structured +// failure fields the child provided. +func (s *remoteA2AState) failureResult(task *a2atype.Task, contextID string) remoteA2AResponse { + state := task.Status.State + text := extractTextFromTask(task) + if text == "" { + text = fmt.Sprintf("Remote agent '%s' ended in state %q.", s.name, state) + } + // The child's own reported session takes precedence for linkage when the + // task carries one; otherwise fall back to the id we sent the call under. + sessionID := contextID + if task.ContextID != "" { + sessionID = task.ContextID + } + resp := remoteA2AResponse{ + Error: text, + ErrorType: "subagent_" + string(state), + SubagentSessionID: sessionID, + SubagentTaskID: string(task.ID), + SubagentTaskState: string(state), + } + applyStructuredFailure(&resp, structuredFailureFields(task, text)) + return resp +} + +// applyStructuredFailure overlays child-provided failure fields onto the +// response; a child-supplied value wins over the default (e.g. a domain-specific +// error_type replaces the generic subagent_). +func applyStructuredFailure(resp *remoteA2AResponse, fields map[string]any) { + if v, ok := fields["error_type"].(string); ok && v != "" { + resp.ErrorType = v + } + if v, ok := fields["error_step"].(string); ok && v != "" { + resp.ErrorStep = v + } + if v, ok := fields["error_message"].(string); ok && v != "" { + resp.ErrorMessage = v + } + if v, ok := fields["remediation_hint"].(string); ok && v != "" { + resp.RemediationHint = v + } +} + +// structuredFailureFields extracts child-provided failure fields from the +// task's status-message data parts, or from a JSON object embedded in the +// failure text (optionally fenced in a markdown code block). +func structuredFailureFields(task *a2atype.Task, text string) map[string]any { + fields := make(map[string]any) + if task.Status.Message != nil { + for _, part := range task.Status.Message.Parts { + if dp, ok := part.(a2atype.DataPart); ok { + collectFailureKeys(fields, dp.Data) + } + } + } + if len(fields) == 0 { + if payload := parseJSONObject(text); payload != nil { + collectFailureKeys(fields, payload) + } + } + return fields +} + +func collectFailureKeys(dst map[string]any, src map[string]any) { + for _, key := range structuredFailureKeys { + if value, ok := src[key]; ok && value != nil && value != "" { + dst[key] = value + } + } +} + +// parseJSONObject attempts to decode text (or its first fenced code block) +// as a JSON object; it returns nil when text carries none. +func parseJSONObject(text string) map[string]any { + candidate := strings.TrimSpace(text) + if idx := strings.Index(candidate, "```"); idx >= 0 { + candidate = candidate[idx+3:] + candidate = strings.TrimPrefix(candidate, "json") + if end := strings.Index(candidate, "```"); end >= 0 { + candidate = candidate[:end] + } + candidate = strings.TrimSpace(candidate) + } + if !strings.HasPrefix(candidate, "{") { + return nil + } + var payload map[string]any + if err := json.Unmarshal([]byte(candidate), &payload); err != nil { + return nil + } + return payload +} + // handleInputRequired pauses parent agent execution via RequestConfirmation. -func (s *remoteA2AState) handleInputRequired(ctx adkagent.Context, task *a2atype.Task) map[string]any { +// contextID is the id this call was sent under; it (or the paused task's own +// context_id, preferred when present) is reported back as SubagentSessionID so +// the UI can link the pending Activity panel to the paused sub-agent 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 +601,15 @@ 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, + sessionID := contextID + if task.ContextID != "" { + sessionID = task.ContextID + } + return remoteA2AResponse{ + Status: "pending", + WaitingFor: "subagent_approval", + Subagent: s.name, + SubagentSessionID: sessionID, } } diff --git a/go/adk/pkg/tools/remote_a2a_tool_state_test.go b/go/adk/pkg/tools/remote_a2a_tool_state_test.go new file mode 100644 index 0000000000..d22336e3b2 --- /dev/null +++ b/go/adk/pkg/tools/remote_a2a_tool_state_test.go @@ -0,0 +1,259 @@ +package tools + +import ( + "testing" + + a2atype "github.com/a2aproject/a2a-go/a2a" +) + +// TestContextIDForCall covers session isolation: with isolateSessions the tool +// mints a fresh context_id (== sub-agent session id) per call; without it the +// stable per-tool id is reused so calls share one sub-agent session. +func TestContextIDForCall(t *testing.T) { + t.Run("shared reuses the same id across calls", func(t *testing.T) { + s := &remoteA2AState{lastContextID: "stable-ctx", isolateSessions: false} + if got := s.contextIDForCall(); got != "stable-ctx" { + t.Errorf("first call = %q, want stable-ctx", got) + } + if got := s.contextIDForCall(); got != "stable-ctx" { + t.Errorf("second call = %q, want stable-ctx", got) + } + }) + + t.Run("isolated mints a fresh non-empty id per call", func(t *testing.T) { + s := &remoteA2AState{lastContextID: "stable-ctx", isolateSessions: true} + first := s.contextIDForCall() + second := s.contextIDForCall() + if first == "" || second == "" { + t.Fatalf("isolated ids must be non-empty, got %q and %q", first, second) + } + if first == second { + t.Errorf("isolated calls must differ, both = %q", first) + } + if first == "stable-ctx" || second == "stable-ctx" { + t.Errorf("isolated calls must not reuse lastContextID, got %q / %q", first, second) + } + }) +} + +// TestNewKAgentRemoteA2AToolStampID: isolated tools expose no stable pre-known +// session id (the per-call id rides the function_response instead), while shared +// tools return their stable id for function_call stamping. +func TestNewKAgentRemoteA2AToolStampID(t *testing.T) { + _, shared, err := NewKAgentRemoteA2ATool(RemoteA2AToolConfig{Name: "t", Description: "d", BaseURL: "http://x"}) + if err != nil { + t.Fatalf("shared tool: %v", err) + } + if shared == "" { + t.Error("shared tool must return a non-empty stamp session id") + } + + _, isolated, err := NewKAgentRemoteA2ATool(RemoteA2AToolConfig{Name: "t", Description: "d", BaseURL: "http://x", IsolateSessions: true}) + if err != nil { + t.Fatalf("isolated tool: %v", err) + } + if isolated != "" { + t.Errorf("isolated tool must return empty stamp session id, got %q", isolated) + } +} + +func taskWithState(state a2atype.TaskState, statusText string) *a2atype.Task { + task := &a2atype.Task{ID: "task-1", ContextID: "child-ctx"} + task.Status.State = state + if statusText != "" { + task.Status.Message = a2atype.NewMessage(a2atype.MessageRoleAgent, a2atype.TextPart{Text: statusText}) + } + return task +} + +func TestProcessResult_Completed(t *testing.T) { + s := &remoteA2AState{name: "child"} + task := taskWithState(a2atype.TaskStateCompleted, "") + task.Artifacts = []*a2atype.Artifact{ + {Parts: a2atype.ContentParts{a2atype.TextPart{Text: "the answer"}}}, + } + + got, err := s.processResult(nil, task, "ctx-1") + if err != nil { + t.Fatalf("processResult() error = %v", err) + } + if got.Result != "the answer" { + t.Errorf("result = %v, want %q", got.Result, "the answer") + } + if got.SubagentSessionID != "ctx-1" { + t.Errorf("subagent_session_id = %v, want ctx-1", got.SubagentSessionID) + } + if got.SubagentTaskID != "task-1" { + t.Errorf("subagent_task_id = %v, want task-1", got.SubagentTaskID) + } + if got.Error != "" { + t.Error("completed task must not carry an error field") + } +} + +func TestProcessResult_TerminalFailures(t *testing.T) { + tests := []struct { + name string + state a2atype.TaskState + wantErrorType string + }{ + {name: "failed", state: a2atype.TaskStateFailed, wantErrorType: "subagent_failed"}, + {name: "canceled", state: a2atype.TaskStateCanceled, wantErrorType: "subagent_canceled"}, + {name: "rejected", state: a2atype.TaskStateRejected, wantErrorType: "subagent_rejected"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + s := &remoteA2AState{name: "child"} + got, err := s.processResult(nil, taskWithState(tt.state, "boom"), "ctx-1") + if err != nil { + t.Fatalf("processResult() error = %v", err) + } + if got.Error != "boom" { + t.Errorf("error = %v, want boom", got.Error) + } + if got.ErrorType != tt.wantErrorType { + t.Errorf("error_type = %v, want %v", got.ErrorType, tt.wantErrorType) + } + if got.SubagentTaskID != "task-1" { + t.Errorf("subagent_task_id = %v, want task-1", got.SubagentTaskID) + } + if got.SubagentTaskState != string(tt.state) { + t.Errorf("subagent_task_state = %v, want %v", got.SubagentTaskState, tt.state) + } + // Task.ContextID takes precedence for linkage when present. + if got.SubagentSessionID != "child-ctx" { + t.Errorf("subagent_session_id = %v, want child-ctx", got.SubagentSessionID) + } + }) + } +} + +func TestProcessResult_NonTerminalIsNotSuccess(t *testing.T) { + for _, state := range []a2atype.TaskState{ + a2atype.TaskStateWorking, + a2atype.TaskStateSubmitted, + a2atype.TaskStateAuthRequired, + } { + t.Run(string(state), func(t *testing.T) { + s := &remoteA2AState{name: "child"} + got, err := s.processResult(nil, taskWithState(state, ""), "ctx-1") + if err != nil { + t.Fatalf("processResult() error = %v", err) + } + if got.ErrorType != "subagent_non_terminal" { + t.Errorf("error_type = %v, want subagent_non_terminal", got.ErrorType) + } + if got.SubagentSessionID != "ctx-1" { + t.Errorf("subagent_session_id = %v, want ctx-1", got.SubagentSessionID) + } + if got.Result != "" { + t.Error("non-terminal task must not be treated as success") + } + }) + } +} + +func TestProcessResult_MessageAndNil(t *testing.T) { + s := &remoteA2AState{name: "child"} + msg := a2atype.NewMessage(a2atype.MessageRoleAgent, a2atype.TextPart{Text: "direct"}) + got, err := s.processResult(nil, msg, "ctx-1") + if err != nil { + t.Fatalf("processResult(message) error = %v", err) + } + if got.Result != "direct" { + t.Errorf("result = %v, want direct", got.Result) + } + if got.SubagentSessionID != "ctx-1" { + t.Errorf("message subagent_session_id = %v, want ctx-1", got.SubagentSessionID) + } + + got, err = s.processResult(nil, nil, "ctx-1") + if err != nil { + t.Fatalf("processResult(nil) error = %v", err) + } + if got.Error == "" { + t.Error("nil result must produce an error") + } + if got.SubagentSessionID != "ctx-1" { + t.Errorf("no-result subagent_session_id = %v, want ctx-1", got.SubagentSessionID) + } +} + +// TestHandleInputRequired_NilTaskReportsSession covers the early-return branch +// (no ctx interaction) and asserts the session id is reported. +func TestHandleInputRequired_NilTaskReportsSession(t *testing.T) { + s := &remoteA2AState{name: "child"} + got := s.handleInputRequired(nil, nil, "ctx-1") + if got.Error == "" { + t.Error("input_required without task must produce an error") + } + if got.SubagentSessionID != "ctx-1" { + t.Errorf("subagent_session_id = %v, want ctx-1", got.SubagentSessionID) + } +} + +func TestFailureResult_StructuredPassthrough(t *testing.T) { + t.Run("from data part, child value wins", func(t *testing.T) { + s := &remoteA2AState{name: "child"} + task := taskWithState(a2atype.TaskStateFailed, "deploy failed") + task.Status.Message.Parts = append(task.Status.Message.Parts, a2atype.DataPart{ + Data: map[string]any{ + "error_type": "helm_upgrade_failed", + "error_step": "upgrade", + "error_message": "release stuck", + "remediation_hint": "rollback first", + "unrelated": "dropped", + }, + }) + got := s.failureResult(task, "ctx-1") + if got.ErrorType != "helm_upgrade_failed" { + t.Errorf("error_type = %v, want helm_upgrade_failed", got.ErrorType) + } + if got.ErrorStep != "upgrade" || got.ErrorMessage != "release stuck" || got.RemediationHint != "rollback first" { + t.Errorf("structured fields not passed through: %+v", got) + } + }) + + t.Run("from fenced JSON failure text", func(t *testing.T) { + s := &remoteA2AState{name: "child"} + task := taskWithState(a2atype.TaskStateFailed, + "The step failed:\n```json\n{\"error_type\":\"quota_exceeded\",\"remediation_hint\":\"increase quota\"}\n```") + got := s.failureResult(task, "ctx-1") + if got.ErrorType != "quota_exceeded" { + t.Errorf("error_type = %v, want quota_exceeded", got.ErrorType) + } + if got.RemediationHint != "increase quota" { + t.Errorf("remediation_hint = %v, want increase quota", got.RemediationHint) + } + }) + + t.Run("plain text keeps default error_type", func(t *testing.T) { + s := &remoteA2AState{name: "child"} + got := s.failureResult(taskWithState(a2atype.TaskStateFailed, "plain failure"), "ctx-1") + if got.ErrorType != "subagent_failed" { + t.Errorf("error_type = %v, want subagent_failed", got.ErrorType) + } + }) +} + +func TestParseJSONObject(t *testing.T) { + tests := []struct { + name string + in string + want bool + }{ + {name: "plain object", in: `{"error_type":"x"}`, want: true}, + {name: "fenced object", in: "```json\n{\"error_type\":\"x\"}\n```", want: true}, + {name: "fenced without lang", in: "```\n{\"error_type\":\"x\"}\n```", want: true}, + {name: "prose", in: "it broke", want: false}, + {name: "malformed json", in: "{not json", want: false}, + {name: "empty", in: "", want: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := parseJSONObject(tt.in); (got != nil) != tt.want { + t.Errorf("parseJSONObject(%q) present = %v, want %v", tt.in, got != nil, tt.want) + } + }) + } +} diff --git a/go/api/adk/types.go b/go/api/adk/types.go index 71b4250316..06f800209b 100644 --- a/go/api/adk/types.go +++ b/go/api/adk/types.go @@ -382,6 +382,10 @@ type RemoteAgentConfig struct { Url string `json:"url"` Headers map[string]string `json:"headers,omitempty"` Description string `json:"description,omitempty"` + // IsolateSessions mints a fresh A2A context_id per call so each invocation + // runs in its own isolated sub-agent session (enables parallel fan-out). + // When false (default) all calls reuse one stable context_id. + 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 15b08f0ca8..ed123acbd2 100644 --- a/go/api/config/crd/bases/kagent.dev_agents.yaml +++ b/go/api/config/crd/bases/kagent.dev_agents.yaml @@ -13222,6 +13222,16 @@ spec: rule: (has(self.value) && !has(self.valueFrom)) || (!has(self.value) && has(self.valueFrom)) type: array + isolateSessions: + description: |- + IsolateSessions applies only to Agent tools (type=Agent). When true, each + call the parent makes to this sub-agent uses a fresh A2A context_id, so + every invocation runs in its own isolated sub-agent session. This enables + parallel fan-out (e.g. batch dispatch) where calls must not share + history/state. When false (default) all calls to the sub-agent reuse one + stable context_id (session continuity for stateful sub-agents). Honored by + the Go agent runtime. + type: boolean mcpServer: properties: allowedHeaders: @@ -13292,6 +13302,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 for Agent tools + 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 8dd8560b8c..db93d577d0 100644 --- a/go/api/config/crd/bases/kagent.dev_sandboxagents.yaml +++ b/go/api/config/crd/bases/kagent.dev_sandboxagents.yaml @@ -10879,6 +10879,16 @@ spec: rule: (has(self.value) && !has(self.valueFrom)) || (!has(self.value) && has(self.valueFrom)) type: array + isolateSessions: + description: |- + IsolateSessions applies only to Agent tools (type=Agent). When true, each + call the parent makes to this sub-agent uses a fresh A2A context_id, so + every invocation runs in its own isolated sub-agent session. This enables + parallel fan-out (e.g. batch dispatch) where calls must not share + history/state. When false (default) all calls to the sub-agent reuse one + stable context_id (session continuity for stateful sub-agents). Honored by + the Go agent runtime. + type: boolean mcpServer: properties: allowedHeaders: @@ -10949,6 +10959,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 for Agent tools + 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 2303687590..2f161ef1bd 100644 --- a/go/api/v1alpha2/agent_types.go +++ b/go/api/v1alpha2/agent_types.go @@ -508,6 +508,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 for Agent tools",rule="!(has(self.isolateSessions) && self.type != 'Agent')" type Tool struct { // +optional Type ToolProviderType `json:"type,omitempty"` @@ -516,6 +517,16 @@ type Tool struct { // +optional Agent *TypedReference `json:"agent,omitempty"` + // IsolateSessions applies only to Agent tools (type=Agent). When true, each + // call the parent makes to this sub-agent uses a fresh A2A context_id, so + // every invocation runs in its own isolated sub-agent session. This enables + // parallel fan-out (e.g. batch dispatch) where calls must not share + // history/state. When false (default) all calls to the sub-agent reuse one + // stable context_id (session continuity for stateful sub-agents). Honored by + // the Go agent runtime. + // +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 98d89ec396..88c9bc9364 100644 --- a/go/api/v1alpha2/zz_generated.deepcopy.go +++ b/go/api/v1alpha2/zz_generated.deepcopy.go @@ -1949,6 +1949,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..0d6ae9a0a1 100644 --- a/go/core/internal/controller/translator/agent/compiler.go +++ b/go/core/internal/controller/translator/agent/compiler.go @@ -358,11 +358,16 @@ func (a *adkApiTranslator) translateInlineAgent(ctx context.Context, agent v1alp } } + isolateSessions := false + if tool.IsolateSessions != nil { + isolateSessions = *tool.IsolateSessions + } 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: isolateSessions, }) default: return nil, nil, nil, fmt.Errorf("unknown agent type: %s", toolSpec.Type) diff --git a/go/core/internal/controller/translator/agent/isolate_sessions_test.go b/go/core/internal/controller/translator/agent/isolate_sessions_test.go new file mode 100644 index 0000000000..de9d5f1397 --- /dev/null +++ b/go/core/internal/controller/translator/agent/isolate_sessions_test.go @@ -0,0 +1,95 @@ +package agent_test + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + schemev1 "k8s.io/client-go/kubernetes/scheme" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + "github.com/kagent-dev/kagent/go/api/v1alpha2" + agenttranslator "github.com/kagent-dev/kagent/go/core/internal/controller/translator/agent" +) + +// Test_AdkApiTranslator_IsolateSessions verifies the Tool.IsolateSessions flag is +// carried into the ADK RemoteAgentConfig the Go runtime consumes. +func Test_AdkApiTranslator_IsolateSessions(t *testing.T) { + ctx := context.Background() + scheme := schemev1.Scheme + require.NoError(t, v1alpha2.AddToScheme(scheme)) + + declarativeSpec := func(tools ...*v1alpha2.Tool) v1alpha2.AgentSpec { + return v1alpha2.AgentSpec{ + Type: v1alpha2.AgentType_Declarative, + Description: "test agent", + Declarative: &v1alpha2.DeclarativeAgentSpec{ + SystemMessage: "Test", + ModelConfig: "default-model", + Tools: tools, + }, + } + } + + agentTool := func(isolate *bool) *v1alpha2.Tool { + return &v1alpha2.Tool{ + Type: v1alpha2.ToolProviderType_Agent, + Agent: &v1alpha2.TypedReference{Name: "specialist", Kind: "Agent"}, + IsolateSessions: isolate, + } + } + + modelConfig := &v1alpha2.ModelConfig{ + ObjectMeta: metav1.ObjectMeta{Name: "default-model", Namespace: "test"}, + Spec: v1alpha2.ModelConfigSpec{Provider: "OpenAI", Model: "gpt-4o"}, + } + testNamespace := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "test"}} + specialist := &v1alpha2.Agent{ + ObjectMeta: metav1.ObjectMeta{Name: "specialist", Namespace: "test"}, + Spec: declarativeSpec(), + } + + boolPtr := func(b bool) *bool { return &b } + + tests := []struct { + name string + isolate *bool + want bool + }{ + {name: "isolateSessions true", isolate: boolPtr(true), want: true}, + {name: "isolateSessions false", isolate: boolPtr(false), want: false}, + {name: "isolateSessions unset defaults to false", isolate: nil, want: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + parent := &v1alpha2.Agent{ + ObjectMeta: metav1.ObjectMeta{Name: "parent", Namespace: "test"}, + Spec: declarativeSpec(agentTool(tt.isolate)), + } + kubeClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(modelConfig, testNamespace, specialist). + Build() + + translator := agenttranslator.NewAdkApiTranslator( + kubeClient, + types.NamespacedName{Name: "default-model", Namespace: "test"}, + nil, + "", + nil, + ) + + inputs, err := translator.CompileAgent(ctx, parent) + require.NoError(t, err) + require.NotNil(t, inputs) + require.NotNil(t, inputs.Config) + require.Len(t, inputs.Config.RemoteAgents, 1) + assert.Equal(t, tt.want, inputs.Config.RemoteAgents[0].IsolateSessions) + }) + } +} diff --git a/helm/kagent-crds/templates/kagent.dev_agents.yaml b/helm/kagent-crds/templates/kagent.dev_agents.yaml index 15b08f0ca8..ed123acbd2 100644 --- a/helm/kagent-crds/templates/kagent.dev_agents.yaml +++ b/helm/kagent-crds/templates/kagent.dev_agents.yaml @@ -13222,6 +13222,16 @@ spec: rule: (has(self.value) && !has(self.valueFrom)) || (!has(self.value) && has(self.valueFrom)) type: array + isolateSessions: + description: |- + IsolateSessions applies only to Agent tools (type=Agent). When true, each + call the parent makes to this sub-agent uses a fresh A2A context_id, so + every invocation runs in its own isolated sub-agent session. This enables + parallel fan-out (e.g. batch dispatch) where calls must not share + history/state. When false (default) all calls to the sub-agent reuse one + stable context_id (session continuity for stateful sub-agents). Honored by + the Go agent runtime. + type: boolean mcpServer: properties: allowedHeaders: @@ -13292,6 +13302,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 for Agent tools + 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 8dd8560b8c..db93d577d0 100644 --- a/helm/kagent-crds/templates/kagent.dev_sandboxagents.yaml +++ b/helm/kagent-crds/templates/kagent.dev_sandboxagents.yaml @@ -10879,6 +10879,16 @@ spec: rule: (has(self.value) && !has(self.valueFrom)) || (!has(self.value) && has(self.valueFrom)) type: array + isolateSessions: + description: |- + IsolateSessions applies only to Agent tools (type=Agent). When true, each + call the parent makes to this sub-agent uses a fresh A2A context_id, so + every invocation runs in its own isolated sub-agent session. This enables + parallel fan-out (e.g. batch dispatch) where calls must not share + history/state. When false (default) all calls to the sub-agent reuse one + stable context_id (session continuity for stateful sub-agents). Honored by + the Go agent runtime. + type: boolean mcpServer: properties: allowedHeaders: @@ -10949,6 +10959,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 for Agent tools + 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 e556e4639b..1712a75297 100644 --- a/python/packages/kagent-adk/src/kagent/adk/types.py +++ b/python/packages/kagent-adk/src/kagent/adk/types.py @@ -238,6 +238,10 @@ class RemoteAgentConfig(BaseModel): headers: dict[str, Any] | None = None timeout: float = DEFAULT_TIMEOUT description: str = "" + # Mints a fresh A2A context_id per call so each invocation runs in its own + # isolated sub-agent session (parallel fan-out). Honored by the Go runtime; + # accepted here for config-schema parity. + isolate_sessions: bool = False class BaseLLM(BaseModel):