From 68ad5953d728da407985a657126bb6f1bd1d472c Mon Sep 17 00:00:00 2001 From: Thanh_Dang Date: Wed, 24 Jun 2026 10:49:31 +0700 Subject: [PATCH] fix(pipeline): defer non-tool messages until all tool results are emitted (#1270) Multi-tool responses could interleave synthetic user messages (loop warnings, nudges) between tool results, breaking OpenAI-compatible providers that require contiguous tool_result grouping. Fix both sequential and parallel tool execution paths in tool_stage.go to defer non-tool messages until after all tool results are appended. Fix sanitizeHistory lookahead in loop_history_sanitize.go to reorder interleaved non-tool messages in persisted session history. Closes #1177 --- internal/agent/loop_history_edge_test.go | 116 +++++++++++++++++++++++ internal/agent/loop_history_sanitize.go | 57 ++++++++--- internal/pipeline/stages_test.go | 102 ++++++++++++++++++++ internal/pipeline/tool_stage.go | 44 +++++++-- 4 files changed, 299 insertions(+), 20 deletions(-) diff --git a/internal/agent/loop_history_edge_test.go b/internal/agent/loop_history_edge_test.go index e4244230..dfcf13ec 100644 --- a/internal/agent/loop_history_edge_test.go +++ b/internal/agent/loop_history_edge_test.go @@ -2,6 +2,7 @@ package agent import ( "fmt" + "slices" "strings" "testing" @@ -264,3 +265,118 @@ func TestSanitizeHistory_EmptyToolCallID(t *testing.T) { t.Fatalf("expected 3 messages, got %d", len(result)) } } + +// --- sanitizeHistory: interleaved user warning between tool results (#1177) --- +// When a synthetic user message (loop warning) appears between tool results +// for the same assistant, it should be deferred after all tool results. + +func TestSanitizeHistory_InterleavedUserWarningBetweenToolResults(t *testing.T) { + msgs := []providers.Message{ + {Role: "user", Content: "do work"}, + {Role: "assistant", Content: "", ToolCalls: []providers.ToolCall{ + {ID: "tc1", Name: "read_file", Arguments: map[string]any{}}, + {ID: "tc2", Name: "read_file", Arguments: map[string]any{}}, + {ID: "tc3", Name: "read_file", Arguments: map[string]any{}}, + }}, + {Role: "tool", Content: "result1", ToolCallID: "tc1"}, + {Role: "user", Content: "loop warning: read-only streak"}, // interleaved! + {Role: "tool", Content: "result2", ToolCallID: "tc2"}, + {Role: "tool", Content: "result3", ToolCallID: "tc3"}, + {Role: "user", Content: "next question"}, + } + + result, dropped := sanitizeHistory(msgs) + + // All tool results should be contiguous, warning deferred after them. + // Expected order: user, assistant, tool(tc1), tool(tc2), tool(tc3), user(warning), user(next) + // The two consecutive user messages will be merged by role-alternation fix. + if dropped < 0 { + t.Fatalf("unexpected negative dropped: %d", dropped) + } + + // Verify tool results are contiguous (no non-tool message between them) + toolStart := -1 + toolEnd := -1 + for i, m := range result { + if m.Role == "tool" { + if toolStart == -1 { + toolStart = i + } + toolEnd = i + } + } + if toolStart >= 0 { + for i := toolStart; i <= toolEnd; i++ { + if result[i].Role != "tool" { + t.Fatalf("non-tool message at index %d between tool results (role=%s, content=%q)", + i, result[i].Role, result[i].Content) + } + } + } + + // Verify all 3 tool call IDs are present + toolIDs := make(map[string]bool) + for _, m := range result { + if m.Role == "tool" { + toolIDs[m.ToolCallID] = true + } + } + for _, id := range []string{"tc1", "tc2", "tc3"} { + if !toolIDs[id] { + t.Fatalf("missing tool result for %s", id) + } + } + + // Verify warning appears after all tool results + warningIdx := slices.IndexFunc(result, func(m providers.Message) bool { + return m.Role == "user" && strings.Contains(m.Content, "loop warning") + }) + if warningIdx >= 0 && warningIdx <= toolEnd { + t.Fatalf("warning (index %d) should appear after last tool result (index %d)", warningIdx, toolEnd) + } +} + +// --- sanitizeHistory: multiple warnings interleaved between tool results (#1177) --- + +func TestSanitizeHistory_MultipleWarningsBetweenToolResults(t *testing.T) { + msgs := []providers.Message{ + {Role: "user", Content: "go"}, + {Role: "assistant", Content: "", ToolCalls: []providers.ToolCall{ + {ID: "tc1", Name: "exec", Arguments: map[string]any{}}, + {ID: "tc2", Name: "exec", Arguments: map[string]any{}}, + }}, + {Role: "tool", Content: "r1", ToolCallID: "tc1"}, + {Role: "user", Content: "warning1"}, + {Role: "user", Content: "warning2"}, + {Role: "tool", Content: "r2", ToolCallID: "tc2"}, + } + + result, _ := sanitizeHistory(msgs) + + // Tool results must be contiguous + toolStart := -1 + toolEnd := -1 + for i, m := range result { + if m.Role == "tool" { + if toolStart == -1 { + toolStart = i + } + toolEnd = i + } + } + if toolStart >= 0 { + for i := toolStart; i <= toolEnd; i++ { + if result[i].Role != "tool" { + t.Fatalf("non-tool message at index %d between tool results", i) + } + } + } + + // Both tool results present + if !slices.ContainsFunc(result, func(m providers.Message) bool { return m.ToolCallID == "tc1" }) { + t.Fatal("missing tc1") + } + if !slices.ContainsFunc(result, func(m providers.Message) bool { return m.ToolCallID == "tc2" }) { + t.Fatal("missing tc2") + } +} diff --git a/internal/agent/loop_history_sanitize.go b/internal/agent/loop_history_sanitize.go index 9022b45c..516667f8 100644 --- a/internal/agent/loop_history_sanitize.go +++ b/internal/agent/loop_history_sanitize.go @@ -113,22 +113,36 @@ func sanitizeHistory(msgs []providers.Message) ([]providers.Message, int) { result = append(result, msg) - // Collect matching tool results that follow - for i+1 < len(msgs) && msgs[i+1].Role == "tool" { - i++ - toolMsg := msgs[i] - if queue, ok := idQueue[toolMsg.ToolCallID]; ok && len(queue) > 0 { - newID := queue[0] - idQueue[toolMsg.ToolCallID] = queue[1:] - toolMsg.ToolCallID = newID - result = append(result, toolMsg) - delete(expectedIDs, newID) + // Collect matching tool results that follow. + // Non-tool messages (warnings, nudges) interleaved between tool results + // are deferred until all tool results for this assistant turn are collected, + // maintaining contiguous tool_result grouping (#1177). + var deferredNonTool []providers.Message + for i+1 < len(msgs) { + next := msgs[i+1] + if next.Role == "tool" { + i++ + if queue, ok := idQueue[next.ToolCallID]; ok && len(queue) > 0 { + newID := queue[0] + idQueue[next.ToolCallID] = queue[1:] + next.ToolCallID = newID + result = append(result, next) + delete(expectedIDs, newID) + } else { + slog.Debug("sanitizeHistory: dropping mismatched tool result", + "tool_call_id", next.ToolCallID) + dropped++ + } + } else if next.Role != "assistant" && hasPendingToolResultAhead(msgs, i+2, idQueue) { + // Non-tool, non-assistant message with more tool results ahead — defer it. + i++ + deferredNonTool = append(deferredNonTool, next) } else { - slog.Debug("sanitizeHistory: dropping mismatched tool result", - "tool_call_id", toolMsg.ToolCallID) - dropped++ + break } } + // Flush deferred non-tool messages after all tool results. + result = append(result, deferredNonTool...) // Synthesize missing tool results for _, tc := range msg.ToolCalls { @@ -182,6 +196,23 @@ func sanitizeHistory(msgs []providers.Message) ([]providers.Message, int) { return result, dropped } +// hasPendingToolResultAhead returns true if there is at least one tool-role +// message after position start whose ToolCallID is still expected (present +// in idQueue). Stops scanning at the next assistant message. +func hasPendingToolResultAhead(msgs []providers.Message, start int, idQueue map[string][]string) bool { + for j := start; j < len(msgs); j++ { + if msgs[j].Role == "assistant" { + return false + } + if msgs[j].Role == "tool" { + if queue := idQueue[msgs[j].ToolCallID]; len(queue) > 0 { + return true + } + } + } + return false +} + func (l *Loop) maybeSummarize(ctx context.Context, sessionKey string) { history := l.sessions.GetHistory(ctx, sessionKey) diff --git a/internal/pipeline/stages_test.go b/internal/pipeline/stages_test.go index 4360b038..5eb9f24b 100644 --- a/internal/pipeline/stages_test.go +++ b/internal/pipeline/stages_test.go @@ -3231,3 +3231,105 @@ func TestThinkStage_AllowedTools_MixedFunctionAndNativeTools_NoPanic(t *testing. } } +// --- Issue #1177: non-tool messages deferred until all tool results emitted --- + +func TestToolStage_Sequential_DefersNonToolMessages(t *testing.T) { + t.Parallel() + deps := &PipelineDeps{ + ExecuteToolCall: func(_ context.Context, _ *RunState, tc providers.ToolCall) ([]providers.Message, error) { + msgs := []providers.Message{ + {Role: "tool", Content: "result:" + tc.Name, ToolCallID: tc.ID}, + } + // Simulate a warning injected after first tool + if tc.Name == "tool_a" { + msgs = append(msgs, providers.Message{Role: "user", Content: "loop warning: read-only streak"}) + } + return msgs, nil + }, + } + stage := NewToolStage(deps) + state := defaultState() + state.Think.LastResponse = &providers.ChatResponse{ + ToolCalls: []providers.ToolCall{ + {ID: "1", Name: "tool_a"}, + {ID: "2", Name: "tool_b"}, + {ID: "3", Name: "tool_c"}, + }, + } + + if err := stage.Execute(context.Background(), state); err != nil { + t.Fatalf("Execute() error: %v", err) + } + + pending := state.Messages.Pending() + // Expect: 3 tool results first, then 1 user warning at the end + if len(pending) != 4 { + t.Fatalf("pending len = %d, want 4", len(pending)) + } + // First 3 must be tool role + for i := 0; i < 3; i++ { + if pending[i].Role != "tool" { + t.Errorf("pending[%d].Role = %q, want tool", i, pending[i].Role) + } + } + // Last must be the deferred user warning + if pending[3].Role != "user" { + t.Errorf("pending[3].Role = %q, want user", pending[3].Role) + } + if !strings.Contains(pending[3].Content, "loop warning") { + t.Errorf("pending[3].Content = %q, want loop warning", pending[3].Content) + } +} + +func TestToolStage_Parallel_DefersNonToolMessages(t *testing.T) { + t.Parallel() + deps := &PipelineDeps{ + ExecuteToolCall: func(_ context.Context, _ *RunState, _ providers.ToolCall) ([]providers.Message, error) { + t.Fatal("ExecuteToolCall must NOT be called when parallel path is active") + return nil, nil + }, + ExecuteToolRaw: func(_ context.Context, tc providers.ToolCall) (providers.Message, any, error) { + return providers.Message{Role: "tool", Content: "raw:" + tc.Name, ToolCallID: tc.ID}, nil, nil + }, + ProcessToolResult: func(_ context.Context, _ *RunState, tc providers.ToolCall, rawMsg providers.Message, _ any) []providers.Message { + msgs := []providers.Message{rawMsg} + // Inject a warning after processing tool_b + if tc.Name == "tool_b" { + msgs = append(msgs, providers.Message{Role: "user", Content: "nudge: consider alternatives"}) + } + return msgs + }, + ParallelEligibleToolCall: func(providers.ToolCall) bool { return true }, + } + stage := NewToolStage(deps) + state := defaultState() + state.Think.LastResponse = &providers.ChatResponse{ + ToolCalls: []providers.ToolCall{ + {ID: "1", Name: "tool_a"}, + {ID: "2", Name: "tool_b"}, + {ID: "3", Name: "tool_c"}, + }, + } + + if err := stage.Execute(context.Background(), state); err != nil { + t.Fatalf("Execute() error: %v", err) + } + + pending := state.Messages.Pending() + // Expect: 3 tool results first, then 1 user nudge at the end + if len(pending) != 4 { + t.Fatalf("pending len = %d, want 4", len(pending)) + } + for i := 0; i < 3; i++ { + if pending[i].Role != "tool" { + t.Errorf("pending[%d].Role = %q, want tool", i, pending[i].Role) + } + } + if pending[3].Role != "user" { + t.Errorf("pending[3].Role = %q, want user", pending[3].Role) + } + if !strings.Contains(pending[3].Content, "nudge") { + t.Errorf("pending[3].Content = %q, want nudge", pending[3].Content) + } +} + diff --git a/internal/pipeline/tool_stage.go b/internal/pipeline/tool_stage.go index aa2ab2c6..115e89d0 100644 --- a/internal/pipeline/tool_stage.go +++ b/internal/pipeline/tool_stage.go @@ -58,14 +58,20 @@ func (s *ToolStage) Execute(ctx context.Context, state *RunState) error { } // Sequential fallback: ExecuteToolCall handles both I/O and state mutation. + // Non-tool messages (warnings, nudges) are deferred until all tool results + // are emitted to maintain correct tool_result grouping for OpenAI-compatible + // providers. See https://github.com/nextlevelbuilder/goclaw/issues/1177 cumulativeWaitMs := 0 + var deferredNonTool []providers.Message for _, tc := range toolCalls { if s.shouldStopBeforeTool(ctx, state) { + appendDeferredMessages(state, deferredNonTool) return nil } if s.deps.SequentialToolCall != nil && s.deps.SequentialToolCall(tc) { cumulativeWaitMs += toolCallTimeMs(tc) if cumulativeWaitMs > maxSequentialWaitBatchMs { + appendDeferredMessages(state, deferredNonTool) s.result = AbortRun return nil } @@ -82,9 +88,7 @@ func (s *ToolStage) Execute(ctx context.Context, state *RunState) error { if err != nil { return fmt.Errorf("execute tool %s: %w", tc.Name, err) } - for _, msg := range msgs { - state.Messages.AppendPending(msg) - } + appendToolBatchMessages(state, msgs, &deferredNonTool) state.Tool.TotalToolCalls++ // Hook: async PostToolUse — fire and forget with detached context. @@ -102,15 +106,18 @@ func (s *ToolStage) Execute(ctx context.Context, state *RunState) error { } if state.Tool.LoopKilled { + appendDeferredMessages(state, deferredNonTool) s.result = BreakLoop return nil } if ctx.Err() != nil { + appendDeferredMessages(state, deferredNonTool) s.result = AbortRun return nil } } + appendDeferredMessages(state, deferredNonTool) s.checkExitConditions(state) return nil } @@ -277,11 +284,13 @@ func (s *ToolStage) executeParallel(ctx context.Context, state *RunState, prefli "limit", defaultParallelToolCallLimit, "duration_ms", time.Since(startedAt).Milliseconds()) - // Phase 2: sequential state mutation (safe, deterministic order) + // Phase 2: sequential state mutation (safe, deterministic order). + // Non-tool messages deferred until all tool results are emitted (#1177). resultByIndex := make(map[int]rawResult, len(results)) for _, r := range results { resultByIndex[r.index] = r } + var deferredNonTool []providers.Message for _, item := range preflight.ordered { tc := item.tc if item.blocked != nil { @@ -297,9 +306,7 @@ func (s *ToolStage) executeParallel(ctx context.Context, state *RunState, prefli return fmt.Errorf("execute tool %s: %w", tc.Name, r.err) } processed := s.deps.ProcessToolResult(ctx, state, tc, r.msg, r.rawData) - for _, msg := range processed { - state.Messages.AppendPending(msg) - } + appendToolBatchMessages(state, processed, &deferredNonTool) state.Tool.TotalToolCalls++ // Hook: async PostToolUse for parallel path — fire and forget. @@ -318,15 +325,38 @@ func (s *ToolStage) executeParallel(ctx context.Context, state *RunState, prefli } if state.Tool.LoopKilled { + appendDeferredMessages(state, deferredNonTool) s.result = BreakLoop return nil } } + appendDeferredMessages(state, deferredNonTool) s.checkExitConditions(state) return nil } +// appendToolBatchMessages appends tool-role messages immediately and defers +// non-tool messages (warnings, nudges) so that all tool results for a single +// assistant turn stay contiguous. This prevents OpenAI-compatible providers +// from rejecting the transcript due to interleaved user messages (#1177). +func appendToolBatchMessages(state *RunState, msgs []providers.Message, deferred *[]providers.Message) { + for _, msg := range msgs { + if msg.Role == "tool" { + state.Messages.AppendPending(msg) + } else { + *deferred = append(*deferred, msg) + } + } +} + +// appendDeferredMessages flushes deferred non-tool messages into pending. +func appendDeferredMessages(state *RunState, deferred []providers.Message) { + for _, msg := range deferred { + state.Messages.AppendPending(msg) + } +} + // checkExitConditions checks read-only streak and tool budget. func (s *ToolStage) checkExitConditions(state *RunState) { if state.Tool.LoopKilled {