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
This commit is contained in:
Thanh_Dang authored and GitHub committed 2026-06-24 10:49:31 +07:00
1 parent 06ef06ca37
commit 68ad5953d7
4 files changed
+299 -20

No files matched your search

+116
View File
@@ -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")
}
}
+44 -13
View File
@@ -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)
+102
View File
@@ -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)
}
}
+37 -7
View File
@@ -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 {