background_tasks: tenant-scope TaskExecution, and register the tenant listeners in the Celery worker process

[antosubash]

Why

background_tasks already carries the enqueuing tenant through Celery headers (background_tasks/tenant_context.py), but TaskExecution rows (models.py:24) are not tenant-scoped, and the Celery worker process never runs register_listeners — so even after adding MultiTenantMixin, every signal-driven write in the worker stays unscoped unless that gap is closed too.

Tables

• TaskExecution (models.py:24) — adopt MultiTenantMixin: yes. Task rows are per-customer activity (args/kwargs/results can contain tenant data) and the admin screen (/admin background-tasks view) must not show one tenant's jobs to another's operators — except where jobs are genuinely platform-wide (beat sweeps, see below).
  • No unique constraints exist today (__table_args__ at models.py:76-82 is a plain non-unique (status, queued_at) index) — SM024 not triggered, but add tenant_id to that composite index since the admin list/sweep queries filter by status per tenant.
  • celery_task_id has a plain (non-unique) index (models.py:38) — fine as-is since Celery ids are already globally unique; no change needed for tenancy correctness, but tenant-scoped lookups by this column still need tenant_id in the query.

Code paths

• Worker process has no tenant listeners: register_listeners attaches its hooks to the plain SQLAlchemy Session class (db_state.sync_session_class, framework/db/simple_module_db/session.py), so in the web process sync_db's sessions are filtered like any other. The Celery worker (scripts/run_worker.py) never builds the app, so register_listeners never runs there and every query in a task body or signal handler (signals.py _apply/upsert_by_celery_id) is unscoped and unfiltered — even with the tenant restored into current_tenant_id. Fix: call register_listeners (with tenant_strict set from the same setting) at worker boot, and add a regression test that a tenant-scoped query in a worker-context session is filtered.
• tenant_context.py already restores current_tenant_id around the task body (restore_tenant/release_tenant, tenant_context.py:49-70) via signals.py:184,205 — that context is available for (a) above at prerun time, but on_task_publish (signals.py:121-152) fires from the enqueuing process/thread where current_tenant_id should already be set by the request; stamp_tenant (tenant_context.py:26-36) captures it onto headers but the row written at publish time (signals.py:141-152) does not stamp tenant_id onto the TaskExecution row itself.
• Admin reads: service.py/api_admin.py (list_executions, get_execution, retry_failed_executions, retry_execution — endpoints/api_admin.py:34-97) must become tenant-scoped for tenant operators; a platform-wide view (all tenants' jobs) may still be wanted for platform admins and would need explicit all_tenants().
• Beat/maintenance job: sweep_stuck_tasks (tasks.py:58-82) scans TaskExecution across all running rows with stale heartbeats — this is inherently cross-tenant maintenance and must wrap its query in all_tenants(); it runs in the worker, so it depends on the worker-listener fix above.
• worker_inspector.py/workers_state.py (poll_workers, used by get_workers at api_admin.py:100-108) reports live Celery broker state, not DB rows — inherently platform-wide, no tenant scoping needed, but confirm no per-task args are echoed to a tenant-scoped viewer.
• Retry: retry_service.py re-enqueues tasks; must preserve/restore the original row's tenant on re-publish, not the retrying operator's tenant, so a platform admin retrying a stuck job doesn't run it as themselves.

Migration

• Add tenant_id (nullable initially, or NOT NULL with backfill) to background_tasks_task_execution. Existing rows have no captured tenant; backfill to a "platform" sentinel or NULL, consistent with whatever audit_log's migration decides (same fail-open-for-history problem).
• Add tenant_id to the (status, queued_at) index.
• The worker-boot listener registration is a prerequisite and belongs in the framework/background_tasks boot path, not per module.

Tests

• Read isolation: tenant A cannot see tenant B's TaskExecution rows via list_executions/get_execution.
• Write isolation / tagging: a task enqueued under tenant_context(A) produces a TaskExecution row with tenant_id=A end-to-end through publish → prerun → success/failure (exercising the actual sync-session write path, not just the ORM mixin).
• Beat job: sweep_stuck_tasks flips stuck rows across all tenants (verify it doesn't silently skip other tenants once scoping exists) and does not raise TenantIsolationError when run with no request context.
• Retry preserves original tenant: retrying a failed task queued by tenant A re-executes as tenant A even when triggered by a platform admin with no active tenant.
• Regression test: in a worker-context session (no create_app), a query on a MultiTenantMixin model inside tenant_context(x) is filtered, and outside any tenant raises under strict mode.

Out of scope / open questions

• Whether platform admins get an all_tenants() cross-tenant view of /admin background jobs, or only ever see their own tenant's — needs a product decision mirroring the audit_log admin-screen question.
• Any other process that touches the DB without create_app (CLI commands, scripts) has the same gap: no listeners unless it calls register_listeners itself. Worth a framework-wide audit.

---

Context: part of the SaaS tenancy work in #370 (fail-closed isolation, tenant_context()/all_tenants(), the tenants module). Design: docs/plans/2026-09-27-saas-tenancy-design.md and docs/framework/multi-tenancy.md on that branch.