diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go index bc6d2b39b..d6593daab 100644 --- a/pkg/agent/agent.go +++ b/pkg/agent/agent.go @@ -121,6 +121,7 @@ const ( messageKindThought = "thought" messageKindToolFeedback = "tool_feedback" messageKindToolCalls = "tool_calls" + messageKindFinalReply = "final_reply" metadataKeyAccountID = "account_id" metadataKeyGuildID = "guild_id" metadataKeyTeamID = "team_id" @@ -260,7 +261,21 @@ func (al *AgentLoop) Run(ctx context.Context) error { return } if continued != "" { - al.PublishResponseIfNeeded(ctx, target.Channel, target.ChatID, target.SessionKey, continued) + al.publishResponseWithContextIfNeeded( + ctx, + target.Channel, + target.ChatID, + target.SessionKey, + continued, + &bus.InboundContext{ + Channel: m.Context.Channel, + ChatID: m.Context.ChatID, + TopicID: m.Context.TopicID, + Raw: map[string]string{ + metadataKeyMessageKind: messageKindFinalReply, + }, + }, + ) } return } @@ -554,13 +569,20 @@ func (al *AgentLoop) runAgentLoop( opts.Dispatch.SessionKey, opts.Dispatch.SessionScope, ) + outboundCtx := outboundContextFromInbound( + opts.Dispatch.InboundContext, + opts.Dispatch.Channel(), + opts.Dispatch.ChatID(), + opts.Dispatch.ReplyToMessageID(), + ) + if result.preferNewOutboundReply { + if outboundCtx.Raw == nil { + outboundCtx.Raw = make(map[string]string, 1) + } + outboundCtx.Raw[metadataKeyMessageKind] = messageKindFinalReply + } al.bus.PublishOutbound(ctx, bus.OutboundMessage{ - Context: outboundContextFromInbound( - opts.Dispatch.InboundContext, - opts.Dispatch.Channel(), - opts.Dispatch.ChatID(), - opts.Dispatch.ReplyToMessageID(), - ), + Context: outboundCtx, AgentID: agentID, SessionKey: sessionKey, Scope: scope, diff --git a/pkg/agent/agent_outbound.go b/pkg/agent/agent_outbound.go index f4a01adfd..d429d8254 100644 --- a/pkg/agent/agent_outbound.go +++ b/pkg/agent/agent_outbound.go @@ -41,6 +41,14 @@ func (al *AgentLoop) publishResponseOrError( } func (al *AgentLoop) PublishResponseIfNeeded(ctx context.Context, channel, chatID, sessionKey, response string) { + al.publishResponseWithContextIfNeeded(ctx, channel, chatID, sessionKey, response, nil) +} + +func (al *AgentLoop) publishResponseWithContextIfNeeded( + ctx context.Context, + channel, chatID, sessionKey, response string, + inboundCtx *bus.InboundContext, +) { if response == "" { return } @@ -75,7 +83,7 @@ func (al *AgentLoop) PublishResponseIfNeeded(ctx context.Context, channel, chatI } msg := bus.OutboundMessage{ - Context: bus.NewOutboundContext(channel, chatID, ""), + Context: outboundContextFromInbound(inboundCtx, channel, chatID, ""), Content: response, } if sessionKey != "" { diff --git a/pkg/agent/agent_steering.go b/pkg/agent/agent_steering.go index 9b136e7cd..8a1e1dc3a 100644 --- a/pkg/agent/agent_steering.go +++ b/pkg/agent/agent_steering.go @@ -58,7 +58,21 @@ func (al *AgentLoop) runTurnWithSteering(ctx context.Context, initialMsg bus.Inb // Publish final response if finalResponse != "" { - al.PublishResponseIfNeeded(ctx, target.Channel, target.ChatID, target.SessionKey, finalResponse) + al.publishResponseWithContextIfNeeded( + ctx, + target.Channel, + target.ChatID, + target.SessionKey, + finalResponse, + &bus.InboundContext{ + Channel: initialMsg.Context.Channel, + ChatID: initialMsg.Context.ChatID, + TopicID: initialMsg.Context.TopicID, + Raw: map[string]string{ + metadataKeyMessageKind: messageKindFinalReply, + }, + }, + ) } } diff --git a/pkg/agent/pipeline_execute.go b/pkg/agent/pipeline_execute.go index 0f71c7432..5e4729cfa 100644 --- a/pkg/agent/pipeline_execute.go +++ b/pkg/agent/pipeline_execute.go @@ -587,6 +587,7 @@ toolLoop: } if steerMsgs := al.dequeueSteeringMessagesForScope(ts.sessionKey); len(steerMsgs) > 0 { + exec.sawSteering = true exec.pendingMessages = append(exec.pendingMessages, steerMsgs...) } @@ -655,6 +656,7 @@ toolLoop: // This covers the case where tools were partially executed and skipped due to steering, // but one tool had ResponseHandled=false (so allResponsesHandled=false). if len(exec.pendingMessages) > 0 { + exec.sawSteering = true logger.InfoCF("agent", "Pending steering after partial tool execution; continuing turn", map[string]any{ "agent_id": ts.agent.ID, @@ -667,6 +669,7 @@ toolLoop: // Poll for newly arrived steering if steerMsgs := al.dequeueSteeringMessagesForScope(ts.sessionKey); len(steerMsgs) > 0 { + exec.sawSteering = true logger.InfoCF("agent", "Steering arrived after tool delivery; continuing turn", map[string]any{ "agent_id": ts.agent.ID, diff --git a/pkg/agent/pipeline_finalize.go b/pkg/agent/pipeline_finalize.go index 1f407825e..06c1dc6b6 100644 --- a/pkg/agent/pipeline_finalize.go +++ b/pkg/agent/pipeline_finalize.go @@ -32,9 +32,10 @@ func (p *Pipeline) Finalize( } ts.setPhase(TurnPhaseCompleted) return turnResult{ - finalContent: finalContent, - status: turnStatus, - followUps: append([]bus.InboundMessage(nil), ts.followUps...), + finalContent: finalContent, + status: turnStatus, + followUps: append([]bus.InboundMessage(nil), ts.followUps...), + preferNewOutboundReply: exec.sawSteering, }, nil } @@ -75,8 +76,9 @@ func (p *Pipeline) Finalize( ts.setPhase(TurnPhaseCompleted) return turnResult{ - finalContent: finalContent, - status: turnStatus, - followUps: append([]bus.InboundMessage(nil), ts.followUps...), + finalContent: finalContent, + status: turnStatus, + followUps: append([]bus.InboundMessage(nil), ts.followUps...), + preferNewOutboundReply: exec.sawSteering, }, nil } diff --git a/pkg/agent/pipeline_llm.go b/pkg/agent/pipeline_llm.go index 496fcd7e4..cc62ff9a0 100644 --- a/pkg/agent/pipeline_llm.go +++ b/pkg/agent/pipeline_llm.go @@ -475,6 +475,7 @@ func (p *Pipeline) CallLLM( responseContent = exec.response.ReasoningContent } if steerMsgs := al.dequeueSteeringMessagesForScope(ts.sessionKey); len(steerMsgs) > 0 { + exec.sawSteering = true logger.InfoCF("agent", "Steering arrived after direct LLM response; continuing turn", map[string]any{ "agent_id": ts.agent.ID, diff --git a/pkg/agent/steering_test.go b/pkg/agent/steering_test.go index 23d34840e..10f668b88 100644 --- a/pkg/agent/steering_test.go +++ b/pkg/agent/steering_test.go @@ -851,6 +851,97 @@ func TestAgentLoop_Run_AutoContinuesLateSteeringMessage(t *testing.T) { } } +func TestAgentLoop_RunTurnWithSteering_PublishesFinalReplyAsNewMessage(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + ModelName: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + sessionKey := session.BuildMainSessionKey(routing.DefaultAgentID) + provider := &blockingDirectProvider{ + firstStarted: make(chan struct{}), + releaseFirst: make(chan struct{}), + firstResp: "stale direct response", + finalResp: "fresh response after steering", + } + + msgBus := bus.NewMessageBus() + al := NewAgentLoop(cfg, msgBus, provider) + + msg := testInboundMessage(bus.InboundMessage{ + Context: bus.InboundContext{ + Channel: "telegram", + ChatID: "-1001234567890", + ChatType: "group", + TopicID: "6", + SenderID: "user-1", + MessageID: "475", + }, + SessionKey: sessionKey, + Content: "initial request", + }) + + done := make(chan struct{}) + go func() { + al.runTurnWithSteering(context.Background(), msg) + close(done) + }() + + select { + case <-provider.firstStarted: + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for first LLM call to start") + } + + if err := al.Steer(providers.Message{Role: "user", Content: "follow-up instruction"}); err != nil { + t.Fatalf("Steer failed: %v", err) + } + close(provider.releaseFirst) + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("timeout waiting for steering turn to finish") + } + + var finalOutbound bus.OutboundMessage + found := false +drain: + for { + select { + case outbound := <-msgBus.OutboundChan(): + if outbound.Content == "fresh response after steering" { + finalOutbound = outbound + found = true + } + default: + break drain + } + } + + if !found { + t.Fatal("expected final outbound response") + } + if got := finalOutbound.Context.Raw[metadataKeyMessageKind]; got != messageKindFinalReply { + t.Fatalf("message kind = %q, want %q", got, messageKindFinalReply) + } + if finalOutbound.Context.TopicID != "6" { + t.Fatalf("topic_id = %q, want 6", finalOutbound.Context.TopicID) + } +} + func TestAgentLoop_Run_QueuedVoiceMessageIsTranscribedBeforeSteering(t *testing.T) { tmpDir := t.TempDir() cfg := &config.Config{ diff --git a/pkg/agent/turn_coord.go b/pkg/agent/turn_coord.go index 2826e662c..c91389600 100644 --- a/pkg/agent/turn_coord.go +++ b/pkg/agent/turn_coord.go @@ -89,11 +89,13 @@ func (al *AgentLoop) runTurn(ctx context.Context, ts *turnState, pipeline *Pipel // We do NOT call dequeueSteeringMessagesForScope here because // steering was already consumed from al.steering by ExecuteTools. if len(exec.pendingMessages) > 0 { + exec.sawSteering = true pendingMessages = append(pendingMessages, exec.pendingMessages...) exec.pendingMessages = nil } } else if !ts.opts.SkipInitialSteeringPoll { if steerMsgs := al.dequeueSteeringMessagesForScopeWithFallback(ts.sessionKey); len(steerMsgs) > 0 { + exec.sawSteering = true pendingMessages = append(pendingMessages, steerMsgs...) } } diff --git a/pkg/agent/turn_state.go b/pkg/agent/turn_state.go index b769ebcd0..478a49ce7 100644 --- a/pkg/agent/turn_state.go +++ b/pkg/agent/turn_state.go @@ -81,9 +81,10 @@ const ( // ============================================================================= type turnResult struct { - finalContent string - status TurnEndStatus - followUps []bus.InboundMessage + finalContent string + status TurnEndStatus + followUps []bus.InboundMessage + preferNewOutboundReply bool } // ============================================================================= @@ -118,6 +119,7 @@ type turnExecution struct { // Turn output finalContent string + sawSteering bool // Iteration tracking iteration int diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index f7f625271..772b02147 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -167,7 +167,9 @@ func outboundMessageBypassesPlaceholderEdit(msg bus.OutboundMessage) bool { return false } kind := strings.TrimSpace(msg.Context.Raw["message_kind"]) - return strings.EqualFold(kind, "thought") || strings.EqualFold(kind, "tool_calls") + return strings.EqualFold(kind, "thought") || + strings.EqualFold(kind, "tool_calls") || + strings.EqualFold(kind, "final_reply") } func outboundMediaChannel(msg bus.OutboundMediaMessage) string { @@ -387,6 +389,9 @@ func (m *Manager) preSend(ctx context.Context, name string, msg bus.OutboundMess if deleter, ok := ch.(MessageDeleter); ok { deleter.DeleteMessage(ctx, chatID, entry.id) // best effort } + if strings.EqualFold(strings.TrimSpace(msg.Context.Raw["message_kind"]), "final_reply") { + dismissTrackedToolFeedbackMessage(ctx, ch, chatID, &msg.Context) + } return nil, false } if editor, ok := ch.(MessageEditor); ok { diff --git a/pkg/channels/manager_test.go b/pkg/channels/manager_test.go index 5aeabc888..3b1945f1c 100644 --- a/pkg/channels/manager_test.go +++ b/pkg/channels/manager_test.go @@ -1306,6 +1306,48 @@ func TestPreSend_ThoughtPlaceholderDeleteAndSkipsEdit(t *testing.T) { } } +func TestPreSend_FinalReplyPlaceholderDeleteAndDismissesTrackedFeedback(t *testing.T) { + m := newTestManager() + + ch := &mockDeletingMessageEditor{ + mockMessageEditor: mockMessageEditor{ + editFn: func(_ context.Context, _, _, _ string) error { + t.Fatal("expected final reply to bypass placeholder edit") + return nil + }, + }, + } + + m.RecordPlaceholder("test", "123", "456") + + msg := testOutboundMessage(bus.OutboundMessage{ + Channel: "test", + ChatID: "123", + Content: "final aggregated reply", + Context: bus.InboundContext{ + Channel: "test", + ChatID: "123", + Raw: map[string]string{ + "message_kind": "final_reply", + }, + }, + }) + + msgIDs, handled := m.preSend(context.Background(), "test", msg, ch) + if handled { + t.Fatalf("expected preSend to fall through so the channel can send a new final reply, got %v", msgIDs) + } + if ch.deleteCalls != 1 { + t.Fatalf("expected placeholder deletion, got %d delete calls", ch.deleteCalls) + } + if ch.deletedChatID != "123" || ch.deletedMessageID != "456" { + t.Fatalf("unexpected placeholder deletion target: %s/%s", ch.deletedChatID, ch.deletedMessageID) + } + if ch.dismissedChatID != "123" { + t.Fatalf("expected tracked tool feedback dismissal, got %q", ch.dismissedChatID) + } +} + func TestSendWithRetry_ToolCallsPlaceholderDeleteAndFallsThroughToSend(t *testing.T) { m := newTestManager()