diff --git a/internal/jobs/e2e_cohort_sweep.go b/internal/jobs/e2e_cohort_sweep.go
new file mode 100644
index 0000000..7275605
--- /dev/null
+++ b/internal/jobs/e2e_cohort_sweep.go
@@ -0,0 +1,272 @@
+package jobs
+
+// e2e_cohort_sweep.go — belt-and-suspenders reaper for STALE synthetic
+// test-cohort teams.
+//
+// PROBLEM SHAPE
+//
+// CI's E2E suite mints an ephemeral team via the api's /internal/e2e/account
+// endpoint (api PR #260). Every such team is tagged teams.is_test_cohort=true
+// (api migration 067). The happy path is that the SAME CI run deletes its
+// account at the end of the run — but a CI run that crashes, times out, or is
+// hard-cancelled mid-flight dies BEFORE it reaps its account. The leftover
+// is_test_cohort team then lingers: it holds real provisioned resources +
+// k8s namespaces, and it is a stray row in every operator view.
+//
+// THIS JOB is the durable backstop. Every cohortSweepInterval it finds
+// is_test_cohort teams older than cohortSweepTTL that are not already in a
+// terminal/destroying state, and purges each one the SAME way the team-
+// deletion executor (team_deletion_executor.go) and the api reap path do:
+// flip the team to deletion_pending, then run the executor's idempotent
+// per-team teardown (deprovision resources, delete k8s namespaces, scrub
+// PII, tombstone). It reuses the executor wholesale — no second copy of the
+// purge cascade.
+//
+// SAFETY — this NEVER touches a non-cohort team
+//
+// The candidate query filters on `is_test_cohort = true`, so by construction
+// only cohort teams are ever selected. As a defence in depth (the WHERE
+// clause is the single point of failure for "did we pick the right rows"),
+// each candidate is RE-CHECKED with isTestCohort() immediately before the
+// purge; a row that is no longer is_test_cohort=true is skipped and counted
+// under outcome="skipped_not_cohort" (an alert-able disagreement between the
+// candidate query and the guard). No real team is is_test_cohort=true today
+// (migration 067 defaults every row to false), so in practice this job is
+// inert — it only ever matches a future synthetic/CI cohort row.
+//
+// SHORT TTL
+//
+// CI runs complete in minutes, so cohortSweepTTL is deliberately short
+// (2h). That is long enough that a still-running CI job's freshly-created
+// account is never reaped out from under it, yet short enough that a leaked
+// account is reclaimed the same hour. The TTL is a named const so the
+// integration test can reason about "stale vs fresh" without magic numbers.
+//
+// IDEMPOTENCY + FAIL-SAFE
+//
+// processTeam is fully idempotent (a re-run over an already-tombstoned team
+// is a no-op). A per-team failure is isolated: it stays in deletion_pending
+// for the next tick (or the orphan-sweep reconciler's PASS 1), is counted
+// under outcome="failed", and never stops the sweep from attempting the next
+// team. A nil executor disables the job with a single WARN (matching every
+// other worker's fail-open posture). A top-level candidate-query failure
+// returns an error so River retries the dispatch.
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"github.com/riverqueue/river"
+	"go.opentelemetry.io/otel"
+
+	"instant.dev/worker/internal/metrics"
+)
+
+// cohortSweepInterval is how often the job runs. Hourly is the right cadence
+// for a CI-account backstop: CI runs finish in minutes, so a leaked account
+// older than the (also short) TTL is reclaimed within the hour, while an
+// hourly cluster-wide query over the teams table is negligible load.
+const cohortSweepInterval = 1 * time.Hour
+
+// cohortSweepTTL is how old an is_test_cohort team must be before this job
+// will purge it. Short on purpose — CI runs are minutes, so 2h comfortably
+// covers the longest legitimate E2E run without ever reaping an account a
+// still-running CI job is actively using.
+const cohortSweepTTL = 2 * time.Hour
+
+// cohortSweepBatchLimit caps how many stale cohort teams the job purges per
+// run. A healthy steady-state has zero (CI reaps its own accounts); a backlog
+// (a CI outage that leaked many runs) is drained a batch per tick rather than
+// all at once — the per-team teardown does real deprovision/k8s work.
+const cohortSweepBatchLimit = 25
+
+// E2E cohort-sweep outcome labels for instant_e2e_cohort_swept_total. Bounded
+// set — every per-team decision picks exactly one. The literal strings are
+// part of the metric contract (see metrics.go E2ECohortSweptTotal).
+const (
+	cohortSweepOutcomeSwept            = "swept"
+	cohortSweepOutcomeFailed           = "failed"
+	cohortSweepOutcomeSkippedNotCohort = "skipped_not_cohort"
+)
+
+// E2ECohortSweepArgs is the periodic-job payload — empty, every run is a full
+// sweep.
+type E2ECohortSweepArgs struct{}
+
+// Kind implements river.JobArgs.
+func (E2ECohortSweepArgs) Kind() string { return "e2e_cohort_sweep" }
+
+// E2ECohortSweepWorker is the River worker.
+type E2ECohortSweepWorker struct {
+	river.WorkerDefaults[E2ECohortSweepArgs]
+	db       *sql.DB
+	executor teamTeardownExecutor // nil = job skipped (fail open)
+}
+
+// NewE2ECohortSweepWorker constructs the worker. executor may be nil — the
+// job then logs a single WARN per tick and reaps nothing, matching the fail-
+// open posture of every other worker. In production it is the same
+// *TeamDeletionExecutorWorker the orphan-sweep reconciler reuses.
+func NewE2ECohortSweepWorker(db *sql.DB, executor teamTeardownExecutor) *E2ECohortSweepWorker {
+	return &E2ECohortSweepWorker{db: db, executor: executor}
+}
+
+// Work runs one full sweep.
+func (w *E2ECohortSweepWorker) Work(ctx context.Context, job *river.Job[E2ECohortSweepArgs]) error {
+	ctx, span := otel.Tracer("instant.dev/worker").Start(ctx, "job.e2e_cohort_sweep")
+	defer span.End()
+
+	start := time.Now()
+
+	if w.executor == nil {
+		slog.Warn("jobs.e2e_cohort_sweep.skipped_no_executor",
+			"detail", "team-deletion executor unavailable — stale CI cohort teams will not be reaped this tick")
+		return nil
+	}
+
+	candidates, err := w.fetchStaleCohortTeams(ctx)
+	if err != nil {
+		return fmt.Errorf("E2ECohortSweepWorker: candidate query failed: %w", err)
+	}
+
+	var swept, failed, skipped int
+	for _, c := range candidates {
+		switch w.purgeTeam(ctx, c) {
+		case cohortSweepOutcomeSwept:
+			swept++
+		case cohortSweepOutcomeFailed:
+			failed++
+		case cohortSweepOutcomeSkippedNotCohort:
+			skipped++
+		}
+	}
+
+	// Idle-tick noise discipline (worker convention #1): an all-zero sweep is
+	// the steady state (CI reaps its own accounts) — DEBUG. Any non-zero
+	// counter is operational signal — INFO.
+	level := slog.LevelInfo
+	if swept == 0 && failed == 0 && skipped == 0 {
+		level = slog.LevelDebug
+	}
+	slog.Log(ctx, level, "jobs.e2e_cohort_sweep.completed",
+		"swept", swept,
+		"failed", failed,
+		"skipped_not_cohort", skipped,
+		"ttl", cohortSweepTTL.String(),
+		"duration_ms", time.Since(start).Milliseconds(),
+		"job_id", job.ID,
+	)
+	return nil
+}
+
+// fetchStaleCohortTeams returns every is_test_cohort team older than the TTL
+// that is not already in a terminal/destroying state. Capped by
+// cohortSweepBatchLimit.
+//
+// status filter: a team already 'tombstoned' is done; one in
+// 'deletion_pending' is mid-teardown (the orphan-sweep reconciler's PASS 1
+// owns it). We sweep everything else — 'active' (the normal leaked-CI-account
+// case) and 'deletion_requested' (a CI run that asked to delete but died
+// before the 30-day executor ran; for a CI account we skip the grace).
+func (w *E2ECohortSweepWorker) fetchStaleCohortTeams(ctx context.Context) ([]teamPendingDeletion, error) {
+	cutoff := time.Now().Add(-cohortSweepTTL)
+	rows, err := w.db.QueryContext(ctx, fmt.Sprintf(`
+		SELECT id, COALESCE(deletion_requested_at, created_at, now())
+		  FROM teams
+		 WHERE is_test_cohort = true
+		   AND created_at < $1
+		   AND status NOT IN ('tombstoned', 'deletion_pending')
+		 ORDER BY created_at ASC
+		 LIMIT %d
+	`, cohortSweepBatchLimit), cutoff)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = rows.Close() }()
+
+	var out []teamPendingDeletion
+	for rows.Next() {
+		var c teamPendingDeletion
+		if scanErr := rows.Scan(&c.teamID, &c.deletionRequestedAt); scanErr != nil {
+			return nil, fmt.Errorf("scan stale cohort team: %w", scanErr)
+		}
+		out = append(out, c)
+	}
+	return out, rows.Err()
+}
+
+// purgeTeam runs the per-team purge and returns the outcome label. It first
+// re-asserts the team is still is_test_cohort (defence in depth — the WHERE
+// clause already filtered, but a flip between the candidate SELECT and here
+// must NEVER let the destructive path touch a non-cohort team), then flips the
+// team to deletion_pending and runs the executor's idempotent teardown.
+func (w *E2ECohortSweepWorker) purgeTeam(ctx context.Context, c teamPendingDeletion) string {
+	// DEFENSIVE GUARD. isTestCohort fails SAFE-FOR-REAL-TEAMS: on any DB
+	// error it returns (false, err), and we then SKIP rather than purge — a
+	// destructive teardown must never run on an unconfirmed row. Only a row
+	// confirmed is_test_cohort=true proceeds.
+	flagged, err := isTestCohort(ctx, w.db, c.teamID)
+	if err != nil {
+		slog.Warn("jobs.e2e_cohort_sweep.cohort_recheck_failed",
+			"team_id", c.teamID.String(), "error", err,
+			"detail", "could not confirm is_test_cohort at purge time — skipping (never purge an unconfirmed team)")
+		metrics.E2ECohortSweptTotal.WithLabelValues(cohortSweepOutcomeSkippedNotCohort).Inc()
+		return cohortSweepOutcomeSkippedNotCohort
+	}
+	if !flagged {
+		// The candidate query selected a row the guard says is NOT cohort —
+		// a disagreement the alert keys on. Skip; never purge a non-cohort team.
+		slog.Error("jobs.e2e_cohort_sweep.non_cohort_candidate_blocked",
+			"team_id", c.teamID.String(),
+			"detail", "candidate query selected a team that is not is_test_cohort — purge blocked by defensive guard; candidate query may have regressed")
+		metrics.E2ECohortSweptTotal.WithLabelValues(cohortSweepOutcomeSkippedNotCohort).Inc()
+		return cohortSweepOutcomeSkippedNotCohort
+	}
+
+	// Flip to deletion_pending so processTeam's idempotent teardown tombstones
+	// it (the executor's step-0 flip only fires from 'deletion_requested'; for
+	// a CI account we skip the 30-day grace and set the in-flight state here).
+	// Guarded on the cohort flag a SECOND time inside the UPDATE so even a TOCTOU
+	// flip between the SELECT/recheck and this write cannot mark a non-cohort
+	// team for destruction.
+	res, err := w.db.ExecContext(ctx, `
+		UPDATE teams
+		   SET status = 'deletion_pending'
+		 WHERE id = $1
+		   AND is_test_cohort = true
+		   AND status NOT IN ('tombstoned', 'deletion_pending')
+	`, c.teamID)
+	if err != nil {
+		slog.Error("jobs.e2e_cohort_sweep.mark_deletion_pending_failed",
+			"team_id", c.teamID.String(), "error", err)
+		metrics.E2ECohortSweptTotal.WithLabelValues(cohortSweepOutcomeFailed).Inc()
+		return cohortSweepOutcomeFailed
+	}
+	if n, _ := res.RowsAffected(); n == 0 {
+		// The row changed state underneath us (already pending/tombstoned, or
+		// the cohort flag cleared). Re-check guard already passed, so this is a
+		// benign race — count as skipped, not failed.
+		slog.Warn("jobs.e2e_cohort_sweep.mark_deletion_pending_noop",
+			"team_id", c.teamID.String(),
+			"detail", "team changed state between candidate select and purge — skipping this tick")
+		metrics.E2ECohortSweptTotal.WithLabelValues(cohortSweepOutcomeSkippedNotCohort).Inc()
+		return cohortSweepOutcomeSkippedNotCohort
+	}
+
+	if perr := w.executor.processTeam(ctx, c); perr != nil {
+		slog.Error("jobs.e2e_cohort_sweep.purge_failed",
+			"team_id", c.teamID.String(), "error", perr,
+			"detail", "team stays in deletion_pending for next tick / orphan-sweep PASS 1")
+		metrics.E2ECohortSweptTotal.WithLabelValues(cohortSweepOutcomeFailed).Inc()
+		return cohortSweepOutcomeFailed
+	}
+
+	slog.Info("jobs.e2e_cohort_sweep.team_swept",
+		"team_id", c.teamID.String(),
+		"detail", "stale CI test-cohort team purged + tombstoned (belt-and-suspenders backstop)")
+	metrics.E2ECohortSweptTotal.WithLabelValues(cohortSweepOutcomeSwept).Inc()
+	return cohortSweepOutcomeSwept
+}
diff --git a/internal/jobs/e2e_cohort_sweep_integration_test.go b/internal/jobs/e2e_cohort_sweep_integration_test.go
new file mode 100644
index 0000000..1390dbd
--- /dev/null
+++ b/internal/jobs/e2e_cohort_sweep_integration_test.go
@@ -0,0 +1,149 @@
+package jobs
+
+// e2e_cohort_sweep_integration_test.go — REAL-Postgres proof that the
+// e2e_cohort_sweep reaper purges ONLY stale synthetic test-cohort teams and
+// leaves a fresh cohort team AND a normal (non-cohort) team completely
+// untouched.
+//
+// This is the safety-critical assertion: the job runs a DESTRUCTIVE per-team
+// teardown (deprovision + scrub PII + tombstone). A regression that widened
+// the candidate query — dropping the is_test_cohort filter, the TTL, or the
+// status guard — would tombstone real customer teams. So the test seeds three
+// teams and asserts the surgical outcome:
+//
+//   - STALE cohort team (is_test_cohort=true, created 3h ago)  → tombstoned,
+//     resource secrets scrubbed, user PII scrubbed, audit row emitted.
+//   - FRESH cohort team (is_test_cohort=true, created 5m ago)  → UNTOUCHED
+//     (younger than the 2h TTL — a CI run may still be using it).
+//   - NORMAL team (is_test_cohort=false, created 3h ago)       → UNTOUCHED
+//     (the is_test_cohort filter is the only thing protecting real customers).
+//
+// The executor's S3 / k8s / gRPC legs are nil-skipped (fail-open per
+// NewTeamDeletionExecutorWorker — CI has no bucket/cluster/provisioner). The
+// DB cascade (candidate scan, deletion_pending flip, scrub tx, audit emit) is
+// REAL and is the integration target.
+//
+// GATING: testhelpers.SetupTestDB skips under -short / no-DB, so this runs
+// only where a real platform Postgres (with api mig 067 teams.is_test_cohort
+// applied) is supplied via TEST_DATABASE_URL — same posture as the sibling
+// *_integration_test.go files.
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/riverqueue/river"
+	"github.com/riverqueue/river/rivertype"
+
+	"instant.dev/worker/internal/testhelpers"
+)
+
+func fakeCohortSweepJob() *river.Job[E2ECohortSweepArgs] {
+	return &river.Job[E2ECohortSweepArgs]{JobRow: &rivertype.JobRow{ID: 1}}
+}
+
+// TestIntegration_E2ECohortSweep_PurgesOnlyStaleCohort is the load-bearing
+// safety test: only the stale cohort team is swept; the fresh cohort team and
+// the normal team are left fully intact.
+func TestIntegration_E2ECohortSweep_PurgesOnlyStaleCohort(t *testing.T) {
+	db, cleanup := testhelpers.SetupTestDB(t)
+	defer cleanup()
+
+	// Neutralize cohort-flag debris from prior runs on the shared local DB —
+	// the sweep scans ALL cohort teams under a batch limit, so leftover debris
+	// would crowd out the seeded team. Safe: no real team is ever is_test_cohort.
+	testhelpers.ClearPreexistingTestCohortTeams(t, db)
+
+	// 1. STALE cohort team — 3h old (> 2h TTL), is_test_cohort=true.
+	stale := testhelpers.SeedCohortTeamWithAge(t, db, 3*time.Hour)
+	staleRes := testhelpers.SeedResourceWithSecret(t, db, stale, "postgres")
+	staleUser := testhelpers.SeedUser(t, db, stale, "cohort-stale-"+stale.String()[:8]+"@example.com")
+
+	// 2. FRESH cohort team — 5m old (< 2h TTL), is_test_cohort=true. A CI run
+	//    may still be using it; the sweep must NOT reap it.
+	fresh := testhelpers.SeedCohortTeamWithAge(t, db, 5*time.Minute)
+	freshRes := testhelpers.SeedResourceWithSecret(t, db, fresh, "postgres")
+
+	// 3. NORMAL team — 3h old but is_test_cohort=false. A real customer; the
+	//    cohort filter is the ONLY thing keeping the destroy path off it.
+	normal := testhelpers.SeedTeam(t, db, "pro")
+	normalRes := testhelpers.SeedResourceWithSecret(t, db, normal, "postgres")
+	normalUser := testhelpers.SeedUser(t, db, normal, "cohort-normal-"+normal.String()[:8]+"@example.com")
+
+	// provisioner / s3 / k8s all nil → fail-open; pure-DB cascade only.
+	executor := NewTeamDeletionExecutorWorker(db, nil, nil, nil, "")
+	w := NewE2ECohortSweepWorker(db, executor)
+
+	if err := w.Work(context.Background(), fakeCohortSweepJob()); err != nil {
+		t.Fatalf("Work: %v", err)
+	}
+
+	// ── STALE cohort team: fully purged ──────────────────────────────────────
+	if got := testhelpers.TeamStatus(t, db, stale); got != "tombstoned" {
+		t.Errorf("stale cohort team status = %q, want \"tombstoned\" (must be swept)", got)
+	}
+	connURL, keyPrefix := testhelpers.ResourceSecretFields(t, db, staleRes)
+	if connURL.Valid {
+		t.Errorf("stale cohort resource connection_url = %q, want NULL (scrubbed)", connURL.String)
+	}
+	if keyPrefix != "" {
+		t.Errorf("stale cohort resource key_prefix = %q, want empty (scrubbed)", keyPrefix)
+	}
+	email := testhelpers.UserEmail(t, db, staleUser)
+	wantPrefix := "deleted-" + staleUser.String()
+	if email != wantPrefix+"@tombstoned.invalid" {
+		t.Errorf("stale cohort user email = %q, want %q@tombstoned.invalid (scrubbed)", email, wantPrefix)
+	}
+	if n := testhelpers.CountAuditLogByTeam(t, db, stale, auditKindTombstoned); n < 1 {
+		t.Errorf("stale cohort %s audit rows = %d, want >= 1", auditKindTombstoned, n)
+	}
+
+	// ── FRESH cohort team: UNTOUCHED (within TTL) ────────────────────────────
+	if got := testhelpers.TeamStatus(t, db, fresh); got != "active" {
+		t.Errorf("fresh cohort team status = %q, want \"active\" (within TTL — must NOT be swept)", got)
+	}
+	freshConn, _ := testhelpers.ResourceSecretFields(t, db, freshRes)
+	if !freshConn.Valid || freshConn.String == "" {
+		t.Error("fresh cohort resource connection_url was scrubbed — the TTL guard failed (would reap an in-flight CI account)")
+	}
+	if n := testhelpers.CountAuditLogByTeam(t, db, fresh, auditKindTombstoned); n != 0 {
+		t.Errorf("fresh cohort %s audit rows = %d, want 0 (within TTL)", auditKindTombstoned, n)
+	}
+
+	// ── NORMAL team: UNTOUCHED (not cohort) ──────────────────────────────────
+	if got := testhelpers.TeamStatus(t, db, normal); got != "active" {
+		t.Errorf("normal team status = %q, want \"active\" (NOT a cohort team — must NEVER be swept)", got)
+	}
+	normalConn, _ := testhelpers.ResourceSecretFields(t, db, normalRes)
+	if !normalConn.Valid || normalConn.String == "" {
+		t.Error("normal team resource connection_url was scrubbed — the is_test_cohort filter failed (real-customer data-loss regression)")
+	}
+	normalEmail := testhelpers.UserEmail(t, db, normalUser)
+	if normalEmail != "cohort-normal-"+normal.String()[:8]+"@example.com" {
+		t.Errorf("normal team user email = %q, want the original (NOT scrubbed)", normalEmail)
+	}
+	if n := testhelpers.CountAuditLogByTeam(t, db, normal, auditKindTombstoned); n != 0 {
+		t.Errorf("normal team %s audit rows = %d, want 0 (must never be tombstoned)", auditKindTombstoned, n)
+	}
+}
+
+// TestIntegration_E2ECohortSweep_NilExecutorNoOps proves the fail-open arm: a
+// nil executor disables the job (no candidate is touched), matching every
+// other worker's posture when an external dependency is unavailable.
+func TestIntegration_E2ECohortSweep_NilExecutorNoOps(t *testing.T) {
+	db, cleanup := testhelpers.SetupTestDB(t)
+	defer cleanup()
+
+	testhelpers.ClearPreexistingTestCohortTeams(t, db)
+	stale := testhelpers.SeedCohortTeamWithAge(t, db, 3*time.Hour)
+
+	w := NewE2ECohortSweepWorker(db, nil) // nil executor → fail-open skip
+	if err := w.Work(context.Background(), fakeCohortSweepJob()); err != nil {
+		t.Fatalf("Work: %v", err)
+	}
+
+	if got := testhelpers.TeamStatus(t, db, stale); got != "active" {
+		t.Errorf("stale cohort team status = %q with nil executor, want \"active\" (job must no-op)", got)
+	}
+}
diff --git a/internal/jobs/e2e_cohort_sweep_test.go b/internal/jobs/e2e_cohort_sweep_test.go
new file mode 100644
index 0000000..00f5a5c
--- /dev/null
+++ b/internal/jobs/e2e_cohort_sweep_test.go
@@ -0,0 +1,283 @@
+package jobs
+
+// e2e_cohort_sweep_test.go — hermetic (sqlmock) tests for the
+// E2ECohortSweepWorker. Internal-package test so it can inject the unexported
+// teamTeardownExecutor seam (fakeTeardownExecutor, defined in
+// orphan_sweep_reconciler_test.go) and drive every per-team outcome arm
+// without a real Postgres.
+//
+// Scenarios covered:
+//   1. nil executor → fail-open no-op (no candidate query issued).
+//   2. candidate-query error → Work returns an error (River retries).
+//   3. happy path → a stale cohort team is flipped to deletion_pending and
+//      handed to the executor (outcome="swept").
+//   4. recheck-says-not-cohort → the destructive path is BLOCKED
+//      (outcome="skipped_not_cohort"); the executor is NEVER called.
+//   5. mark-deletion_pending 0-rows → benign race, skipped (no executor call).
+//   6. executor teardown error → outcome="failed"; team left for next tick.
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	sqlmock "github.com/DATA-DOG/go-sqlmock"
+	"github.com/google/uuid"
+)
+
+const cohortCandidateQueryRe = `SELECT id, COALESCE\(deletion_requested_at, created_at, now\(\)\)\s+FROM teams\s+WHERE is_test_cohort = true`
+
+// cohortRecheckRe matches isTestCohort's SELECT.
+const cohortRecheckRe = `SELECT is_test_cohort FROM teams WHERE id = \$1`
+
+// cohortFlipRe matches the deletion_pending UPDATE.
+const cohortFlipRe = `UPDATE teams\s+SET status = 'deletion_pending'\s+WHERE id = \$1\s+AND is_test_cohort = true`
+
+func TestE2ECohortSweep_NilExecutor_NoOp(t *testing.T) {
+	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
+	if err != nil {
+		t.Fatalf("sqlmock.New: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	// No query expectations registered — a nil executor must short-circuit
+	// before the candidate scan.
+	w := NewE2ECohortSweepWorker(db, nil)
+	if err := w.Work(context.Background(), fakeCohortSweepJob()); err != nil {
+		t.Fatalf("Work with nil executor: %v", err)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("unexpected DB calls with nil executor: %v", err)
+	}
+}
+
+func TestE2ECohortSweep_CandidateQueryError_ReturnsError(t *testing.T) {
+	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
+	if err != nil {
+		t.Fatalf("sqlmock.New: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	mock.ExpectQuery(cohortCandidateQueryRe).WillReturnError(errors.New("boom"))
+
+	w := NewE2ECohortSweepWorker(db, &fakeTeardownExecutor{})
+	if err := w.Work(context.Background(), fakeCohortSweepJob()); err == nil {
+		t.Fatal("Work: want error on candidate-query failure, got nil")
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("expectations: %v", err)
+	}
+}
+
+func TestE2ECohortSweep_NoCandidates_IdleTick(t *testing.T) {
+	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
+	if err != nil {
+		t.Fatalf("sqlmock.New: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	// Zero candidates → the all-zero (DEBUG) completion branch, no per-team work.
+	mock.ExpectQuery(cohortCandidateQueryRe).
+		WillReturnRows(sqlmock.NewRows([]string{"id", "ts"}))
+
+	exec := &fakeTeardownExecutor{}
+	w := NewE2ECohortSweepWorker(db, exec)
+	if err := w.Work(context.Background(), fakeCohortSweepJob()); err != nil {
+		t.Fatalf("Work: %v", err)
+	}
+	if len(exec.processed) != 0 {
+		t.Errorf("executor.processed = %v, want [] (no candidates)", exec.processed)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("expectations: %v", err)
+	}
+}
+
+func TestE2ECohortSweep_CandidateScanError_ReturnsError(t *testing.T) {
+	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
+	if err != nil {
+		t.Fatalf("sqlmock.New: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	// A row whose column shape (1 col) does not match the 2-col Scan → the
+	// row-scan error arm of fetchStaleCohortTeams.
+	mock.ExpectQuery(cohortCandidateQueryRe).
+		WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(uuid.New()))
+
+	w := NewE2ECohortSweepWorker(db, &fakeTeardownExecutor{})
+	if err := w.Work(context.Background(), fakeCohortSweepJob()); err == nil {
+		t.Fatal("Work: want error on row-scan failure, got nil")
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("expectations: %v", err)
+	}
+}
+
+func TestE2ECohortSweep_HappyPath_Sweeps(t *testing.T) {
+	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
+	if err != nil {
+		t.Fatalf("sqlmock.New: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	teamID := uuid.New()
+	mock.ExpectQuery(cohortCandidateQueryRe).
+		WillReturnRows(sqlmock.NewRows([]string{"id", "ts"}).AddRow(teamID, time.Now()))
+	// Recheck → still cohort.
+	mock.ExpectQuery(cohortRecheckRe).WithArgs(teamID).
+		WillReturnRows(sqlmock.NewRows([]string{"is_test_cohort"}).AddRow(true))
+	// Flip → 1 row affected.
+	mock.ExpectExec(cohortFlipRe).WithArgs(teamID).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+
+	exec := &fakeTeardownExecutor{}
+	w := NewE2ECohortSweepWorker(db, exec)
+	if err := w.Work(context.Background(), fakeCohortSweepJob()); err != nil {
+		t.Fatalf("Work: %v", err)
+	}
+	if len(exec.processed) != 1 || exec.processed[0] != teamID {
+		t.Errorf("executor.processed = %v, want [%s]", exec.processed, teamID)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("expectations: %v", err)
+	}
+}
+
+func TestE2ECohortSweep_RecheckNotCohort_BlocksPurge(t *testing.T) {
+	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
+	if err != nil {
+		t.Fatalf("sqlmock.New: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	teamID := uuid.New()
+	mock.ExpectQuery(cohortCandidateQueryRe).
+		WillReturnRows(sqlmock.NewRows([]string{"id", "ts"}).AddRow(teamID, time.Now()))
+	// Recheck → NOT cohort (the flag flipped under us). The destructive path
+	// must be blocked: NO flip UPDATE, NO executor call.
+	mock.ExpectQuery(cohortRecheckRe).WithArgs(teamID).
+		WillReturnRows(sqlmock.NewRows([]string{"is_test_cohort"}).AddRow(false))
+
+	exec := &fakeTeardownExecutor{}
+	w := NewE2ECohortSweepWorker(db, exec)
+	if err := w.Work(context.Background(), fakeCohortSweepJob()); err != nil {
+		t.Fatalf("Work: %v", err)
+	}
+	if len(exec.processed) != 0 {
+		t.Errorf("executor.processed = %v, want [] (non-cohort team must NOT be purged)", exec.processed)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("expectations: %v", err)
+	}
+}
+
+func TestE2ECohortSweep_RecheckError_Skips(t *testing.T) {
+	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
+	if err != nil {
+		t.Fatalf("sqlmock.New: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	teamID := uuid.New()
+	mock.ExpectQuery(cohortCandidateQueryRe).
+		WillReturnRows(sqlmock.NewRows([]string{"id", "ts"}).AddRow(teamID, time.Now()))
+	mock.ExpectQuery(cohortRecheckRe).WithArgs(teamID).WillReturnError(errors.New("db blip"))
+
+	exec := &fakeTeardownExecutor{}
+	w := NewE2ECohortSweepWorker(db, exec)
+	if err := w.Work(context.Background(), fakeCohortSweepJob()); err != nil {
+		t.Fatalf("Work: %v", err)
+	}
+	if len(exec.processed) != 0 {
+		t.Errorf("executor.processed = %v, want [] (unconfirmed team must NOT be purged)", exec.processed)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("expectations: %v", err)
+	}
+}
+
+func TestE2ECohortSweep_FlipZeroRows_Skips(t *testing.T) {
+	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
+	if err != nil {
+		t.Fatalf("sqlmock.New: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	teamID := uuid.New()
+	mock.ExpectQuery(cohortCandidateQueryRe).
+		WillReturnRows(sqlmock.NewRows([]string{"id", "ts"}).AddRow(teamID, time.Now()))
+	mock.ExpectQuery(cohortRecheckRe).WithArgs(teamID).
+		WillReturnRows(sqlmock.NewRows([]string{"is_test_cohort"}).AddRow(true))
+	// Flip → 0 rows (state changed under us). Benign race → skip, no executor.
+	mock.ExpectExec(cohortFlipRe).WithArgs(teamID).
+		WillReturnResult(sqlmock.NewResult(0, 0))
+
+	exec := &fakeTeardownExecutor{}
+	w := NewE2ECohortSweepWorker(db, exec)
+	if err := w.Work(context.Background(), fakeCohortSweepJob()); err != nil {
+		t.Fatalf("Work: %v", err)
+	}
+	if len(exec.processed) != 0 {
+		t.Errorf("executor.processed = %v, want [] (0-row flip is a benign race)", exec.processed)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("expectations: %v", err)
+	}
+}
+
+func TestE2ECohortSweep_FlipError_CountsFailed(t *testing.T) {
+	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
+	if err != nil {
+		t.Fatalf("sqlmock.New: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	teamID := uuid.New()
+	mock.ExpectQuery(cohortCandidateQueryRe).
+		WillReturnRows(sqlmock.NewRows([]string{"id", "ts"}).AddRow(teamID, time.Now()))
+	mock.ExpectQuery(cohortRecheckRe).WithArgs(teamID).
+		WillReturnRows(sqlmock.NewRows([]string{"is_test_cohort"}).AddRow(true))
+	mock.ExpectExec(cohortFlipRe).WithArgs(teamID).WillReturnError(errors.New("flip failed"))
+
+	exec := &fakeTeardownExecutor{}
+	w := NewE2ECohortSweepWorker(db, exec)
+	if err := w.Work(context.Background(), fakeCohortSweepJob()); err != nil {
+		t.Fatalf("Work: %v (per-team failure must NOT fail the sweep)", err)
+	}
+	if len(exec.processed) != 0 {
+		t.Errorf("executor.processed = %v, want [] (flip failed before purge)", exec.processed)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("expectations: %v", err)
+	}
+}
+
+func TestE2ECohortSweep_ExecutorError_CountsFailed(t *testing.T) {
+	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
+	if err != nil {
+		t.Fatalf("sqlmock.New: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	teamID := uuid.New()
+	mock.ExpectQuery(cohortCandidateQueryRe).
+		WillReturnRows(sqlmock.NewRows([]string{"id", "ts"}).AddRow(teamID, time.Now()))
+	mock.ExpectQuery(cohortRecheckRe).WithArgs(teamID).
+		WillReturnRows(sqlmock.NewRows([]string{"is_test_cohort"}).AddRow(true))
+	mock.ExpectExec(cohortFlipRe).WithArgs(teamID).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+
+	exec := &fakeTeardownExecutor{failFor: map[uuid.UUID]error{teamID: errors.New("teardown boom")}}
+	w := NewE2ECohortSweepWorker(db, exec)
+	if err := w.Work(context.Background(), fakeCohortSweepJob()); err != nil {
+		t.Fatalf("Work: %v (a per-team teardown error must be isolated)", err)
+	}
+	if len(exec.processed) != 1 {
+		t.Errorf("executor.processed = %v, want the team attempted once", exec.processed)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("expectations: %v", err)
+	}
+}
diff --git a/internal/jobs/workers.go b/internal/jobs/workers.go
index d40835a..5e82dde 100644
--- a/internal/jobs/workers.go
+++ b/internal/jobs/workers.go
@@ -690,6 +690,18 @@ func StartWorkers(ctx context.Context, db *sql.DB, rdb *redis.Client, cfg *confi
 		orphanReconciler = orphanReconciler.WithPodStateProvider(podStateClient)
 	}
 	river.AddWorker(workers, WithObservability(orphanReconciler, nrApp))
+	// E2E cohort sweep — hourly belt-and-suspenders reaper for STALE
+	// synthetic test-cohort teams (teams.is_test_cohort=true, api migration
+	// 067). CI mints an ephemeral account per E2E run via the api's
+	// /internal/e2e/account endpoint and is supposed to delete it at the end
+	// of the run; a CI run that dies mid-flight leaks the account. This job
+	// purges any is_test_cohort team older than cohortSweepTTL (2h) by reusing
+	// the SAME team-deletion executor's idempotent per-team teardown the
+	// orphan-sweep reconciler reuses. INERT in practice — no real team is
+	// is_test_cohort=true (migration 067 defaults false). The executor is
+	// always non-nil here (it was constructed above), so the job never
+	// fail-open-skips in production. See e2e_cohort_sweep.go.
+	river.AddWorker(workers, WithObservability(NewE2ECohortSweepWorker(db, teamDeletionExecutor), nrApp))
 	// Provisioner-reconciler (W5-A). Every 2min, recovers or abandons
 	// stuck pending resources.
 	//
@@ -1306,6 +1318,19 @@ func buildPeriodicJobs(cfg *config.Config) []*river.PeriodicJob {
 			},
 			&river.PeriodicJobOpts{RunOnStart: true},
 		),
+		// E2E cohort sweep — hourly. Belt-and-suspenders reaper for stale CI
+		// test-cohort teams (teams.is_test_cohort=true) leaked by a CI run
+		// that died before deleting its own /internal/e2e/account. Routed to
+		// the reconcile queue so a default-queue fan-out cannot starve it.
+		// RunOnStart=true so a worker restart immediately reclaims accounts
+		// that leaked while it was down. Inert until cohort accounts exist.
+		river.NewPeriodicJob(
+			river.PeriodicInterval(cohortSweepInterval),
+			func() (river.JobArgs, *river.InsertOpts) {
+				return E2ECohortSweepArgs{}, reconcileInsertOpts(cohortSweepInterval)
+			},
+			&river.PeriodicJobOpts{RunOnStart: true},
+		),
 		// Provisioner-reconciler (W5-A) — every 2min, reconcile queue.
 		// RunOnStart=true.
 		river.NewPeriodicJob(
diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go
index 7ff1111..6f874ce 100644
--- a/internal/metrics/metrics.go
+++ b/internal/metrics/metrics.go
@@ -701,6 +701,38 @@ var (
 		Help: "Orphan-sweep reconciler reap attempts that failed (k8s API error or DB write failure), labelled by reason.",
 	}, []string{"reason"})
 
+	// E2ECohortSweptTotal counts the e2e_cohort_sweep job's per-team outcomes.
+	// The job is the belt-and-suspenders backstop that purges STALE synthetic
+	// test-cohort teams (teams.is_test_cohort=true, api migration 067) — the
+	// ephemeral accounts a CI run creates via the api's /internal/e2e/account
+	// endpoint and is supposed to delete on its own. If a CI run dies before
+	// reaping its account, this job tombstones the leftover team after a short
+	// TTL so a half-cleaned cohort team never lingers (and never re-enters a
+	// real funnel scan).
+	//
+	// outcome label (bounded set):
+	//   "swept"              — team purged + tombstoned via the team-deletion
+	//                          executor's idempotent per-team teardown.
+	//   "failed"            — purge attempt errored; the team stays in
+	//                          deletion_pending for the next tick / operator.
+	//   "skipped_not_cohort" — DEFENSIVE guard tripped: a candidate that the
+	//                          WHERE clause selected was re-checked at purge
+	//                          time and was NOT is_test_cohort (the row flipped
+	//                          underneath us). A non-zero rate here means the
+	//                          candidate query and the guard disagree — alert.
+	//
+	// NR alert (mandatory, rule 25):
+	//   sum(rate(instant_e2e_cohort_swept_total{outcome="skipped_not_cohort"}[1h])) > 0
+	//     → P1 page. The sweep nearly purged a non-cohort team — the candidate
+	//     query may have regressed; investigate before the next tick.
+	//   sum(rate(instant_e2e_cohort_swept_total{outcome="failed"}[6h])) > 0
+	//     → P2 page. A cohort team could not be torn down; CI accounts are
+	//     leaking compute/DB rows.
+	E2ECohortSweptTotal = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "instant_e2e_cohort_swept_total",
+		Help: "e2e_cohort_sweep per-team outcomes (swept/failed/skipped_not_cohort). Backstop reaper for stale CI test-cohort teams.",
+	}, []string{"outcome"})
+
 	// PGPool* gauges expose worker's *sql.DB pool state. Sampled every
 	// 5s by the exporter started from main.go. See api/internal/metrics
 	// for the matching counterparts in the api process.
diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go
index c3afae3..60fcfdb 100644
--- a/internal/metrics/metrics_test.go
+++ b/internal/metrics/metrics_test.go
@@ -94,6 +94,12 @@ func TestAllMetrics_AreRegistered(t *testing.T) {
 	PropagationUnknownKindTotal.WithLabelValues("unknown").Add(0)
 	OrphanSweepReapedTotal.WithLabelValues("team_tombstoned").Add(0)
 	OrphanSweepReapFailedTotal.WithLabelValues("team_tombstoned").Add(0)
+	// Prime all three e2e_cohort_sweep outcome label values so /metrics
+	// exposes them from process start (lazy emit otherwise leaves the panel
+	// empty until the first real cohort sweep fires).
+	E2ECohortSweptTotal.WithLabelValues("swept").Add(0)
+	E2ECohortSweptTotal.WithLabelValues("failed").Add(0)
+	E2ECohortSweptTotal.WithLabelValues("skipped_not_cohort").Add(0)
 	DeployJobFailedDetectedTotal.WithLabelValues("BackoffLimitExceeded").Add(0)
 	// Prime all four DeployAutopsyCapturedTotal outcome label values so
 	// /metrics exposes them from process start (lazy emit otherwise leaves
diff --git a/internal/testhelpers/cohort.go b/internal/testhelpers/cohort.go
index f19ac5f..a610344 100644
--- a/internal/testhelpers/cohort.go
+++ b/internal/testhelpers/cohort.go
@@ -11,6 +11,7 @@ package testhelpers
 
 import (
 	"database/sql"
+	"fmt"
 	"testing"
 	"time"
 
@@ -29,6 +30,45 @@ func SetTeamTestCohort(t *testing.T, db *sql.DB, teamID uuid.UUID, flagged bool)
 	}
 }
 
+// ClearPreexistingTestCohortTeams flips is_test_cohort=false on every team that
+// currently carries it — clearing the accumulated debris that prior cohort
+// skip-guard / sweep test runs leave behind on a shared local DB. The
+// e2e_cohort_sweep reaper scans ALL cohort teams globally under a batch limit,
+// so a test that seeds its own cohort teams must first neutralize pre-existing
+// ones or the batch fills with debris and the seeded team is never reached.
+//
+// This is SAFE: no real team is ever is_test_cohort=true (api migration 067
+// defaults every row to false; only synthetic/CI accounts flip it), so the flag
+// on a shared test DB is purely test state. The reset is scoped to the flag
+// only — it does not delete or otherwise mutate the rows.
+func ClearPreexistingTestCohortTeams(t *testing.T, db *sql.DB) {
+	t.Helper()
+	if _, err := db.Exec(
+		`UPDATE teams SET is_test_cohort = false WHERE is_test_cohort = true`,
+	); err != nil {
+		tFatalf(t, "ClearPreexistingTestCohortTeams: %v", err)
+	}
+}
+
+// SeedCohortTeamWithAge inserts an 'active' team flagged is_test_cohort=true
+// whose created_at is backdated by `age` — the exact shape the e2e_cohort_sweep
+// reaper keys on. A positive age yields a STALE cohort team (a leaked CI
+// account past the sweep TTL); a small/zero age yields a FRESH one (a CI run
+// still in flight, which the sweep must NOT reap). Returns the team id.
+func SeedCohortTeamWithAge(t *testing.T, db *sql.DB, age time.Duration) uuid.UUID {
+	t.Helper()
+	id := uuid.New()
+	if _, err := db.Exec(`
+		INSERT INTO teams (id, name, plan_tier, status, is_test_cohort, created_at)
+		VALUES ($1, $2, 'free', 'active', true, now() - $3::interval)
+	`, id, "itest-cohort-"+id.String()[:8], fmt.Sprintf("%d seconds", int64(age.Seconds()))); err != nil {
+		tFatalf(t, "SeedCohortTeamWithAge: %v", err)
+		return uuid.Nil
+	}
+	t.Cleanup(func() { _, _ = db.Exec(`DELETE FROM teams WHERE id = $1`, id) })
+	return id
+}
+
 // SeedPrimaryUser inserts a primary (is_primary=true) user for a team — the
 // recipient the expiry-warning and weekly-digest scans join on. Distinct from
 // SeedUser, which leaves is_primary at its default (false).
diff --git a/internal/testhelpers/cohort_smoke_test.go b/internal/testhelpers/cohort_smoke_test.go
index 54ea1ed..c50024d 100644
--- a/internal/testhelpers/cohort_smoke_test.go
+++ b/internal/testhelpers/cohort_smoke_test.go
@@ -55,6 +55,14 @@ func TestCohortErrorArms(t *testing.T) {
 	// SetTeamTestCohort returns nothing; the closed-DB Exec error fires tFatalf.
 	SetTeamTestCohort(t, closed, id, true)
 
+	if got := SeedCohortTeamWithAge(t, closed, 3*time.Hour); got != uuid.Nil {
+		t.Fatalf("SeedCohortTeamWithAge on closed db = %v, want Nil", got)
+	}
+
+	// ClearPreexistingTestCohortTeams returns nothing; the closed-DB Exec
+	// error fires tFatalf.
+	ClearPreexistingTestCohortTeams(t, closed)
+
 	if got := SeedPrimaryUser(t, closed, id, "smoke"); got != uuid.Nil {
 		t.Fatalf("SeedPrimaryUser on closed db = %v, want Nil", got)
 	}
@@ -77,9 +85,9 @@ func TestCohortErrorArms(t *testing.T) {
 		t.Fatal("CheckoutNotified on closed db = true, want false")
 	}
 
-	// 8 distinct fallible call paths each recorded at least one Fatalf.
-	if len(*fatals) < 8 {
-		t.Fatalf("expected >=8 recorded Fatalf arms against the closed DB, got %d: %v", len(*fatals), *fatals)
+	// 10 distinct fallible call paths each recorded at least one Fatalf.
+	if len(*fatals) < 10 {
+		t.Fatalf("expected >=10 recorded Fatalf arms against the closed DB, got %d: %v", len(*fatals), *fatals)
 	}
 }
 
@@ -107,6 +115,24 @@ func TestIntegration_CohortRoundTrip(t *testing.T) {
 		t.Fatal("SetTeamTestCohort(false) did not clear is_test_cohort")
 	}
 
+	// SeedCohortTeamWithAge: a stale (3h-old) is_test_cohort team — the
+	// e2e_cohort_sweep candidate shape. Read back the flag + age to exercise
+	// the happy path.
+	staleCohort := SeedCohortTeamWithAge(t, db, 3*time.Hour)
+	if staleCohort == uuid.Nil {
+		t.Fatal("SeedCohortTeamWithAge returned nil uuid")
+	}
+	if got := isTestCohort(t, db, staleCohort); !got {
+		t.Fatal("SeedCohortTeamWithAge did not set is_test_cohort=true")
+	}
+
+	// ClearPreexistingTestCohortTeams flips the flag off across the DB — the
+	// staleCohort team above must read back is_test_cohort=false afterwards.
+	ClearPreexistingTestCohortTeams(t, db)
+	if got := isTestCohort(t, db, staleCohort); got {
+		t.Fatal("ClearPreexistingTestCohortTeams did not clear is_test_cohort")
+	}
+
 	// SeedPrimaryUser: inserts an is_primary=true user with a globally-unique addr.
 	primary := SeedPrimaryUser(t, db, team, "cohort-primary")
 	if primary == uuid.Nil {
diff --git a/internal/testhelpers/testhelpers.go b/internal/testhelpers/testhelpers.go
index 0541306..bcda99d 100644
--- a/internal/testhelpers/testhelpers.go
+++ b/internal/testhelpers/testhelpers.go
@@ -163,6 +163,12 @@ func ensureSchema(t *testing.T, db *sql.DB) {
 		// works against a bare DB AND a fully api-migrated one.
 		`ALTER TABLE teams ADD COLUMN IF NOT EXISTS is_test_cohort BOOLEAN NOT NULL DEFAULT false`,
 		`ALTER TABLE teams ADD COLUMN IF NOT EXISTS stripe_customer_id TEXT`,
+		// deletion-lifecycle columns the team-deletion executor + the
+		// e2e_cohort_sweep reaper read/write (status flip, tombstone stamp,
+		// grace-window scan). Idempotent adds so the harness works on a bare DB
+		// AND a fully api-migrated one (where these come from the deletion migs).
+		`ALTER TABLE teams ADD COLUMN IF NOT EXISTS deletion_requested_at TIMESTAMPTZ`,
+		`ALTER TABLE teams ADD COLUMN IF NOT EXISTS tombstoned_at TIMESTAMPTZ`,
 
 		// resources — the entitlement reconciler reads tier / applied_conn_limit.
 		`CREATE TABLE IF NOT EXISTS resources (