Add user_id and agent_id filtering to cron jobs with token usage tracking and UUID-based agent resolution

Extend CronJob model with UserID field for multi-tenant isolation. Add agentID and userID filter parameters to ListJobs across all store implementations (file, pg). Support agent lookup by UUID in resolver for cron jobs that store agent_id as UUID. Change cron job handler signature to return CronJobResult struct with content, token usage (input/output tokens), and duration. Track execution metrics

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)
 		},
 	}

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
 	}
 }
 

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)
 		}

internal/gateway/methods/cron.go

@@ -40,7 +40,7 @@ func (m *CronMethods) handleList(_ context.Context, client *gateway.Client, req
 		json.Unmarshal(req.Params, &params)
 	}
 
-	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

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)

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
 	})
 }
 

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
 	}

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)

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;

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);