From e62d9e20c2f897e9ed0d16a61038f5adfc85fe21 Mon Sep 17 00:00:00 2001 From: Anton Bogdanovich <27antonb@gmail.com> Date: Tue, 5 May 2026 19:05:59 -0700 Subject: [PATCH] fix(cron): suppress feedback for scheduled turns --- pkg/agent/agent_message.go | 53 ++++++++++++++++++++++++++++++++++++++ pkg/agent/agent_test.go | 52 +++++++++++++++++++++++++++++++++++++ pkg/tools/cron.go | 34 ++++++++++++++++++------ pkg/tools/cron_test.go | 16 ++++++++++++ 4 files changed, 147 insertions(+), 8 deletions(-) diff --git a/pkg/agent/agent_message.go b/pkg/agent/agent_message.go index 96b0b0817..5aba83da9 100644 --- a/pkg/agent/agent_message.go +++ b/pkg/agent/agent_message.go @@ -43,6 +43,21 @@ func (al *AgentLoop) ProcessDirect( func (al *AgentLoop) ProcessDirectWithChannel( ctx context.Context, content, sessionKey, channel, chatID string, +) (string, error) { + return al.processDirectWithChannel(ctx, content, sessionKey, channel, chatID, false) +} + +func (al *AgentLoop) ProcessScheduledWithChannel( + ctx context.Context, + content, sessionKey, channel, chatID string, +) (string, error) { + return al.processDirectWithChannel(ctx, content, sessionKey, channel, chatID, true) +} + +func (al *AgentLoop) processDirectWithChannel( + ctx context.Context, + content, sessionKey, channel, chatID string, + scheduled bool, ) (string, error) { if err := al.ensureHooksInitialized(ctx); err != nil { return "", err @@ -61,10 +76,48 @@ func (al *AgentLoop) ProcessDirectWithChannel( Content: content, SessionKey: sessionKey, } + if scheduled { + return al.processScheduledMessage(ctx, msg) + } return al.processMessage(ctx, msg) } +func (al *AgentLoop) processScheduledMessage(ctx context.Context, msg bus.InboundMessage) (string, error) { + msg = bus.NormalizeInboundMessage(msg) + route, agent, routeErr := al.resolveMessageRoute(msg) + if routeErr != nil { + return "", routeErr + } + allocation := al.allocateRouteSession(route, msg) + sessionKey := resolveScopeKey(allocation.SessionKey, msg.SessionKey) + + if tool, ok := agent.Tools.Get("message"); ok { + if resetter, ok := tool.(interface{ ResetSentInRound(sessionKey string) }); ok { + resetter.ResetSentInRound(sessionKey) + } + } + + return al.runAgentLoop(ctx, agent, processOptions{ + Dispatch: DispatchRequest{ + SessionKey: sessionKey, + SessionAliases: buildSessionAliases(sessionKey, append(allocation.SessionAliases, msg.SessionKey)...), + InboundContext: cloneInboundContext(&msg.Context), + RouteResult: cloneResolvedRoute(&route), + SessionScope: session.CloneScope(&allocation.Scope), + UserMessage: msg.Content, + Media: append([]string(nil), msg.Media...), + }, + SenderID: msg.SenderID, + SenderDisplayName: msg.Sender.DisplayName, + DefaultResponse: defaultResponse, + EnableSummary: false, + SendResponse: false, + SuppressToolFeedback: true, + NoHistory: true, + }) +} + func (al *AgentLoop) ProcessHeartbeat( ctx context.Context, content, channel, chatID string, diff --git a/pkg/agent/agent_test.go b/pkg/agent/agent_test.go index a75919912..3e2fc7266 100644 --- a/pkg/agent/agent_test.go +++ b/pkg/agent/agent_test.go @@ -4002,6 +4002,58 @@ func TestProcessHeartbeat_DoesNotPublishToolFeedback(t *testing.T) { } } +func TestProcessScheduledWithChannel_DoesNotPublishToolFeedback(t *testing.T) { + tmpDir := t.TempDir() + heartbeatFile := filepath.Join(tmpDir, "scheduled-task.txt") + if err := os.WriteFile(heartbeatFile, []byte("scheduled task"), 0o644); err != nil { + t.Fatalf("WriteFile() error = %v", err) + } + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + ModelName: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + ToolFeedback: config.ToolFeedbackConfig{ + Enabled: true, + MaxArgsLength: 300, + }, + }, + }, + Tools: config.ToolsConfig{ + ReadFile: config.ReadFileToolConfig{ + Enabled: true, + }, + }, + } + + msgBus := bus.NewMessageBus() + provider := &toolFeedbackProvider{filePath: heartbeatFile} + al := NewAgentLoop(cfg, msgBus, provider) + + response, err := al.ProcessScheduledWithChannel( + context.Background(), + "run scheduled task", + "agent:cron-test", + "telegram", + "chat-1", + ) + if err != nil { + t.Fatalf("ProcessScheduledWithChannel() error = %v", err) + } + if response != "HEARTBEAT_OK" { + t.Fatalf("ProcessScheduledWithChannel() response = %q, want %q", response, "HEARTBEAT_OK") + } + + select { + case outbound := <-msgBus.OutboundChan(): + t.Fatalf("expected no outbound tool feedback during scheduled turn, got %+v", outbound) + case <-time.After(200 * time.Millisecond): + } +} + func TestProcessMessage_PublishesToolFeedbackWhenEnabled(t *testing.T) { tmpDir := t.TempDir() heartbeatFile := filepath.Join(tmpDir, "tool-feedback.txt") diff --git a/pkg/tools/cron.go b/pkg/tools/cron.go index a9547eba9..d62284763 100644 --- a/pkg/tools/cron.go +++ b/pkg/tools/cron.go @@ -23,6 +23,10 @@ type JobExecutor interface { PublishResponseIfNeeded(ctx context.Context, channel, chatID, sessionKey, response string) } +type scheduledJobExecutor interface { + ProcessScheduledWithChannel(ctx context.Context, content, sessionKey, channel, chatID string) (string, error) +} + // CronTool provides scheduling capabilities for the agent type CronTool struct { cronService *cron.CronService @@ -344,14 +348,28 @@ func (t *CronTool) ExecuteJob(ctx context.Context, job *cron.CronJob) string { sessionKey := fmt.Sprintf("agent:cron-%s-%s", job.ID, uuid.New().String()) - // Call agent with the job message - response, err := t.executor.ProcessDirectWithChannel( - ctx, - job.Payload.Message, - sessionKey, - channel, - chatID, - ) + // Call agent with the job message. Scheduled agent turns should not emit + // interactive progress/tool-feedback messages; they should only publish a + // final response when the job has something actionable to say. + var response string + var err error + if scheduledExecutor, ok := t.executor.(scheduledJobExecutor); ok { + response, err = scheduledExecutor.ProcessScheduledWithChannel( + ctx, + job.Payload.Message, + sessionKey, + channel, + chatID, + ) + } else { + response, err = t.executor.ProcessDirectWithChannel( + ctx, + job.Payload.Message, + sessionKey, + channel, + chatID, + ) + } if err != nil { return fmt.Sprintf("Error: %v", err) } diff --git a/pkg/tools/cron_test.go b/pkg/tools/cron_test.go index 0e527c98a..7c7c6882f 100644 --- a/pkg/tools/cron_test.go +++ b/pkg/tools/cron_test.go @@ -21,6 +21,7 @@ type stubJobExecutor struct { lastKey string lastChan string lastChatID string + scheduledUsed bool publishedResp string publishedChan string publishedChatID string @@ -38,6 +39,18 @@ func (s *stubJobExecutor) ProcessDirectWithChannel( return s.response, s.err } +func (s *stubJobExecutor) ProcessScheduledWithChannel( + _ context.Context, + content, sessionKey, channel, chatID string, +) (string, error) { + s.scheduledUsed = true + s.lastPrompt = content + s.lastKey = sessionKey + s.lastChan = channel + s.lastChatID = chatID + return s.response, s.err +} + func (s *stubJobExecutor) PublishResponseIfNeeded( _ context.Context, channel, chatID, sessionKey, response string, @@ -282,6 +295,9 @@ func TestCronTool_ExecuteJobPublishesAgentResponse(t *testing.T) { if executor.lastPrompt != "send me a poem" { t.Fatalf("prompt = %q, want original message", executor.lastPrompt) } + if !executor.scheduledUsed { + t.Fatal("expected cron agent job to use scheduled executor path") + } if executor.publishedResp != "generated reply" { t.Fatalf("published response = %q, want generated reply", executor.publishedResp) }