diff --git a/pkg/agent/agent_tool_test.go b/pkg/agent/agent_tool_test.go index 2164ec5fd..a95447b21 100644 --- a/pkg/agent/agent_tool_test.go +++ b/pkg/agent/agent_tool_test.go @@ -375,6 +375,420 @@ func TestAgentTool_Execute(t *testing.T) { ) } +func TestAgentTool_Execute_NestedApproval(t *testing.T) { + t.Parallel() + + t.Run( + "nested agent approval surfaces as InterruptedError", + func(t *testing.T) { + t.Parallel() + + deleteTool := agent.FunctionTool[struct{}]( + "delete_file", + "Delete a file", + func(_ context.Context, _ struct{}) (agent.ToolResult, error) { + return agent.ToolResult{Content: "file deleted"}, nil + }, + ) + + innerProvider := &mockProvider{ + responses: []*llm.ChatCompletionResponse{ + toolCallResponse(llm.ToolCall{ + ID: "inner_tc1", + Function: llm.FunctionCall{Name: "delete_file", Arguments: `{}`}, + }), + }, + } + + innerAgent := agent.New( + "file_manager", + newTestClient(innerProvider), + agent.WithModel("test-model"), + agent.WithTools(deleteTool), + agent.WithApproval(agent.ApprovalConfig{ + ToolNames: []string{"delete_file"}, + }), + ) + + outerProvider := &mockProvider{ + responses: []*llm.ChatCompletionResponse{ + toolCallResponse(llm.ToolCall{ + ID: "outer_tc1", + Function: llm.FunctionCall{Name: "file_expert", Arguments: `{"input":"delete the file"}`}, + }), + }, + } + + outerAgent := agent.New( + "assistant", + newTestClient(outerProvider), + agent.WithModel("test-model"), + agent.WithTools(innerAgent.AsTool("file_expert", "Manage files")), + ) + + _, err := outerAgent.Run( + context.Background(), + []llm.Message{userMessage("Delete the file")}, + ) + + require.Error(t, err) + var interrupted *agent.InterruptedError + require.ErrorAs(t, err, &interrupted) + assert.Len(t, interrupted.PendingApprovals, 1) + assert.Equal(t, "delete_file", interrupted.PendingApprovals[0].Function.Name) + assert.Equal(t, "file_manager", interrupted.Agent.Name()) + }, + ) + + t.Run( + "nested agent approval can be resumed with approve", + func(t *testing.T) { + t.Parallel() + + var toolExecuted bool + + deleteTool := agent.FunctionTool[struct{}]( + "delete_file", + "Delete a file", + func(_ context.Context, _ struct{}) (agent.ToolResult, error) { + toolExecuted = true + return agent.ToolResult{Content: "file deleted"}, nil + }, + ) + + innerProvider := &mockProvider{ + responses: []*llm.ChatCompletionResponse{ + toolCallResponse(llm.ToolCall{ + ID: "inner_tc1", + Function: llm.FunctionCall{Name: "delete_file", Arguments: `{}`}, + }), + stopResponse("File has been deleted."), + }, + } + + innerAgent := agent.New( + "file_manager", + newTestClient(innerProvider), + agent.WithModel("test-model"), + agent.WithTools(deleteTool), + agent.WithApproval(agent.ApprovalConfig{ + ToolNames: []string{"delete_file"}, + }), + ) + + outerProvider := &mockProvider{ + responses: []*llm.ChatCompletionResponse{ + toolCallResponse(llm.ToolCall{ + ID: "outer_tc1", + Function: llm.FunctionCall{Name: "file_expert", Arguments: `{"input":"delete the file"}`}, + }), + stopResponse("Done, the file has been deleted."), + }, + } + + outerAgent := agent.New( + "assistant", + newTestClient(outerProvider), + agent.WithModel("test-model"), + agent.WithTools(innerAgent.AsTool("file_expert", "Manage files")), + ) + + _, err := outerAgent.Run( + context.Background(), + []llm.Message{userMessage("Delete the file")}, + ) + + var interrupted *agent.InterruptedError + require.ErrorAs(t, err, &interrupted) + assert.False(t, toolExecuted) + + result, err := agent.Resume( + context.Background(), + interrupted, + agent.ResumeInput{ + Approvals: map[string]agent.ApprovalResult{ + "inner_tc1": {Approved: true}, + }, + }, + ) + + require.NoError(t, err) + assert.True(t, toolExecuted) + assert.Equal(t, "Done, the file has been deleted.", result.FinalMessage().Text()) + assert.Equal(t, "assistant", result.LastAgent.Name()) + }, + ) + + t.Run( + "nested agent rejection resumes outer agent", + func(t *testing.T) { + t.Parallel() + + deleteTool := agent.FunctionTool[struct{}]( + "delete_file", + "Delete a file", + func(_ context.Context, _ struct{}) (agent.ToolResult, error) { + t.Fatal("tool should not be called") + return agent.ToolResult{}, nil + }, + ) + + innerProvider := &mockProvider{ + responses: []*llm.ChatCompletionResponse{ + toolCallResponse(llm.ToolCall{ + ID: "inner_tc1", + Function: llm.FunctionCall{Name: "delete_file", Arguments: `{}`}, + }), + stopResponse("OK, I won't delete the file."), + }, + } + + innerAgent := agent.New( + "file_manager", + newTestClient(innerProvider), + agent.WithModel("test-model"), + agent.WithTools(deleteTool), + agent.WithApproval(agent.ApprovalConfig{ + ToolNames: []string{"delete_file"}, + }), + ) + + outerProvider := &mockProvider{ + responses: []*llm.ChatCompletionResponse{ + toolCallResponse(llm.ToolCall{ + ID: "outer_tc1", + Function: llm.FunctionCall{Name: "file_expert", Arguments: `{"input":"delete the file"}`}, + }), + stopResponse("The file manager declined."), + }, + } + + outerAgent := agent.New( + "assistant", + newTestClient(outerProvider), + agent.WithModel("test-model"), + agent.WithTools(innerAgent.AsTool("file_expert", "Manage files")), + ) + + _, err := outerAgent.Run( + context.Background(), + []llm.Message{userMessage("Delete the file")}, + ) + + var interrupted *agent.InterruptedError + require.ErrorAs(t, err, &interrupted) + + result, err := agent.Resume( + context.Background(), + interrupted, + agent.ResumeInput{ + Approvals: map[string]agent.ApprovalResult{ + "inner_tc1": {Approved: false, Message: "User denied deletion."}, + }, + }, + ) + + require.NoError(t, err) + assert.Equal(t, "The file manager declined.", result.FinalMessage().Text()) + assert.Equal(t, "assistant", result.LastAgent.Name()) + }, + ) + + t.Run( + "nested agent approval with parallel sibling tools", + func(t *testing.T) { + t.Parallel() + + var siblingCalled bool + + type Params struct{} + siblingTool := agent.FunctionTool[Params]( + "list_files", + "List files", + func(_ context.Context, _ Params) (agent.ToolResult, error) { + siblingCalled = true + return agent.ToolResult{Content: "file1.txt, file2.txt"}, nil + }, + ) + + deleteTool := agent.FunctionTool[struct{}]( + "delete_file", + "Delete a file", + func(_ context.Context, _ struct{}) (agent.ToolResult, error) { + return agent.ToolResult{Content: "file deleted"}, nil + }, + ) + + innerProvider := &mockProvider{ + responses: []*llm.ChatCompletionResponse{ + toolCallResponse(llm.ToolCall{ + ID: "inner_tc1", + Function: llm.FunctionCall{Name: "delete_file", Arguments: `{}`}, + }), + stopResponse("File has been deleted."), + }, + } + + innerAgent := agent.New( + "file_manager", + newTestClient(innerProvider), + agent.WithModel("test-model"), + agent.WithTools(deleteTool), + agent.WithApproval(agent.ApprovalConfig{ + ToolNames: []string{"delete_file"}, + }), + ) + + outerProvider := &mockProvider{ + responses: []*llm.ChatCompletionResponse{ + toolCallResponse( + llm.ToolCall{ + ID: "outer_tc1", + Function: llm.FunctionCall{Name: "list_files", Arguments: `{}`}, + }, + llm.ToolCall{ + ID: "outer_tc2", + Function: llm.FunctionCall{Name: "file_expert", Arguments: `{"input":"delete the file"}`}, + }, + ), + stopResponse("Files listed and deleted."), + }, + } + + outerAgent := agent.New( + "assistant", + newTestClient(outerProvider), + agent.WithModel("test-model"), + agent.WithTools( + siblingTool, + innerAgent.AsTool("file_expert", "Manage files"), + ), + ) + + _, err := outerAgent.Run( + context.Background(), + []llm.Message{userMessage("List and delete files")}, + ) + + var interrupted *agent.InterruptedError + require.ErrorAs(t, err, &interrupted) + assert.True(t, siblingCalled) + + result, err := agent.Resume( + context.Background(), + interrupted, + agent.ResumeInput{ + Approvals: map[string]agent.ApprovalResult{ + "inner_tc1": {Approved: true}, + }, + }, + ) + + require.NoError(t, err) + assert.Equal(t, "Files listed and deleted.", result.FinalMessage().Text()) + }, + ) + + t.Run( + "three-level nesting A to B to C preserves full chain", + func(t *testing.T) { + t.Parallel() + + var toolExecuted bool + + dangerTool := agent.FunctionTool[struct{}]( + "danger", + "Dangerous operation", + func(_ context.Context, _ struct{}) (agent.ToolResult, error) { + toolExecuted = true + return agent.ToolResult{Content: "danger executed"}, nil + }, + ) + + cProvider := &mockProvider{ + responses: []*llm.ChatCompletionResponse{ + toolCallResponse(llm.ToolCall{ + ID: "c_tc1", + Function: llm.FunctionCall{Name: "danger", Arguments: `{}`}, + }), + stopResponse("C done."), + }, + } + + agentC := agent.New( + "agent_c", + newTestClient(cProvider), + agent.WithModel("test-model"), + agent.WithTools(dangerTool), + agent.WithApproval(agent.ApprovalConfig{ + ToolNames: []string{"danger"}, + }), + ) + + bProvider := &mockProvider{ + responses: []*llm.ChatCompletionResponse{ + toolCallResponse(llm.ToolCall{ + ID: "b_tc1", + Function: llm.FunctionCall{Name: "call_c", Arguments: `{"input":"do danger"}`}, + }), + stopResponse("B done."), + }, + } + + agentB := agent.New( + "agent_b", + newTestClient(bProvider), + agent.WithModel("test-model"), + agent.WithTools(agentC.AsTool("call_c", "Call agent C")), + ) + + aProvider := &mockProvider{ + responses: []*llm.ChatCompletionResponse{ + toolCallResponse(llm.ToolCall{ + ID: "a_tc1", + Function: llm.FunctionCall{Name: "call_b", Arguments: `{"input":"delegate to C"}`}, + }), + stopResponse("A done."), + }, + } + + agentA := agent.New( + "agent_a", + newTestClient(aProvider), + agent.WithModel("test-model"), + agent.WithTools(agentB.AsTool("call_b", "Call agent B")), + ) + + _, err := agentA.Run( + context.Background(), + []llm.Message{userMessage("start")}, + ) + + require.Error(t, err) + var interrupted *agent.InterruptedError + require.ErrorAs(t, err, &interrupted) + assert.Equal(t, "agent_c", interrupted.Agent.Name()) + assert.Equal(t, "danger", interrupted.PendingApprovals[0].Function.Name) + assert.False(t, toolExecuted) + + result, err := agent.Resume( + context.Background(), + interrupted, + agent.ResumeInput{ + Approvals: map[string]agent.ApprovalResult{ + "c_tc1": {Approved: true}, + }, + }, + ) + + require.NoError(t, err) + assert.True(t, toolExecuted) + assert.Equal(t, "A done.", result.FinalMessage().Text()) + assert.Equal(t, "agent_a", result.LastAgent.Name()) + }, + ) +} + func TestAgentTool_InterfaceSatisfaction(t *testing.T) { t.Parallel() diff --git a/pkg/agent/errors.go b/pkg/agent/errors.go index df59c79b3..595597f3f 100644 --- a/pkg/agent/errors.go +++ b/pkg/agent/errors.go @@ -42,12 +42,37 @@ type ( Messages []llm.Message Usage llm.Usage Turns int + + outerState *outerLoopState } needsApprovalError struct { allToolCalls []llm.ToolCall pendingApprovals []llm.ToolCall } + + nestedInterruptionError struct { + inner *InterruptedError + toolCallID string + allToolCalls []llm.ToolCall + completedCalls []completedCall + } + + outerLoopState struct { + agent *Agent + messages []llm.Message + usage llm.Usage + turns int + allToolCalls []llm.ToolCall + toolCallID string + completedCalls []completedCall + innerInterrupt *InterruptedError + } + + completedCall struct { + toolCallID string + result ToolResult + } ) func (e *MaxTurnsExceededError) Error() string { @@ -69,3 +94,7 @@ func (e *InterruptedError) Error() string { func (e *needsApprovalError) Error() string { return fmt.Sprintf("%d tool call(s) require approval", len(e.pendingApprovals)) } + +func (e *nestedInterruptionError) Error() string { + return fmt.Sprintf("nested agent interrupted: %s", e.inner.Error()) +} diff --git a/pkg/agent/run.go b/pkg/agent/run.go index a0df7cbe9..17d6c7f47 100644 --- a/pkg/agent/run.go +++ b/pkg/agent/run.go @@ -396,6 +396,42 @@ func coreLoop(ctx context.Context, startAgent *Agent, inputMessages []llm.Messag }, ) } + + if nie, ok := errors.AsType[*nestedInterruptionError](err); ok { + s.logger.InfoCtx( + ctx, + "nested agent interrupted, approval required", + log.Int("pending_count", len(nie.inner.PendingApprovals)), + log.String("nested_agent", nie.inner.Agent.name), + ) + + msgsCopy := make([]llm.Message, len(s.messages)) + copy(msgsCopy, s.messages) + + return s.finishRun( + ctx, + nil, + &InterruptedError{ + ToolCalls: nie.inner.ToolCalls, + PendingApprovals: nie.inner.PendingApprovals, + Agent: nie.inner.Agent, + Messages: nie.inner.Messages, + Usage: nie.inner.Usage, + Turns: nie.inner.Turns, + outerState: &outerLoopState{ + agent: s.agent, + messages: msgsCopy, + usage: s.totalUsage, + turns: s.turns, + allToolCalls: nie.allToolCalls, + toolCallID: nie.toolCallID, + completedCalls: nie.completedCalls, + innerInterrupt: nie.inner, + }, + }, + ) + } + return s.finishRun(ctx, nil, err) } @@ -553,6 +589,24 @@ func executeWithHandoff( tr, err := executeSingleTool(ctx, tracer, agent, toolCalls[i], descriptors[i].(Tool), onEvent, logger) if err != nil { + if ie, ok := errors.AsType[*InterruptedError](err); ok { + var completed []completedCall + for j := range results { + completed = append( + completed, + completedCall{ + toolCallID: toolCalls[j].ID, + result: results[j].Result, + }, + ) + } + return nil, nil, msgs, &nestedInterruptionError{ + inner: ie, + toolCallID: toolCalls[i].ID, + allToolCalls: toolCalls, + completedCalls: completed, + } + } return nil, nil, msgs, err } @@ -622,6 +676,44 @@ func executeParallel( wg.Wait() + for i, entry := range entries { + var ie *InterruptedError + if entry.err != nil && errors.As(entry.err, &ie) { + var completed []completedCall + for j, other := range entries { + if j == i { + continue + } + if other.err != nil { + completed = append( + completed, + completedCall{ + toolCallID: toolCalls[j].ID, + result: ToolResult{ + Content: fmt.Sprintf("Error: %s", other.err.Error()), + IsError: true, + }, + }, + ) + continue + } + completed = append( + completed, + completedCall{ + toolCallID: toolCalls[j].ID, + result: other.result, + }, + ) + } + return nil, nil, &nestedInterruptionError{ + inner: ie, + toolCallID: toolCalls[i].ID, + allToolCalls: toolCalls, + completedCalls: completed, + } + } + } + var ( results []ToolCallResult msgs []llm.Message @@ -710,6 +802,12 @@ func executeSingleTool( result, err := tool.Execute(toolCtx, tc.Function.Arguments) if err != nil { + if _, ok := errors.AsType[*InterruptedError](err); ok { + toolSpan.SetAttributes(attribute.Bool("tool.interrupted", true)) + toolSpan.End() + return ToolResult{}, err + } + toolSpan.RecordError(err) toolSpan.SetStatus(codes.Error, err.Error()) toolSpan.End() @@ -812,6 +910,10 @@ func runOutputGuardrails(ctx context.Context, agent *Agent, message llm.Message) // are not re-evaluated because the messages were already validated in the // original Run call. func Resume(ctx context.Context, interrupted *InterruptedError, input ResumeInput) (*Result, error) { + if interrupted.outerState != nil { + return resumeNested(ctx, interrupted, input) + } + tracer := otel.GetTracerProvider().Tracer(tracerName) logger := interrupted.Agent.logger @@ -965,6 +1067,86 @@ func Resume(ctx context.Context, interrupted *InterruptedError, input ResumeInpu ) } +func resumeNested(ctx context.Context, interrupted *InterruptedError, input ResumeInput) (*Result, error) { + outer := interrupted.outerState + logger := outer.agent.logger + + logger.InfoCtx( + ctx, + "resuming nested agent interruption", + log.String("outer_agent", outer.agent.name), + log.String("inner_agent", interrupted.Agent.name), + ) + + innerResult, err := Resume(ctx, outer.innerInterrupt, input) + if err != nil { + var innerIE *InterruptedError + if errors.As(err, &innerIE) { + return nil, &InterruptedError{ + ToolCalls: innerIE.ToolCalls, + PendingApprovals: innerIE.PendingApprovals, + Agent: innerIE.Agent, + Messages: innerIE.Messages, + Usage: innerIE.Usage, + Turns: innerIE.Turns, + outerState: &outerLoopState{ + agent: outer.agent, + messages: outer.messages, + usage: outer.usage, + turns: outer.turns, + allToolCalls: outer.allToolCalls, + toolCallID: outer.toolCallID, + completedCalls: outer.completedCalls, + innerInterrupt: innerIE, + }, + } + } + return nil, fmt.Errorf("cannot resume nested agent: %w", err) + } + + completedMap := make(map[string]ToolResult, len(outer.completedCalls)) + for _, cc := range outer.completedCalls { + completedMap[cc.toolCallID] = cc.result + } + + messages := make([]llm.Message, len(outer.messages)) + copy(messages, outer.messages) + + for _, tc := range outer.allToolCalls { + var content string + if tc.ID == outer.toolCallID { + content = innerResult.FinalMessage().Text() + } else if cr, ok := completedMap[tc.ID]; ok { + content = cr.Content + } else { + content = "Error: tool execution was interrupted" + } + + messages = append( + messages, + llm.Message{ + Role: llm.RoleTool, + ToolCallID: tc.ID, + Parts: []llm.Part{llm.TextPart{Text: content}}, + }, + ) + } + + return coreLoop( + ctx, + outer.agent, + messages, + runOpts{ + callLLM: blockingCallLLM, + onEvent: noopEvent, + skipInputGuardrails: true, + skipSessionLoad: true, + initialUsage: outer.usage, + initialTurns: outer.turns, + }, + ) +} + func emitHook(agent *Agent, fn func(RunHooks)) { for _, h := range agent.hooks { fn(h)