diff --git a/cmd/gateway_cron.go b/cmd/gateway_cron.go index 5fc4ea0a..77db869e 100644 --- a/cmd/gateway_cron.go +++ b/cmd/gateway_cron.go @@ -32,12 +32,34 @@ var cronHeartbeatWakeFn func(agentID string) // so the stateless-reset behavior can be unit-tested without filesystem effects. var cronCLISessionReset = providers.ResetCLISession -func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg *config.Config, channelMgr *channels.Manager, sessionMgr store.SessionStore, agentStore store.AgentStore, providerStore store.ProviderStore, providerReg *providers.Registry) func(job *store.CronJob) (*store.CronJobResult, error) { +// cronTenantContext scopes a context to the job's tenant. It sets BOTH the +// tenant ID and the tenant SLUG: tenant-scoped filesystem paths +// (skills-store, workspace, media via config.TenantScopedDir) key off the +// slug, and resolve to an id-based path when the slug is absent — a different +// directory than where HTTP/WS upload materialized the files. Without the +// slug, a cron agent turn sees NONE of its tenant's managed skills. tenantStore +// may be nil (older wiring) — then only the tenant ID is set, preserving prior +// behavior. Master tenant needs no slug (TenantScopedDir returns the base). +func cronTenantContext(ctx context.Context, tenantStore store.TenantStore, tenantID uuid.UUID) context.Context { + ctx = store.WithTenantID(ctx, tenantID) + if tenantStore == nil || tenantID == uuid.Nil || tenantID == store.MasterTenantID { + return ctx + } + tenant, err := tenantStore.GetTenant(ctx, tenantID) + if err != nil || tenant == nil || tenant.Slug == "" { + slog.Warn("cron: could not resolve tenant slug; tenant-scoped skills/workspace may be invisible", + "tenant_id", tenantID, "error", err) + return ctx + } + return store.WithTenantSlug(ctx, tenant.Slug) +} + +func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg *config.Config, channelMgr *channels.Manager, sessionMgr store.SessionStore, agentStore store.AgentStore, tenantStore store.TenantStore, providerStore store.ProviderStore, providerReg *providers.Registry) func(job *store.CronJob) (*store.CronJobResult, error) { return func(job *store.CronJob) (*store.CronJobResult, error) { agentID := job.AgentID if agentID == "" && agentStore != nil { // Resolve real default agent from DB instead of using literal "default" string. - tenantCtx := store.WithTenantID(context.Background(), job.TenantID) + tenantCtx := cronTenantContext(context.Background(), tenantStore, job.TenantID) if defaultAgent, err := agentStore.GetDefault(tenantCtx); err == nil { agentID = defaultAgent.AgentKey } else { @@ -48,7 +70,7 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg } else if id, err := uuid.Parse(agentID); err == nil && agentStore != nil { // Resolve agentKey from UUID so session key uses agentKey // (consistent with chat/WS/team paths, fixes cache invalidation mismatch). - cronCtx := store.WithTenantID(context.Background(), job.TenantID) + cronCtx := cronTenantContext(context.Background(), tenantStore, job.TenantID) if ag, err := agentStore.GetByID(cronCtx, id); err == nil { agentID = ag.AgentKey } @@ -72,7 +94,7 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg // Deterministic command payload: run the shell command in-process WITHOUT // an LLM/agent turn (zero model tokens). Gated by cron.command_enabled. if job.Payload.IsCommand() { - return runCommandCronJob(cfg, job, msgBus, peerKind) + return runCommandCronJob(cfg, job, tenantStore, msgBus, peerKind) } // Build cron context so the agent knows delivery target and requester. @@ -97,7 +119,7 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg jobTimeout := cfg.Cron.JobTimeoutDuration() cronCtx, cancelCron := context.WithTimeout(context.Background(), jobTimeout) defer cancelCron() - cronCtx = store.WithTenantID(cronCtx, job.TenantID) + cronCtx = cronTenantContext(cronCtx, tenantStore, job.TenantID) if job.Payload.CredentialUserID != "" { cronCtx = store.WithCredentialUserID(cronCtx, job.Payload.CredentialUserID) } @@ -232,7 +254,7 @@ func deliverCronOutput(msgBus *bus.MessageBus, job *store.CronJob, content strin // stderr) like an agent turn. On failure it returns an error so the run is // recorded as "error" and retried per cron.max_retries — failures are NOT // delivered, mirroring the agent path where only successful output is announced. -func runCommandCronJob(cfg *config.Config, job *store.CronJob, msgBus *bus.MessageBus, peerKind string) (*store.CronJobResult, error) { +func runCommandCronJob(cfg *config.Config, job *store.CronJob, tenantStore store.TenantStore, msgBus *bus.MessageBus, peerKind string) (*store.CronJobResult, error) { if !cfg.Cron.CommandEnabled { return nil, fmt.Errorf("cron command payloads are disabled; set cron.command_enabled=true to allow them") } @@ -246,7 +268,7 @@ func runCommandCronJob(cfg *config.Config, job *store.CronJob, msgBus *bus.Messa cmdTimeout = time.Duration(spec.TimeoutSeconds) * time.Second } // The job timeout is a hard ceiling above the per-command timeout. - ctx, cancel := context.WithTimeout(store.WithTenantID(context.Background(), job.TenantID), cfg.Cron.JobTimeoutDuration()) + ctx, cancel := context.WithTimeout(cronTenantContext(context.Background(), tenantStore, job.TenantID), cfg.Cron.JobTimeoutDuration()) defer cancel() res := cronexec.Run(ctx, cronexec.Spec{ diff --git a/cmd/gateway_cron_command_test.go b/cmd/gateway_cron_command_test.go index bfa46075..af938ac4 100644 --- a/cmd/gateway_cron_command_test.go +++ b/cmd/gateway_cron_command_test.go @@ -39,7 +39,7 @@ func commandCronJob(spec *store.CronCommandSpec, deliver bool) *store.CronJob { // A command payload must be refused unless cron.command_enabled is set. func TestCronJobHandler_CommandDisabled(t *testing.T) { - handler := makeCronJobHandler(nil, nil, commandCronConfig(false), nil, nil, nil, nil, nil) + handler := makeCronJobHandler(nil, nil, commandCronConfig(false), nil, nil, nil, nil, nil, nil) if _, err := handler(commandCronJob(&store.CronCommandSpec{Argv: []string{"sh", "-c", "echo hi"}}, false)); err == nil { t.Fatal("expected error when cron.command_enabled is false") } @@ -50,7 +50,7 @@ func TestCronJobHandler_CommandSuccessDelivers(t *testing.T) { mb := bus.New() defer mb.Close() - handler := makeCronJobHandler(nil, mb, commandCronConfig(true), nil, nil, nil, nil, nil) + handler := makeCronJobHandler(nil, mb, commandCronConfig(true), nil, nil, nil, nil, nil, nil) result, err := handler(commandCronJob(&store.CronCommandSpec{Argv: []string{"sh", "-c", "printf hello"}}, true)) if err != nil { t.Fatalf("command cron returned error: %v", err) @@ -79,7 +79,7 @@ func TestCronJobHandler_CommandFailureNotDelivered(t *testing.T) { mb := bus.New() defer mb.Close() - handler := makeCronJobHandler(nil, mb, commandCronConfig(true), nil, nil, nil, nil, nil) + handler := makeCronJobHandler(nil, mb, commandCronConfig(true), nil, nil, nil, nil, nil, nil) result, err := handler(commandCronJob(&store.CronCommandSpec{Argv: []string{"sh", "-c", "echo boom 1>&2; exit 3"}}, true)) if err == nil { t.Fatal("expected error for non-zero command exit") @@ -97,7 +97,7 @@ func TestCronJobHandler_CommandFailureNotDelivered(t *testing.T) { // An empty argv is rejected before execution. func TestCronJobHandler_CommandInvalidSpec(t *testing.T) { - handler := makeCronJobHandler(nil, nil, commandCronConfig(true), nil, nil, nil, nil, nil) + handler := makeCronJobHandler(nil, nil, commandCronConfig(true), nil, nil, nil, nil, nil, nil) if _, err := handler(commandCronJob(&store.CronCommandSpec{}, false)); err == nil { t.Fatal("expected error for empty argv") } diff --git a/cmd/gateway_cron_test.go b/cmd/gateway_cron_test.go index 49ba5bd9..e63991c8 100644 --- a/cmd/gateway_cron_test.go +++ b/cmd/gateway_cron_test.go @@ -2,6 +2,7 @@ package cmd import ( "context" + "fmt" "testing" "time" @@ -43,6 +44,7 @@ func TestCronJobHandlerInjectsPayloadCredentialUserID(t *testing.T) { nil, nil, nil, + nil, ) result, err := handler(&store.CronJob{ @@ -141,6 +143,7 @@ func TestCronJobHandlerSuppressesNoReplyDelivery(t *testing.T) { nil, nil, nil, + nil, ) result, err := handler(&store.CronJob{ @@ -230,7 +233,7 @@ func TestCronJobHandler_StatelessResetsSession(t *testing.T) { ) defer sched.Stop() - handler := makeCronJobHandler(sched, nil, &config.Config{}, nil, fakeStore, nil, nil, nil) + handler := makeCronJobHandler(sched, nil, &config.Config{}, nil, fakeStore, nil, nil, nil, nil) if _, err := handler(&store.CronJob{ ID: uuid.NewString(), @@ -253,3 +256,69 @@ func TestCronJobHandler_StatelessResetsSession(t *testing.T) { }) } } + +// fakeTenantStore implements only GetTenant; embedding the interface satisfies +// the rest (calling any other method would nil-panic, which none of these tests do). +type fakeTenantStore struct { + store.TenantStore + byID map[uuid.UUID]*store.TenantData + err error +} + +func (f *fakeTenantStore) GetTenant(_ context.Context, id uuid.UUID) (*store.TenantData, error) { + if f.err != nil { + return nil, f.err + } + return f.byID[id], nil +} + +func TestCronTenantContext_InjectsSlugForNonMasterTenant(t *testing.T) { + tid := uuid.Must(uuid.NewV7()) + ts := &fakeTenantStore{byID: map[uuid.UUID]*store.TenantData{ + tid: {ID: tid, Slug: "family-pilot"}, + }} + + ctx := cronTenantContext(context.Background(), ts, tid) + + if got := store.TenantIDFromContext(ctx); got != tid { + t.Errorf("tenant id = %v, want %v", got, tid) + } + // The slug is what tenant-scoped skills-store/workspace paths key off; without + // it a cron agent turn sees none of its tenant's managed skills. + if got := store.TenantSlugFromContext(ctx); got != "family-pilot" { + t.Errorf("tenant slug = %q, want %q (skills-store would resolve to the wrong dir)", got, "family-pilot") + } +} + +func TestCronTenantContext_MasterTenantNeedsNoSlug(t *testing.T) { + // Master tenant paths resolve to the base dir regardless of slug; the store + // must not even be consulted. + ts := &fakeTenantStore{err: fmt.Errorf("GetTenant must not be called for master")} + ctx := cronTenantContext(context.Background(), ts, store.MasterTenantID) + if got := store.TenantIDFromContext(ctx); got != store.MasterTenantID { + t.Errorf("tenant id = %v, want master", got) + } +} + +func TestCronTenantContext_NilStore_TenantIDOnly(t *testing.T) { + tid := uuid.Must(uuid.NewV7()) + ctx := cronTenantContext(context.Background(), nil, tid) + if got := store.TenantIDFromContext(ctx); got != tid { + t.Errorf("tenant id = %v, want %v", got, tid) + } + if got := store.TenantSlugFromContext(ctx); got != "" { + t.Errorf("slug = %q, want empty when store is nil", got) + } +} + +func TestCronTenantContext_LookupError_FallsBackToIDOnly(t *testing.T) { + tid := uuid.Must(uuid.NewV7()) + ts := &fakeTenantStore{err: fmt.Errorf("db down")} + ctx := cronTenantContext(context.Background(), ts, tid) + if got := store.TenantSlugFromContext(ctx); got != "" { + t.Errorf("slug = %q, want empty on lookup error", got) + } + if got := store.TenantIDFromContext(ctx); got != tid { + t.Errorf("tenant id = %v, want %v (must still scope by id)", got, tid) + } +} diff --git a/cmd/gateway_heartbeat.go b/cmd/gateway_heartbeat.go index aec569ed..8dd6522c 100644 --- a/cmd/gateway_heartbeat.go +++ b/cmd/gateway_heartbeat.go @@ -42,7 +42,7 @@ func startCronAndHeartbeat( heartbeatMethods *methods.HeartbeatMethods, ) *heartbeat.Ticker { // Start cron service with job handler (routes through scheduler's cron lane) - pgStores.Cron.SetOnJob(makeCronJobHandler(sched, msgBus, cfg, channelMgr, pgStores.Sessions, pgStores.Agents, pgStores.Providers, providerRegistry)) + pgStores.Cron.SetOnJob(makeCronJobHandler(sched, msgBus, cfg, channelMgr, pgStores.Sessions, pgStores.Agents, pgStores.Tenants, pgStores.Providers, providerRegistry)) pgStores.Cron.SetOnEvent(func(event store.CronEvent) { server.BroadcastEvent(*protocol.NewEvent(protocol.EventCron, event)) })