diff --git a/cmd/gateway_consumer_normal.go b/cmd/gateway_consumer_normal.go index 99d1d7f9..fc76f340 100644 --- a/cmd/gateway_consumer_normal.go +++ b/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{ diff --git a/docs/05-channels-messaging.md b/docs/05-channels-messaging.md index 83c26476..56eeca4f 100644 --- a/docs/05-channels-messaging.md +++ b/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. diff --git a/docs/project-changelog.md b/docs/project-changelog.md index 3eef23b6..337b0f4f 100644 --- a/docs/project-changelog.md +++ b/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** diff --git a/internal/channels/chat_behavior.go b/internal/channels/chat_behavior.go index df853c11..c10ef33a 100644 --- a/internal/channels/chat_behavior.go +++ b/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: diff --git a/internal/channels/chat_behavior_events_test.go b/internal/channels/chat_behavior_events_test.go index 442e29e5..c60c573e 100644 --- a/internal/channels/chat_behavior_events_test.go +++ b/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) + } + }) + } } diff --git a/internal/channels/events.go b/internal/channels/events.go index 8b71741e..d36c1223 100644 --- a/internal/channels/events.go +++ b/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) } } diff --git a/internal/channels/manager.go b/internal/channels/manager.go index f5a325b5..d143fd6d 100644 --- a/internal/channels/manager.go +++ b/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 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 diff --git a/internal/channels/runs.go b/internal/channels/runs.go index 2e1010d7..cc3ee6ac 100644 --- a/internal/channels/runs.go +++ b/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