From d7fbcc6bcd44af8cdcab1b6c67b9663fabc80cd2 Mon Sep 17 00:00:00 2001 From: Raphael Simon Date: Tue, 7 Jul 2026 18:58:53 -0700 Subject: [PATCH] fix(runtime): advance run attempt before resumed tool calls --- runtime/agent/runtime/helpers.go | 5 +- .../workflow_bookkeeping_contract_test.go | 198 ++++++++++++++++++ runtime/agent/runtime/workflow_helpers.go | 6 +- 3 files changed, 205 insertions(+), 4 deletions(-) diff --git a/runtime/agent/runtime/helpers.go b/runtime/agent/runtime/helpers.go index b74459f4..6bafc5c4 100644 --- a/runtime/agent/runtime/helpers.go +++ b/runtime/agent/runtime/helpers.go @@ -144,8 +144,9 @@ func nestedRunIDSuffix(toolCallID string) string { // // Attempt is required to avoid ID collisions when the same run executes multiple // tool batches within a single logical turn (for example when callers set TurnID -// to a constant run identifier). The runtime increments Attempt on each planner -// iteration so generated IDs remain unique within the run. +// to a constant run identifier). The workflow stamps RunContext.Attempt with the +// planner turn attempt before executing that turn's generated tool calls, so +// generated IDs remain unique within the run. func generateDeterministicToolCallID(runID, turnID string, attempt int, toolName tools.Ident, index int) string { if runID == "" { runID = unknownID diff --git a/runtime/agent/runtime/workflow_bookkeeping_contract_test.go b/runtime/agent/runtime/workflow_bookkeeping_contract_test.go index eef449d4..77c12f75 100644 --- a/runtime/agent/runtime/workflow_bookkeeping_contract_test.go +++ b/runtime/agent/runtime/workflow_bookkeeping_contract_test.go @@ -300,6 +300,204 @@ func TestRunLoopRetryableBookkeepingTerminalFailureResumes(t *testing.T) { require.Equal(t, model.ConversationRoleSystem, wfCtx.lastPlannerCall.Input.Messages[2].Role) } +func TestRunLoopProviderEmptyToolCallIDsAdvanceAcrossResumeAttempts(t *testing.T) { + cases := []struct { + name string + tool tools.Ident + want []string + }{ + { + name: "tool unavailable resumes", + tool: tools.ToolUnavailable, + want: []string{ + "run-1/turn-1/attempt-1/runtime-tool_unavailable/0", + "run-1/turn-1/attempt-2/runtime-tool_unavailable/0", + "run-1/turn-1/attempt-3/runtime-tool_unavailable/0", + }, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + rt := New(WithLogger(telemetry.NoopLogger{})) + agentID := agent.Ident("agent-1") + var resumeAttempts []int + rt.agents[agentID] = AgentRegistration{ + ID: agentID, + Planner: &stubPlanner{resume: func(ctx context.Context, input *planner.PlanResumeInput) (*planner.PlanResult, error) { + resumeAttempts = append(resumeAttempts, input.RunContext.Attempt) + switch len(resumeAttempts) { + case 1, 2: + require.Len(t, input.ToolOutputs, len(resumeAttempts)) + return &planner.PlanResult{ + ToolCalls: []planner.ToolRequest{{Name: tc.tool}}, + }, nil + case 3: + require.Len(t, input.ToolOutputs, len(tc.want)) + return &planner.PlanResult{ + FinalResponse: &planner.FinalResponse{ + Message: &model.Message{ + Role: model.ConversationRoleAssistant, + Parts: []model.Part{model.TextPart{Text: "done"}}, + }, + }, + }, nil + default: + require.FailNow(t, "unexpected resume attempt") + } + return nil, nil + }}, + } + wfCtx := &routeWorkflowContext{ + ctx: context.Background(), + runID: "run-1", + hookRuntime: rt, + plannerRoutes: map[string]func(context.Context, *PlanActivityInput) (*PlanActivityOutput, error){ + "resume": func(ctx context.Context, input *PlanActivityInput) (*PlanActivityOutput, error) { + return rt.PlanResumeActivity(ctx, input) + }, + }, + } + base := &planner.PlanInput{ + RunContext: run.Context{ + RunID: "run-1", + SessionID: "sess-1", + TurnID: "turn-1", + Attempt: 1, + }, + } + input := &RunInput{ + AgentID: agentID, + RunID: "run-1", + SessionID: "sess-1", + TurnID: "turn-1", + } + initial := &planner.PlanResult{ + ToolCalls: []planner.ToolRequest{{Name: tc.tool}}, + } + + out, err := rt.runLoop( + wfCtx, + AgentRegistration{ID: agentID, ExecuteToolActivity: "execute", ResumeActivityName: "resume"}, + input, + base, + initial, + nil, + model.TokenUsage{}, + policy.CapsState{}, + time.Time{}, + time.Time{}, + 2, + "turn-1", + nil, + nil, + 0, + ) + require.NoError(t, err) + require.NotNil(t, out) + require.Equal(t, []int{2, 3, 4}, resumeAttempts) + require.Len(t, out.ToolEvents, len(tc.want)) + for i, want := range tc.want { + require.Equal(t, want, out.ToolEvents[i].ToolCallID) + } + require.NotEqual(t, out.ToolEvents[1].ToolCallID, out.ToolEvents[2].ToolCallID) + }) + } +} + +func TestRunLoopProviderEmptyToolCallIDsUseBatchIndexes(t *testing.T) { + cases := []struct { + name string + calls []planner.ToolRequest + want []string + }{ + { + name: "two tool unavailable calls", + calls: []planner.ToolRequest{ + {Name: tools.ToolUnavailable}, + {Name: tools.ToolUnavailable}, + }, + want: []string{ + "run-1/turn-1/attempt-1/runtime-tool_unavailable/0", + "run-1/turn-1/attempt-1/runtime-tool_unavailable/1", + }, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + rt := New(WithLogger(telemetry.NoopLogger{})) + agentID := agent.Ident("agent-1") + rt.agents[agentID] = AgentRegistration{ + ID: agentID, + Planner: &stubPlanner{resume: func(ctx context.Context, input *planner.PlanResumeInput) (*planner.PlanResult, error) { + require.Len(t, input.ToolOutputs, len(tc.want)) + return &planner.PlanResult{ + FinalResponse: &planner.FinalResponse{ + Message: &model.Message{ + Role: model.ConversationRoleAssistant, + Parts: []model.Part{model.TextPart{Text: "done"}}, + }, + }, + }, nil + }}, + } + wfCtx := &routeWorkflowContext{ + ctx: context.Background(), + runID: "run-1", + hookRuntime: rt, + plannerRoutes: map[string]func(context.Context, *PlanActivityInput) (*PlanActivityOutput, error){ + "resume": func(ctx context.Context, input *PlanActivityInput) (*PlanActivityOutput, error) { + return rt.PlanResumeActivity(ctx, input) + }, + }, + } + base := &planner.PlanInput{ + RunContext: run.Context{ + RunID: "run-1", + SessionID: "sess-1", + TurnID: "turn-1", + Attempt: 1, + }, + } + input := &RunInput{ + AgentID: agentID, + RunID: "run-1", + SessionID: "sess-1", + TurnID: "turn-1", + } + initial := &planner.PlanResult{ + ToolCalls: tc.calls, + } + + out, err := rt.runLoop( + wfCtx, + AgentRegistration{ID: agentID, ExecuteToolActivity: "execute", ResumeActivityName: "resume"}, + input, + base, + initial, + nil, + model.TokenUsage{}, + policy.CapsState{}, + time.Time{}, + time.Time{}, + 2, + "turn-1", + nil, + nil, + 0, + ) + require.NoError(t, err) + require.NotNil(t, out) + require.Len(t, out.ToolEvents, len(tc.want)) + for i, want := range tc.want { + require.Equal(t, want, out.ToolEvents[i].ToolCallID) + } + require.NotEqual(t, out.ToolEvents[0].ToolCallID, out.ToolEvents[1].ToolCallID) + }) + } +} + func TestRunLoopMixedBudgetedAndBookkeepingStillResumes(t *testing.T) { rt := New(WithLogger(telemetry.NoopLogger{})) diff --git a/runtime/agent/runtime/workflow_helpers.go b/runtime/agent/runtime/workflow_helpers.go index 47d3a9eb..673a3e6d 100644 --- a/runtime/agent/runtime/workflow_helpers.go +++ b/runtime/agent/runtime/workflow_helpers.go @@ -378,9 +378,9 @@ func (r *Runtime) buildNextResumeRequest( toolOutputs []*planner.ToolOutput, nextAttempt *int, ) (PlanActivityInput, error) { + attempt := *nextAttempt resumeCtx := base.RunContext - resumeCtx.Attempt = *nextAttempt - *nextAttempt++ + resumeCtx.Attempt = attempt plannerMsgs := cloneMessages(base.Messages) if err := transcript.ValidatePlannerTranscript(plannerMsgs); err != nil { return PlanActivityInput{}, fmt.Errorf("invalid resume transcript for run %s: %w", base.RunContext.RunID, err) @@ -400,5 +400,7 @@ func (r *Runtime) buildNextResumeRequest( if err := enforcePlanActivityInputBudget(out); err != nil { return PlanActivityInput{}, err } + base.RunContext.Attempt = attempt + *nextAttempt = attempt + 1 return out, nil }