Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 12 additions & 1 deletion python/packages/kagent-adk/src/kagent/adk/_hitl.py
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,18 @@ def visible_tools(request: ToolApprovalRequest | AskUserRequest) -> list[HitlToo

def remote_hitl_hint(state: RemoteHitlState) -> str:
"""Build the parent confirmation hint for a paused child task."""
names = [tool.name for tool in visible_tools(state.hitl_request)]
request = state.hitl_request
if isinstance(request, AskUserRequest):
# questions carries the real text whether or not nested is set (see
# build_hitl_status_message), so prefer it over the bare tool name.
question_text = " ".join(
question["question"]
for question in request.questions
if isinstance(question.get("question"), str) and question["question"]
)
if question_text:
return f"Remote agent '{state.subagent_name}' asks: {question_text}"
names = [tool.name for tool in visible_tools(request)]
if names:
return f"Remote agent '{state.subagent_name}' requires approval for tool(s): {', '.join(names)}"
return f"Remote agent '{state.subagent_name}' requires human input before continuing."
Expand Down
50 changes: 50 additions & 0 deletions python/packages/kagent-adk/tests/unittests/test_hitl.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,9 @@
from kagent.adk._hitl import (
RemoteHitlState,
build_hitl_status_message,
build_remote_hitl_state,
build_resume_hitl_message,
remote_hitl_hint,
)


Expand Down Expand Up @@ -285,6 +287,54 @@ def test_resume_rejects_non_input_required_task():
)


def test_remote_hitl_hint_tool_approval():
task = _stored_task(ToolApprovalRequest(tools=[_tool("child-confirm", "delete_pod")]))
state = build_remote_hitl_state(task, "k8s_agent")

assert state is not None
assert remote_hitl_hint(state) == "Remote agent 'k8s_agent' requires approval for tool(s): delete_pod"


def test_remote_hitl_hint_ask_user():
task = _stored_task(
AskUserRequest(id="confirm-1", questions=[{"question": "What is the GitHub owner/org for the repo?"}])
)
state = build_remote_hitl_state(task, "github_agent")

assert state is not None
assert remote_hitl_hint(state) == "Remote agent 'github_agent' asks: What is the GitHub owner/org for the repo?"


def test_remote_hitl_hint_ask_user_nested():
"""A two-level nested ask_user pause should also surface the real question
from the top-level questions field, not just the bare 'ask_user' tool name."""
question = "What is the GitHub owner/org for the repo?"
task = _stored_task(
AskUserRequest(
id="confirm-1",
questions=[{"question": question}],
nested=NestedHitlRequest(
subagent_name="grandchild_agent",
task_id="grandchild-task",
context_id="grandchild-context",
tools=[
HitlTool(
id="confirm-2",
call_id="confirm-2",
name="ask_user",
args={"questions": [{"question": question}]},
)
],
),
)
)
state = build_remote_hitl_state(task, "github_agent")

assert state is not None
assert state.hitl_request.nested is not None
assert remote_hitl_hint(state) == f"Remote agent 'github_agent' asks: {question}"


def test_resume_rejects_input_required_task_without_public_hitl_request():
task = Task(
id="task-1",
Expand Down
Loading