diff --git a/docs/08-scheduling-cron.md b/docs/08-scheduling-cron.md index 7f23b5b0..d62de27a 100644 --- a/docs/08-scheduling-cron.md +++ b/docs/08-scheduling-cron.md @@ -122,6 +122,7 @@ Cancel commands for Telegram and other channels. - **Empty outbound**: On cancel, an empty outbound message is published to trigger cleanup (stop typing indicator, clear reactions) - **Trace finalization**: When `ctx.Err() != nil`, trace finalization falls back to `context.Background()` for the final DB write. Status is set to `"cancelled"` - **Context survival**: Context values (traceID, collector) survive cancellation -- only the Done channel fires +- **Background workers (ticker/cron) — tenant ctx injection required**: Jobs started from `context.Background()` carry no tenant. Before calling any tenant-scoped store method (e.g. `GetTeam`, `GetTask`, `GetByID`), the worker MUST inject `store.WithTenantID(ctx, tenantID)` derived from the row-level `tenant_id` (e.g. `RecoveredTaskInfo.TenantID`, `TeamTaskData.TenantID`). Callers must also nil-check returned entities — some stores (e.g. `PGTeamStore.GetTeam`) return `(nil, nil)` when tenant is missing rather than an error. See `internal/tasks/task_ticker.go` for the reference pattern - **Generation counter**: Each `SessionQueue` tracks a generation counter. When reset (e.g., during SIGUSR1 in-process restart), old generations are ignored, preventing stale completions from interfering with new requests --- diff --git a/internal/store/pg/teams_tasks.go b/internal/store/pg/teams_tasks.go index dac6e2c9..4b825df1 100644 --- a/internal/store/pg/teams_tasks.go +++ b/internal/store/pg/teams_tasks.go @@ -22,7 +22,7 @@ import ( const taskLockDuration = 60 * time.Minute // taskSelectCols is the shared SELECT column list for task queries (must match scanTaskRowsJoined). -const taskSelectCols = `t.id, t.team_id, t.subject, t.description, t.status, t.owner_agent_id, t.blocked_by, t.priority, t.result, t.user_id, t.channel, +const taskSelectCols = `t.id, t.team_id, t.tenant_id, t.subject, t.description, t.status, t.owner_agent_id, t.blocked_by, t.priority, t.result, t.user_id, t.channel, t.task_type, t.task_number, COALESCE(t.identifier,''), t.created_by_agent_id, COALESCE(t.assignee_user_id,''), t.parent_id, COALESCE(t.chat_id,''), t.metadata, t.locked_at, t.lock_expires_at, COALESCE(t.progress_percent,0), COALESCE(t.progress_step,''), t.followup_at, COALESCE(t.followup_count,0), COALESCE(t.followup_max,0), COALESCE(t.followup_message,''), COALESCE(t.followup_channel,''), COALESCE(t.followup_chat_id,''), @@ -560,7 +560,7 @@ func scanTaskRowsJoined(rows *sql.Rows) ([]store.TeamTaskData, error) { var followupCount, followupMax int var followupMessage, followupChannel, followupChatID string if err := rows.Scan( - &d.ID, &d.TeamID, &d.Subject, &desc, &d.Status, + &d.ID, &d.TeamID, &d.TenantID, &d.Subject, &desc, &d.Status, &ownerID, pq.Array(&blockedBy), &d.Priority, &result, &userID, &channel, &d.TaskType, &d.TaskNumber, &identifier, &createdByAgentID, &assigneeUserID, &parentID, diff --git a/internal/store/sqlitestore/teams_tasks.go b/internal/store/sqlitestore/teams_tasks.go index 0699666f..85540e4b 100644 --- a/internal/store/sqlitestore/teams_tasks.go +++ b/internal/store/sqlitestore/teams_tasks.go @@ -24,7 +24,7 @@ const taskLockDuration = 30 * time.Minute const maxListTasksRows = 30 // taskSelectCols is the shared SELECT column list for task queries. -const taskSelectCols = `t.id, t.team_id, t.subject, t.description, t.status, t.owner_agent_id, t.blocked_by, t.priority, t.result, t.user_id, t.channel, +const taskSelectCols = `t.id, t.team_id, t.tenant_id, t.subject, t.description, t.status, t.owner_agent_id, t.blocked_by, t.priority, t.result, t.user_id, t.channel, t.task_type, t.task_number, COALESCE(t.identifier,''), t.created_by_agent_id, COALESCE(t.assignee_user_id,''), t.parent_id, COALESCE(t.chat_id,''), t.metadata, t.locked_at, t.lock_expires_at, COALESCE(t.progress_percent,0), COALESCE(t.progress_step,''), t.followup_at, COALESCE(t.followup_count,0), COALESCE(t.followup_max,0), COALESCE(t.followup_message,''), COALESCE(t.followup_channel,''), COALESCE(t.followup_chat_id,''), @@ -510,7 +510,7 @@ func scanTaskRowsJoined(rows *sql.Rows) ([]store.TeamTaskData, error) { var followupMessage, followupChannel, followupChatID string createdAt, updatedAt := scanTimePair() if err := rows.Scan( - &d.ID, &d.TeamID, &d.Subject, &desc, &d.Status, + &d.ID, &d.TeamID, &d.TenantID, &d.Subject, &desc, &d.Status, &ownerID, &blockedByJSON, &d.Priority, &result, &userID, &channel, &d.TaskType, &d.TaskNumber, &identifier, &createdByAgentID, &assigneeUserID, &parentID, diff --git a/internal/store/team_store.go b/internal/store/team_store.go index 61ec76f7..d954530e 100644 --- a/internal/store/team_store.go +++ b/internal/store/team_store.go @@ -94,6 +94,7 @@ type TeamMemberData struct { type TeamTaskData struct { BaseModel TeamID uuid.UUID `json:"team_id" db:"team_id"` + TenantID uuid.UUID `json:"tenant_id" db:"tenant_id"` Subject string `json:"subject" db:"subject"` Description string `json:"description,omitempty" db:"description"` Status string `json:"status" db:"status"` diff --git a/internal/tasks/task_ticker.go b/internal/tasks/task_ticker.go index fbe85509..e5775317 100644 --- a/internal/tasks/task_ticker.go +++ b/internal/tasks/task_ticker.go @@ -195,27 +195,59 @@ func (t *TaskTicker) notifyLeaders(ctx context.Context, tasks []store.RecoveredT } // Cache team+lead lookups (same team may have multiple scopes). - teamCache := map[uuid.UUID]*store.TeamData{} - leadCache := map[uuid.UUID]*store.AgentData{} + // Composite key includes TenantID to clarify multi-tenant intent (UUIDs are globally + // unique but composite key makes the isolation boundary explicit and future-proof). + type teamCacheKey struct { + TeamID uuid.UUID + TenantID uuid.UUID + } + type leadCacheKey struct { + AgentID uuid.UUID + TenantID uuid.UUID + } + teamCache := map[teamCacheKey]*store.TeamData{} + leadCache := map[leadCacheKey]*store.AgentData{} for scope, scopeTasks := range byScope { - team := teamCache[scope.TeamID] + // Inject tenant into ctx so store lookups (GetTeam, GetByID, GetTask) apply the + // correct tenant filter. Without this, PGTeamStore.GetTeam silently returns + // (nil, nil) when ctx has no tenant — causing a nil-deref panic on team.LeadAgentID. + scopeCtx := ctx + if scope.TenantID != uuid.Nil { + scopeCtx = store.WithTenantID(ctx, scope.TenantID) + } + + teamKey := teamCacheKey{TeamID: scope.TeamID, TenantID: scope.TenantID} + team := teamCache[teamKey] if team == nil { var err error - team, err = t.teams.GetTeam(ctx, scope.TeamID) + team, err = t.teams.GetTeam(scopeCtx, scope.TeamID) if err != nil { + slog.Warn("task_ticker: get team failed", "team_id", scope.TeamID, "tenant_id", scope.TenantID, "error", err) continue } - teamCache[scope.TeamID] = team + if team == nil { + // GetTeam can return (nil, nil) when tenant ctx is missing or team is deleted. + slog.Warn("task_ticker: team not found (nil)", "team_id", scope.TeamID, "tenant_id", scope.TenantID) + continue + } + teamCache[teamKey] = team } - lead := leadCache[team.LeadAgentID] + + leadKey := leadCacheKey{AgentID: team.LeadAgentID, TenantID: scope.TenantID} + lead := leadCache[leadKey] if lead == nil { var err error - lead, err = t.agents.GetByID(ctx, team.LeadAgentID) + lead, err = t.agents.GetByID(scopeCtx, team.LeadAgentID) if err != nil { + slog.Warn("task_ticker: get lead agent failed", "agent_id", team.LeadAgentID, "tenant_id", scope.TenantID, "error", err) continue } - leadCache[team.LeadAgentID] = lead + if lead == nil { + slog.Warn("task_ticker: lead agent not found (nil)", "agent_id", team.LeadAgentID, "tenant_id", scope.TenantID) + continue + } + leadCache[leadKey] = lead } // Build batched task list with clear actionable hints. @@ -238,7 +270,7 @@ func (t *TaskTicker) notifyLeaders(ctx context.Context, tasks []store.RecoveredT // Resolve PeerKind from first task's metadata for correct session routing (#266). var peerKind string var fullTask *store.TeamTaskData - if task, err := t.teams.GetTask(ctx, scopeTasks[0].ID); err == nil { + if task, err := t.teams.GetTask(scopeCtx, scopeTasks[0].ID); err == nil { fullTask = task if fullTask != nil && fullTask.Metadata != nil { if pk, ok := fullTask.Metadata["peer_kind"].(string); ok { @@ -304,12 +336,27 @@ func (t *TaskTicker) processFollowups(ctx context.Context) { byTeam[task.TeamID] = append(byTeam[task.TeamID], task) } for teamID, teamTasks := range byTeam { - team, err := t.teams.GetTeam(ctx, teamID) + if len(teamTasks) == 0 { + continue + } + tenantID := teamTasks[0].TenantID + scopeCtx := ctx + if tenantID != uuid.Nil { + scopeCtx = store.WithTenantID(ctx, tenantID) + } + team, err := t.teams.GetTeam(scopeCtx, teamID) if err != nil { + slog.Warn("task_ticker: followups get team failed", + "team_id", teamID, "tenant_id", tenantID, "error", err) + continue + } + if team == nil { + slog.Warn("task_ticker: followups team not found (nil)", + "team_id", teamID, "tenant_id", tenantID) continue } interval := followupInterval(*team) - t.processTeamFollowups(ctx, teamTasks, interval) + t.processTeamFollowups(scopeCtx, teamTasks, interval) } } diff --git a/internal/tasks/task_ticker_test.go b/internal/tasks/task_ticker_test.go index d93802f1..432c6ca5 100644 --- a/internal/tasks/task_ticker_test.go +++ b/internal/tasks/task_ticker_test.go @@ -18,6 +18,12 @@ import ( // ─── minimal stub stores ─────────────────────────────────────────────────── +// getTeamCall records ctx metadata for each GetTeam call (tenant propagation assertions). +type getTeamCall struct { + TeamID uuid.UUID + TenantID uuid.UUID +} + // stubTeamStore satisfies store.TeamStore by embedding the interface and only // overriding the methods actually exercised by TaskTicker. type stubTeamStore struct { @@ -37,11 +43,18 @@ type stubTeamStore struct { recoverErr error staleErr error - recovered []store.RecoveredTaskInfo - stale []store.RecoveredTaskInfo - inReview []store.RecoveredTaskInfo - orphans []store.RecoveredTaskInfo + recovered []store.RecoveredTaskInfo + stale []store.RecoveredTaskInfo + inReview []store.RecoveredTaskInfo + orphans []store.RecoveredTaskInfo followupTasks []store.TeamTaskData + + // function-based dispatch — nil means default behavior (return error "not found") + getTeamFunc func(ctx context.Context, id uuid.UUID) (*store.TeamData, error) + getTaskFunc func(ctx context.Context, id uuid.UUID) (*store.TeamTaskData, error) + + // captured calls for tenant propagation assertions (protected by mu) + getTeamCalls []getTeamCall } func (s *stubTeamStore) ForceRecoverAllTasks(_ context.Context) ([]store.RecoveredTaskInfo, error) { @@ -79,17 +92,38 @@ func (s *stubTeamStore) IncrementFollowupCount(_ context.Context, _ uuid.UUID, _ return nil } -func (s *stubTeamStore) GetTeam(_ context.Context, _ uuid.UUID) (*store.TeamData, error) { +func (s *stubTeamStore) GetTeam(ctx context.Context, id uuid.UUID) (*store.TeamData, error) { + s.mu.Lock() + s.getTeamCalls = append(s.getTeamCalls, getTeamCall{ + TeamID: id, + TenantID: store.TenantIDFromContext(ctx), + }) + s.mu.Unlock() + if s.getTeamFunc != nil { + return s.getTeamFunc(ctx, id) + } return nil, errors.New("not found") } -func (s *stubTeamStore) GetTask(_ context.Context, _ uuid.UUID) (*store.TeamTaskData, error) { +func (s *stubTeamStore) GetTask(ctx context.Context, id uuid.UUID) (*store.TeamTaskData, error) { + if s.getTaskFunc != nil { + return s.getTaskFunc(ctx, id) + } return nil, errors.New("not found") } -// stubAgentStore satisfies store.AgentStore — unused by ticker logic. +// stubAgentStore satisfies store.AgentStore — unused by most ticker logic. +// GetByID is overridable via getByIDFunc for tests exercising the lead-agent lookup path. type stubAgentStore struct { store.AgentStore + getByIDFunc func(ctx context.Context, id uuid.UUID) (*store.AgentData, error) +} + +func (s *stubAgentStore) GetByID(ctx context.Context, id uuid.UUID) (*store.AgentData, error) { + if s.getByIDFunc != nil { + return s.getByIDFunc(ctx, id) + } + return nil, errors.New("not found") } // ─── NewTaskTicker ───────────────────────────────────────────────────────── @@ -491,3 +525,229 @@ func TestFollowupOutboundMessage_OmitsLocalKeyWhenMissing(t *testing.T) { t.Fatalf("expected metadata to be nil, got %#v", got.Metadata) } } + +// ─── notifyLeaders: nil-team and tenant propagation ────────────────────────── + +// TestNotifyLeaders_NilTeam_NoPanic verifies that when GetTeam returns (nil, nil) +// the scope is skipped gracefully — no panic, no publishes. +func TestNotifyLeaders_NilTeam_NoPanic(t *testing.T) { + tenantID := uuid.New() + teamID := uuid.New() + + ts := &stubTeamStore{ + getTeamFunc: func(_ context.Context, _ uuid.UUID) (*store.TeamData, error) { + return nil, nil // simulate PGTeamStore returning nil without error + }, + } + as := &stubAgentStore{} + mb := bus.New() + tt := NewTaskTicker(ts, as, mb, 3600) + + // Must not panic. + tt.notifyLeaders(context.Background(), []store.RecoveredTaskInfo{ + {ID: uuid.New(), TeamID: teamID, TenantID: tenantID}, + }, "recovered", "hint") + + ts.mu.Lock() + callCount := len(ts.getTeamCalls) + capturedTenant := uuid.Nil + if callCount > 0 { + capturedTenant = ts.getTeamCalls[0].TenantID + } + ts.mu.Unlock() + + if callCount != 1 { + t.Errorf("GetTeam call count = %d, want 1", callCount) + } + if capturedTenant != tenantID { + t.Errorf("GetTeam ctx TenantID = %v, want %v", capturedTenant, tenantID) + } +} + +// TestNotifyLeaders_MultiTenantCacheIsolation verifies that tasks from different tenants +// each trigger their own GetTeam call with the correct tenant in ctx. +func TestNotifyLeaders_MultiTenantCacheIsolation(t *testing.T) { + tenantX := uuid.New() + tenantY := uuid.New() + teamT1 := uuid.New() + teamT2 := uuid.New() + leadID := uuid.New() + + ts := &stubTeamStore{ + getTeamFunc: func(_ context.Context, id uuid.UUID) (*store.TeamData, error) { + return &store.TeamData{ + BaseModel: store.BaseModel{ID: id}, + LeadAgentID: leadID, + }, nil + }, + } + as := &stubAgentStore{ + // short-circuit before publish; return error so notifyLeaders skips publishing + getByIDFunc: func(_ context.Context, _ uuid.UUID) (*store.AgentData, error) { + return nil, errors.New("not found") + }, + } + mb := bus.New() + tt := NewTaskTicker(ts, as, mb, 3600) + + tasks := []store.RecoveredTaskInfo{ + {ID: uuid.New(), TeamID: teamT1, TenantID: tenantX}, + {ID: uuid.New(), TeamID: teamT2, TenantID: tenantY}, + } + tt.notifyLeaders(context.Background(), tasks, "recovered", "hint") + + ts.mu.Lock() + calls := make([]getTeamCall, len(ts.getTeamCalls)) + copy(calls, ts.getTeamCalls) + ts.mu.Unlock() + + if len(calls) != 2 { + t.Fatalf("GetTeam call count = %d, want 2", len(calls)) + } + + // Build map: teamID → tenantID from captured calls. + captured := map[uuid.UUID]uuid.UUID{} + for _, c := range calls { + captured[c.TeamID] = c.TenantID + } + + if captured[teamT1] != tenantX { + t.Errorf("team T1: ctx TenantID = %v, want %v", captured[teamT1], tenantX) + } + if captured[teamT2] != tenantY { + t.Errorf("team T2: ctx TenantID = %v, want %v", captured[teamT2], tenantY) + } +} + +// TestNotifyLeaders_CachedOnSameTenant verifies that two scopes with the same team+tenant +// result in only one GetTeam call (cache hit on composite key). +func TestNotifyLeaders_CachedOnSameTenant(t *testing.T) { + tenantID := uuid.New() + teamID := uuid.New() + leadID := uuid.New() + + ts := &stubTeamStore{ + getTeamFunc: func(_ context.Context, id uuid.UUID) (*store.TeamData, error) { + return &store.TeamData{ + BaseModel: store.BaseModel{ID: id}, + LeadAgentID: leadID, + }, nil + }, + } + as := &stubAgentStore{ + getByIDFunc: func(_ context.Context, _ uuid.UUID) (*store.AgentData, error) { + return nil, errors.New("not found") + }, + } + mb := bus.New() + tt := NewTaskTicker(ts, as, mb, 3600) + + // Two tasks same team+tenant but different ChatID → two scopes, same cache key. + tasks := []store.RecoveredTaskInfo{ + {ID: uuid.New(), TeamID: teamID, TenantID: tenantID, ChatID: "chat-A"}, + {ID: uuid.New(), TeamID: teamID, TenantID: tenantID, ChatID: "chat-B"}, + } + tt.notifyLeaders(context.Background(), tasks, "recovered", "hint") + + ts.mu.Lock() + callCount := len(ts.getTeamCalls) + ts.mu.Unlock() + + if callCount != 1 { + t.Errorf("GetTeam call count = %d, want 1 (cache hit)", callCount) + } +} + +// ─── processFollowups: multi-tenant tenant ctx and nil-team guard ───────────── + +// TestProcessFollowups_MultiTenantTenantCtx verifies that tasks from two different +// tenants each trigger GetTeam with the correct tenant in ctx. +func TestProcessFollowups_MultiTenantTenantCtx(t *testing.T) { + tenantX := uuid.New() + tenantY := uuid.New() + teamT1 := uuid.New() + teamT2 := uuid.New() + + ts := &stubTeamStore{ + followupTasks: []store.TeamTaskData{ + { + BaseModel: store.BaseModel{ID: uuid.New()}, + TeamID: teamT1, + TenantID: tenantX, + FollowupChannel: "telegram", + FollowupChatID: "chat-1", + FollowupMessage: "ping", + }, + { + BaseModel: store.BaseModel{ID: uuid.New()}, + TeamID: teamT2, + TenantID: tenantY, + FollowupChannel: "telegram", + FollowupChatID: "chat-2", + FollowupMessage: "ping", + }, + }, + getTeamFunc: func(_ context.Context, id uuid.UUID) (*store.TeamData, error) { + return &store.TeamData{BaseModel: store.BaseModel{ID: id}}, nil + }, + } + as := &stubAgentStore{} + mb := bus.New() + tt := NewTaskTicker(ts, as, mb, 3600) + + tt.processFollowups(context.Background()) + + ts.mu.Lock() + calls := make([]getTeamCall, len(ts.getTeamCalls)) + copy(calls, ts.getTeamCalls) + ts.mu.Unlock() + + if len(calls) != 2 { + t.Fatalf("GetTeam call count = %d, want 2", len(calls)) + } + + captured := map[uuid.UUID]uuid.UUID{} + for _, c := range calls { + captured[c.TeamID] = c.TenantID + } + + if captured[teamT1] != tenantX { + t.Errorf("team T1: ctx TenantID = %v, want %v", captured[teamT1], tenantX) + } + if captured[teamT2] != tenantY { + t.Errorf("team T2: ctx TenantID = %v, want %v", captured[teamT2], tenantY) + } +} + +// TestProcessFollowups_NilTeam_NoPanic verifies that when GetTeam returns (nil, nil) +// processFollowups skips without panicking, and no followup is sent. +func TestProcessFollowups_NilTeam_NoPanic(t *testing.T) { + tenantID := uuid.New() + teamID := uuid.New() + + ts := &stubTeamStore{ + followupTasks: []store.TeamTaskData{ + { + BaseModel: store.BaseModel{ID: uuid.New()}, + TeamID: teamID, + TenantID: tenantID, + FollowupChannel: "telegram", + FollowupChatID: "chat-1", + FollowupMessage: "ping", + }, + }, + getTeamFunc: func(_ context.Context, _ uuid.UUID) (*store.TeamData, error) { + return nil, nil // simulate nil without error + }, + } + as := &stubAgentStore{} + mb := bus.New() + tt := NewTaskTicker(ts, as, mb, 3600) + + // Must not panic. + tt.processFollowups(context.Background()) + + if ts.incrementCalls.Load() != 0 { + t.Errorf("IncrementFollowupCount = %d, want 0 (nil team skipped)", ts.incrementCalls.Load()) + } +}