fix(cron): suppress feedback for scheduled turns
This commit is contained in:
parent
eb4e187550
commit
e62d9e20c2
4 changed files with 147 additions and 8 deletions
|
|
@ -43,6 +43,21 @@ func (al *AgentLoop) ProcessDirect(
|
||||||
func (al *AgentLoop) ProcessDirectWithChannel(
|
func (al *AgentLoop) ProcessDirectWithChannel(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
content, sessionKey, channel, chatID string,
|
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) {
|
) (string, error) {
|
||||||
if err := al.ensureHooksInitialized(ctx); err != nil {
|
if err := al.ensureHooksInitialized(ctx); err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
|
|
@ -61,10 +76,48 @@ func (al *AgentLoop) ProcessDirectWithChannel(
|
||||||
Content: content,
|
Content: content,
|
||||||
SessionKey: sessionKey,
|
SessionKey: sessionKey,
|
||||||
}
|
}
|
||||||
|
if scheduled {
|
||||||
|
return al.processScheduledMessage(ctx, msg)
|
||||||
|
}
|
||||||
|
|
||||||
return al.processMessage(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(
|
func (al *AgentLoop) ProcessHeartbeat(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
content, channel, chatID string,
|
content, channel, chatID string,
|
||||||
|
|
|
||||||
|
|
@ -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) {
|
func TestProcessMessage_PublishesToolFeedbackWhenEnabled(t *testing.T) {
|
||||||
tmpDir := t.TempDir()
|
tmpDir := t.TempDir()
|
||||||
heartbeatFile := filepath.Join(tmpDir, "tool-feedback.txt")
|
heartbeatFile := filepath.Join(tmpDir, "tool-feedback.txt")
|
||||||
|
|
|
||||||
|
|
@ -23,6 +23,10 @@ type JobExecutor interface {
|
||||||
PublishResponseIfNeeded(ctx context.Context, channel, chatID, sessionKey, response string)
|
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
|
// CronTool provides scheduling capabilities for the agent
|
||||||
type CronTool struct {
|
type CronTool struct {
|
||||||
cronService *cron.CronService
|
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())
|
sessionKey := fmt.Sprintf("agent:cron-%s-%s", job.ID, uuid.New().String())
|
||||||
|
|
||||||
// Call agent with the job message
|
// Call agent with the job message. Scheduled agent turns should not emit
|
||||||
response, err := t.executor.ProcessDirectWithChannel(
|
// 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,
|
ctx,
|
||||||
job.Payload.Message,
|
job.Payload.Message,
|
||||||
sessionKey,
|
sessionKey,
|
||||||
channel,
|
channel,
|
||||||
chatID,
|
chatID,
|
||||||
)
|
)
|
||||||
|
} else {
|
||||||
|
response, err = t.executor.ProcessDirectWithChannel(
|
||||||
|
ctx,
|
||||||
|
job.Payload.Message,
|
||||||
|
sessionKey,
|
||||||
|
channel,
|
||||||
|
chatID,
|
||||||
|
)
|
||||||
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Sprintf("Error: %v", err)
|
return fmt.Sprintf("Error: %v", err)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -21,6 +21,7 @@ type stubJobExecutor struct {
|
||||||
lastKey string
|
lastKey string
|
||||||
lastChan string
|
lastChan string
|
||||||
lastChatID string
|
lastChatID string
|
||||||
|
scheduledUsed bool
|
||||||
publishedResp string
|
publishedResp string
|
||||||
publishedChan string
|
publishedChan string
|
||||||
publishedChatID string
|
publishedChatID string
|
||||||
|
|
@ -38,6 +39,18 @@ func (s *stubJobExecutor) ProcessDirectWithChannel(
|
||||||
return s.response, s.err
|
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(
|
func (s *stubJobExecutor) PublishResponseIfNeeded(
|
||||||
_ context.Context,
|
_ context.Context,
|
||||||
channel, chatID, sessionKey, response string,
|
channel, chatID, sessionKey, response string,
|
||||||
|
|
@ -282,6 +295,9 @@ func TestCronTool_ExecuteJobPublishesAgentResponse(t *testing.T) {
|
||||||
if executor.lastPrompt != "send me a poem" {
|
if executor.lastPrompt != "send me a poem" {
|
||||||
t.Fatalf("prompt = %q, want original message", executor.lastPrompt)
|
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" {
|
if executor.publishedResp != "generated reply" {
|
||||||
t.Fatalf("published response = %q, want generated reply", executor.publishedResp)
|
t.Fatalf("published response = %q, want generated reply", executor.publishedResp)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue