fix(channels): gate initial intermediate replies

cmd/gateway_consumer_normal.go

@@ -480,7 +480,10 @@ func processNormalMessage(
 		}
 
 		// Clean up run tracking (in case HandleAgentEvent didn't fire for terminal events)
+		interimDelivered := 0
+		lastInterimReply := ""
 		if deps.ChannelMgr != nil {
+			interimDelivered, lastInterimReply = deps.ChannelMgr.InterimDeliverySnapshot(rID)
 			deps.ChannelMgr.UnregisterRun(rID)
 		}
 
@@ -539,11 +542,11 @@ func processNormalMessage(
 			return
 		}
 
-		interimReplyEnabled := blockReplyEnabled || channels.ShouldDeliverGeneratedProgress(chatBehavior, streaming)
-		// Dedup: if block replies were delivered and the final content matches the last
+		// Dedup: if interim replies were delivered and the final content matches the last
 		// block reply, suppress the final message to avoid duplicate delivery.
-		// Only applies when an interim reply mode is enabled (otherwise nothing was delivered).
-		if interimReplyEnabled && outcome.Result.BlockReplies > 0 && outcome.Result.Content == outcome.Result.LastBlockReply && len(outcome.Result.Media) == 0 {
+		// This uses channel delivery state, not emitted pipeline events, because streaming
+		// runs and quick-ack-disabled initial block replies can intentionally suppress them.
+		if interimDelivered > 0 && outcome.Result.Content == lastInterimReply && len(outcome.Result.Media) == 0 {
 			slog.Debug("inbound: dedup final message (matches last block reply)",
 				"channel", channel, "run_id", rID)
 			deps.MsgBus.PublishOutbound(bus.OutboundMessage{

docs/05-channels-messaging.md

@@ -78,7 +78,7 @@ Normal channel messages pass through the shared inbound debouncer before agent e
 - `quick_ack.mode = "llm_generated"` uses the existing main-turn `block.reply` event as the natural progress response for non-streaming channel runs. It does not make a separate LLM call.
 - `quick_ack.templates` are fallback messages. In generated mode, the first template is sent only if no generated `block.reply` arrives before `quick_ack.min_delay_ms`.
 - `quick_ack.mode = "fixed_template"` preserves the older fixed-template acknowledgement behavior.
-- `quick_ack.mode = "off"` disables chat-behavior quick acknowledgement without disabling explicit `gateway.block_reply`.
+- `quick_ack.mode = "off"` disables chat-behavior quick acknowledgement. When explicit `gateway.block_reply` is on, the first pre-tool `block.reply` is treated as the immediate acknowledgement and suppressed; later tool-iteration `block.reply` messages can still be delivered as intermediate replies.
 - `final_split` splits long final text replies into a bounded number of paragraph messages.
 - Per-channel `chat_behavior` overrides inherit the gateway config unless a field is explicitly set.
 

docs/project-changelog.md

@@ -6,6 +6,24 @@ Significant changes, features, and fixes in reverse chronological order.
 
 ## 2026-05-31
 
+### Channel intermediate reply gating
+
+**Fixes**
+
+- Quick acknowledgement off now suppresses the first pre-tool `block.reply`
+  even when explicit `gateway.block_reply` is enabled, so the initial
+  acknowledgement does not leak through the Intermediate Replies path.
+- Final reply dedup now uses channel-delivered interim state instead of raw
+  pipeline `block.reply` emit counts, avoiding false suppression when an
+  interim event was skipped.
+
+**Tests**
+
+- Added channel event coverage for quick acknowledgement disabled and
+  `quick_ack.mode = "off"` with explicit intermediate replies enabled.
+
+---
+
 ### Agent-scoped git credentials (issue #117)
 
 **New**

internal/channels/chat_behavior.go

@@ -174,6 +174,13 @@ func ShouldDeliverGeneratedProgress(behavior ResolvedChatBehavior, streaming boo
 		effectiveQuickAckMode(behavior.QuickAck.Mode) == QuickAckModeLLMGenerated
 }
 
+func ShouldSuppressInitialBlockReply(behavior ResolvedChatBehavior, streaming bool) bool {
+	if streaming || !behavior.Enabled {
+		return false
+	}
+	return !behavior.QuickAck.Enabled || effectiveQuickAckMode(behavior.QuickAck.Mode) == QuickAckModeOff
+}
+
 func normalizeQuickAckMode(mode string) string {
 	switch strings.TrimSpace(mode) {
 	case QuickAckModeFixedTemplate:

internal/channels/chat_behavior_events_test.go

@@ -168,13 +168,14 @@ func TestHandleAgentEvent_GeneratedModeFallsBackWithoutBlockReply(t *testing.T)
 	}
 }
 
-func TestHandleAgentEvent_QuickAckModeOffPreservesExplicitBlockReply(t *testing.T) {
+func TestHandleAgentEvent_FixedQuickAckPreservesInitialExplicitBlockReply(t *testing.T) {
 	behavior := ResolvedChatBehavior{
 		Enabled: true,
 		QuickAck: ResolvedQuickAckConfig{
-			Enabled:   true,
-			Mode:      QuickAckModeOff,
-			Templates: []string{"Fallback."},
+			Enabled:    true,
+			Mode:       QuickAckModeFixedTemplate,
+			MinDelayMs: 50,
+			Templates:  []string{"Fallback."},
 		},
 	}
 
@@ -195,4 +196,79 @@ func TestHandleAgentEvent_QuickAckModeOffPreservesExplicitBlockReply(t *testing.
 	if got.Content != "Explicit block reply." {
 		t.Fatalf("explicit block reply content = %q", got.Content)
 	}
+
+	ctx, cancel = context.WithTimeout(context.Background(), 75*time.Millisecond)
+	defer cancel()
+	if got, ok := mb.SubscribeOutbound(ctx); ok {
+		t.Fatalf("fallback emitted after explicit block reply: %+v", got)
+	}
+}
+
+func TestHandleAgentEvent_QuickAckDisabledSuppressesInitialExplicitBlockReply(t *testing.T) {
+	for _, tc := range []struct {
+		name  string
+		quick ResolvedQuickAckConfig
+	}{
+		{
+			name:  "enabled_false",
+			quick: ResolvedQuickAckConfig{Enabled: false, Templates: []string{"Fallback."}},
+		},
+		{
+			name:  "mode_off",
+			quick: ResolvedQuickAckConfig{Enabled: true, Mode: QuickAckModeOff, Templates: []string{"Fallback."}},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			behavior := ResolvedChatBehavior{
+				Enabled:  true,
+				QuickAck: tc.quick,
+			}
+
+			mb := bus.New()
+			mgr := NewManager(mb)
+			mgr.RegisterChannel("test", &chatBehaviorTestChannel{name: "test"})
+			mgr.RegisterRunWithBehavior("run-1", "test", "chat-1", "msg-1", nil, uuid.Nil, false, true, true, behavior)
+
+			mgr.HandleAgentEvent(protocol.AgentEventRunStarted, "run-1", nil)
+			mgr.HandleAgentEvent(protocol.AgentEventBlockReply, "run-1", map[string]string{"content": "Initial quick acknowledgement."})
+
+			ctx, cancel := context.WithTimeout(context.Background(), 25*time.Millisecond)
+			defer cancel()
+			if got, ok := mb.SubscribeOutbound(ctx); ok {
+				t.Fatalf("initial explicit block reply emitted with quick ack disabled: %+v", got)
+			}
+
+			delivered, last := mgr.InterimDeliverySnapshot("run-1")
+			if delivered != 0 || last != "" {
+				t.Fatalf("interim delivery snapshot = (%d, %q), want none", delivered, last)
+			}
+
+			mgr.HandleAgentEvent(protocol.AgentEventBlockReply, "run-1", map[string]string{"content": "Second iteration update."})
+
+			ctx, cancel = context.WithTimeout(context.Background(), time.Second)
+			defer cancel()
+			got, ok := mb.SubscribeOutbound(ctx)
+			if !ok {
+				t.Fatal("expected second explicit block reply outbound message")
+			}
+			if got.Content != "Second iteration update." {
+				t.Fatalf("explicit block reply content = %q", got.Content)
+			}
+			delivered, last = mgr.InterimDeliverySnapshot("run-1")
+			if delivered != 1 || last != "Second iteration update." {
+				t.Fatalf("interim delivery snapshot = (%d, %q), want delivered second reply", delivered, last)
+			}
+
+			mgr.HandleAgentEvent(protocol.AgentEventRunCompleted, "run-1", nil)
+			delivered, last = mgr.InterimDeliverySnapshot("run-1")
+			if delivered != 1 || last != "Second iteration update." {
+				t.Fatalf("completed interim delivery snapshot = (%d, %q), want delivered second reply", delivered, last)
+			}
+			mgr.UnregisterRun("run-1")
+			delivered, last = mgr.InterimDeliverySnapshot("run-1")
+			if delivered != 0 || last != "" {
+				t.Fatalf("unregistered interim delivery snapshot = (%d, %q), want none", delivered, last)
+			}
+		})
+	}
 }

internal/channels/events.go

@@ -280,18 +280,28 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) {
 		}
 		rc.mu.Lock()
 		streaming := rc.Streaming
+		rc.blockReplySeen++
+		isInitialBlockReply := rc.blockReplySeen == 1
+		blockReplyEnabled := rc.BlockReplyEnabled
+		chatBehavior := rc.ChatBehavior
 		rc.mu.Unlock()
 
 		if streaming {
 			return // streaming already delivered via chunks
 		}
-		if !rc.BlockReplyEnabled && !ShouldDeliverGeneratedProgress(rc.ChatBehavior, streaming) {
+		generatedProgress := ShouldDeliverGeneratedProgress(chatBehavior, streaming)
+		if !blockReplyEnabled && !generatedProgress {
+			return
+		}
+		if isInitialBlockReply && blockReplyEnabled && !generatedProgress && ShouldSuppressInitialBlockReply(chatBehavior, streaming) {
 			return
 		}
 
 		m.cancelQuickAck(rc)
 		rc.mu.Lock()
 		rc.blockReplySent = true
+		rc.interimDelivered++
+		rc.lastInterimReply = content
 		rc.mu.Unlock()
 
 		// Build outbound metadata: copy routing fields but strip reply_to_message_id
@@ -365,6 +375,7 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) {
 	// Clean up on terminal events
 	if eventType == protocol.AgentEventRunCompleted || eventType == protocol.AgentEventRunFailed || eventType == protocol.AgentEventRunCancelled {
 		m.cancelQuickAck(rc)
+		m.snapshotCompletedRun(runID, rc)
 		m.runs.Delete(runID)
 	}
 }

internal/channels/manager.go

@@ -43,6 +43,9 @@ type RunContext struct {
 	ackSent           bool
 	ackCancelled      bool
 	blockReplySent    bool
+	blockReplySeen    int
+	interimDelivered  int
+	lastInterimReply  string
 	streamBuffer      string        // accumulated streaming text (chunks are deltas)
 	inToolPhase       bool          // true after tool.call, reset on next chunk (new LLM iteration)
 	stream            ChannelStream // per-run stream handle (replaces per-chat sync.Map in channel impls)
@@ -52,6 +55,11 @@ type RunContext struct {
 	tagParseSkipped   bool          // true after first chunk with no <think> tags (skip re-parsing)
 }
 
+type interimDeliverySnapshot struct {
+	delivered int
+	lastReply string
+}
+
 // Manager manages all registered channels, handling their lifecycle
 // and routing outbound messages to the correct channel.
 type Manager struct {
@@ -59,6 +67,7 @@ type Manager struct {
 	health           map[string]ChannelHealth
 	bus              *bus.MessageBus
 	runs             sync.Map // runID string → *RunContext
+	completedRuns    sync.Map // runID string → interimDeliverySnapshot
 	dispatchTask     *asyncTask
 	mu               sync.RWMutex
 	contactCollector *store.ContactCollector

internal/channels/runs.go

@@ -17,6 +17,7 @@ func (m *Manager) RegisterRun(runID, channelName, chatID, messageID string, meta
 // RegisterRunWithBehavior associates a run ID with channel context and
 // resolved delivery behavior so event handlers do not read mutable config mid-run.
 func (m *Manager) RegisterRunWithBehavior(runID, channelName, chatID, messageID string, metadata map[string]string, tenantID uuid.UUID, streaming, blockReply, toolStatus bool, chatBehavior ResolvedChatBehavior) {
+	m.completedRuns.Delete(runID)
 	m.runs.Store(runID, &RunContext{
 		ChannelName:       channelName,
 		ChatID:            chatID,
@@ -37,6 +38,40 @@ func (m *Manager) UnregisterRun(runID string) {
 			m.cancelQuickAck(rc)
 		}
 	}
+	m.completedRuns.Delete(runID)
+}
+
+func (m *Manager) InterimDeliverySnapshot(runID string) (int, string) {
+	val, ok := m.runs.Load(runID)
+	if ok {
+		rc, ok := val.(*RunContext)
+		if !ok {
+			return 0, ""
+		}
+		rc.mu.Lock()
+		defer rc.mu.Unlock()
+		return rc.interimDelivered, rc.lastInterimReply
+	}
+	if val, ok := m.completedRuns.Load(runID); ok {
+		if snapshot, ok := val.(interimDeliverySnapshot); ok {
+			return snapshot.delivered, snapshot.lastReply
+		}
+	}
+	return 0, ""
+}
+
+func (m *Manager) snapshotCompletedRun(runID string, rc *RunContext) {
+	rc.mu.Lock()
+	snapshot := interimDeliverySnapshot{
+		delivered: rc.interimDelivered,
+		lastReply: rc.lastInterimReply,
+	}
+	rc.mu.Unlock()
+	if snapshot.delivered == 0 && snapshot.lastReply == "" {
+		m.completedRuns.Delete(runID)
+		return
+	}
+	m.completedRuns.Store(runID, snapshot)
 }
 
 // IsStreamingChannel checks if a named channel implements StreamingChannel