From 7844df74f52f09b6fa56f0f3f3dbbf6942116d44 Mon Sep 17 00:00:00 2001 From: MToan <58007568+taminhtoan2601@users.noreply.github.com> Date: Tue, 7 Jul 2026 05:52:17 +0200 Subject: [PATCH] fix(cron): scope cron context to tenant slug so managed skills resolve (#1378) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cron job execution set the tenant ID on its context but not the tenant slug. Tenant-scoped filesystem paths (skills-store, workspace, media via config.TenantScopedDir) key off the slug and fall back to an id-based path when it is absent — a different directory than where HTTP/WS skill upload materialized the files (which sets the slug). As a result a cron agent turn in a non-master tenant saw NONE of its tenant's managed skills: skill_search returned 0 results and the agent, unable to run the skill, produced an ungrounded answer. Add cronTenantContext() which resolves the tenant slug via TenantStore and sets both WithTenantID and WithTenantSlug. Master tenant and nil-store/lookup-failure paths fall back to id-only (prior behavior). Thread TenantStore into makeCronJobHandler and runCommandCronJob. Tested: added unit tests for cronTenantContext (slug injected for non-master; master skips lookup; nil store and lookup error fall back to id-only). Verified end-to-end on a live tenant: before, a daily-agenda cron guessed an empty day; after, it read the real event from the DB. Note: other background executors that build a context from a tenant ID (e.g. heartbeat) likely share this gap and are worth an audit. --- cmd/gateway_cron.go | 36 ++++++++++++---- cmd/gateway_cron_command_test.go | 8 ++-- cmd/gateway_cron_test.go | 71 +++++++++++++++++++++++++++++++- cmd/gateway_heartbeat.go | 2 +- 4 files changed, 104 insertions(+), 13 deletions(-) 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)) })