From 034187ac9cb31e925f63285824fe72306f9ae988 Mon Sep 17 00:00:00 2001
From: viettranx <viettranx@gmail.com>
Date: Sun, 12 Apr 2026 17:17:51 +0700
Subject: [PATCH] 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
---
 internal/pipeline/observe_stage.go |  7 +++--
 internal/pipeline/stages_test.go   | 44 ++++++++++++++++++++++++++----
 2 files changed, 44 insertions(+), 7 deletions(-)

diff --git a/internal/pipeline/observe_stage.go b/internal/pipeline/observe_stage.go
index 8d2c7bfa..d24ac88d 100644
--- a/internal/pipeline/observe_stage.go
+++ b/internal/pipeline/observe_stage.go
@@ -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
 	}
diff --git a/internal/pipeline/stages_test.go b/internal/pipeline/stages_test.go
index 86fce166..b778b1f1 100644
--- a/internal/pipeline/stages_test.go
+++ b/internal/pipeline/stages_test.go
@@ -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{}