diff --git a/cmd/cron_cmd.go b/cmd/cron_cmd.go index e5a76aa6..d6f8fabe 100644 --- a/cmd/cron_cmd.go +++ b/cmd/cron_cmd.go @@ -40,7 +40,7 @@ func cronListCmd() *cobra.Command { return } svc := loadCronStore() - jobs := svc.ListJobs(showDisabled) + jobs := svc.ListJobs(showDisabled, "", "") printCronJobs(jobs, jsonOutput) }, } diff --git a/cmd/gateway_consumer.go b/cmd/gateway_consumer.go index 452b7838..ab301f2e 100644 --- a/cmd/gateway_consumer.go +++ b/cmd/gateway_consumer.go @@ -330,12 +330,12 @@ func resolveCronAgent(agentID string, agents *agent.Router, cfg *config.Config) } // makeCronJobHandler creates a cron job handler that sends job messages through the agent. -func makeCronJobHandler(agents *agent.Router, msgBus *bus.MessageBus, cfg *config.Config) func(job *store.CronJob) (string, error) { - return func(job *store.CronJob) (string, error) { +func makeCronJobHandler(agents *agent.Router, msgBus *bus.MessageBus, cfg *config.Config) func(job *store.CronJob) (*store.CronJobResult, error) { + return func(job *store.CronJob) (*store.CronJobResult, error) { agentID := resolveCronAgent(job.AgentID, agents, cfg) loop, err := agents.Get(agentID) if err != nil { - return "", fmt.Errorf("agent %s not found: %w", agentID, err) + return nil, fmt.Errorf("agent %s not found: %w", agentID, err) } sessionKey := sessions.BuildCronSessionKey(agentID, job.ID, fmt.Sprintf("cron-%s", job.ID)) @@ -349,11 +349,12 @@ func makeCronJobHandler(agents *agent.Router, msgBus *bus.MessageBus, cfg *confi Message: job.Payload.Message, Channel: channel, ChatID: job.Payload.To, + UserID: job.UserID, RunID: fmt.Sprintf("cron-%s", job.ID), Stream: false, }) if err != nil { - return "", err + return nil, err } // If job wants delivery to a channel, publish outbound @@ -365,7 +366,15 @@ func makeCronJobHandler(agents *agent.Router, msgBus *bus.MessageBus, cfg *confi }) } - return result.Content, nil + cronResult := &store.CronJobResult{ + Content: result.Content, + } + if result.Usage != nil { + cronResult.InputTokens = result.Usage.PromptTokens + cronResult.OutputTokens = result.Usage.CompletionTokens + } + + return cronResult, nil } } diff --git a/internal/agent/resolver.go b/internal/agent/resolver.go index 3a025510..c99b31f5 100644 --- a/internal/agent/resolver.go +++ b/internal/agent/resolver.go @@ -5,6 +5,7 @@ import ( "fmt" "log/slog" + "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/bootstrap" "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/config" @@ -53,7 +54,14 @@ func NewManagedResolver(deps ResolverDeps) ResolverFunc { return func(agentKey string) (Agent, error) { ctx := context.Background() - ag, err := deps.AgentStore.GetByKey(ctx, agentKey) + // Support lookup by UUID (e.g. from cron jobs that store agent_id as UUID) + var ag *store.AgentData + var err error + if id, parseErr := uuid.Parse(agentKey); parseErr == nil { + ag, err = deps.AgentStore.GetByID(ctx, id) + } else { + ag, err = deps.AgentStore.GetByKey(ctx, agentKey) + } if err != nil { return nil, fmt.Errorf("agent not found: %s", agentKey) } diff --git a/internal/gateway/methods/cron.go b/internal/gateway/methods/cron.go index 46814062..3275477f 100644 --- a/internal/gateway/methods/cron.go +++ b/internal/gateway/methods/cron.go @@ -40,7 +40,7 @@ func (m *CronMethods) handleList(_ context.Context, client *gateway.Client, req json.Unmarshal(req.Params, ¶ms) } - jobs := m.service.ListJobs(params.IncludeDisabled) + jobs := m.service.ListJobs(params.IncludeDisabled, "", "") client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{ "jobs": jobs, @@ -75,7 +75,7 @@ func (m *CronMethods) handleCreate(_ context.Context, client *gateway.Client, re return } - job, err := m.service.AddJob(params.Name, params.Schedule, params.Message, params.Deliver, params.Channel, params.To, params.AgentID) + job, err := m.service.AddJob(params.Name, params.Schedule, params.Message, params.Deliver, params.Channel, params.To, params.AgentID, "") if err != nil { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, err.Error())) return diff --git a/internal/store/cron_store.go b/internal/store/cron_store.go index 5bcd263d..5e008c69 100644 --- a/internal/store/cron_store.go +++ b/internal/store/cron_store.go @@ -7,6 +7,7 @@ type CronJob struct { ID string `json:"id"` Name string `json:"name"` AgentID string `json:"agentId,omitempty"` + UserID string `json:"userId,omitempty"` Enabled bool `json:"enabled"` Schedule CronSchedule `json:"schedule"` Payload CronPayload `json:"payload"` @@ -52,6 +53,14 @@ type CronRunLogEntry struct { Summary string `json:"summary,omitempty"` } +// CronJobResult is the output of a cron job handler execution. +type CronJobResult struct { + Content string `json:"content"` + InputTokens int `json:"inputTokens,omitempty"` + OutputTokens int `json:"outputTokens,omitempty"` + DurationMS int64 `json:"durationMs,omitempty"` +} + // CronJobPatch holds optional fields for updating a job. type CronJobPatch struct { Name string `json:"name,omitempty"` @@ -67,9 +76,9 @@ type CronJobPatch struct { // CronStore manages scheduled jobs. type CronStore interface { - AddJob(name string, schedule CronSchedule, message string, deliver bool, channel, to, agentID string) (*CronJob, error) + AddJob(name string, schedule CronSchedule, message string, deliver bool, channel, to, agentID, userID string) (*CronJob, error) GetJob(jobID string) (*CronJob, bool) - ListJobs(includeDisabled bool) []CronJob + ListJobs(includeDisabled bool, agentID, userID string) []CronJob RemoveJob(jobID string) error UpdateJob(jobID string, patch CronJobPatch) (*CronJob, error) EnableJob(jobID string, enabled bool) error @@ -81,7 +90,7 @@ type CronStore interface { Stop() // Job execution - SetOnJob(handler func(job *CronJob) (string, error)) + SetOnJob(handler func(job *CronJob) (*CronJobResult, error)) RunJob(jobID string, force bool) (ran bool, reason string, err error) // Due job detection (for scheduler) diff --git a/internal/store/file/cron.go b/internal/store/file/cron.go index d091ca29..4b7b8d5e 100644 --- a/internal/store/file/cron.go +++ b/internal/store/file/cron.go @@ -19,7 +19,7 @@ func NewFileCronStore(svc *cron.Service) *FileCronStore { // Service returns the underlying cron.Service for direct access during migration. func (f *FileCronStore) Service() *cron.Service { return f.svc } -func (f *FileCronStore) AddJob(name string, schedule store.CronSchedule, message string, deliver bool, channel, to, agentID string) (*store.CronJob, error) { +func (f *FileCronStore) AddJob(name string, schedule store.CronSchedule, message string, deliver bool, channel, to, agentID, userID string) (*store.CronJob, error) { cronSched := toCronSchedule(schedule) job, err := f.svc.AddJob(name, cronSched, message, deliver, channel, to, agentID) if err != nil { @@ -38,7 +38,7 @@ func (f *FileCronStore) GetJob(jobID string) (*store.CronJob, bool) { return &result, true } -func (f *FileCronStore) ListJobs(includeDisabled bool) []store.CronJob { +func (f *FileCronStore) ListJobs(includeDisabled bool, agentID, userID string) []store.CronJob { jobs := f.svc.ListJobs(includeDisabled) result := make([]store.CronJob, len(jobs)) for i, j := range jobs { @@ -87,10 +87,17 @@ func (f *FileCronStore) Status() map[string]interface{} { func (f *FileCronStore) Start() error { return f.svc.Start() } func (f *FileCronStore) Stop() { f.svc.Stop() } -func (f *FileCronStore) SetOnJob(handler func(job *store.CronJob) (string, error)) { +func (f *FileCronStore) SetOnJob(handler func(job *store.CronJob) (*store.CronJobResult, error)) { f.svc.SetOnJob(func(j *cron.Job) (string, error) { sj := cronJobToStore(j) - return handler(&sj) + result, err := handler(&sj) + if err != nil { + return "", err + } + if result != nil { + return result.Content, nil + } + return "", nil }) } diff --git a/internal/store/pg/cron.go b/internal/store/pg/cron.go index 03c30c5b..1fc3c775 100644 --- a/internal/store/pg/cron.go +++ b/internal/store/pg/cron.go @@ -23,7 +23,7 @@ const defaultCronCacheTTL = 2 * time.Minute type PGCronStore struct { db *sql.DB mu sync.Mutex - onJob func(job *store.CronJob) (string, error) + onJob func(job *store.CronJob) (*store.CronJobResult, error) running bool stop chan struct{} @@ -47,7 +47,7 @@ func (s *PGCronStore) SetRetryConfig(cfg cron.RetryConfig) { s.retryCfg = cfg } -func (s *PGCronStore) AddJob(name string, schedule store.CronSchedule, message string, deliver bool, channel, to, agentID string) (*store.CronJob, error) { +func (s *PGCronStore) AddJob(name string, schedule store.CronSchedule, message string, deliver bool, channel, to, agentID, userID string) (*store.CronJob, error) { payload := store.CronPayload{ Kind: "agent_turn", Message: message, Deliver: deliver, Channel: channel, To: to, } @@ -79,14 +79,24 @@ func (s *PGCronStore) AddJob(name string, schedule store.CronSchedule, message s } } + var userIDPtr *string + if userID != "" { + userIDPtr = &userID + } + + var intervalMS *int64 + if schedule.EveryMS != nil { + intervalMS = schedule.EveryMS + } + nextRun := computeNextRun(&schedule, now) _, err := s.db.Exec( - `INSERT INTO cron_jobs (id, agent_id, name, enabled, schedule_kind, cron_expression, run_at, timezone, - payload, delete_after_run, next_run_at, created_at, updated_at) - VALUES ($1, $2, $3, true, $4, $5, $6, $7, $8, $9, $10, $11, $12)`, - id, agentUUID, name, scheduleKind, cronExpr, runAt, tz, - payloadJSON, deleteAfterRun, nextRun, now, now, + `INSERT INTO cron_jobs (id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone, + interval_ms, payload, delete_after_run, next_run_at, created_at, updated_at) + VALUES ($1, $2, $3, $4, true, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)`, + id, agentUUID, userIDPtr, name, scheduleKind, cronExpr, runAt, tz, + intervalMS, payloadJSON, deleteAfterRun, nextRun, now, now, ) if err != nil { return nil, fmt.Errorf("create cron job: %w", err) @@ -110,16 +120,35 @@ func (s *PGCronStore) GetJob(jobID string) (*store.CronJob, bool) { return job, true } -func (s *PGCronStore) ListJobs(includeDisabled bool) []store.CronJob { - q := `SELECT id, agent_id, name, enabled, schedule_kind, cron_expression, run_at, timezone, - payload, delete_after_run, next_run_at, last_run_at, last_status, last_error, - created_at, updated_at FROM cron_jobs` +func (s *PGCronStore) ListJobs(includeDisabled bool, agentID, userID string) []store.CronJob { + q := `SELECT id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone, + interval_ms, payload, delete_after_run, next_run_at, last_run_at, last_status, last_error, + created_at, updated_at FROM cron_jobs WHERE 1=1` + + var args []interface{} + argIdx := 1 + if !includeDisabled { - q += " WHERE enabled = true" + q += fmt.Sprintf(" AND enabled = $%d", argIdx) + args = append(args, true) + argIdx++ } + if agentID != "" { + if aid, err := uuid.Parse(agentID); err == nil { + q += fmt.Sprintf(" AND agent_id = $%d", argIdx) + args = append(args, aid) + argIdx++ + } + } + if userID != "" { + q += fmt.Sprintf(" AND user_id = $%d", argIdx) + args = append(args, userID) + argIdx++ + } + q += " ORDER BY created_at DESC" - rows, err := s.db.Query(q) + rows, err := s.db.Query(q, args...) if err != nil { return nil } @@ -309,7 +338,7 @@ func (s *PGCronStore) Stop() { s.running = false } -func (s *PGCronStore) SetOnJob(handler func(job *store.CronJob) (string, error)) { +func (s *PGCronStore) SetOnJob(handler func(job *store.CronJob) (*store.CronJobResult, error)) { s.mu.Lock() defer s.mu.Unlock() s.onJob = handler @@ -330,7 +359,11 @@ func (s *PGCronStore) RunJob(jobID string, force bool) (bool, string, error) { } result, err := handler(job) - return true, result, err + content := "" + if result != nil { + content = result.Content + } + return true, content, err } func (s *PGCronStore) GetDueJobs(now time.Time) []store.CronJob { @@ -356,8 +389,8 @@ func (s *PGCronStore) GetDueJobs(now time.Time) []store.CronJob { // refreshJobCache reloads all enabled jobs from DB. Must be called with mu held. func (s *PGCronStore) refreshJobCache() { rows, err := s.db.Query( - `SELECT id, agent_id, name, enabled, schedule_kind, cron_expression, run_at, timezone, - payload, delete_after_run, next_run_at, last_run_at, last_status, last_error, + `SELECT id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone, + interval_ms, payload, delete_after_run, next_run_at, last_run_at, last_status, last_error, created_at, updated_at FROM cron_jobs WHERE enabled = true`) if err != nil { return @@ -387,8 +420,8 @@ func (s *PGCronStore) InvalidateCache() { func (s *PGCronStore) scanJob(id uuid.UUID) (*store.CronJob, error) { row := s.db.QueryRow( - `SELECT id, agent_id, name, enabled, schedule_kind, cron_expression, run_at, timezone, - payload, delete_after_run, next_run_at, last_run_at, last_status, last_error, + `SELECT id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone, + interval_ms, payload, delete_after_run, next_run_at, last_run_at, last_status, last_error, created_at, updated_at FROM cron_jobs WHERE id = $1`, id) return scanCronSingleRow(row) } @@ -427,10 +460,24 @@ func (s *PGCronStore) checkAndRunDueJobs() { } jobCopy := job - result, attempts, err := cron.ExecuteWithRetry(func() (string, error) { - return handler(&jobCopy) + startTime := time.Now() + + // Wrap handler to fit ExecuteWithRetry's (string, error) signature + var lastResult *store.CronJobResult + resultStr, attempts, err := cron.ExecuteWithRetry(func() (string, error) { + r, e := handler(&jobCopy) + if e != nil { + return "", e + } + lastResult = r + if r != nil { + return r.Content, nil + } + return "", nil }, s.retryCfg) + durationMS := time.Since(startTime).Milliseconds() + if attempts > 1 { slog.Info("cron job retried", "id", job.ID, "attempts", attempts, "success", err == nil) } @@ -444,18 +491,29 @@ func (s *PGCronStore) checkAndRunDueJobs() { lastError = &errStr } + // Extract token usage from handler result + var inputTokens, outputTokens int + if lastResult != nil { + inputTokens = lastResult.InputTokens + outputTokens = lastResult.OutputTokens + } + // Log run logID := uuid.Must(uuid.NewV7()) var summary *string if err == nil { - s := cron.TruncateOutput(result) + s := cron.TruncateOutput(resultStr) summary = &s } if id, parseErr := uuid.Parse(job.ID); parseErr == nil { + var agentUUID *uuid.UUID + if aid, aidErr := uuid.Parse(job.AgentID); aidErr == nil { + agentUUID = &aid + } s.db.Exec( - `INSERT INTO cron_run_logs (id, job_id, status, error, summary, ran_at) - VALUES ($1, $2, $3, $4, $5, $6)`, - logID, id, status, lastError, summary, now, + `INSERT INTO cron_run_logs (id, job_id, agent_id, status, error, summary, duration_ms, input_tokens, output_tokens, ran_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`, + logID, id, agentUUID, status, lastError, summary, durationMS, inputTokens, outputTokens, now, ) } @@ -489,15 +547,17 @@ type cronRowScanner interface { func scanCronRow(row cronRowScanner) (*store.CronJob, error) { var id uuid.UUID var agentID *uuid.UUID + var userID *string var name, scheduleKind string var enabled, deleteAfterRun bool var cronExpr, tz, lastStatus, lastError *string var runAt, nextRunAt, lastRunAt *time.Time + var intervalMS *int64 var payloadJSON []byte var createdAt, updatedAt time.Time - err := row.Scan(&id, &agentID, &name, &enabled, &scheduleKind, &cronExpr, &runAt, &tz, - &payloadJSON, &deleteAfterRun, &nextRunAt, &lastRunAt, &lastStatus, &lastError, + err := row.Scan(&id, &agentID, &userID, &name, &enabled, &scheduleKind, &cronExpr, &runAt, &tz, + &intervalMS, &payloadJSON, &deleteAfterRun, &nextRunAt, &lastRunAt, &lastStatus, &lastError, &createdAt, &updatedAt) if err != nil { return nil, err @@ -522,6 +582,9 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) { if agentID != nil { job.AgentID = agentID.String() } + if userID != nil { + job.UserID = *userID + } if cronExpr != nil { job.Schedule.Expr = *cronExpr } @@ -529,6 +592,9 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) { ms := runAt.UnixMilli() job.Schedule.AtMS = &ms } + if intervalMS != nil { + job.Schedule.EveryMS = intervalMS + } if tz != nil { job.Schedule.TZ = *tz } diff --git a/internal/tools/cron.go b/internal/tools/cron.go index 6692b506..6aad3e26 100644 --- a/internal/tools/cron.go +++ b/internal/tools/cron.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "time" "github.com/nextlevelbuilder/goclaw/internal/store" ) @@ -108,21 +109,24 @@ func (t *CronTool) Execute(ctx context.Context, args map[string]interface{}) *Re return ErrorResult("action parameter is required") } + agentID := resolveAgentIDString(ctx) + userID := store.UserIDFromContext(ctx) + switch action { case "status": return t.handleStatus() case "list": - return t.handleList(args) + return t.handleList(args, agentID, userID) case "add": - return t.handleAdd(args) + return t.handleAdd(ctx, args, agentID, userID) case "update": - return t.handleUpdate(args) + return t.handleUpdate(args, agentID, userID) case "remove": - return t.handleRemove(args) + return t.handleRemove(args, agentID, userID) case "run": - return t.handleRun(args) + return t.handleRun(args, agentID, userID) case "runs": - return t.handleRuns(args) + return t.handleRuns(args, agentID, userID) default: return ErrorResult(fmt.Sprintf("unknown action: %s", action)) } @@ -134,9 +138,9 @@ func (t *CronTool) handleStatus() *Result { return NewResult(string(data)) } -func (t *CronTool) handleList(args map[string]interface{}) *Result { +func (t *CronTool) handleList(args map[string]interface{}, agentID, userID string) *Result { includeDisabled, _ := args["includeDisabled"].(bool) - jobs := t.cronStore.ListJobs(includeDisabled) + jobs := t.cronStore.ListJobs(includeDisabled, agentID, userID) result := map[string]interface{}{ "jobs": jobs, @@ -146,7 +150,7 @@ func (t *CronTool) handleList(args map[string]interface{}) *Result { return NewResult(string(data)) } -func (t *CronTool) handleAdd(args map[string]interface{}) *Result { +func (t *CronTool) handleAdd(ctx context.Context, args map[string]interface{}, agentID, userID string) *Result { jobObj, ok := args["job"].(map[string]interface{}) if !ok { return ErrorResult("job object is required for add action") @@ -179,6 +183,9 @@ func (t *CronTool) handleAdd(args map[string]interface{}) *Result { case "at": if v, ok := numberFromMap(scheduleObj, "atMs"); ok { ms := int64(v) + if ms <= time.Now().UnixMilli() { + return ErrorResult(fmt.Sprintf("job.schedule.atMs is in the past (%d). Use a future Unix timestamp in milliseconds. Current time is %d ms", ms, time.Now().UnixMilli())) + } schedule.AtMS = &ms } else { return ErrorResult("job.schedule.atMs is required for 'at' schedule") @@ -204,9 +211,23 @@ func (t *CronTool) handleAdd(args map[string]interface{}) *Result { deliver, _ := jobObj["deliver"].(bool) channel, _ := jobObj["channel"].(string) to, _ := jobObj["to"].(string) - agentID, _ := jobObj["agentId"].(string) - job, err := t.cronStore.AddJob(name, schedule, message, deliver, channel, to, agentID) + // Auto-fill channel and to from context if deliver is requested but not specified + if deliver { + if channel == "" { + channel = ToolChannelFromCtx(ctx) + } + if to == "" { + to = ToolChatIDFromCtx(ctx) + } + } + + // Use agent ID from job object if explicitly provided, otherwise from context + if explicit, _ := jobObj["agentId"].(string); explicit != "" { + agentID = explicit + } + + job, err := t.cronStore.AddJob(name, schedule, message, deliver, channel, to, agentID, userID) if err != nil { return ErrorResult(fmt.Sprintf("failed to create cron job: %v", err)) } @@ -215,12 +236,35 @@ func (t *CronTool) handleAdd(args map[string]interface{}) *Result { return NewResult(string(data)) } -func (t *CronTool) handleUpdate(args map[string]interface{}) *Result { +// checkJobOwnership validates that the job belongs to the current agent+user scope. +// In standalone mode (empty agentID/userID), all jobs are accessible. +func (t *CronTool) checkJobOwnership(jobID, agentID, userID string) (*store.CronJob, *Result) { + job, ok := t.cronStore.GetJob(jobID) + if !ok { + return nil, ErrorResult(fmt.Sprintf("job %s not found", jobID)) + } + + // In managed mode, verify ownership + if agentID != "" && job.AgentID != agentID { + return nil, ErrorResult(fmt.Sprintf("job %s not found", jobID)) + } + if userID != "" && job.UserID != userID { + return nil, ErrorResult(fmt.Sprintf("job %s not found", jobID)) + } + + return job, nil +} + +func (t *CronTool) handleUpdate(args map[string]interface{}, agentID, userID string) *Result { jobID := resolveJobID(args) if jobID == "" { return ErrorResult("jobId is required for update action") } + if _, errResult := t.checkJobOwnership(jobID, agentID, userID); errResult != nil { + return errResult + } + patchObj, ok := args["patch"].(map[string]interface{}) if !ok { return ErrorResult("patch object is required for update action") @@ -240,12 +284,16 @@ func (t *CronTool) handleUpdate(args map[string]interface{}) *Result { return NewResult(string(data)) } -func (t *CronTool) handleRemove(args map[string]interface{}) *Result { +func (t *CronTool) handleRemove(args map[string]interface{}, agentID, userID string) *Result { jobID := resolveJobID(args) if jobID == "" { return ErrorResult("jobId is required for remove action") } + if _, errResult := t.checkJobOwnership(jobID, agentID, userID); errResult != nil { + return errResult + } + if err := t.cronStore.RemoveJob(jobID); err != nil { return ErrorResult(fmt.Sprintf("failed to remove cron job: %v", err)) } @@ -254,12 +302,16 @@ func (t *CronTool) handleRemove(args map[string]interface{}) *Result { return NewResult(string(data)) } -func (t *CronTool) handleRun(args map[string]interface{}) *Result { +func (t *CronTool) handleRun(args map[string]interface{}, agentID, userID string) *Result { jobID := resolveJobID(args) if jobID == "" { return ErrorResult("jobId is required for run action") } + if _, errResult := t.checkJobOwnership(jobID, agentID, userID); errResult != nil { + return errResult + } + runMode, _ := args["runMode"].(string) force := runMode == "force" @@ -279,9 +331,16 @@ func (t *CronTool) handleRun(args map[string]interface{}) *Result { return NewResult(string(data)) } -func (t *CronTool) handleRuns(args map[string]interface{}) *Result { +func (t *CronTool) handleRuns(args map[string]interface{}, agentID, userID string) *Result { jobID := resolveJobID(args) + // Validate ownership if a specific job is requested + if jobID != "" { + if _, errResult := t.checkJobOwnership(jobID, agentID, userID); errResult != nil { + return errResult + } + } + limit := 20 if v, ok := numberFromMap(args, "limit"); ok { limit = int(v) diff --git a/migrations/000003_cron_jobs_user_id.down.sql b/migrations/000003_cron_jobs_user_id.down.sql new file mode 100644 index 00000000..1845efbe --- /dev/null +++ b/migrations/000003_cron_jobs_user_id.down.sql @@ -0,0 +1,4 @@ +DROP INDEX IF EXISTS idx_cron_jobs_agent_user; +DROP INDEX IF EXISTS idx_cron_jobs_user_id; +ALTER TABLE cron_jobs DROP COLUMN IF EXISTS interval_ms; +ALTER TABLE cron_jobs DROP COLUMN IF EXISTS user_id; diff --git a/migrations/000003_cron_jobs_user_id.up.sql b/migrations/000003_cron_jobs_user_id.up.sql new file mode 100644 index 00000000..e454ec74 --- /dev/null +++ b/migrations/000003_cron_jobs_user_id.up.sql @@ -0,0 +1,4 @@ +ALTER TABLE cron_jobs ADD COLUMN IF NOT EXISTS user_id TEXT; +ALTER TABLE cron_jobs ADD COLUMN IF NOT EXISTS interval_ms BIGINT; +CREATE INDEX IF NOT EXISTS idx_cron_jobs_user_id ON cron_jobs (user_id); +CREATE INDEX IF NOT EXISTS idx_cron_jobs_agent_user ON cron_jobs (agent_id, user_id);