feat(worker): cohort skip-guards — no-op every team-iterating job for is_test_cohort teams (PR-1 follow-up)

internal/jobs/billing_reconciler.go

@@ -822,6 +822,7 @@ func (w *BillingReconcilerWorker) Work(ctx context.Context, job *river.Job[Billi
 		  FROM teams
 		 WHERE stripe_customer_id IS NOT NULL
 		   AND stripe_customer_id != ''
+		   AND NOT is_test_cohort
 		 ORDER BY id
 		 LIMIT $1
 	`, billingReconcilerBatchLimit)
@@ -1204,6 +1205,7 @@ func (w *BillingReconcilerWorker) runOrphanSweep(ctx context.Context) (scanned,
 		  JOIN teams t ON t.id = pc.team_id
 		 WHERE pc.subscription_id IS NOT NULL
 		   AND pc.subscription_id != ''
+		   AND NOT t.is_test_cohort
 		   AND pc.created_at < $1
 		 ORDER BY pc.created_at
 		 LIMIT $2

internal/jobs/checkout_reconcile.go

@@ -148,9 +148,10 @@ func (w *CheckoutReconcileWorker) Work(ctx context.Context, job *river.Job[Check
 
 	rows, err := w.db.QueryContext(ctx, `
 		SELECT subscription_id, team_id, customer_email, plan_tier
-		FROM pending_checkouts
+		FROM pending_checkouts pc
 		WHERE resolved_at IS NULL
 		  AND failure_notified_at IS NULL
+		  AND `+testCohortNotExistsClause("pc.team_id")+`
 		  AND created_at < $1
 		ORDER BY created_at ASC
 		LIMIT $2

internal/jobs/churn_predictor.go

@@ -87,14 +87,14 @@ const churnInactivityWindow = 7 * 24 * time.Hour
 // churnDedupeWindow is the minimum gap between two churn.risk_flagged
 // rows for the same team. 30 days is the brief's window. Rationale:
 //
-//   * Slower than 30d would let a team get re-flagged month after month
+//   - Slower than 30d would let a team get re-flagged month after month
 //     without intervention — operationally noisy and emotionally
 //     repetitive ("you again missed us") for the customer.
-//   * Faster than 30d (e.g. weekly) would risk sending the same
+//   - Faster than 30d (e.g. weekly) would risk sending the same
 //     "we miss you" copy four times in a month to a customer who has
 //     decided not to return; the perception cost (spammy) outweighs
 //     the recall benefit.
-//   * 30d also aligns with monthly billing cycles — most B2B SaaS
+//   - 30d also aligns with monthly billing cycles — most B2B SaaS
 //     reactivation playbooks rate-limit on a monthly cadence for the
 //     same reason.
 //
@@ -210,6 +210,7 @@ func (w *ChurnPredictorWorker) Work(ctx context.Context, job *river.Job[ChurnPre
 			)
 		LEFT JOIN resources r ON r.team_id = t.id AND r.status = 'active'
 		WHERE t.plan_tier != 'team'
+		  AND NOT t.is_test_cohort
 		  AND NOT EXISTS (
 			SELECT 1 FROM audit_log f
 			WHERE f.team_id = t.id
@@ -300,10 +301,10 @@ func (w *ChurnPredictorWorker) Work(ctx context.Context, job *river.Job[ChurnPre
 		// the brief enumerated lives here: tier, last_activity_days_ago,
 		// active_resource_count, email.
 		meta := map[string]any{
-			"tier":                    r.planTier,
-			"last_activity_days_ago":  daysSince,
-			"active_resource_count":   r.activeResourceCount,
-			"email":                   r.ownerEmail.String,
+			"tier":                   r.planTier,
+			"last_activity_days_ago": daysSince,
+			"active_resource_count":  r.activeResourceCount,
+			"email":                  r.ownerEmail.String,
 		}
 		metaBytes, mErr := churnMetaMarshal(meta)
 		if mErr != nil {

internal/jobs/email.go

@@ -120,6 +120,7 @@ func (w *WeeklyDigestWorker) Work(ctx context.Context, job *river.Job[WeeklyDige
 		FROM users u
 		JOIN teams t ON t.id = u.team_id
 		WHERE t.plan_tier != 'anonymous'
+		  AND NOT t.is_test_cohort
 		  AND NOT EXISTS (
 		    SELECT 1 FROM audit_log a
 		    WHERE a.team_id = t.id
@@ -228,11 +229,12 @@ func (w *WeeklyDigestWorker) buildResourceDigestCounts(ctx context.Context, team
 // to if a per-row count is missing.
 //
 // Metadata shape — buildDigestWeekly (event_email_mapping.go) reads these:
-//   email                   — recipient address (also the forwarder's resolver)
-//   team_name               — display name for the email greeting (may be "")
-//   total_active_resources  — sum across all resource_types; 0 means "empty week"
-//   resource_breakdown      — JSON-encoded array of {resource_type, count}
-//                             (templates that don't render a table can ignore)
+//
+//	email                   — recipient address (also the forwarder's resolver)
+//	team_name               — display name for the email greeting (may be "")
+//	total_active_resources  — sum across all resource_types; 0 means "empty week"
+//	resource_breakdown      — JSON-encoded array of {resource_type, count}
+//	                          (templates that don't render a table can ignore)
 func emitWeeklyDigestAudit(ctx context.Context, db *sql.DB, teamID uuid.UUID, email, teamName string, stats []DigestResourceCount) error {
 	var total int64
 	for _, s := range stats {
@@ -268,4 +270,3 @@ func emitWeeklyDigestAudit(ctx context.Context, db *sql.DB, teamID uuid.UUID, em
 	}
 	return nil
 }
-

internal/jobs/expire_imminent.go

@@ -167,6 +167,7 @@ func (w *ExpireImminentWorker) Work(ctx context.Context, job *river.Job[ExpireIm
 		  AND r.expires_at IS NOT NULL
 		  AND r.expires_at > $1
 		  AND r.expires_at < $2
+		  AND `+testCohortNotExistsClause("r.team_id")+`
 		  AND NOT EXISTS (
 			SELECT 1
 			FROM audit_log al
@@ -303,4 +304,3 @@ func (w *ExpireImminentWorker) Work(ctx context.Context, job *river.Job[ExpireIm
 	)
 	return nil
 }
-

internal/jobs/expiry_reminder.go

@@ -181,6 +181,7 @@ func (w *ExpiryReminderWorker) Work(ctx context.Context, job *river.Job[ExpiryRe
 		WHERE r.team_id IS NOT NULL
 		  AND r.tier = 'free'
 		  AND r.status = 'active'
+		  AND `+testCohortNotExistsClause("r.team_id")+`
 		  AND r.expires_at IS NOT NULL
 		  AND r.expires_at > $1
 		  AND r.expires_at <= $2

internal/jobs/payment_grace_reminder.go

@@ -113,10 +113,11 @@ func (w *PaymentGraceReminderWorker) Work(ctx context.Context, job *river.Job[Pa
 	// first so a backlog drains in FIFO order.
 	rows, err := w.db.QueryContext(ctx, `
 		SELECT id, team_id, expires_at
-		FROM payment_grace_periods
+		FROM payment_grace_periods pgp
 		WHERE status = 'active'
 		  AND expires_at > $1
 		  AND (last_reminder_at IS NULL OR last_reminder_at < $2)
+		  AND `+testCohortNotExistsClause("pgp.team_id")+`
 		ORDER BY last_reminder_at ASC NULLS FIRST, started_at ASC
 		LIMIT $3
 	`, now, cutoff, paymentGraceReminderBatchLimit)
@@ -186,9 +187,9 @@ func (w *PaymentGraceReminderWorker) Work(ctx context.Context, job *river.Job[Pa
 
 		hoursRemaining := math.Round(r.expiresAt.Sub(now).Hours()*10) / 10
 		meta := map[string]any{
-			"grace_id":         r.id.String(),
-			"hours_remaining":  hoursRemaining,
-			"grace_ends_at":    r.expiresAt.UTC().Format(time.RFC3339),
+			"grace_id":        r.id.String(),
+			"hours_remaining": hoursRemaining,
+			"grace_ends_at":   r.expiresAt.UTC().Format(time.RFC3339),
 		}
 		metaBytes, mErr := graceReminderMetaMarshal(meta)
 		if mErr != nil {

internal/jobs/payment_grace_terminator.go

@@ -149,9 +149,10 @@ func (w *PaymentGraceTerminatorWorker) Work(ctx context.Context, job *river.Job[
 	// Candidate query: active grace rows whose clock has expired.
 	rows, err := w.db.QueryContext(ctx, `
 		SELECT id, team_id, expires_at
-		FROM payment_grace_periods
+		FROM payment_grace_periods pgp
 		WHERE status = 'active'
 		  AND expires_at < $1
+		  AND `+testCohortNotExistsClause("pgp.team_id")+`
 		ORDER BY expires_at ASC
 		LIMIT $2
 	`, now, paymentGraceTerminatorBatchLimit)
@@ -308,14 +309,15 @@ func (w *PaymentGraceTerminatorWorker) terminate(ctx context.Context, teamID uui
 // HS256 algorithm and the claim set below.
 //
 // Claims:
-//   sub  — the team id being terminated (allows the api to assert the
-//          path :id matches the JWT subject; prevents a stolen token
-//          from terminating a different team)
-//   iss  — "instanode-worker" so the api can route on issuer
-//   iat  — issued-at (UTC seconds)
-//   exp  — issued-at + 5 minutes (short window — single-shot use)
-//   aud  — "internal-teams-terminate" so the same secret can't be
-//          re-used by an attacker against a different internal route
+//
+//	sub  — the team id being terminated (allows the api to assert the
+//	       path :id matches the JWT subject; prevents a stolen token
+//	       from terminating a different team)
+//	iss  — "instanode-worker" so the api can route on issuer
+//	iat  — issued-at (UTC seconds)
+//	exp  — issued-at + 5 minutes (short window — single-shot use)
+//	aud  — "internal-teams-terminate" so the same secret can't be
+//	       re-used by an attacker against a different internal route
 func signWorkerInternalJWT(secret, teamID string) (string, error) {
 	if secret == "" {
 		return "", errors.New("empty secret")

internal/jobs/quota.go

@@ -241,9 +241,10 @@ func (w *EnforceStorageQuotaWorker) runRedisEvictionLoop(ctx context.Context) (i
 	for {
 		rows, err := w.db.QueryContext(ctx, `
 			SELECT id, token, tier, storage_bytes
-			FROM resources
+			FROM resources r
 			WHERE status = $1
 			  AND resource_type = 'redis'
+			  AND `+testCohortNotExistsClause("r.team_id")+`
 			  AND id::text > $2
 			ORDER BY id::text ASC
 			LIMIT $3
@@ -362,9 +363,10 @@ func (w *EnforceStorageQuotaWorker) runSuspendLoop(ctx context.Context) ([]strin
 			SELECT id, token, resource_type, tier, storage_bytes,
 			       COALESCE(provider_resource_id, ''),
 			       team_id, COALESCE(name, '')
-			FROM resources
+			FROM resources r
 			WHERE status = $1
 			  AND resource_type IN ('postgres', 'redis', 'mongodb')
+			  AND `+testCohortNotExistsClause("r.team_id")+`
 			  AND id::text > $2
 			ORDER BY id::text ASC
 			LIMIT $3
@@ -528,9 +530,10 @@ func (w *EnforceStorageQuotaWorker) runUnsuspendLoop(ctx context.Context, skipID
 			SELECT id, token, resource_type, tier, storage_bytes,
 			       COALESCE(provider_resource_id, ''),
 			       team_id, COALESCE(name, '')
-			FROM resources
+			FROM resources r
 			WHERE status = $1
 			  AND resource_type IN ('postgres', 'redis', 'mongodb')
+			  AND `+testCohortNotExistsClause("r.team_id")+`
 			  AND id::text > $2
 			ORDER BY id::text ASC
 			LIMIT $3

internal/jobs/quota_wall_nudge.go

@@ -145,6 +145,7 @@ func (w *QuotaWallNudgeWorker) Work(ctx context.Context, job *river.Job[QuotaWal
 			SELECT id, plan_tier
 			FROM teams
 			WHERE plan_tier NOT IN ('team', 'anonymous', 'free')
+			  AND NOT is_test_cohort
 			  AND id::text > $1
 			ORDER BY id::text ASC
 			LIMIT $2

internal/jobs/test_cohort.go

@@ -0,0 +1,133 @@
+package jobs
+
+// test_cohort.go — shared skip-guards for synthetic test-cohort teams.
+//
+// teams.is_test_cohort (api migration 067) tags durable per-tier teams the
+// continuous synthetic-monitoring program seeds (synthetic+hobby@instanode.dev,
+// etc.). Those teams provision real resources on a continuous cadence so the
+// synthetic runner can assert the live funnel — but to every background job
+// they would otherwise look like real customers: the billing reconciler would
+// flag them as drift (no real Razorpay subscription), the quota/expiry/churn/
+// digest jobs would email synthetic addresses (Brevo-rejected noise + ledger
+// pollution) and skew the real 2%/20% funnel.
+//
+// This file is the worker-side follow-up to api PR #246 (W0): every
+// team-iterating job that charges, churns, emails, or quota-nudges a team MUST
+// no-op for is_test_cohort teams. The plan enumerating the jobs is
+// docs/sessions/2026-06-04/TEST-ACCOUNTS-AND-NR-SYNTHETICS-PLAN.md §1.6.
+//
+// # The two skip mechanisms
+//
+//  1. SQL-driven scans (a SELECT over teams / resources / a team-keyed table)
+//     add the cohort filter inline so a test-cohort team never enters the
+//     candidate set. SQL-driven jobs join teams and add `AND NOT t.is_test_cohort`,
+//     or — when the candidate row carries a team_id but does not already join
+//     teams — gate via a NOT EXISTS subselect (testCohortNotExistsClause).
+//
+//  2. Per-team Go checks (a job that already holds a team_id mid-loop) call
+//     isTestCohort(ctx, db, teamID) and `continue` on true.
+//
+// # Inert by default
+//
+// No real team carries is_test_cohort=true today (migration 067 defaults every
+// row to false and only a future seeder flips the synthetic teams). So every
+// guard added here is a pure no-op for all real teams — a safety guard, not a
+// feature, and it deliberately does NOT depend on FLOW_SYNTHETIC_ENABLED (that
+// flag gates the future synthetic RUNNER, not these guards).
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"log/slog"
+
+	"github.com/google/uuid"
+)
+
+// testCohortNotExistsClause is the SQL fragment that excludes synthetic
+// test-cohort teams from a candidate scan whose driving table carries a
+// `team_id` but does not already join teams. Drop it into the WHERE clause of
+// such a scan; it filters out any row whose owning team is flagged
+// is_test_cohort=true.
+//
+// It is a correlated NOT EXISTS (rather than `team_id NOT IN (SELECT …)`)
+// because NOT IN is NULL-unsafe: an anonymous resource carries team_id IS NULL,
+// and `NULL NOT IN (…)` evaluates to NULL (treated as false), which would
+// silently drop every anonymous row from the scan. NOT EXISTS keeps NULL-
+// team_id rows in the result set — they cannot be a synthetic team (synthetic
+// teams are always real, seeded teams), so an anonymous row must never be
+// filtered by the cohort guard. The correlation alias is the literal `r` so the
+// fragment slots into the expiry reaper's existing `FROM resources r`; jobs
+// whose driving alias differs format the team-id column in via %s.
+//
+// Usage (resources reaper, alias r):
+//
+//	WHERE … AND NOT EXISTS (
+//	    SELECT 1 FROM teams tc WHERE tc.id = r.team_id AND tc.is_test_cohort
+//	)
+//
+// Kept as a formatting helper so the (single, fixed) `tc` subselect alias can
+// never collide with the caller's own table aliases.
+func testCohortNotExistsClause(teamIDExpr string) string {
+	return fmt.Sprintf(
+		"NOT EXISTS (SELECT 1 FROM teams tc WHERE tc.id = %s AND tc.is_test_cohort)",
+		teamIDExpr,
+	)
+}
+
+// isTestCohort reports whether the given team is a synthetic test-cohort team
+// (teams.is_test_cohort = true). It is the per-team Go guard for jobs that
+// already hold a team_id mid-loop and cannot push the filter into their
+// candidate SELECT (e.g. the payment-grace and checkout reconcilers, which scan
+// payment_grace_periods / pending_checkouts and only carry the team_id forward).
+//
+// Fail-CLOSED is wrong here and fail-OPEN is wrong too — instead it fails
+// SAFE-FOR-REAL-TEAMS: on any DB error (or a team row that has vanished) it
+// returns (false, err). The caller treats a non-nil err as "could not confirm
+// synthetic — process the team normally", so a transient platform-DB blip never
+// silently suppresses a REAL customer's charge / dunning email. Because no real
+// team is ever is_test_cohort=true, treating "unknown" as "not synthetic" can
+// only ever mis-handle a synthetic team during a DB outage (a tolerable,
+// self-correcting miss), never a real one.
+func isTestCohort(ctx context.Context, db *sql.DB, teamID uuid.UUID) (bool, error) {
+	var flagged bool
+	err := db.QueryRowContext(ctx,
+		`SELECT is_test_cohort FROM teams WHERE id = $1`, teamID,
+	).Scan(&flagged)
+	if errors.Is(err, sql.ErrNoRows) {
+		// Team row gone (deleted between the candidate SELECT and this check).
+		// Not synthetic by any meaningful definition; let the caller's own
+		// row-vanished handling take over.
+		return false, nil
+	}
+	if err != nil {
+		return false, fmt.Errorf("isTestCohort: %w", err)
+	}
+	return flagged, nil
+}
+
+// skipIfTestCohort is the convenience wrapper the per-team-loop jobs use. It
+// returns true when the team should be SKIPPED (it is synthetic). On a DB error
+// it logs at WARN under jobKind and returns false — "could not confirm, so
+// process normally" — so a platform-DB blip never suppresses a real customer's
+// money/email path (see isTestCohort's fail-safe-for-real-teams rationale).
+func skipIfTestCohort(ctx context.Context, db *sql.DB, teamID uuid.UUID, jobKind string) bool {
+	flagged, err := isTestCohort(ctx, db, teamID)
+	if err != nil {
+		slog.Warn(jobKind+".test_cohort_check_failed",
+			"team_id", teamID.String(),
+			"error", err,
+			"note", "could not confirm is_test_cohort — processing team normally (no real team is synthetic, so this only ever risks one synthetic tick during a DB blip)",
+		)
+		return false
+	}
+	if flagged {
+		slog.Debug(jobKind+".test_cohort_skipped",
+			"team_id", teamID.String(),
+			"note", "synthetic test-cohort team — no-op (no charge/churn/email/quota effect)",
+		)
+		return true
+	}
+	return false
+}