Propagate nested agent approval interruptions through tool pipeline
Signed-off-by: Bryan Frimin <bryan@getprobo.com>
This commit is contained in:
@@ -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) {
|
func TestAgentTool_InterfaceSatisfaction(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
|
|||||||
@@ -42,12 +42,37 @@ type (
|
|||||||
Messages []llm.Message
|
Messages []llm.Message
|
||||||
Usage llm.Usage
|
Usage llm.Usage
|
||||||
Turns int
|
Turns int
|
||||||
|
|
||||||
|
outerState *outerLoopState
|
||||||
}
|
}
|
||||||
|
|
||||||
needsApprovalError struct {
|
needsApprovalError struct {
|
||||||
allToolCalls []llm.ToolCall
|
allToolCalls []llm.ToolCall
|
||||||
pendingApprovals []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 {
|
func (e *MaxTurnsExceededError) Error() string {
|
||||||
@@ -69,3 +94,7 @@ func (e *InterruptedError) Error() string {
|
|||||||
func (e *needsApprovalError) Error() string {
|
func (e *needsApprovalError) Error() string {
|
||||||
return fmt.Sprintf("%d tool call(s) require approval", len(e.pendingApprovals))
|
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())
|
||||||
|
}
|
||||||
|
|||||||
182
pkg/agent/run.go
182
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)
|
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)
|
tr, err := executeSingleTool(ctx, tracer, agent, toolCalls[i], descriptors[i].(Tool), onEvent, logger)
|
||||||
if err != nil {
|
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
|
return nil, nil, msgs, err
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -622,6 +676,44 @@ func executeParallel(
|
|||||||
|
|
||||||
wg.Wait()
|
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 (
|
var (
|
||||||
results []ToolCallResult
|
results []ToolCallResult
|
||||||
msgs []llm.Message
|
msgs []llm.Message
|
||||||
@@ -710,6 +802,12 @@ func executeSingleTool(
|
|||||||
|
|
||||||
result, err := tool.Execute(toolCtx, tc.Function.Arguments)
|
result, err := tool.Execute(toolCtx, tc.Function.Arguments)
|
||||||
if err != nil {
|
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.RecordError(err)
|
||||||
toolSpan.SetStatus(codes.Error, err.Error())
|
toolSpan.SetStatus(codes.Error, err.Error())
|
||||||
toolSpan.End()
|
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
|
// are not re-evaluated because the messages were already validated in the
|
||||||
// original Run call.
|
// original Run call.
|
||||||
func Resume(ctx context.Context, interrupted *InterruptedError, input ResumeInput) (*Result, error) {
|
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)
|
tracer := otel.GetTracerProvider().Tracer(tracerName)
|
||||||
logger := interrupted.Agent.logger
|
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)) {
|
func emitHook(agent *Agent, fn func(RunHooks)) {
|
||||||
for _, h := range agent.hooks {
|
for _, h := range agent.hooks {
|
||||||
fn(h)
|
fn(h)
|
||||||
|
|||||||
Reference in New Issue
Block a user