fix(pipeline): block_reply dedup falsely suppresses final message (#838)

Only count BlockReplies when tool calls are present (matching when
EmitBlockReply actually fires in think_stage). Final answers without
tool calls must not increment the counter, otherwise gateway dedup
falsely suppresses delivery on non-streaming channels.

- Update ObserveStage to check len(resp.ToolCalls) > 0
- Rename test to reflect new behavior (tool calls required)
- Add regression test for #838 scenario

Closes #838
This commit is contained in:
viettranx committed 2026-04-12 17:17:51 +07:00
1 parent 45b9240b3a
commit 034187ac9c
2 files changed
+44 -7

No files matched your search

+5 -2
View File
@@ -30,8 +30,11 @@ func (s *ObserveStage) Execute(_ context.Context, state *RunState) error {
return nil
}
// 2. Track block replies (every response with content counts as a block)
if resp.Content != "" {
// 2. Track block replies — only count tool-iteration responses where
// EmitBlockReply actually fires (think_stage emits block.reply only when
// tool calls are present). The final answer (no tool calls) must NOT be
// counted, otherwise the gateway dedup check falsely suppresses delivery.
if resp.Content != "" && len(resp.ToolCalls) > 0 {
state.Observe.BlockReplies++
state.Observe.LastBlockReply = resp.Content
}
+39 -5
View File
@@ -978,18 +978,24 @@ func TestObserveStage_FinalContent_NotSetWhenToolCalls(t *testing.T) {
}
}
func TestObserveStage_BlockReplies_IncrementedPerContentResponse(t *testing.T) {
func TestObserveStage_BlockReplies_IncrementedOnlyWithToolCalls(t *testing.T) {
t.Parallel()
deps := &PipelineDeps{}
stage := NewObserveStage(deps)
state := defaultState()
// first response with content
state.Think.LastResponse = &providers.ChatResponse{Content: "reply 1", FinishReason: "stop"}
// first tool iteration response (has tool calls → should count)
state.Think.LastResponse = &providers.ChatResponse{
Content: "reply 1",
ToolCalls: []providers.ToolCall{{ID: "1", Name: "tool_a"}},
}
_ = stage.Execute(context.Background(), state)
// second response with content
state.Think.LastResponse = &providers.ChatResponse{Content: "reply 2", FinishReason: "stop"}
// second tool iteration response (has tool calls → should count)
state.Think.LastResponse = &providers.ChatResponse{
Content: "reply 2",
ToolCalls: []providers.ToolCall{{ID: "2", Name: "tool_b"}},
}
_ = stage.Execute(context.Background(), state)
if state.Observe.BlockReplies != 2 {
@@ -1000,6 +1006,34 @@ func TestObserveStage_BlockReplies_IncrementedPerContentResponse(t *testing.T) {
}
}
// Regression test for #838: final answer (no tool calls) must NOT increment
// BlockReplies, otherwise gateway dedup falsely suppresses delivery.
func TestObserveStage_FinalAnswer_NoToolCalls_BlockRepliesNotIncremented(t *testing.T) {
t.Parallel()
deps := &PipelineDeps{}
stage := NewObserveStage(deps)
state := defaultState()
// Final answer: content but NO tool calls
state.Think.LastResponse = &providers.ChatResponse{
Content: "Here is your answer.",
FinishReason: "stop",
ToolCalls: nil,
}
_ = stage.Execute(context.Background(), state)
if state.Observe.BlockReplies != 0 {
t.Errorf("BlockReplies = %d, want 0 for final answer without tool calls", state.Observe.BlockReplies)
}
if state.Observe.LastBlockReply != "" {
t.Errorf("LastBlockReply = %q, want empty for final answer", state.Observe.LastBlockReply)
}
// FinalContent should still be set
if state.Observe.FinalContent != "Here is your answer." {
t.Errorf("FinalContent = %q, want answer text", state.Observe.FinalContent)
}
}
func TestObserveStage_NilResponse_NoOp(t *testing.T) {
t.Parallel()
deps := &PipelineDeps{}