feat(background_tasks): tracking modes and resource-visible task status (#710, #709)

CHANGELOG.md

@@ -18,6 +18,11 @@ All notable changes to this project are documented in this file. The format is b
 - Users auth emails support ordered host template directories through the
   restart-required DB setting `mail_template_dirs`, with per-file bundled
   fallback and optional autoescaped HTML alternatives (#712).
+- Background-task resource references and resource-policy status visibility (#709);
+  migration `b31a70971001` adds nullable resource columns and a paired-reference constraint.
+- **Behaviour change:** configurable task execution tracking (`full`, `terminal`,
+  `failures`, `off`) and DB-backed `track_publish` for worker-first tracking (#710).
+  Defaults preserve existing tracking; reduced modes retain tenant/owner context.
 - Public durable delivery context (`in_delivery()` and `IN_DELIVERY`), inherited
   by handler child tasks and reset after delivery (#682).
 - Transactional, tenant-scoped `sharing.contracts.forget_resource` for resource

docs/database/migrations.md

@@ -21,7 +21,7 @@ The history is **not** a single linear chain. It has several roots and several h
 
 | Head | Owning module | Label Alembic reports |
 |---|---|---|
-| `a626b1c8d023` | users (hashed bearer tokens) | `api_tokens`, `background_tasks`, `users` |
+| `b31a70971001` | background_tasks (resource references) | `api_tokens`, `background_tasks`, `users` |
 | `a7c2e91d4b58` | audit_log | `audit_log` |
 | `f1a9c3d7b2e4` | tenants | `tenants` |
 | `d263ea727663` | framework (`sm_lease`) | `leases` |

docs/modules/background_tasks.md

@@ -120,7 +120,53 @@ def rebuild(report_id: int) -> None:
 
 The enqueuing user is recorded as `owner_user_id` from the `sm_owner_id` message header (stamped from `current_user_id` at publish, next to the tenant header). A task that enqueues another task passes the owner on, and a retry keeps the original owner rather than adopting the operator. Beat and other platform publishes have no owner.
 
-`GET /api/background_tasks/tasks/{celery_task_id}/status` returns a `TaskStatusDTO`: status, the progress fields, timestamps and the exception's type name. It never returns args, kwargs, result, traceback or an error message. It is visible to the owner while acting in the row's tenant (any tenant when multi-tenancy is off) and to holders of `background_tasks.view`; everyone else, and every unknown id, gets `404` so the route does not reveal which ids exist. The task detail page shows a progress bar once a report exists.
+`GET /api/background_tasks/tasks/{celery_task_id}/status` returns a `TaskStatusDTO`: status, the progress fields, timestamps and the exception's type name. It never returns args, kwargs, result, traceback or an error message. Holders of `background_tasks.view` may see any task. Everyone else must act in the task's active tenant when multi-tenancy is enabled, and must either own the task or have `view` access to its attached resource. Unknown tasks, unknown resource types, malformed IDs, missing/deleted resources and denied access all return the same `404`. Resource authorization is checked afresh through the registered policy's filtered loader, so revoking a share removes visibility immediately. The task detail page shows a progress bar once a report exists.
+
+Attach a resource explicitly to a signature before enqueueing it (#709):
+
+```python
+from background_tasks import enqueue_after_commit, with_task_resource
+
+job = with_task_resource(
+    import_images.s(collection_id),
+    resource_type="images.collection",
+    resource_id=str(collection_id),
+)
+task_id = enqueue_after_commit(db, job)
+```
+
+The helper returns a cloned signature and merges validated headers. The type must be a dotted resource type name, and the ID a nonempty string of at most 255 characters without control characters. Register the type in `app.state.sm.resource_types` through the module's `register_resource_types` hook. The reference survives worker-only tracking and administrative retries; unrelated child tasks never inherit it automatically. Resource viewers receive only the status DTO, with no retry or admin authority. Progress messages are visible to these viewers: avoid secrets in them. Migration `b31a70971001` adds nullable `resource_type` / `resource_id` columns with a both-or-neither constraint; existing tasks keep their owner/operator visibility.
+
+## Execution tracking
+
+The DB-backed, restart-required settings `execution_tracking` (default `full`) and `track_publish` (default `True`) control execution history (#710). They deliberately have no environment source. Web and worker processes freeze them into Celery configuration at startup; the publish signal never queries settings.
+
+| Mode | Records |
+|---|---|
+| `full` | Publish, prerun, terminal transitions, retries, postrun and heartbeats (existing behavior) |
+| `terminal` | Success, final failure and revocation only |
+| `failures` | Final failure and revocation only; no intermediate retries |
+| `off` | No execution tracking |
+
+Resolution is per-message header → `@shared_task(sm_tracking=...)` option → configured default. Public `TrackingMode` values and `SM_TRACKING_HEADER` avoid hardcoded header names:
+
+```python
+from celery import shared_task
+from background_tasks import SM_TRACKING_HEADER, TrackingMode
+
+
+@shared_task(sm_tracking=TrackingMode.FAILURES)
+def resize_image(image_id): ...
+
+
+# Override this one message, retaining the task's usual default otherwise.
+resize_image.s(image_id).set(headers={SM_TRACKING_HEADER: TrackingMode.OFF}).apply_async()
+```
+
+All reduced modes skip publish/prerun/postrun tracking and `heartbeat()` without opening a tracking session. `track_publish=False` also skips the publisher-side DB write for `full`: the worker creates the row from message metadata, retaining owner, tenant, queue, arguments and resource. Tenant/owner message stamping and worker context restoration/cleanup still run in every mode. Invalid tracking overrides fall back to `full`, so malformed metadata cannot hide a failure. The internal outbox sweep defaults to `failures` through this same policy.
+
+Reduced modes lose live progress, stuck detection and some history/retry information. Worker-first tracking returns `404` until the worker creates a row; reduced modes may have no row until a recorded terminal event, or ever for successful failures-only/off tasks. Broker publishing itself remains synchronous. Administrative retries still reserve their own attempt rows transactionally before publishing, and are always published with `full` tracking so the reserved `pending` row can advance.
+
 
 ## Retained task and app logs
 
@@ -169,7 +215,7 @@ Entry values are plain Celery [beat entries](https://docs.celeryq.dev/en/stable/
 
 Beat persists its schedule (`celery.beat:PersistentScheduler`) so it survives restarts.
 
-The worker and beat read the same DB-backed settings the web process does: `scripts/run_worker.py` (and the copy `smpy new` scaffolds) calls `background_tasks.runner.build_runner_celery()`, which hydrates `BackgroundTasksSettings` from the database at boot exactly as the web lifespan does. Precedence is env → DB → default: an `SM_BG_TASKS_*` variable that is set wins over a stored value for the same field, and the database supplies everything the environment leaves unset. So an admin-edited queue or time limit reaches the worker after a restart, unless the deployment pins that field in its environment. Because the web process behaves identically, this adds no new trust boundary; still, treat write access to these settings as control of the workers. If the database cannot be read the worker starts from env and defaults and logs a warning.
+The worker and beat read the same DB-backed settings the web process does: `scripts/run_worker.py` (and the copy `smpy new` scaffolds) calls `background_tasks.runner.build_runner_celery()`, which hydrates `BackgroundTasksSettings` from the database at boot exactly as the web lifespan does. Except for the DB-only `execution_tracking` and `track_publish` fields, precedence is env → DB → default: an `SM_BG_TASKS_*` variable that is set wins over a stored value for the same field, and the database supplies everything the environment leaves unset. So an admin-edited queue or time limit reaches the worker after a restart, unless the deployment pins that field in its environment. Because the web process behaves identically, this adds no new trust boundary; still, treat write access to these settings as control of the workers. If the database cannot be read the worker starts from env and defaults and logs a warning.
 
 ## Routing and queues
 
@@ -411,7 +457,7 @@ All require `background_tasks.view`; retry additionally needs `background_tasks.
 | `GET /api/background_tasks/admin/executions/{execution_id}/logs` | `?level=&page=&per_page=` → `LogListResponse` | the task's retained lines, newest first; `level` is a minimum |
 | `GET /api/background_tasks/admin/logs` | `?q=&level=&task_id=&logger=&page=&per_page=` → `LogListResponse` | search across retained lines |
 | `POST /api/background_tasks/admin/executions/{execution_id}/retry` | → `TaskExecutionDetail` | 409 unless status is `failed` or `stuck` |
-| `GET /api/background_tasks/tasks/{celery_task_id}/status` | → `TaskStatusDTO` | no permission gate: owner in the active tenant or `view`; otherwise 404 |
+| `GET /api/background_tasks/tasks/{celery_task_id}/status` | → `TaskStatusDTO` | no permission gate: owner/resource viewer in the active tenant or `background_tasks.view`; otherwise 404 |
 | `GET /api/background_tasks/admin/workers` | → `WorkerSnapshot` | live ping; consumed by the `Workers` page |
 | `GET /api/background_tasks/admin/outbox` | `?status=&page=&per_page=` → `OutboxListResponse` | undelivered rows (pending, running, dead) and per-status counts |
 | `POST /api/background_tasks/admin/outbox/{id}/retry` | → `OutboxItem` | `manage`; 409 unless `dead`; resets attempts and dispatches after commit |
@@ -442,7 +488,7 @@ from background_tasks.contracts.schemas import (
 
 | Class | Purpose |
 |---|---|
-| `TaskFailed` (event) | Published when a task transitions to `failed`. Fields: `task_execution_id`, `task_name`, `exception_type`. |
+| `TaskFailed` (event) | Published when a task transitions to `failed` and the failure is recorded, so not for a task in `off` tracking mode (there is no row to point at). Fields: `task_execution_id`, `task_name`, `exception_type`. |
 | `TaskRetried` (event) | Published when an admin retries via the UI. Fields: `original_id`, `new_id`, `task_name`. |
 | `TaskExecutionListItem` | Compact row for the listing table. |
 | `TaskExecutionDetail` | Full record incl. `args`, `kwargs`, `result`, `traceback`. |
@@ -466,6 +512,7 @@ from background_tasks.contracts.schemas import (
 | `worker` | `str(255) \| None` | hostname that ran it |
 | `retries` | `int` | |
 | `retried_from_id` | `UUID \| None` | self-FK; tracks retry chains |
+| `resource_type` / `resource_id` | `str(255) \| None` | optional both-or-neither resource reference (#709) |
 | `owner_user_id` | `str(64) \| None` | indexed; user whose request enqueued the task (#536) |
 | `progress_current` / `progress_total` | `int \| None` | latest `heartbeat(progress=...)` report |
 | `progress_message` / `progress_phase` | `str(255) \| None` / `str(64) \| None` | not audited |
@@ -500,6 +547,8 @@ DB-backed; defaults are in `BackgroundTasksSettings`. Several are marked `requir
 | `task_routes_override`, tuning fields | unset | restart required; see [Routing and queues](#routing-and-queues) and [Worker tuning](#worker-tuning) |
 | `task_always_eager` | `False` | runs tasks in-process (test mode) |
 | `task_eager_propagates` | `True` | re-raises eager-mode errors |
+| `execution_tracking` | `full` | `full`, `terminal`, `failures`, `off`; DB-only, restart required |
+| `track_publish` | `True` | publish-side tracking for full mode; DB-only, restart required |
 | `stuck_after_seconds` | `300` | heartbeat-staleness threshold; a task's own `stuck_after` overrides it — see [Long-running tasks](#long-running-tasks-heartbeats-and-stuck-thresholds) |
 | `stuck_sweep_interval_seconds` | `60` | beat cadence for the stuck sweep |
 | `purge_interval_seconds` | `86_400` | beat cadence for old-row purge |
@@ -563,6 +612,8 @@ It's synchronous; the view route calls it via `asyncio.to_thread`.
 
 ## Celery signal hooks
 
+The table below describes full tracking with `track_publish=True`; reduced modes gate these writes as described under [Execution tracking](#execution-tracking).
+
 Synchronous handlers in `signals.py`, all bound through Celery's `signals.<name>.connect`, persisting through a separate **sync** SQLAlchemy engine (`sync_db.py`) so they don't have to spin up an event loop:
 
 | Signal | What it writes |

host/migrations/versions/b31a70971001_background_tasks_resource_reference.py

@@ -0,0 +1,32 @@
+"""Task resource references (#709).
+
+Revision ID: b31a70971001
+Revises: a626b1c8d023 (background_tasks@head)
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "b31a70971001"
+down_revision = "a626b1c8d023"
+branch_labels = None
+depends_on = None
+
+_TABLE = "background_tasks_task_execution"
+_CONSTRAINT = "ck_background_tasks_task_execution_task_resource_pair"
+
+
+def upgrade() -> None:
+    with op.batch_alter_table(_TABLE) as batch:
+        batch.add_column(sa.Column("resource_type", sa.String(255), nullable=True))
+        batch.add_column(sa.Column("resource_id", sa.String(255), nullable=True))
+        batch.create_check_constraint(
+            op.f(_CONSTRAINT), "(resource_type IS NULL) = (resource_id IS NULL)"
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table(_TABLE) as batch:
+        batch.drop_constraint(op.f(_CONSTRAINT), type_="check")
+        batch.drop_column("resource_id")
+        batch.drop_column("resource_type")

host/tests/test_background_tasks_resource_migration.py

@@ -0,0 +1,64 @@
+"""Resource-reference migration preserves legacy rows and roundtrips on SQLite."""
+
+from importlib import import_module
+
+import pytest
+import sqlalchemy as sa
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+from background_tasks.models import TaskExecution
+from sqlalchemy.exc import IntegrityError
+
+
+def test_resource_reference_migration_roundtrip(tmp_path, monkeypatch):
+    migration = import_module(
+        "host.migrations.versions.b31a70971001_background_tasks_resource_reference"
+    )
+    engine = sa.create_engine(f"sqlite:///{tmp_path / 'migration.db'}")
+    with engine.begin() as connection:
+        connection.exec_driver_sql(
+            "CREATE TABLE background_tasks_task_execution (id INTEGER PRIMARY KEY, "
+            "task_name VARCHAR(255), owner_user_id VARCHAR(64))"
+        )
+        connection.exec_driver_sql(
+            "INSERT INTO background_tasks_task_execution VALUES (1, 'demo.image', 'owner')"
+        )
+        monkeypatch.setattr(
+            migration,
+            "op",
+            Operations(
+                MigrationContext.configure(
+                    connection, opts={"target_metadata": TaskExecution.metadata}
+                )
+            ),
+        )
+        migration.upgrade()
+        assert {
+            item["name"]
+            for item in sa.inspect(connection).get_check_constraints(
+                "background_tasks_task_execution"
+            )
+        } == {"ck_background_tasks_task_execution_task_resource_pair"}
+        row = connection.exec_driver_sql(
+            "SELECT owner_user_id, resource_type, resource_id FROM background_tasks_task_execution"
+        ).one()
+        assert tuple(row) == ("owner", None, None)
+        connection.exec_driver_sql(
+            "UPDATE background_tasks_task_execution SET resource_type='demo.image', resource_id='7'"
+        )
+        with pytest.raises(IntegrityError):
+            connection.exec_driver_sql(
+                "UPDATE background_tasks_task_execution SET resource_id=NULL"
+            )
+        migration.downgrade()
+        assert [
+            col["name"]
+            for col in sa.inspect(connection).get_columns("background_tasks_task_execution")
+        ] == ["id", "task_name", "owner_user_id"]
+        assert (
+            connection.exec_driver_sql(
+                "SELECT owner_user_id FROM background_tasks_task_execution"
+            ).scalar_one()
+            == "owner"
+        )
+    engine.dispose()

host/tests/test_migration_graph.py

@@ -38,6 +38,7 @@
     "b5d3f08a6e17": "background_tasks",
     "e878882924e7": "background_tasks",
     "bb906e1a1089": "background_tasks",  # task progress + owner (#536)
+    "b31a70971001": "background_tasks",  # resource references (#709)
     "1f5985daa10b": "background_tasks",  # retained log records (#537)
     "a625f0b7c912": "users",  # refresh-token families (#625)
     "a626b1c8d023": "users",  # hashed bearer tokens (#626)
@@ -57,7 +58,7 @@
 # head revision -> module owning the tables it creates or alters.
 _HEADS = {
     "a7c2e91d4b58": "audit_log",
-    "a626b1c8d023": "users",  # also carries ``api_tokens`` and ``background_tasks``
+    "b31a70971001": "background_tasks",  # also carries ``api_tokens`` and ``users``
     "f1a9c3d7b2e4": "tenants",
     "f6a3b9d2e8c4": "keycloak",
     "f733c7938fd1": "sharing",
@@ -97,6 +98,7 @@
     "bb906e1a1089": "api_tokens",  # also its own ``background_tasks``
     "1f5985daa10b": "api_tokens",  # also inherits ``background_tasks``
     "a625f0b7c912": "api_tokens",  # also inherits ``background_tasks``
+    "b31a70971001": "api_tokens",
     "a626b1c8d023": "api_tokens",  # also inherits ``background_tasks``
     "51b2058d8afd": "api_tokens",  # ancestors of a labelled revision inherit its label too
 }

modules/background_tasks/background_tasks/__init__.py

@@ -3,11 +3,14 @@
 from typing import TYPE_CHECKING, Any
 
 from background_tasks.after_commit import enqueue_after_commit
+from background_tasks.constants import SM_TRACKING_HEADER
 from background_tasks.log_context import (
     bind_task_context,
     get_log_context,
     install_log_filter,
 )
+from background_tasks.task_resource import with_task_resource
+from background_tasks.tracking import TrackingMode
 
 if TYPE_CHECKING:
     from background_tasks.task_heartbeat import heartbeat
@@ -30,10 +33,13 @@ def __getattr__(name: str) -> Any:
 
 
 __all__ = [
+    "SM_TRACKING_HEADER",
+    "TrackingMode",
     "bind_task_context",
     "enqueue_after_commit",
     "get_log_context",
     "heartbeat",
     "install_log_filter",
     "settings_for",
+    "with_task_resource",
 ]