fix(cron): publish agent response to outbound bus for cron-triggered jobs
When a cron job triggers agent execution via ProcessDirectWithChannel, the agent response was silently discarded — the code assumed AgentLoop would auto-publish it, but SendResponse is false on this path. Delegate to PublishResponseIfNeeded (exported from AgentLoop) so the response reaches the originating channel (e.g. Telegram) only when the message tool did not already deliver content in the same round. Also adds a "directive" message type to CronPayload, allowing cron jobs to instruct the agent to execute a task rather than echo static text.
This commit is contained in:
parent
60d7ec20a5
commit
b6826101e7
4 changed files with 148 additions and 12 deletions
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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"`
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue