diff --git a/docs/architecture/human-in-the-loop.md b/docs/architecture/human-in-the-loop.md index 2953e027bc..bec1138e99 100644 --- a/docs/architecture/human-in-the-loop.md +++ b/docs/architecture/human-in-the-loop.md @@ -72,6 +72,21 @@ request may also carry `nested` child-agent correlation: `subagent_name`, The server rejects responses that use the wrong task/context, omit required decisions, duplicate IDs, or answer an operation that is no longer pending. +The status message also carries a text part that describes the pause. Every +runtime composes that text the same way, and a client receives it whether or not +it activated the extension: + +- an `ask_user` pause renders its questions, joined with `; `; +- any other pause renders the hints its tools supplied, followed by the tool + names in parentheses: `Deleting this file requires approval (delete_file, + restart_pod)`; +- with no hints, it names the tools: `Approval is required for tool(s): + delete_file`; +- a pause that carries no tool call renders `Human input is required before the + agent can continue.` + +The `hint` field of an activated payload holds that same string. + ## Ownership | Component | Responsibility | diff --git a/go/adk/cmd/main.go b/go/adk/cmd/main.go index 84140537bb..dc6316489e 100644 --- a/go/adk/cmd/main.go +++ b/go/adk/cmd/main.go @@ -207,9 +207,7 @@ func main() { }, } } - agentCard.Capabilities = a2atype.AgentCapabilities{ - Streaming: stream, - } + agentCard.Capabilities.Streaming = stream // Delegate the actor-local A2A server and task store to app.New. kagentApp, err := app.New(app.AppConfig{ diff --git a/go/adk/pkg/a2a/agentcard.go b/go/adk/pkg/a2a/agentcard.go index 93dc48416a..58c42cd3fd 100644 --- a/go/adk/pkg/a2a/agentcard.go +++ b/go/adk/pkg/a2a/agentcard.go @@ -22,11 +22,8 @@ func EnrichAgentCard(card *a2atype.AgentCard, agent adkagent.Agent) { if card.Description == "" && agent.Description() != "" { card.Description = agent.Description() } - // If the agent card does not have the HITL extension, add it. - // Kagent's harness always supports it. - if !hasHITLExtension(card.Capabilities.Extensions) { - card.Capabilities.Extensions = append(card.Capabilities.Extensions, apia2a.HITLExtension()) - } + + EnsureHITLExtension(card) // Default to JSONRPC when no interface is explicitly configured. if len(card.SupportedInterfaces) == 0 { @@ -36,6 +33,16 @@ func EnrichAgentCard(card *a2atype.AgentCard, agent adkagent.Agent) { } } +// EnsureHITLExtension declares the optional HITL extension on the card so a client +// can discover it and negotiate. Kagent's harness always supports it, and the +// declaration does not depend on whether an ADK agent was supplied. +func EnsureHITLExtension(card *a2atype.AgentCard) { + if card == nil || hasHITLExtension(card.Capabilities.Extensions) { + return + } + card.Capabilities.Extensions = append(card.Capabilities.Extensions, apia2a.HITLExtension()) +} + func hasHITLExtension(extensions []a2atype.AgentExtension) bool { for _, extension := range extensions { if extension.URI == HITLExtensionURI { diff --git a/go/adk/pkg/a2a/agentcard_test.go b/go/adk/pkg/a2a/agentcard_test.go new file mode 100644 index 0000000000..f411f0f6e2 --- /dev/null +++ b/go/adk/pkg/a2a/agentcard_test.go @@ -0,0 +1,68 @@ +package a2a + +import ( + "iter" + "testing" + + a2atype "github.com/a2aproject/a2a-go/v2/a2a" + adkagent "google.golang.org/adk/v2/agent" + "google.golang.org/adk/v2/session" +) + +func TestEnsureHITLExtension(t *testing.T) { + t.Run("attaches when absent", func(t *testing.T) { + card := &a2atype.AgentCard{Name: "agent"} + EnsureHITLExtension(card) + if !hasHITLExtension(card.Capabilities.Extensions) { + t.Fatalf("extensions = %#v, want the HITL extension", card.Capabilities.Extensions) + } + }) + + t.Run("idempotent", func(t *testing.T) { + card := &a2atype.AgentCard{Name: "agent"} + EnsureHITLExtension(card) + EnsureHITLExtension(card) + if len(card.Capabilities.Extensions) != 1 { + t.Errorf("len(extensions) = %d, want 1", len(card.Capabilities.Extensions)) + } + }) + + t.Run("preserves unrelated extensions", func(t *testing.T) { + card := &a2atype.AgentCard{ + Name: "agent", + Capabilities: a2atype.AgentCapabilities{ + Extensions: []a2atype.AgentExtension{{URI: "https://example.com/extensions/other/v1"}}, + }, + } + EnsureHITLExtension(card) + if len(card.Capabilities.Extensions) != 2 { + t.Fatalf("len(extensions) = %d, want 2", len(card.Capabilities.Extensions)) + } + if card.Capabilities.Extensions[0].URI != "https://example.com/extensions/other/v1" { + t.Errorf("extensions[0].URI = %q, want the pre-existing extension", card.Capabilities.Extensions[0].URI) + } + }) + + t.Run("nil card", func(t *testing.T) { + EnsureHITLExtension(nil) + }) +} + +func TestEnrichAgentCardDeclaresHITLExtension(t *testing.T) { + agent, err := adkagent.New(adkagent.Config{ + Name: "adk_agent", + Run: func(adkagent.InvocationContext) iter.Seq2[*session.Event, error] { + return func(yield func(*session.Event, error) bool) {} + }, + }) + if err != nil { + t.Fatalf("adkagent.New: %v", err) + } + card := &a2atype.AgentCard{Name: "agent"} + + EnrichAgentCard(card, agent) + + if !hasHITLExtension(card.Capabilities.Extensions) { + t.Fatalf("extensions = %#v, want the HITL extension", card.Capabilities.Extensions) + } +} diff --git a/go/adk/pkg/a2a/hitl.go b/go/adk/pkg/a2a/hitl.go index 0496b56828..6a68a52344 100644 --- a/go/adk/pkg/a2a/hitl.go +++ b/go/adk/pkg/a2a/hitl.go @@ -21,6 +21,9 @@ const ( HITLTypeToolApprovalResponse = apia2a.HITLTypeToolApprovalResponse HITLTypeAskUserResponse = apia2a.HITLTypeAskUserResponse KAgentMetadataKeyPrefix = "kagent_" + + // genericHITLText is the status text for a pause that names no tool. + genericHITLText = "Human input is required before the agent can continue." ) var hitlAgentExtension = apia2a.HITLExtension() @@ -391,6 +394,59 @@ func (tool confirmationTool) asHitlTool() apia2a.HITLTool { return apia2a.HITLTool{ID: tool.approvalID, CallID: tool.callID, Name: tool.name, Args: tool.args} } +// askUserQuestions returns the answerable questions of an ask_user pause, or nil. +// A pause qualifies when it holds exactly one ask_user call, and that call carries +// at least one question with text. A question without text cannot be rendered, and +// an answer for it cannot be correlated, so it is dropped. +func askUserQuestions(tools []apia2a.HITLTool) []apia2a.HITLQuestion { + if len(tools) != 1 || tools[0].Name != "ask_user" { + return nil + } + questions := publicAskUserQuestions(tools[0].Args["questions"]) + answerable := make([]apia2a.HITLQuestion, 0, len(questions)) + for _, question := range questions { + if question.Question != "" { + answerable = append(answerable, question) + } + } + if len(answerable) == 0 { + return nil + } + return answerable +} + +// pendingQuestionText joins the questions an ask_user call is waiting on. +func pendingQuestionText(questions []apia2a.HITLQuestion) string { + texts := make([]string, 0, len(questions)) + for _, question := range questions { + texts = append(texts, question.Question) + } + return strings.Join(texts, "; ") +} + +// hitlStatusText renders a pause as one human-readable line. An ask_user pause speaks +// for itself; every other pause names its tools so no pending tool stays hidden. +func hitlStatusText(tools []apia2a.HITLTool, hints []string) string { + if questions := pendingQuestionText(askUserQuestions(tools)); questions != "" { + return questions + } + names := make([]string, 0, len(tools)) + for _, tool := range tools { + if tool.Name != "" { + names = append(names, tool.Name) + } + } + switch { + case len(hints) > 0 && len(names) > 0: + return fmt.Sprintf("%s (%s)", strings.Join(hints, "; "), strings.Join(names, ", ")) + case len(hints) > 0: + return strings.Join(hints, "; ") + case len(names) > 0: + return fmt.Sprintf("Approval is required for tool(s): %s", strings.Join(names, ", ")) + } + return genericHITLText +} + // BuildHITLStatusMessage: ADK confirmation DataParts → public HITL Message extension. func BuildHITLStatusMessage(message *a2atype.Message, activated bool) *a2atype.Message { if message == nil { @@ -398,7 +454,7 @@ func BuildHITLStatusMessage(message *a2atype.Message, activated bool) *a2atype.M } var tools []apia2a.HITLTool var remote *RemoteHitlState - hint := "Human input is required before the agent can continue." + var hints []string for _, part := range message.Parts { data := asDataPart(part) if data == nil || part.Metadata == nil { @@ -412,7 +468,7 @@ func BuildHITLStatusMessage(message *a2atype.Message, activated bool) *a2atype.M tool := parseConfirmationTool(data) tools = append(tools, tool.asHitlTool()) if tool.hint != "" { - hint = tool.hint + hints = append(hints, tool.hint) } if candidate := ParseRemoteHitlState(tool.payload); candidate != nil { remote = candidate @@ -422,7 +478,11 @@ func BuildHITLStatusMessage(message *a2atype.Message, activated bool) *a2atype.M return message } - public := a2atype.NewMessage(a2atype.MessageRoleAgent, a2atype.NewTextPart(hint)) + // The text part carries the same information as the typed payload, so a client + // that did not activate the extension still learns what it is being asked. + text := hitlStatusText(tools, hints) + + public := a2atype.NewMessage(a2atype.MessageRoleAgent, a2atype.NewTextPart(text)) public.TaskID, public.ContextID = message.TaskID, message.ContextID if !activated { return public @@ -444,14 +504,17 @@ func BuildHITLStatusMessage(message *a2atype.Message, activated bool) *a2atype.M Questions: remote.AskUserRequest.Questions, Nested: nested, }) } - if len(tools) == 1 && tools[0].Name == "ask_user" { + // An ask_user call with no answerable question becomes an approval: an empty + // question list gives a request that no response can satisfy, because resume + // requires answers. + if questions := askUserQuestions(tools); questions != nil { return AttachHitlExtension(public, &apia2a.AskUserRequest{ Type: HITLTypeAskUserRequest, ID: tools[0].ID, - Questions: publicAskUserQuestions(tools[0].Args["questions"]), + Questions: questions, }) } return AttachHitlExtension(public, &apia2a.ToolApprovalRequest{ - Type: HITLTypeToolApprovalRequest, Hint: hint, Tools: tools, Nested: nested, + Type: HITLTypeToolApprovalRequest, Hint: text, Tools: tools, Nested: nested, }) } diff --git a/go/adk/pkg/a2a/hitl_test.go b/go/adk/pkg/a2a/hitl_test.go index a92a92bcf5..def5393a0d 100644 --- a/go/adk/pkg/a2a/hitl_test.go +++ b/go/adk/pkg/a2a/hitl_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "slices" + "strings" "testing" a2atype "github.com/a2aproject/a2a-go/v2/a2a" @@ -29,6 +30,50 @@ func confirmationPart(id, toolName, toolID string, args, payload map[string]any) }, map[string]any{"adk_type": "function_call", "adk_is_long_running": true}) } +func hintedConfirmationPart(id, toolName, toolID, hint string) *a2atype.Part { + return dataPart(map[string]any{ + "name": toolconfirmation.FunctionCallName, + "id": id, + "args": map[string]any{ + "originalFunctionCall": map[string]any{"name": toolName, "id": toolID}, + "toolConfirmation": map[string]any{"hint": hint}, + }, + }, map[string]any{"adk_type": "function_call", "adk_is_long_running": true}) +} + +func hintlessConfirmationPart(id, toolName, toolID string) *a2atype.Part { + return dataPart(map[string]any{ + "name": toolconfirmation.FunctionCallName, + "id": id, + "args": map[string]any{ + "originalFunctionCall": map[string]any{"name": toolName, "id": toolID}, + }, + }, map[string]any{"adk_type": "function_call", "adk_is_long_running": true}) +} + +func askUserConfirmationPart(id, toolID string, questions []any, hint string) *a2atype.Part { + return dataPart(map[string]any{ + "name": toolconfirmation.FunctionCallName, + "id": id, + "args": map[string]any{ + "originalFunctionCall": map[string]any{ + "name": "ask_user", "id": toolID, "args": map[string]any{"questions": questions}, + }, + "toolConfirmation": map[string]any{"hint": hint}, + }, + }, map[string]any{"adk_type": "function_call", "adk_is_long_running": true}) +} + +func messageText(message *a2atype.Message) string { + var text strings.Builder + for _, part := range message.Parts { + if content, ok := part.Content.(a2atype.Text); ok { + text.WriteString(string(content)) + } + } + return text.String() +} + func hitlDecisionMessage(payload any) *a2atype.Message { message := a2atype.NewMessage(a2atype.MessageRoleUser, a2atype.NewTextPart("Decision")) return AttachHitlExtension(message, payload) @@ -97,6 +142,36 @@ func TestBuildHITLStatusMessage(t *testing.T) { } }) + t.Run("ask user without questions asks for approval", func(t *testing.T) { + internal := a2atype.NewMessage(a2atype.MessageRoleAgent, + hintlessConfirmationPart("confirm-2", "ask_user", "call-2")) + public := BuildHITLStatusMessage(internal, true) + if GetAskUserRequest(public) != nil { + t.Fatal("an ask_user call without questions must not become an ask_user request") + } + payload := GetToolApprovalRequest(public) + if payload == nil || payload.Tools[0].Name != "ask_user" { + t.Fatalf("payload = %#v", payload) + } + if want := "Approval is required for tool(s): ask_user"; messageText(public) != want { + t.Errorf("text = %q, want %q", messageText(public), want) + } + }) + + t.Run("ask user drops questions without text", func(t *testing.T) { + questions := []any{map[string]any{"question": ""}, map[string]any{"question": "Which cluster?"}} + internal := a2atype.NewMessage(a2atype.MessageRoleAgent, + confirmationPart("confirm-2", "ask_user", "call-2", map[string]any{"questions": questions}, nil)) + public := BuildHITLStatusMessage(internal, true) + payload := GetAskUserRequest(public) + if payload == nil || len(payload.Questions) != 1 || payload.Questions[0].Question != "Which cluster?" { + t.Fatalf("payload = %#v", payload) + } + if want := "Which cluster?"; messageText(public) != want { + t.Errorf("text = %q, want %q", messageText(public), want) + } + }) + t.Run("nested subagent", func(t *testing.T) { remote := RemoteHitlState{ TaskID: "child-task", ContextID: "child-context", SubagentName: "k8s_agent", @@ -124,6 +199,62 @@ func TestBuildHITLStatusMessage(t *testing.T) { if GetToolApprovalRequest(public) != nil { t.Fatalf("unexpected payload on inactive client") } + if want := "Please confirm (delete_file)"; messageText(public) != want { + t.Errorf("text = %q, want %q", messageText(public), want) + } + }) + + t.Run("not activated without a tool hint names the tool", func(t *testing.T) { + public := BuildHITLStatusMessage(a2atype.NewMessage(a2atype.MessageRoleAgent, + hintlessConfirmationPart("confirm-1", "delete_file", "call-1")), false) + if want := "Approval is required for tool(s): delete_file"; messageText(public) != want { + t.Errorf("text = %q, want %q", messageText(public), want) + } + }) + + t.Run("several tools keep every hint and every name", func(t *testing.T) { + internal := a2atype.NewMessage(a2atype.MessageRoleAgent, + confirmationPart("confirm-1", "delete_file", "call-1", nil, nil), + hintlessConfirmationPart("confirm-2", "restart_pod", "call-2")) + want := "Please confirm (delete_file, restart_pod)" + if text := messageText(BuildHITLStatusMessage(internal, false)); text != want { + t.Errorf("text = %q, want %q", text, want) + } + }) + + t.Run("hints join in tool order", func(t *testing.T) { + internal := a2atype.NewMessage(a2atype.MessageRoleAgent, + hintedConfirmationPart("confirm-1", "delete_file", "call-1", "First hint"), + hintedConfirmationPart("confirm-2", "restart_pod", "call-2", "Second hint")) + want := "First hint; Second hint (delete_file, restart_pod)" + if text := messageText(BuildHITLStatusMessage(internal, false)); text != want { + t.Errorf("text = %q, want %q", text, want) + } + }) + + t.Run("activated payload hint repeats the text", func(t *testing.T) { + public := BuildHITLStatusMessage(a2atype.NewMessage(a2atype.MessageRoleAgent, + hintlessConfirmationPart("confirm-1", "delete_file", "call-1")), true) + payload := GetToolApprovalRequest(public) + if payload == nil { + t.Fatal("payload = nil, want a tool approval request") + } + if payload.Hint != messageText(public) { + t.Errorf("hint = %q, text = %q, want them identical", payload.Hint, messageText(public)) + } + }) + + t.Run("not activated ask_user carries the question", func(t *testing.T) { + questions := []any{map[string]any{"question": "Which database?"}} + internal := a2atype.NewMessage(a2atype.MessageRoleAgent, + askUserConfirmationPart("confirm-2", "call-2", questions, "Which database?")) + public := BuildHITLStatusMessage(internal, false) + if GetAskUserRequest(public) != nil { + t.Fatal("unexpected payload on inactive client") + } + if want := "Which database?"; messageText(public) != want { + t.Errorf("text = %q, want the question verbatim %q", messageText(public), want) + } }) t.Run("non-confirmation long-running call", func(t *testing.T) { diff --git a/go/adk/pkg/app/app.go b/go/adk/pkg/app/app.go index edbbba17da..664ba95d98 100644 --- a/go/adk/pkg/app/app.go +++ b/go/adk/pkg/app/app.go @@ -126,18 +126,13 @@ func New(cfg AppConfig, executor a2asrv.AgentExecutor) (*KAgentApp, error) { // Append any caller-supplied handler options. handlerOpts = append(handlerOpts, cfg.HandlerOpts...) - // Enrich agent card with skills derived from the ADK agent. - if cfg.Agent != nil { - a2a.EnrichAgentCard(&cfg.AgentCard, cfg.Agent) - } - serverConfig := server.ServerConfig{ Host: cfg.Host, Port: cfg.Port, ShutdownTimeout: cfg.ShutdownTimeout, } - a2aServer, err := server.NewA2AServer(cfg.AgentCard, executor, log, serverConfig, handlerOpts...) + a2aServer, err := server.NewA2AServer(buildAgentCard(cfg), executor, log, serverConfig, handlerOpts...) if err != nil { return nil, fmt.Errorf("failed to create A2A server: %w", err) } @@ -146,6 +141,17 @@ func New(cfg AppConfig, executor a2asrv.AgentExecutor) (*KAgentApp, error) { return app, nil } +// buildAgentCard returns the card the server serves. The HITL extension is declared +// for every app, whether or not an ADK agent was supplied for skill derivation. +func buildAgentCard(cfg AppConfig) a2atype.AgentCard { + card := cfg.AgentCard + a2a.EnsureHITLExtension(&card) + if cfg.Agent != nil { + a2a.EnrichAgentCard(&card, cfg.Agent) + } + return card +} + // Run starts the A2A server and blocks until a shutdown signal is received. func (a *KAgentApp) Run() error { return a.server.Run() diff --git a/go/adk/pkg/app/app_test.go b/go/adk/pkg/app/app_test.go index 45e85b7d3c..50aeed7d53 100644 --- a/go/adk/pkg/app/app_test.go +++ b/go/adk/pkg/app/app_test.go @@ -3,6 +3,7 @@ package app import ( "context" "iter" + "slices" "testing" "time" @@ -11,6 +12,8 @@ import ( a2ataskstore "github.com/a2aproject/a2a-go/v2/a2asrv/taskstore" "github.com/kagent-dev/kagent/go/adk/pkg/a2a" apia2a "github.com/kagent-dev/kagent/go/api/a2a" + adkagent "google.golang.org/adk/v2/agent" + "google.golang.org/adk/v2/session" ) // fakeExecutor implements a2asrv.AgentExecutor for testing. @@ -168,3 +171,47 @@ func TestBuildAppName_Default(t *testing.T) { t.Errorf("expected %q, got %q", defaultAppName, name) } } + +func TestBuildAgentCard_DeclaresHITLWithoutAgent(t *testing.T) { + card := buildAgentCard(AppConfig{AgentCard: a2atype.AgentCard{Name: "byo-agent"}}) + + if !slices.ContainsFunc(card.Capabilities.Extensions, func(extension a2atype.AgentExtension) bool { + return extension.URI == a2a.HITLExtensionURI + }) { + t.Fatalf("extensions = %#v, want the HITL extension", card.Capabilities.Extensions) + } +} + +func TestBuildAgentCard_DeclaresHITLWithAgent(t *testing.T) { + agent, err := adkagent.New(adkagent.Config{ + Name: "adk_agent", + Description: "an ADK agent", + Run: func(adkagent.InvocationContext) iter.Seq2[*session.Event, error] { + return func(yield func(*session.Event, error) bool) {} + }, + }) + if err != nil { + t.Fatalf("adkagent.New: %v", err) + } + + card := buildAgentCard(AppConfig{AgentCard: a2atype.AgentCard{Name: "adk-agent"}, Agent: agent}) + + if !slices.ContainsFunc(card.Capabilities.Extensions, func(extension a2atype.AgentExtension) bool { + return extension.URI == a2a.HITLExtensionURI + }) { + t.Fatalf("extensions = %#v, want the HITL extension", card.Capabilities.Extensions) + } + if card.Description != "an ADK agent" { + t.Errorf("description = %q, want the agent description", card.Description) + } +} + +func TestBuildAgentCard_LeavesCallerCardUntouched(t *testing.T) { + cfg := AppConfig{AgentCard: a2atype.AgentCard{Name: "byo-agent"}} + + buildAgentCard(cfg) + + if len(cfg.AgentCard.Capabilities.Extensions) != 0 { + t.Errorf("caller extensions = %#v, want the caller's card untouched", cfg.AgentCard.Capabilities.Extensions) + } +} diff --git a/python/packages/kagent-adk/src/kagent/adk/_hitl.py b/python/packages/kagent-adk/src/kagent/adk/_hitl.py index 8f5efbc7a6..7c2782dd3f 100644 --- a/python/packages/kagent-adk/src/kagent/adk/_hitl.py +++ b/python/packages/kagent-adk/src/kagent/adk/_hitl.py @@ -7,6 +7,7 @@ from __future__ import annotations +import logging import uuid from typing import Annotated, Any @@ -23,16 +24,20 @@ NestedHitlRequest, ToolApprovalRequest, ToolApprovalResponse, + ask_user_questions, attach_hitl_extension, get_ask_user_request, get_ask_user_response, get_tool_approval_request, get_tool_approval_response, + hitl_status_text, require_ask_user_response, require_tool_approval_response, ) from pydantic import BaseModel, ConfigDict, Field, ValidationError +logger = logging.getLogger(__name__) + HitlRequest = Annotated[ToolApprovalRequest | AskUserRequest, Field(discriminator="type")] HitlResponse = Annotated[ToolApprovalResponse | AskUserResponse, Field(discriminator="type")] @@ -121,9 +126,16 @@ def remote_hitl_hint(state: RemoteHitlState) -> str: return f"Remote agent '{state.subagent_name}' requires human input before continuing." +def _confirmation_args(data: dict[str, Any], key: str) -> dict[str, Any]: + """Return one sub-object of an ADK confirmation call, or an empty one.""" + args = data.get("args") + value = args.get(key) if isinstance(args, dict) else None + return value if isinstance(value, dict) else {} + + def _tool_from_confirmation_data(data: dict[str, Any]) -> HitlTool: """Parse one ADK adk_request_confirmation DataPart into a public HitlTool.""" - original = data.get("args", {}).get("originalFunctionCall", {}) + original = _confirmation_args(data, "originalFunctionCall") return HitlTool( id=str(data.get("id") or ""), call_id=str(original.get("id") or data.get("id") or ""), @@ -139,13 +151,8 @@ def build_hitl_status_message(parts: list[Part], task_id: str, context_id: str, the confirmation payload. Without activation, only human-readable text is returned. """ message = Message(message_id=str(uuid.uuid4()), role=Role.ROLE_AGENT, task_id=task_id, context_id=context_id) - default_hint = "Human input is required before the agent can continue." - if not activated: - message.parts.append(Part(text=default_hint)) - return message - tools: list[HitlTool] = [] - hint: str | None = None + hints: list[str] = [] remote_state: RemoteHitlState | None = None for part in parts: if not part.HasField("data"): @@ -153,17 +160,28 @@ def build_hitl_status_message(parts: list[Part], task_id: str, context_id: str, data = MessageToDict(part.data) if data.get("name") != REQUEST_CONFIRMATION_FUNCTION_CALL_NAME: continue - tools.append(_tool_from_confirmation_data(data)) - tool_confirmation = data.get("args", {}).get("toolConfirmation", {}) - hint = hint or tool_confirmation.get("hint") + try: + tool = _tool_from_confirmation_data(data) + except ValidationError: + logger.warning("Skipping an ADK confirmation part that carries no usable tool call") + continue + tools.append(tool) + tool_confirmation = _confirmation_args(data, "toolConfirmation") + hint = tool_confirmation.get("hint") + if isinstance(hint, str) and hint: + hints.append(hint) candidate = get_remote_hitl_state(tool_confirmation.get("payload")) if candidate is not None: remote_state = candidate + # The text part carries the same information as the typed payload, so a client + # that did not activate the extension still learns what it is being asked. + text = hitl_status_text(tools, hints) + # Auth-required parts can share the long-running path without being HITL # confirmations. Leave those messages unextended. - if not tools: - message.parts.append(Part(text=hint or default_hint)) + if not tools or not activated: + message.parts.append(Part(text=text)) return message nested: NestedHitlRequest | None = None @@ -182,16 +200,13 @@ def build_hitl_status_message(parts: list[Part], task_id: str, context_id: str, questions=remote_state.hitl_request.questions, nested=nested, ) - elif len(tools) == 1 and tools[0].name == "ask_user": - request = AskUserRequest( - id=tools[0].id, - questions=tools[0].args.get("questions") or [], - ) + elif questions := ask_user_questions(tools): + request = AskUserRequest(id=tools[0].id, questions=questions) else: - request = ToolApprovalRequest(hint=hint, tools=tools, nested=nested) + request = ToolApprovalRequest(hint=text, tools=tools, nested=nested) attach_hitl_extension(message, request) - message.parts.append(Part(text=hint or default_hint)) + message.parts.append(Part(text=text)) return message diff --git a/python/packages/kagent-adk/tests/unittests/test_hitl.py b/python/packages/kagent-adk/tests/unittests/test_hitl.py index e3233bb9c4..d29467ca8c 100644 --- a/python/packages/kagent-adk/tests/unittests/test_hitl.py +++ b/python/packages/kagent-adk/tests/unittests/test_hitl.py @@ -3,7 +3,7 @@ from __future__ import annotations import pytest -from a2a.types import Message, Role, Task, TaskState, TaskStatus +from a2a.types import Message, Part, Role, Task, TaskState, TaskStatus from google.adk.a2a.converters.part_converter import ( convert_a2a_part_to_genai_part, convert_genai_part_to_a2a_part, @@ -21,6 +21,7 @@ ToolApprovalResponse, attach_hitl_extension, get_ask_user_request, + get_tool_approval_request, ) from kagent.adk._approval import make_approval_callback @@ -352,3 +353,124 @@ def test_resume_rejects_input_required_task_without_public_hitl_request(): task, _incoming(ToolApprovalResponse(approvals=[ToolApproval(id="confirm-1", approved=True)])), ) + + +def _confirmation_part(confirmation_id: str, tool_name: str, args: dict, hint: str | None = None) -> Part: + function_call = genai_types.Part( + function_call=genai_types.FunctionCall( + id=confirmation_id, + name=REQUEST_CONFIRMATION_FUNCTION_CALL_NAME, + args={ + "originalFunctionCall": {"id": f"call-{confirmation_id}", "name": tool_name, "args": args}, + "toolConfirmation": {"hint": hint} if hint else {}, + }, + ) + ) + converted = convert_genai_part_to_a2a_part(function_call) + assert converted is not None and not isinstance(converted, list) + return converted + + +def _status_text(message: Message) -> str: + return "".join(part.text for part in message.parts if part.HasField("text")) + + +def test_status_message_without_activation_names_the_tool(): + part = _confirmation_part("confirm-1", "delete_file", {"path": "/tmp/x"}) + + status_message = build_hitl_status_message([part], "task-1", "context-1", activated=False) + + assert get_tool_approval_request(status_message) is None + assert _status_text(status_message) == "Approval is required for tool(s): delete_file" + + +def test_status_message_keeps_the_hint_and_the_tool_name(): + part = _confirmation_part("confirm-1", "delete_file", {"path": "/tmp/x"}, hint="Deleting this file is permanent") + + status_message = build_hitl_status_message([part], "task-1", "context-1", activated=False) + + assert _status_text(status_message) == "Deleting this file is permanent (delete_file)" + + +def test_status_message_keeps_every_hint_across_tools(): + parts = [ + _confirmation_part("confirm-1", "delete_file", {"path": "/tmp/x"}, hint="Deleting is permanent"), + _confirmation_part("confirm-2", "restart_pod", {"name": "api"}), + ] + + status_message = build_hitl_status_message(parts, "task-1", "context-1", activated=False) + + assert _status_text(status_message) == "Deleting is permanent (delete_file, restart_pod)" + + +def test_status_message_skips_a_confirmation_part_without_a_call(): + unusable = convert_genai_part_to_a2a_part( + genai_types.Part(function_call=genai_types.FunctionCall(name=REQUEST_CONFIRMATION_FUNCTION_CALL_NAME, args={})) + ) + assert unusable is not None and not isinstance(unusable, list) + + status_message = build_hitl_status_message([unusable], "task-1", "context-1", activated=False) + + assert _status_text(status_message) == "Human input is required before the agent can continue." + + +def test_status_message_without_activation_carries_the_question(): + part = _confirmation_part( + "confirm-1", + "ask_user", + {"questions": [{"question": "Which namespace?"}]}, + hint="Which namespace?", + ) + + status_message = build_hitl_status_message([part], "task-1", "context-1", activated=False) + + assert get_ask_user_request(status_message) is None + assert _status_text(status_message) == "Which namespace?" + + +def test_status_message_with_activation_keeps_the_same_text(): + part = _confirmation_part("confirm-1", "delete_file", {"path": "/tmp/x"}) + + status_message = build_hitl_status_message([part], "task-1", "context-1", activated=True) + request = get_tool_approval_request(status_message) + + assert request is not None + assert request.tools[0].name == "delete_file" + assert _status_text(status_message) == "Approval is required for tool(s): delete_file" + assert request.hint == _status_text(status_message) + + +def test_status_message_without_confirmation_parts_stays_generic(): + status_message = build_hitl_status_message( + [Part(text="waiting on authorization")], "task-1", "context-1", activated=True + ) + + assert get_tool_approval_request(status_message) is None + assert _status_text(status_message) == "Human input is required before the agent can continue." + + +def test_status_message_ask_user_without_questions_asks_for_approval(): + part = _confirmation_part("confirm-1", "ask_user", {"questions": []}) + + status_message = build_hitl_status_message([part], "task-1", "context-1", activated=True) + request = get_tool_approval_request(status_message) + + assert get_ask_user_request(status_message) is None + assert request is not None + assert request.tools[0].name == "ask_user" + assert _status_text(status_message) == "Approval is required for tool(s): ask_user" + + +def test_status_message_ask_user_drops_questions_without_text(): + part = _confirmation_part( + "confirm-1", + "ask_user", + {"questions": [{"question": ""}, {"question": "Which cluster?"}]}, + ) + + status_message = build_hitl_status_message([part], "task-1", "context-1", activated=True) + request = get_ask_user_request(status_message) + + assert request is not None + assert request.questions == [{"question": "Which cluster?"}] + assert _status_text(status_message) == "Which cluster?" diff --git a/python/packages/kagent-core/src/kagent/core/a2a/__init__.py b/python/packages/kagent-core/src/kagent/core/a2a/__init__.py index 73cc2a6504..30d4ff1215 100644 --- a/python/packages/kagent-core/src/kagent/core/a2a/__init__.py +++ b/python/packages/kagent-core/src/kagent/core/a2a/__init__.py @@ -25,6 +25,7 @@ ToolApproval, ToolApprovalRequest, ToolApprovalResponse, + ask_user_questions, attach_hitl_agent_extension, attach_hitl_extension, get_ask_user_request, @@ -33,6 +34,7 @@ get_tool_approval_request, get_tool_approval_response, hitl_activated, + hitl_status_text, require_ask_user_response, require_tool_approval_response, ) @@ -72,6 +74,8 @@ "ToolApprovalResponse", "AskUserResponse", "hitl_activated", + "hitl_status_text", + "ask_user_questions", "get_hitl_payload", "get_tool_approval_request", "get_ask_user_request", diff --git a/python/packages/kagent-core/src/kagent/core/a2a/_hitl.py b/python/packages/kagent-core/src/kagent/core/a2a/_hitl.py index fd7658f2c3..3e4ccb5a46 100644 --- a/python/packages/kagent-core/src/kagent/core/a2a/_hitl.py +++ b/python/packages/kagent-core/src/kagent/core/a2a/_hitl.py @@ -184,6 +184,53 @@ def require_ask_user_response( return response +GENERIC_HITL_TEXT = "Human input is required before the agent can continue." + + +def ask_user_questions(tools: list[HitlTool]) -> list[dict[str, Any]]: + """Return the answerable questions of an ask_user pause, or an empty list. + + A pause qualifies when it holds exactly one ask_user call, and that call + carries at least one question with text. A question without text cannot be + rendered, and an answer for it cannot be correlated, so it is dropped. + """ + if len(tools) != 1 or tools[0].name != "ask_user": + return [] + questions = tools[0].args.get("questions") + if not isinstance(questions, list): + return [] + return [ + question + for question in questions + if isinstance(question, dict) and isinstance(question.get("question"), str) and question["question"] + ] + + +def _ask_user_question_text(questions: list[dict[str, Any]]) -> str: + """Join the questions an ask_user call is waiting on.""" + return "; ".join(question["question"] for question in questions) + + +def hitl_status_text(tools: list[HitlTool], hints: list[str] | None = None) -> str: + """Render a pause as one human-readable line. + + An ask_user pause speaks for itself; every other pause names its tools so no + pending tool stays hidden, and keeps the hints the tools supplied. + """ + hints = [hint for hint in (hints or []) if hint] + questions = _ask_user_question_text(ask_user_questions(tools)) + if questions: + return questions + names = [tool.name for tool in tools if tool.name] + if hints and names: + return f"{'; '.join(hints)} ({', '.join(names)})" + if hints: + return "; ".join(hints) + if names: + return f"Approval is required for tool(s): {', '.join(names)}" + return GENERIC_HITL_TEXT + + def hitl_activated(headers: Mapping[str, Any] | None) -> bool: """True when the client opted in with the exact hitl/v1 URI in A2A-Extensions.""" if not headers: diff --git a/python/packages/kagent-core/tests/test_hitl_utils.py b/python/packages/kagent-core/tests/test_hitl_utils.py index 6cc2f6d238..69eb8bcf33 100644 --- a/python/packages/kagent-core/tests/test_hitl_utils.py +++ b/python/packages/kagent-core/tests/test_hitl_utils.py @@ -15,6 +15,7 @@ get_tool_approval_request, get_tool_approval_response, hitl_activated, + hitl_status_text, ) @@ -96,3 +97,41 @@ def test_parse_ask_user_request() -> None: assert request is not None assert request.id == "question-1" + + +def test_status_text_names_the_tools() -> None: + tools = [ + HitlTool(id="confirmation-1", call_id="call-1", name="delete_file"), + HitlTool(id="confirmation-2", call_id="call-2", name="restart_pod"), + ] + + assert hitl_status_text(tools) == "Approval is required for tool(s): delete_file, restart_pod" + + +def test_status_text_keeps_every_hint_and_every_name() -> None: + tools = [ + HitlTool(id="confirmation-1", call_id="call-1", name="delete_file"), + HitlTool(id="confirmation-2", call_id="call-2", name="restart_pod"), + ] + + assert ( + hitl_status_text(tools, ["Deleting is permanent", "Restarting drops connections"]) + == "Deleting is permanent; Restarting drops connections (delete_file, restart_pod)" + ) + + +def test_status_text_for_ask_user_is_the_question() -> None: + tools = [ + HitlTool( + id="confirmation-1", + call_id="call-1", + name="ask_user", + args={"questions": [{"question": "Which namespace?"}]}, + ) + ] + + assert hitl_status_text(tools, ["Which namespace?"]) == "Which namespace?" + + +def test_status_text_without_tools() -> None: + assert hitl_status_text([]) == "Human input is required before the agent can continue." diff --git a/python/packages/kagent-langgraph/src/kagent/langgraph/_executor.py b/python/packages/kagent-langgraph/src/kagent/langgraph/_executor.py index f3ff8a6f8a..19a88bc1d4 100644 --- a/python/packages/kagent-langgraph/src/kagent/langgraph/_executor.py +++ b/python/packages/kagent-langgraph/src/kagent/langgraph/_executor.py @@ -29,8 +29,10 @@ ) from google.protobuf.json_format import MessageToDict from kagent.core.a2a import ( + AskUserRequest, HitlTool, ToolApprovalRequest, + ask_user_questions, attach_hitl_extension, get_ask_user_request, get_ask_user_response, @@ -39,6 +41,7 @@ get_tool_approval_request, get_tool_approval_response, hitl_activated, + hitl_status_text, now_timestamp, require_ask_user_response, require_tool_approval_response, @@ -69,6 +72,19 @@ class LangGraphAgentExecutorConfig(BaseModel): enable_streaming: bool = True +def _hitl_request(tools: list[HitlTool], text: str) -> AskUserRequest | ToolApprovalRequest: + """An ask_user interrupt becomes a question; anything else becomes an approval. + + An ask_user call with no answerable question becomes an approval too: an + empty question list gives a request that no response can satisfy, because + resume requires one answer per question. + """ + questions = ask_user_questions(tools) + if questions: + return AskUserRequest(id=tools[0].id, questions=questions) + return ToolApprovalRequest(hint=text, tools=tools) + + class LangGraphAgentExecutor(AgentExecutor): """An AgentExecutor that runs LangGraph workflows against A2A requests. @@ -224,29 +240,38 @@ async def _handle_interrupt( action, ) continue - tool_name = action["name"] - tool_args = action["args"] + tool_name = str(action.get("name") or "") # id is the opaque HITL correlation id; call_id is the tool call id. # Graphs typically set both to the LangChain tool call id. - correlation_id = action["id"] - call_id = action.get("call_id") or correlation_id - tools.append(HitlTool(id=correlation_id, call_id=call_id, name=tool_name, args=tool_args)) + correlation_id = str(action.get("id") or "") + if not tool_name or not correlation_id: + logger.warning("Skipping an action_request without a tool name or id: %r", action) + continue + tool_args = action.get("args") + call_id = str(action.get("call_id") or correlation_id) + tools.append( + HitlTool( + id=correlation_id, + call_id=call_id, + name=tool_name, + args=tool_args if isinstance(tool_args, dict) else {}, + ) + ) + # The text part names the tools so a client that did not activate the + # extension still learns what it is being asked. + text = hitl_status_text(tools) status_message = Message( message_id=str(uuid.uuid4()), role=Role.ROLE_AGENT, task_id=task_id, context_id=context_id, - parts=[Part(text="Human approval is required before the agent can continue.")], + parts=[Part(text=text)], ) - if hitl_enabled: - attach_hitl_extension( - status_message, - ToolApprovalRequest( - hint="Human approval is required before the agent can continue.", - tools=tools, - ), - ) + # With no usable action request there is nothing for a client to decide on, + # so the pause stays text-only rather than carrying an empty request. + if hitl_enabled and tools: + attach_hitl_extension(status_message, _hitl_request(tools, text)) await event_queue.enqueue_event( TaskStatusUpdateEvent( diff --git a/python/packages/kagent-langgraph/tests/test_langgraph_executor.py b/python/packages/kagent-langgraph/tests/test_langgraph_executor.py new file mode 100644 index 0000000000..9a94a3bfc3 --- /dev/null +++ b/python/packages/kagent-langgraph/tests/test_langgraph_executor.py @@ -0,0 +1,138 @@ +"""Tests for the LangGraph executor's interrupt handling.""" + +from unittest.mock import AsyncMock, MagicMock + +from kagent.core.a2a import get_ask_user_request, get_tool_approval_request + +from kagent.langgraph._executor import LangGraphAgentExecutor + + +def _executor() -> LangGraphAgentExecutor: + return LangGraphAgentExecutor(graph=MagicMock(), app_name="test__NS__agent") + + +def _interrupt(*names: str) -> list[dict]: + return [{"action_requests": [{"id": f"call-{name}", "name": name, "args": {"path": f"/{name}"}} for name in names]}] + + +def _ask_user_interrupt(*questions: str) -> list[dict]: + return [ + { + "action_requests": [ + { + "id": "call-ask", + "name": "ask_user", + "args": {"questions": [{"question": question} for question in questions]}, + } + ] + } + ] + + +async def _handle(hitl_enabled: bool, *names: str, interrupt_data: list[dict] | None = None): + event_queue = MagicMock() + event_queue.enqueue_event = AsyncMock() + await _executor()._handle_interrupt( + interrupt_data=interrupt_data or _interrupt(*names), + task_id="task-1", + context_id="context-1", + event_queue=event_queue, + hitl_enabled=hitl_enabled, + ) + event_queue.enqueue_event.assert_awaited_once() + return event_queue.enqueue_event.await_args.args[0].status.message + + +def _text(message) -> str: + return "".join(part.text for part in message.parts if part.HasField("text")) + + +async def test_interrupt_without_activation_names_the_tools(): + message = await _handle(False, "delete_file", "restart_pod") + + assert get_tool_approval_request(message) is None + assert _text(message) == "Approval is required for tool(s): delete_file, restart_pod" + + +async def test_interrupt_with_activation_keeps_the_same_text(): + message = await _handle(True, "delete_file") + + request = get_tool_approval_request(message) + + assert request is not None + assert request.tools[0].name == "delete_file" + assert request.hint == "Approval is required for tool(s): delete_file" + assert _text(message) == "Approval is required for tool(s): delete_file" + + +async def test_interrupt_ask_user_carries_the_questions(): + message = await _handle(True, interrupt_data=_ask_user_interrupt("Which namespace?", "Which cluster?")) + + request = get_ask_user_request(message) + + assert request is not None + assert request.questions == [{"question": "Which namespace?"}, {"question": "Which cluster?"}] + assert _text(message) == "Which namespace?; Which cluster?" + + +async def test_interrupt_ask_user_without_activation_still_asks(): + message = await _handle(False, interrupt_data=_ask_user_interrupt("Which namespace?")) + + assert get_ask_user_request(message) is None + assert _text(message) == "Which namespace?" + + +async def test_interrupt_skips_action_requests_without_a_name_or_id(): + interrupt_data = [{"action_requests": [{"args": {"path": "/tmp"}}, {"id": "call-1"}, "not-a-mapping"]}] + + message = await _handle(True, interrupt_data=interrupt_data) + + assert get_tool_approval_request(message) is None + assert _text(message) == "Human input is required before the agent can continue." + + +async def test_interrupt_normalizes_non_dict_args(): + interrupt_data = [{"action_requests": [{"id": "call-1", "name": "delete_file", "args": "/tmp/x"}]}] + + message = await _handle(True, interrupt_data=interrupt_data) + + request = get_tool_approval_request(message) + + assert request is not None + assert request.tools[0].args == {} + assert _text(message) == "Approval is required for tool(s): delete_file" + + +async def test_interrupt_ask_user_without_questions_asks_for_approval(): + interrupt_data = [{"action_requests": [{"id": "call-ask", "name": "ask_user", "args": {"questions": []}}]}] + + message = await _handle(True, interrupt_data=interrupt_data) + + request = get_tool_approval_request(message) + + assert get_ask_user_request(message) is None + assert request is not None + assert request.tools[0].name == "ask_user" + assert _text(message) == "Approval is required for tool(s): ask_user" + + +async def test_interrupt_ask_user_drops_questions_without_text(): + interrupt_data = [ + { + "action_requests": [ + { + "id": "call-ask", + "name": "ask_user", + "args": {"questions": [{"question": ""}, "not-a-mapping", {"question": "Which cluster?"}]}, + } + ] + } + ] + + message = await _handle(True, interrupt_data=interrupt_data) + + request = get_ask_user_request(message) + + assert request is not None + assert request.questions == [{"question": "Which cluster?"}] + assert _text(message) == "Which cluster?"