diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index fb9edda25..f7b0675f1 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -454,7 +454,7 @@ func (al *AgentLoop) Run(ctx context.Context) error { if target == nil { cancelDrain() if finalResponse != "" { - al.publishResponseIfNeeded(ctx, msg.Channel, msg.ChatID, finalResponse) + al.PublishResponseIfNeeded(ctx, msg.Channel, msg.ChatID, finalResponse) } return } @@ -514,7 +514,7 @@ func (al *AgentLoop) Run(ctx context.Context) error { } if finalResponse != "" { - al.publishResponseIfNeeded(ctx, target.Channel, target.ChatID, finalResponse) + al.PublishResponseIfNeeded(ctx, target.Channel, target.ChatID, finalResponse) } }() default: @@ -600,7 +600,7 @@ func (al *AgentLoop) Stop() { al.running.Store(false) } -func (al *AgentLoop) publishResponseIfNeeded(ctx context.Context, channel, chatID, response string) { +func (al *AgentLoop) PublishResponseIfNeeded(ctx context.Context, channel, chatID, response string) { if response == "" { return } diff --git a/pkg/cron/service.go b/pkg/cron/service.go index 77a413133..0cb4932e2 100644 --- a/pkg/cron/service.go +++ b/pkg/cron/service.go @@ -25,6 +25,7 @@ type CronSchedule struct { type CronPayload struct { Kind string `json:"kind"` + Type string `json:"type,omitempty"` // "message" (default) or "directive" Message string `json:"message"` Command string `json:"command,omitempty"` Deliver bool `json:"deliver"` diff --git a/pkg/tools/cron.go b/pkg/tools/cron.go index 154ec75f0..25cf61e30 100644 --- a/pkg/tools/cron.go +++ b/pkg/tools/cron.go @@ -16,6 +16,9 @@ import ( // JobExecutor is the interface for executing cron jobs through the agent type JobExecutor interface { ProcessDirectWithChannel(ctx context.Context, content, sessionKey, channel, chatID string) (string, error) + // PublishResponseIfNeeded sends response to the outbound bus only when the + // agent did not already deliver content through the message tool in this round. + PublishResponseIfNeeded(ctx context.Context, channel, chatID, response string) } // CronTool provides scheduling capabilities for the agent @@ -111,6 +114,11 @@ func (t *CronTool) Parameters() map[string]any { "type": "string", "description": "Job ID (for remove/enable/disable)", }, + "type": map[string]any{ + "type": "string", + "enum": []string{"message", "directive"}, + "description": "Message generation strategy. 'message' (default): content is sent directly as-is. 'directive': content is treated as instructions for an AI agent to execute before delivery.", + }, "deliver": map[string]any{ "type": "boolean", "description": "If true, send message directly to channel. If false, let agent process message (for complex tasks). Default: false", @@ -236,6 +244,13 @@ func (t *CronTool) addJob(ctx context.Context, args map[string]any) *ToolResult t.cronService.UpdateJob(job) } + // Read and set message type (default to empty string which means "message") + msgType, _ := args["type"].(string) + if msgType != "" { + job.Payload.Type = msgType + t.cronService.UpdateJob(job) + } + return SilentResult(fmt.Sprintf("Cron job added: %s (id: %s)", job.Name, job.ID)) } @@ -347,8 +362,13 @@ func (t *CronTool) ExecuteJob(ctx context.Context, job *cron.CronJob) string { return "ok" } - // If deliver=true, send message directly without agent processing - if job.Payload.Deliver { + // Determine message generation strategy + // Type="directive": treat message as instructions for AI agent to execute + // Type="" or "message" (default): static message content + isDirective := job.Payload.Type == "directive" + + // If deliver=true and not directive, send message directly without agent processing + if job.Payload.Deliver && !isDirective { pubCtx, pubCancel := context.WithTimeout(context.Background(), 5*time.Second) defer pubCancel() t.msgBus.PublishOutbound(pubCtx, bus.OutboundMessage{ @@ -359,13 +379,20 @@ func (t *CronTool) ExecuteJob(ctx context.Context, job *cron.CronJob) string { return "ok" } - // For deliver=false, process through agent (for complex tasks) + // For deliver=false OR directive mode, process through agent sessionKey := fmt.Sprintf("cron-%s", job.ID) - // Call agent with job's message + // Prepare the prompt based on type + prompt := job.Payload.Message + if isDirective { + // For directive type, prefix to clarify this is an instruction + prompt = fmt.Sprintf("Please execute the following directive and provide the result:\n\n%s", job.Payload.Message) + } + + // Call agent with the prepared prompt response, err := t.executor.ProcessDirectWithChannel( ctx, - job.Payload.Message, + prompt, sessionKey, channel, chatID, @@ -374,7 +401,8 @@ func (t *CronTool) ExecuteJob(ctx context.Context, job *cron.CronJob) string { return fmt.Sprintf("Error: %v", err) } - // Response is automatically sent via MessageBus by AgentLoop - _ = response // Will be sent by AgentLoop + if response != "" { + t.executor.PublishResponseIfNeeded(ctx, channel, chatID, response) + } return "ok" } diff --git a/pkg/tools/cron_test.go b/pkg/tools/cron_test.go index cd7d39860..6767cc1d3 100644 --- a/pkg/tools/cron_test.go +++ b/pkg/tools/cron_test.go @@ -12,18 +12,59 @@ import ( "github.com/sipeed/picoclaw/pkg/cron" ) -func newTestCronToolWithConfig(t *testing.T, cfg *config.Config) *CronTool { +type stubJobExecutor struct { + response string + err error + alreadySent bool // simulate message tool having already sent in this round + lastPrompt string + lastKey string + lastChan string + lastChatID string + publishedResp string + publishedChan string + publishedChatID string +} + +func (s *stubJobExecutor) ProcessDirectWithChannel( + _ context.Context, + content, sessionKey, channel, chatID string, +) (string, error) { + 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, response string, +) { + if s.alreadySent { + return + } + s.publishedResp = response + s.publishedChan = channel + s.publishedChatID = chatID +} + +func newTestCronToolWithExecutorAndConfig(t *testing.T, executor JobExecutor, cfg *config.Config) *CronTool { t.Helper() storePath := filepath.Join(t.TempDir(), "cron.json") cronService := cron.NewCronService(storePath, nil) msgBus := bus.NewMessageBus() - tool, err := NewCronTool(cronService, nil, msgBus, t.TempDir(), true, 0, cfg) + tool, err := NewCronTool(cronService, executor, msgBus, t.TempDir(), true, 0, cfg) if err != nil { t.Fatalf("NewCronTool() error: %v", err) } return tool } +func newTestCronToolWithConfig(t *testing.T, cfg *config.Config) *CronTool { + t.Helper() + return newTestCronToolWithExecutorAndConfig(t, nil, cfg) +} + func newTestCronTool(t *testing.T) *CronTool { t.Helper() return newTestCronToolWithConfig(t, config.DefaultConfig()) @@ -237,3 +278,69 @@ func TestCronTool_ExecuteJobPublishesErrorWhenExecDisabled(t *testing.T) { t.Fatalf("expected exec disabled message, got: %s", msg.Content) } } + +func TestCronTool_ExecuteJobPublishesAgentResponse(t *testing.T) { + executor := &stubJobExecutor{response: "generated reply"} + tool := newTestCronToolWithExecutorAndConfig(t, executor, config.DefaultConfig()) + + job := &cron.CronJob{ID: "job-1"} + job.Payload.Channel = "telegram" + job.Payload.To = "chat-1" + job.Payload.Message = "send me a poem" + + if got := tool.ExecuteJob(context.Background(), job); got != "ok" { + t.Fatalf("ExecuteJob() = %q, want ok", got) + } + + if executor.lastKey != "cron-job-1" { + t.Fatalf("sessionKey = %q, want cron-job-1", executor.lastKey) + } + if executor.lastChan != "telegram" || executor.lastChatID != "chat-1" { + t.Fatalf("executor target = %s/%s, want telegram/chat-1", executor.lastChan, executor.lastChatID) + } + if executor.lastPrompt != "send me a poem" { + t.Fatalf("prompt = %q, want original message", executor.lastPrompt) + } + if executor.publishedResp != "generated reply" { + t.Fatalf("published response = %q, want generated reply", executor.publishedResp) + } + if executor.publishedChan != "telegram" || executor.publishedChatID != "chat-1" { + t.Fatalf("published target = %s/%s, want telegram/chat-1", executor.publishedChan, executor.publishedChatID) + } +} + +func TestCronTool_ExecuteJobSkipsEmptyAgentResponse(t *testing.T) { + executor := &stubJobExecutor{} + tool := newTestCronToolWithExecutorAndConfig(t, executor, config.DefaultConfig()) + + job := &cron.CronJob{ID: "job-empty"} + job.Payload.Channel = "telegram" + job.Payload.To = "chat-1" + job.Payload.Message = "say nothing" + + if got := tool.ExecuteJob(context.Background(), job); got != "ok" { + t.Fatalf("ExecuteJob() = %q, want ok", got) + } + + if executor.publishedResp != "" { + t.Fatalf("unexpected published response: %q", executor.publishedResp) + } +} + +func TestCronTool_ExecuteJobSkipsWhenMessageToolAlreadySent(t *testing.T) { + executor := &stubJobExecutor{response: "已发送。", alreadySent: true} + tool := newTestCronToolWithExecutorAndConfig(t, executor, config.DefaultConfig()) + + job := &cron.CronJob{ID: "job-msg-sent"} + job.Payload.Channel = "telegram" + job.Payload.To = "chat-1" + job.Payload.Message = "send weather" + + if got := tool.ExecuteJob(context.Background(), job); got != "ok" { + t.Fatalf("ExecuteJob() = %q, want ok", got) + } + + if executor.publishedResp != "" { + t.Fatalf("expected no published response when message tool already sent, got: %q", executor.publishedResp) + } +}