diff --git a/CHANGELOG.md b/CHANGELOG.md
index df5ff91ec..0060cfef2 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/docs/database/migrations.md b/docs/database/migrations.md
index d84a41e0d..096f18cfd 100644
--- a/docs/database/migrations.md
+++ b/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` |
diff --git a/docs/modules/background_tasks.md b/docs/modules/background_tasks.md
index be471d52d..16e122101 100644
--- a/docs/modules/background_tasks.md
+++ b/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 |
diff --git a/host/migrations/versions/b31a70971001_background_tasks_resource_reference.py b/host/migrations/versions/b31a70971001_background_tasks_resource_reference.py
new file mode 100644
index 000000000..8ff1c9de5
--- /dev/null
+++ b/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")
diff --git a/host/tests/test_background_tasks_resource_migration.py b/host/tests/test_background_tasks_resource_migration.py
new file mode 100644
index 000000000..6cf949d29
--- /dev/null
+++ b/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()
diff --git a/host/tests/test_migration_graph.py b/host/tests/test_migration_graph.py
index b82b00031..5c0598042 100644
--- a/host/tests/test_migration_graph.py
+++ b/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
 }
diff --git a/modules/background_tasks/background_tasks/__init__.py b/modules/background_tasks/background_tasks/__init__.py
index 5a4bf365d..7379b357f 100644
--- a/modules/background_tasks/background_tasks/__init__.py
+++ b/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",
 ]
diff --git a/modules/background_tasks/background_tasks/_signal_support.py b/modules/background_tasks/background_tasks/_signal_support.py
index 0e67077af..5bc207062 100644
--- a/modules/background_tasks/background_tasks/_signal_support.py
+++ b/modules/background_tasks/background_tasks/_signal_support.py
@@ -13,13 +13,16 @@
 from datetime import UTC, datetime
 from typing import Any
 
+from celery import current_app
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 from sqlmodel import col
 
-from background_tasks.constants import INTERNAL_TASK_SWEEP_OUTBOX, TaskStatus
 from background_tasks.models import TaskExecution
 from background_tasks.sync_db import sync_session
+from background_tasks.task_resource import message_resource
+from background_tasks.tenant_context import message_owner
+from background_tasks.tracking import records_signal, tracking_mode
 
 logger = logging.getLogger(__name__)
 
@@ -112,13 +115,6 @@ def render_traceback(einfo: Any, exception: BaseException | None) -> str | None:
     return None
 
 
-# Beat ticks that fire every few seconds and are a no-op almost every time;
-# recording each run buried real work (QA, GH #459). Only a run that went wrong
-# is worth a row.
-_UNRECORDED_WHEN_QUIET = frozenset({INTERNAL_TASK_SWEEP_OUTBOX})
-_QUIET_STATUSES = frozenset({None, TaskStatus.PENDING, TaskStatus.RUNNING, TaskStatus.SUCCESS})
-
-
 def apply_signal(
     handler: str,
     *,
@@ -127,6 +123,9 @@ def apply_signal(
     after: Callable[[TaskExecution], None] | None = None,
     tenant_id: str | None = None,
     owner_user_id: str | None = None,
+    task: Any = None,
+    request: Any = None,
+    headers: dict | None = None,
 ) -> TaskExecution | None:
     """Open a sync session, upsert by celery_task_id, log on failure.
 
@@ -136,12 +135,30 @@ def apply_signal(
     ``tenant_id`` and ``owner_user_id`` (the message's) are written only when
     known, so a signal whose message carries neither never blanks a row already
     stamped.
+
+    The task's tracking mode (:mod:`.tracking`) decides first whether this
+    signal is recorded at all. When it is, the row also takes args, kwargs,
+    queue, owner and resource from the message when ``defaults`` lack them, so
+    a worker can create the row itself when the publisher never did.
     """
-    if (
-        defaults.get("task_name") in _UNRECORDED_WHEN_QUIET
-        and defaults.get("status") in _QUIET_STATUSES
-    ):
+    name = defaults.get("task_name", "")
+    mode = tracking_mode(task=task, request=request, headers=headers, name=name)
+    app = getattr(task, "app", None) or current_app
+    if not records_signal(mode, handler, defaults.get("status"), app=app):
         return None
+    request = request if request is not None else getattr(task, "request", None)
+    if request is not None:
+        metadata = {}
+        if getattr(request, "args", None) is not None:
+            metadata["args"] = list(request.args)
+        if getattr(request, "kwargs", None) is not None:
+            metadata["kwargs"] = dict(request.kwargs)
+        delivery = getattr(request, "delivery_info", None) or {}
+        if delivery.get("routing_key"):
+            metadata["queue"] = delivery["routing_key"]
+        defaults = {**metadata, **defaults}
+        owner_user_id = owner_user_id or message_owner(request=request)
+    defaults = {**message_resource(headers=headers, request=request), **defaults}
     if tenant_id:
         defaults = {**defaults, "tenant_id": tenant_id}
     if owner_user_id:
diff --git a/modules/background_tasks/background_tasks/celery_app.py b/modules/background_tasks/background_tasks/celery_app.py
index 9e33cb80c..8f516e505 100644
--- a/modules/background_tasks/background_tasks/celery_app.py
+++ b/modules/background_tasks/background_tasks/celery_app.py
@@ -174,6 +174,8 @@ def build_celery(settings: BackgroundTasksSettings) -> Celery:
             beat_schedule[name] = entry
 
     celery.conf.update(
+        sm_execution_tracking=settings.execution_tracking,
+        sm_track_publish=settings.track_publish,
         broker_url=settings.broker_url,
         result_backend=settings.result_backend,
         task_default_queue=settings.task_default_queue,
diff --git a/modules/background_tasks/background_tasks/constants.py b/modules/background_tasks/background_tasks/constants.py
index f45cc5064..c682cba64 100644
--- a/modules/background_tasks/background_tasks/constants.py
+++ b/modules/background_tasks/background_tasks/constants.py
@@ -199,3 +199,9 @@ class OutboxMode(StrEnum):
 LOG_PER_PAGE = 50
 # Level names the admin filter offers, lowest first.
 LOG_LEVEL_NAMES = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")
+
+# Message metadata shared by publishers and workers.
+SM_TRACKING_HEADER = "sm_tracking"
+SM_RESOURCE_TYPE_HEADER = "sm_resource_type"
+SM_RESOURCE_ID_HEADER = "sm_resource_id"
+RESOURCE_REF_MAX_LENGTH = 255
diff --git a/modules/background_tasks/background_tasks/endpoints/api_status.py b/modules/background_tasks/background_tasks/endpoints/api_status.py
index 6d7e0c1b7..c2b3da859 100644
--- a/modules/background_tasks/background_tasks/endpoints/api_status.py
+++ b/modules/background_tasks/background_tasks/endpoints/api_status.py
@@ -1,8 +1,8 @@
-"""Owner-scoped task status (#536).
+"""Owner- and resource-scoped task status (#536, #709).
 
 Mounted apart from the admin router because it carries no ``PERM_VIEW`` gate:
-the person who started a job may follow it without being a task operator. It
-is marked ``Authenticated()`` (SM031); the ownership check lives in the
+owners and resource viewers may follow a job without being task operators. It
+is marked ``Authenticated()`` (SM031); the visibility check lives in the
 handler, and every refusal is a 404 so the route does not reveal which task
 ids exist.
 """
@@ -13,9 +13,12 @@
 
 from fastapi import APIRouter, Depends, HTTPException, Request
 from simple_module_core.permissions import grants
+from simple_module_core.resources import ACTION_VIEW
 from simple_module_db import RequestSession, current_tenant_id
 from simple_module_db.deps import get_db
+from simple_module_db.resource_policy import load_authorized
 from simple_module_hosting.permissions import ensure_resolved_permissions
+from simple_module_hosting.resource_deps import resource_principal
 from simple_module_hosting.token_guard import Authenticated
 from sqlalchemy import select
 from sqlmodel import col
@@ -30,15 +33,28 @@
 _NOT_FOUND = "Task not found"
 
 
-async def _visible(request: Request, row: TaskExecution, user_id: str) -> bool:
+async def _visible(request: Request, row: TaskExecution, user_id: str, db: RequestSession) -> bool:
     if grants(await ensure_resolved_permissions(request), PERM_VIEW):
         return True
-    if not row.owner_user_id or row.owner_user_id != user_id:
-        return False
-    if not bool(getattr(request.app.state.sm.settings, "multi_tenant", False)):
+    if bool(getattr(request.app.state.sm.settings, "multi_tenant", False)):
+        tenant_id = current_tenant_id.get()
+        if tenant_id is None or row.tenant_id != tenant_id:
+            return False
+    if row.owner_user_id and row.owner_user_id == user_id:
         return True
-    tenant_id = current_tenant_id.get()
-    return tenant_id is not None and row.tenant_id == tenant_id
+    if not row.resource_type or not row.resource_id:
+        return False
+    resource = request.app.state.sm.resource_types.get(row.resource_type)
+    if resource is None:
+        return False
+    try:
+        stmt = resource.select_by_id(row.resource_id)
+    except ValueError:
+        return False
+    obj, allowed = await load_authorized(
+        db, resource.policy, await resource_principal(request), ACTION_VIEW, stmt
+    )
+    return obj is not None and allowed
 
 
 @router.get(
@@ -62,7 +78,7 @@ async def get_task_status(
             .limit(1)
         )
     ).scalar_one_or_none()
-    if row is None or not await _visible(request, row, str(user.id)):
+    if row is None or not await _visible(request, row, str(user.id), db):
         raise HTTPException(status_code=404, detail=_NOT_FOUND)
     error = row.exception_type
     if error and not _TYPE_NAME.fullmatch(error):
diff --git a/modules/background_tasks/background_tasks/models.py b/modules/background_tasks/background_tasks/models.py
index ee24c21cc..882262999 100644
--- a/modules/background_tasks/background_tasks/models.py
+++ b/modules/background_tasks/background_tasks/models.py
@@ -9,7 +9,7 @@
 from simple_module_db.base import create_module_base
 from simple_module_db.mixins import AuditMixin
 from simple_module_db.types import AwareDateTime
-from sqlalchemy import JSON, Column, Index, String
+from sqlalchemy import JSON, CheckConstraint, Column, Index, String
 from sqlmodel import Field
 
 from background_tasks.constants import (
@@ -22,6 +22,7 @@
     OWNER_ID_MAX_LENGTH,
     PROGRESS_MESSAGE_MAX_LENGTH,
     PROGRESS_PHASE_MAX_LENGTH,
+    RESOURCE_REF_MAX_LENGTH,
     TABLE_LOG_RECORD,
     TABLE_OUTBOX,
     TABLE_TASK_EXECUTION,
@@ -86,6 +87,11 @@ class TaskExecution(Base, AuditMixin, table=True):  # ty: ignore[unsupported-bas
     # publishes. It scopes who may read the task's status (#536).
     owner_user_id: str | None = Field(default=None, index=True, max_length=OWNER_ID_MAX_LENGTH)
 
+    # Optional resource this task is about (#709); lets viewers of that resource
+    # read the task's status. Both set or both null (check constraint below).
+    resource_type: str | None = Field(default=None, max_length=RESOURCE_REF_MAX_LENGTH)
+    resource_id: str | None = Field(default=None, max_length=RESOURCE_REF_MAX_LENGTH)
+
     # Progress the task body reports through ``heartbeat(progress=...)``.
     progress_current: int | None = None
     progress_total: int | None = None
@@ -118,6 +124,10 @@ class TaskExecution(Base, AuditMixin, table=True):  # ty: ignore[unsupported-bas
     heartbeat_at: datetime | None = None
 
     __table_args__ = (
+        CheckConstraint(
+            "(resource_type IS NULL) = (resource_id IS NULL)",
+            name="task_resource_pair",
+        ),
         Index(
             "ix_background_tasks_task_execution_status_queued",
             "status",
diff --git a/modules/background_tasks/background_tasks/retry_service.py b/modules/background_tasks/background_tasks/retry_service.py
index d3e5d4fb6..00761c113 100644
--- a/modules/background_tasks/background_tasks/retry_service.py
+++ b/modules/background_tasks/background_tasks/retry_service.py
@@ -20,13 +20,15 @@
 
 from background_tasks._signal_support import now_utc
 from background_tasks.after_commit import after_commit
-from background_tasks.constants import RETRY_ALL_BATCH, TaskStatus
+from background_tasks.constants import RETRY_ALL_BATCH, SM_TRACKING_HEADER, TaskStatus
 from background_tasks.contracts.events import TaskRetried
 from background_tasks.contracts.schemas import RetryFailedResult
 from background_tasks.filters import Conditions, bulk_retry_conditions
 from background_tasks.models import TaskExecution
 from background_tasks.sync_db import sync_session
+from background_tasks.task_resource import resource_headers
 from background_tasks.tenant_context import publish_as_owner
+from background_tasks.tracking import TrackingMode
 
 if TYPE_CHECKING:
     from celery import Celery
@@ -196,6 +198,13 @@ def _publisher(self, row: TaskExecution, task_id: str) -> Callable[[], str]:
             "task_id": task_id,
         }
 
+        # The retry already reserved a ``pending`` row; a reduced tracking
+        # mode would never move it on, so an operator's retry is tracked fully.
+        headers: dict[str, str] = {SM_TRACKING_HEADER: TrackingMode.FULL.value}
+        if row.resource_type and row.resource_id:
+            headers.update(resource_headers(row.resource_type, row.resource_id))
+        options["headers"] = headers
+
         def publish() -> str:
             try:
                 return _send()
@@ -233,6 +242,8 @@ def _new_attempt(self, row: TaskExecution, celery_task_id: str) -> TaskExecution
             queue=row.queue,
             tenant_id=row.tenant_id,
             owner_user_id=row.owner_user_id,
+            resource_type=row.resource_type,
+            resource_id=row.resource_id,
             args=list(row.args or []),
             kwargs=dict(row.kwargs or {}),
             retried_from_id=row.id,
diff --git a/modules/background_tasks/background_tasks/settings.py b/modules/background_tasks/background_tasks/settings.py
index 05f10944a..b52da78c6 100644
--- a/modules/background_tasks/background_tasks/settings.py
+++ b/modules/background_tasks/background_tasks/settings.py
@@ -53,6 +53,7 @@
     forbid_localhost_broker_in_production,
     warn_on_legacy_redis_vars,
 )
+from background_tasks.tracking import TrackingMode
 
 __all__ = ["DEFAULT_RETENTION_DAYS", "ENV_PREFIX", "BackgroundTasksSettings"]
 
@@ -62,6 +63,33 @@
 class BackgroundTasksSettings(CelerySettingsFields):
     """Configuration for the Celery + Redis task runner."""
 
+    execution_tracking: TrackingMode = Field(
+        default=TrackingMode.FULL, json_schema_extra=_CELERY_RESTART
+    )
+    track_publish: bool = Field(default=True, json_schema_extra=_CELERY_RESTART)
+
+    @classmethod
+    def settings_customise_sources(
+        cls, settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings
+    ):
+        # Tracking controls are DB-backed; preserve env wiring for older fields.
+        def db_only(source):
+            def read():
+                return {
+                    k: v
+                    for k, v in source().items()
+                    if k not in {"execution_tracking", "track_publish"}
+                }
+
+            return read
+
+        return (
+            init_settings,
+            db_only(env_settings),
+            db_only(dotenv_settings),
+            db_only(file_secret_settings),
+        )
+
     # ── Durable event outbox (GH #459) ──────────────────────
     outbox_mode: Literal["celery", "inprocess"] = Field(
         default="celery", json_schema_extra=_CELERY_RESTART
diff --git a/modules/background_tasks/background_tasks/signals.py b/modules/background_tasks/background_tasks/signals.py
index 84334cd85..b5765ea27 100644
--- a/modules/background_tasks/background_tasks/signals.py
+++ b/modules/background_tasks/background_tasks/signals.py
@@ -42,7 +42,6 @@
 from background_tasks.models import TaskExecution
 from background_tasks.task_heartbeat import forget_heartbeat
 from background_tasks.tenant_context import (
-    message_owner,
     message_tenant,
     published_owner,
     published_tenant,
@@ -86,6 +85,7 @@ def on_task_publish(
     }
     apply_signal(
         "on_task_publish",
+        headers=headers,
         celery_task_id=task_id,
         defaults=defaults,
         tenant_id=published_tenant(headers),
@@ -106,6 +106,7 @@ def on_task_prerun(
     **_k: Any,
 ) -> None:
     """Flip the row to ``running``, start the heartbeat, bind log context."""
+    restore_tenant(task_id=task_id, task=task or sender)
     args_n, kwargs_n = coerce_args_kwargs(args, kwargs)
     now = now_utc()
     name = task_name_of(sender, task)
@@ -120,11 +121,10 @@ def on_task_prerun(
             "started_at": now,
             "heartbeat_at": now,
         },
-        tenant_id=message_tenant(task=task),
-        owner_user_id=message_owner(task=task),
+        task=task or sender,
+        tenant_id=message_tenant(task=task or sender),
     )
     signal_task_started(task_id=task_id, task_name=name)
-    restore_tenant(task_id=task_id, task=task)
 
 
 @signals.task_postrun.connect
@@ -140,15 +140,18 @@ def on_task_postrun(
     ``task_retry`` which fire *before* postrun; postrun only refreshes the
     heartbeat so the sweep doesn't immediately flip a just-finished row.
     """
-    apply_signal(
-        "on_task_postrun",
-        celery_task_id=task_id,
-        defaults={"task_name": task_name_of(sender, task), "heartbeat_at": now_utc()},
-        tenant_id=message_tenant(task=task),
-    )
-    release_tenant(task_id=task_id)
-    forget_heartbeat(task_id)
-    signal_task_finished(task_id=task_id)
+    try:
+        apply_signal(
+            "on_task_postrun",
+            celery_task_id=task_id,
+            defaults={"task_name": task_name_of(sender, task), "heartbeat_at": now_utc()},
+            task=task or sender,
+            tenant_id=message_tenant(task=task or sender),
+        )
+    finally:
+        release_tenant(task_id=task_id)
+        forget_heartbeat(task_id)
+        signal_task_finished(task_id=task_id)
 
 
 @signals.task_success.connect
@@ -164,6 +167,7 @@ def on_task_success(sender: Any = None, result: Any = None, **_k: Any) -> None:
             "traceback": None,
             "exception_type": None,
         },
+        task=sender,
         tenant_id=message_tenant(task=sender),
     )
 
@@ -189,6 +193,7 @@ def on_task_failure(
             "exception_type": exception_type,
             "finished_at": now_utc(),
         },
+        task=sender,
         tenant_id=message_tenant(task=sender),
     )
 
@@ -225,6 +230,8 @@ def _stamp_reason(row: TaskExecution) -> None:
             "heartbeat_at": now_utc(),
         },
         after=_stamp_reason,
+        task=sender,
+        request=request,
         tenant_id=message_tenant(request=request),
     )
 
@@ -239,6 +246,8 @@ def on_task_revoked(sender: Any = None, request: Any = None, **_k: Any) -> None:
             "status": TaskStatus.REVOKED,
             "finished_at": now_utc(),
         },
+        task=sender,
+        request=request,
         tenant_id=message_tenant(request=request),
     )
 
diff --git a/modules/background_tasks/background_tasks/task_heartbeat.py b/modules/background_tasks/background_tasks/task_heartbeat.py
index f94655bc3..eef08893a 100644
--- a/modules/background_tasks/background_tasks/task_heartbeat.py
+++ b/modules/background_tasks/background_tasks/task_heartbeat.py
@@ -35,6 +35,7 @@
 )
 from background_tasks.models import TaskExecution
 from background_tasks.sync_db import sync_session
+from background_tasks.tracking import TrackingMode, tracking_mode
 
 logger = logging.getLogger(__name__)
 
@@ -99,7 +100,7 @@ def heartbeat(
     reporting on.
     """
     task_id = _current_task_id()
-    if task_id is None:
+    if task_id is None or tracking_mode(task=current_task) != TrackingMode.FULL:
         return False
     try:
         reported = _progress_values(progress, message, phase)
diff --git a/modules/background_tasks/background_tasks/task_resource.py b/modules/background_tasks/background_tasks/task_resource.py
new file mode 100644
index 000000000..7ebad6ec8
--- /dev/null
+++ b/modules/background_tasks/background_tasks/task_resource.py
@@ -0,0 +1,55 @@
+"""Explicit resource references on task messages (#709)."""
+
+from __future__ import annotations
+
+import re
+from typing import TYPE_CHECKING, Any
+
+from background_tasks.constants import (
+    RESOURCE_REF_MAX_LENGTH,
+    SM_RESOURCE_ID_HEADER,
+    SM_RESOURCE_TYPE_HEADER,
+)
+from background_tasks.tenant_context import header_of_request
+
+if TYPE_CHECKING:
+    from celery.canvas import Signature
+
+_TYPE_NAME = re.compile(r"[a-z][a-z0-9_]*(\.[a-z][a-z0-9_]*)+")
+
+
+def resource_headers(resource_type: str, resource_id: str) -> dict[str, str]:
+    """Validate before publishing; never silently truncate an identity."""
+    if (
+        not isinstance(resource_type, str)
+        or len(resource_type) > RESOURCE_REF_MAX_LENGTH
+        or not _TYPE_NAME.fullmatch(resource_type)
+        or not isinstance(resource_id, str)
+        or not resource_id.strip()
+        or len(resource_id) > RESOURCE_REF_MAX_LENGTH
+        or any(ord(c) < 32 or ord(c) == 127 for c in resource_id)
+    ):
+        raise ValueError("Invalid task resource reference")
+    return {SM_RESOURCE_TYPE_HEADER: resource_type, SM_RESOURCE_ID_HEADER: resource_id}
+
+
+def with_task_resource(signature: Signature, *, resource_type: str, resource_id: str) -> Signature:
+    """Clone a signature, preserving its other options and headers."""
+    headers = resource_headers(resource_type, resource_id)
+    return signature.clone().set(headers={**(signature.options.get("headers") or {}), **headers})
+
+
+def message_resource(*, headers: dict | None = None, request: Any = None) -> dict[str, str]:
+    """Accept a validated pair only; malformed external messages grant no access."""
+    kind = (headers or {}).get(SM_RESOURCE_TYPE_HEADER)
+    key = (headers or {}).get(SM_RESOURCE_ID_HEADER)
+    if headers is None:
+        kind = header_of_request(request, SM_RESOURCE_TYPE_HEADER)
+        key = header_of_request(request, SM_RESOURCE_ID_HEADER)
+    if not isinstance(kind, str) or not isinstance(key, str):
+        return {}
+    try:
+        resource_headers(kind, key)
+    except ValueError:
+        return {}
+    return {"resource_type": kind, "resource_id": key}
diff --git a/modules/background_tasks/background_tasks/tasks.py b/modules/background_tasks/background_tasks/tasks.py
index cc195b0a0..b0256ed81 100644
--- a/modules/background_tasks/background_tasks/tasks.py
+++ b/modules/background_tasks/background_tasks/tasks.py
@@ -38,6 +38,7 @@
 from background_tasks.models import OutboxMessage, TaskExecution
 from background_tasks.settings import BackgroundTasksSettings
 from background_tasks.sync_db import sync_session
+from background_tasks.tracking import TrackingMode
 from background_tasks.worker_settings import load_runner_settings
 
 logger = logging.getLogger(__name__)
@@ -174,7 +175,7 @@ def dispatch_outbox(ids: list[str]) -> dict[str, int]:
     return run_dispatch(list(ids))
 
 
-@shared_task(name=INTERNAL_TASK_SWEEP_OUTBOX)
+@shared_task(name=INTERNAL_TASK_SWEEP_OUTBOX, sm_tracking=TrackingMode.FAILURES)
 def sweep_outbox() -> dict[str, int]:
     """Deliver every due outbox row and reclaim expired leases (beat, every 30 s)."""
     from background_tasks.outbox_worker import run_dispatch
diff --git a/modules/background_tasks/background_tasks/tenant_context.py b/modules/background_tasks/background_tasks/tenant_context.py
index 05fe065f9..934ef652c 100644
--- a/modules/background_tasks/background_tasks/tenant_context.py
+++ b/modules/background_tasks/background_tasks/tenant_context.py
@@ -119,7 +119,8 @@ def _tenant_of(task: Any) -> str | None:
     return _tenant_of_request(getattr(task, "request", None))
 
 
-def _header_of_request(request: Any, name: str) -> Any:
+def header_of_request(request: Any, name: str) -> Any:
+    """A message header off a running task's request (``Request`` or publish-side object)."""
     value = getattr(request, name, None)
     # A worker ``Request`` (revoked signal) keeps headers in a dict instead.
     for attr in ("headers", "request_dict"):
@@ -133,14 +134,14 @@ def message_owner(*, task: Any = None, request: Any = None) -> str | None:
     request = request if request is not None else getattr(task, "request", None)
     if request is None:
         return None
-    value = _header_of_request(request, OWNER_HEADER)
+    value = header_of_request(request, OWNER_HEADER)
     return str(value)[:OWNER_ID_MAX_LENGTH] if value else None
 
 
 def _tenant_of_request(request: Any) -> str | None:
     if request is None:
         return None
-    value = _header_of_request(request, TENANT_HEADER)
+    value = header_of_request(request, TENANT_HEADER)
     if not value:
         return None
     if not is_valid_tenant_id(value):
@@ -199,6 +200,7 @@ def release_tenant(*, task_id: str | None) -> None:
     "OWNER_HEADER",
     "TENANT_HEADER",
     "default_tenant",
+    "header_of_request",
     "message_owner",
     "message_tenant",
     "publish_as_owner",
diff --git a/modules/background_tasks/background_tasks/tracking.py b/modules/background_tasks/background_tasks/tracking.py
new file mode 100644
index 000000000..f8866b86a
--- /dev/null
+++ b/modules/background_tasks/background_tasks/tracking.py
@@ -0,0 +1,55 @@
+"""Execution tracking policy, resolved without database access."""
+
+from enum import StrEnum
+from typing import Any
+
+from background_tasks.constants import INTERNAL_TASK_SWEEP_OUTBOX, SM_TRACKING_HEADER, TaskStatus
+from background_tasks.tenant_context import header_of_request
+
+
+class TrackingMode(StrEnum):
+    FULL = "full"
+    TERMINAL = "terminal"
+    FAILURES = "failures"
+    OFF = "off"
+
+
+def tracking_mode(
+    *, task: Any = None, request: Any = None, headers: dict | None = None, name: str = ""
+) -> TrackingMode:
+    """Message override → task option → frozen application default."""
+    from celery import current_app  # lazy: keeps ``import background_tasks`` free of celery
+
+    app = getattr(task, "app", None) or current_app
+    if task is None or isinstance(task, str):
+        task = app.tasks.get(name or task)
+    # A publish carries its own headers; the looked-up task's thread-local request
+    # belongs to whatever instance of it is running now, not to this message.
+    if request is None and headers is None:
+        request = getattr(task, "request", None)
+    explicit = (headers or {}).get(SM_TRACKING_HEADER)
+    if explicit is None:
+        explicit = header_of_request(request, SM_TRACKING_HEADER)
+    value = explicit if explicit is not None else getattr(task, SM_TRACKING_HEADER, None)
+    # Publishers may not have imported the internal task registration yet.
+    if value is None and name == INTERNAL_TASK_SWEEP_OUTBOX:
+        value = TrackingMode.FAILURES
+    if value is None:
+        value = app.conf.get("sm_execution_tracking", TrackingMode.FULL)
+    try:
+        return TrackingMode(value)
+    except (ValueError, TypeError):
+        # A malformed override must not hide failures from operators.
+        return TrackingMode.FULL
+
+
+def records_signal(mode: TrackingMode, handler: str, status: Any, *, app: Any) -> bool:
+    if mode == TrackingMode.FULL:
+        return handler != "on_task_publish" or app.conf.get("sm_track_publish", True)
+    if mode == TrackingMode.OFF:
+        return False
+    return status in (
+        {TaskStatus.FAILED, TaskStatus.REVOKED, TaskStatus.SUCCESS}
+        if mode == TrackingMode.TERMINAL
+        else {TaskStatus.FAILED, TaskStatus.REVOKED}
+    )
diff --git a/modules/background_tasks/tests/test_background_tasks_resource_status.py b/modules/background_tasks/tests/test_background_tasks_resource_status.py
new file mode 100644
index 000000000..54e630218
--- /dev/null
+++ b/modules/background_tasks/tests/test_background_tasks_resource_status.py
@@ -0,0 +1,145 @@
+"""Task resource grants use filtered policies and stay tenant scoped."""
+
+import pytest
+from background_tasks.constants import TaskStatus
+from background_tasks.models import TaskExecution
+from simple_module_core.resources import ResourceType
+from simple_module_core.tenancy import TenantRole
+from simple_module_db.resource_policy import select_by_id
+from sqlalchemy import false
+
+URL = "/api/background_tasks/tasks/{}/status"
+
+
+class SharedPolicy:
+    allowed = True
+
+    def filter(self, principal, action, stmt):
+        return stmt if self.allowed else stmt.where(false())
+
+    def can(self, principal, action, obj):
+        return False  # A share is visible through filter(), never can().
+
+
+async def test_resource_share_revocation_and_tenant_boundary(app, tenant_client, seed_execution):
+    policy = SharedPolicy()
+    app.state.sm.resource_types.register(
+        ResourceType("demo.image", policy, select_by_id(TaskExecution))
+    )
+    async with (
+        tenant_client(TenantRole.MEMBER) as owner,
+        tenant_client(TenantRole.MEMBER, tenant_id=owner.tenant_id) as viewer,
+        tenant_client(TenantRole.MEMBER) as outsider,
+    ):
+        resource = await seed_execution(tenant_id=owner.tenant_id)
+        task = await seed_execution(
+            status=TaskStatus.RUNNING,
+            tenant_id=owner.tenant_id,
+            owner_user_id=owner.user_id,
+            resource_type="demo.image",
+            resource_id=str(resource.id),
+            traceback="secret",
+        )
+        response = await viewer.client.get(URL.format(task.celery_task_id))
+        assert response.status_code == 200, response.text
+        assert (
+            not {"args", "traceback", "resource_type", "resource_id", "result"}
+            & response.json().keys()
+        )
+        assert (await outsider.client.get(URL.format(task.celery_task_id))).status_code == 404
+        policy.allowed = False
+        assert (await viewer.client.get(URL.format(task.celery_task_id))).status_code == 404
+        policy.allowed = True
+        async with app.state.sm.db.session_factory() as session:
+            from sqlalchemy import delete
+
+            await session.execute(delete(TaskExecution).where(TaskExecution.id == resource.id))
+            await session.commit()
+        assert (await viewer.client.get(URL.format(task.celery_task_id))).status_code == 404
+
+
+@pytest.mark.parametrize(
+    "resource_type,resource_id",
+    [
+        ("unknown.type", "abc"),
+        ("demo.image", "bad-id"),
+        ("demo.image", "00000000-0000-0000-0000-000000000000"),
+    ],
+)
+async def test_unavailable_resource_is_the_same_404(
+    resource_type, resource_id, app, tenant_client, seed_execution
+):
+    app.state.sm.resource_types.register(
+        ResourceType("demo.image", SharedPolicy(), select_by_id(TaskExecution))
+    )
+    async with tenant_client(TenantRole.MEMBER) as me:
+        task = await seed_execution(
+            tenant_id=me.tenant_id, resource_type=resource_type, resource_id=resource_id
+        )
+        response = await me.client.get(URL.format(task.celery_task_id))
+        assert response.status_code == 404
+        assert response.json()["detail"] == "Task not found"
+
+
+@pytest.mark.parametrize("headers", [None, {}, {"custom": "value"}])
+def test_resource_signature_is_cloned_and_headers_are_merged(headers):
+    from background_tasks import with_task_resource
+    from celery import signature
+
+    original = signature("demo.images", headers=headers)
+    attached = with_task_resource(original, resource_type="demo.image", resource_id="7")
+    assert original.options["headers"] == headers
+    assert attached.options["headers"] == {
+        **(headers or {}),
+        "sm_resource_type": "demo.image",
+        "sm_resource_id": "7",
+    }
+
+
+@pytest.mark.parametrize(
+    "kind,key",
+    [
+        ("", "7"),
+        ("bad type", "7"),
+        ("demo.image", ""),
+        ("demo.image", "x" * 256),
+        ("demo.image", "x\n"),
+        ("demo.image", 7),
+    ],
+)
+def test_resource_headers_are_validated(kind, key):
+    from background_tasks import with_task_resource
+    from celery import signature
+
+    with pytest.raises(ValueError):
+        with_task_resource(signature("demo.images"), resource_type=kind, resource_id=key)
+
+
+async def test_resource_access_requires_an_active_tenant(app, seed_execution):
+    from simple_module_test.tenant_client import create_user, session_client
+
+    user_id = await create_user(app, "no-tenant-resource@test")
+    app.state.sm.resource_types.register(
+        ResourceType("demo.image", SharedPolicy(), select_by_id(TaskExecution))
+    )
+    resource = await seed_execution()
+    task = await seed_execution(resource_type="demo.image", resource_id=str(resource.id))
+    async with session_client(app, {"user_id": user_id}) as client:
+        assert (await client.get(URL.format(task.celery_task_id))).status_code == 404
+
+
+async def test_unexpected_policy_errors_remain_errors(app, tenant_client, seed_execution):
+    class BrokenPolicy(SharedPolicy):
+        def filter(self, principal, action, stmt):
+            raise RuntimeError("policy failed")
+
+    app.state.sm.resource_types.register(
+        ResourceType("demo.image", BrokenPolicy(), select_by_id(TaskExecution))
+    )
+    async with tenant_client(TenantRole.MEMBER) as me:
+        resource = await seed_execution(tenant_id=me.tenant_id)
+        task = await seed_execution(
+            tenant_id=me.tenant_id, resource_type="demo.image", resource_id=str(resource.id)
+        )
+        with pytest.raises(RuntimeError, match="policy failed"):
+            await me.client.get(URL.format(task.celery_task_id))
diff --git a/modules/background_tasks/tests/test_background_tasks_resource_transport.py b/modules/background_tasks/tests/test_background_tasks_resource_transport.py
new file mode 100644
index 000000000..bb6b8c23e
--- /dev/null
+++ b/modules/background_tasks/tests/test_background_tasks_resource_transport.py
@@ -0,0 +1,129 @@
+"""Resource references survive worker-only tracking and administrative retries."""
+
+from types import SimpleNamespace
+
+import pytest
+from background_tasks import sync_db
+from background_tasks.constants import SM_TRACKING_HEADER
+from background_tasks.models import TaskExecution
+from background_tasks.signals import on_task_failure, on_task_publish
+from sqlalchemy import select
+
+
+@pytest.mark.parametrize("mode", ["full", "terminal", "failures"])
+def test_worker_only_terminal_creation_keeps_resource(mode, sync_sqlite):
+    task = SimpleNamespace(
+        name="demo.image",
+        request=SimpleNamespace(
+            id="terminal-resource",
+            headers={
+                "sm_tracking": mode,
+                "sm_resource_type": "demo.image",
+                "sm_resource_id": "7",
+                "sm_owner_id": "owner",
+                "sm_tenant_id": "org",
+            },
+            args=[7],
+            kwargs={"format": "png"},
+            delivery_info={"routing_key": "images"},
+        ),
+    )
+    on_task_failure(sender=task, task_id=task.request.id, exception=ValueError("bad image"))
+    with sync_db.get_sync_session_factory()() as db:
+        row = db.execute(select(TaskExecution)).scalar_one()
+        assert (row.resource_type, row.resource_id) == ("demo.image", "7")
+        assert (row.owner_user_id, row.tenant_id, row.queue) == ("owner", "org", "images")
+        assert (row.args, row.kwargs) == ([7], {"format": "png"})
+
+
+@pytest.mark.parametrize(
+    "headers",
+    [
+        {"sm_resource_type": "demo.image"},
+        {"sm_resource_type": "bad type", "sm_resource_id": "7"},
+        {"sm_resource_id": "7"},
+    ],
+)
+def test_invalid_message_resource_grants_no_access(headers, sync_sqlite):
+    on_task_publish(sender="demo.image", headers={"id": "invalid-resource", **headers})
+    with sync_db.get_sync_session_factory()() as db:
+        row = db.execute(select(TaskExecution)).scalar_one()
+        assert (row.resource_type, row.resource_id) == (None, None)
+
+
+@pytest.mark.usefixtures("_stub_celery")
+async def test_admin_retry_preserves_resource_in_row_and_message(
+    app, seed_execution, execution_rows, authenticated_client
+):
+    row = await seed_execution(
+        resource_type="demo.image", resource_id="7", owner_user_id="owner", tenant_id="org"
+    )
+    response = await authenticated_client.post(
+        f"/api/background_tasks/admin/executions/{row.id}/retry"
+    )
+    assert response.status_code == 200, response.text
+    child = next(item for item in await execution_rows() if item.id != row.id)
+    assert (child.resource_type, child.resource_id, child.owner_user_id, child.tenant_id) == (
+        "demo.image",
+        "7",
+        "owner",
+        "org",
+    )
+    assert app.state.background_tasks.celery.send_task.call_args.kwargs["headers"] == {
+        "sm_resource_type": "demo.image",
+        "sm_resource_id": "7",
+        # A reduced mode would leave the reserved ``pending`` row stuck.
+        SM_TRACKING_HEADER: "full",
+    }
+
+
+def test_resource_pair_is_required_by_database(sync_sqlite):
+    from sqlalchemy.exc import IntegrityError
+
+    with sync_db.get_sync_session_factory()() as db:
+        db.add(TaskExecution(task_name="demo.image", resource_type="demo.image"))
+        with pytest.raises(IntegrityError):
+            db.flush()
+
+
+def test_off_tracking_keeps_owner_tenant_but_never_inherits_resource(monkeypatch):
+    from background_tasks import _signal_support
+    from background_tasks.signals import on_task_postrun, on_task_prerun
+    from background_tasks.tenant_context import stamp_owner, stamp_tenant
+    from simple_module_db import current_tenant_id
+
+    def forbidden():
+        pytest.fail("OFF tracking opened a session")
+
+    monkeypatch.setattr(_signal_support, "sync_session", forbidden)
+    task = SimpleNamespace(
+        name="demo.parent",
+        request=SimpleNamespace(
+            id="off-parent",
+            headers={
+                "sm_tracking": "off",
+                "sm_tenant_id": "org",
+                "sm_owner_id": "owner",
+                "sm_resource_type": "demo.image",
+                "sm_resource_id": "7",
+            },
+        ),
+    )
+    before = current_tenant_id.get()
+    on_task_prerun(sender=task, task=task, task_id=task.request.id)
+    try:
+        child = {"id": "off-child", "sm_tracking": "off"}
+        on_task_publish(sender="demo.child", headers=child)
+        assert child == {
+            "id": "off-child",
+            "sm_tracking": "off",
+            "sm_tenant_id": "org",
+            "sm_owner_id": "owner",
+        }
+    finally:
+        on_task_postrun(sender=task, task=task, task_id=task.request.id)
+    assert current_tenant_id.get() == before
+    headers = {}
+    stamp_owner(headers)
+    stamp_tenant(headers)
+    assert "sm_owner_id" not in headers
diff --git a/modules/background_tasks/tests/test_background_tasks_tracking_modes.py b/modules/background_tasks/tests/test_background_tasks_tracking_modes.py
new file mode 100644
index 000000000..8e751a251
--- /dev/null
+++ b/modules/background_tasks/tests/test_background_tasks_tracking_modes.py
@@ -0,0 +1,245 @@
+"""Tracking policy must gate sessions, including worker-first execution."""
+
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+from background_tasks import sync_db
+from background_tasks.models import TaskExecution
+from background_tasks.signals import (
+    on_task_failure,
+    on_task_postrun,
+    on_task_prerun,
+    on_task_publish,
+    on_task_retry,
+    on_task_revoked,
+    on_task_success,
+)
+from sqlalchemy import select
+
+
+@pytest.mark.parametrize(
+    "mode,outcome,writes",
+    [
+        (mode, outcome, writes)
+        for mode, outcomes in [
+            ("full", [4, 4, 4]),
+            ("terminal", [1, 1, 1]),
+            ("failures", [0, 1, 1]),
+            ("off", [0, 0, 0]),
+        ]
+        for outcome, writes in zip(["success", "failure", "revoked"], outcomes, strict=True)
+    ],
+)
+def test_lifecycle_writes(mode, outcome, writes, sync_sqlite, monkeypatch):
+    from background_tasks import _signal_support
+    from simple_module_db import current_tenant_id
+
+    opened = Mock(wraps=_signal_support.sync_session)
+    monkeypatch.setattr(_signal_support, "sync_session", opened)
+    headers = {
+        "id": "mode-test",
+        "sm_tracking": mode,
+        "sm_owner_id": "owner",
+        "sm_tenant_id": "org",
+    }
+    request = SimpleNamespace(
+        id="mode-test",
+        headers=headers,
+        args=[7],
+        kwargs={"x": 1},
+        delivery_info={"routing_key": "images"},
+    )
+    task = SimpleNamespace(name="demo.modes", request=request)
+    on_task_publish(sender=task.name, headers=headers, body=([7], {"x": 1}, {}))
+    before = current_tenant_id.get()
+    on_task_prerun(sender=task, task=task, task_id=request.id, args=[7], kwargs={"x": 1})
+    assert current_tenant_id.get() == "org"
+    if outcome == "success":
+        on_task_success(sender=task, result=9)
+    elif outcome == "failure":
+        on_task_failure(sender=task, task_id=request.id, exception=ValueError("bad"))
+    else:
+        on_task_revoked(sender=task, request=request)
+    on_task_postrun(sender=task, task=task, task_id=request.id)
+    assert current_tenant_id.get() == before
+    assert opened.call_count == writes
+    with sync_db.get_sync_session_factory()() as db:
+        row = db.execute(select(TaskExecution)).scalar_one_or_none()
+        if writes:
+            assert row.owner_user_id == "owner"
+            assert row.tenant_id == "org"
+            assert row.args == [7]
+            assert row.kwargs == {"x": 1}
+        else:
+            assert row is None
+
+
+@pytest.mark.parametrize("mode", ["terminal", "failures", "off"])
+def test_retries_and_heartbeat_do_not_open_sessions(mode, monkeypatch):
+    from background_tasks import _signal_support, task_heartbeat
+
+    request = SimpleNamespace(id="retry-test", called_directly=False, headers={"sm_tracking": mode})
+    task = SimpleNamespace(name="demo.retry", request=request)
+
+    def forbidden():
+        pytest.fail("tracking opened a session")
+
+    monkeypatch.setattr(_signal_support, "sync_session", forbidden)
+    monkeypatch.setattr(task_heartbeat, "sync_session", forbidden)
+    monkeypatch.setattr(task_heartbeat, "current_task", task)
+    on_task_retry(sender=task, request=request, reason="again")
+    assert task_heartbeat.heartbeat(force=True) is False
+
+
+def test_tracking_settings_are_db_only(monkeypatch):
+    from background_tasks.settings import BackgroundTasksSettings
+
+    monkeypatch.setenv("SM_BG_TASKS_EXECUTION_TRACKING", "off")
+    monkeypatch.setenv("SM_BG_TASKS_TRACK_PUBLISH", "false")
+    settings = BackgroundTasksSettings()
+    assert settings.execution_tracking == "full"
+    assert settings.track_publish is True
+
+
+@pytest.mark.parametrize("track_publish", [True, False])
+def test_worker_recovers_metadata_without_publish_tracking(track_publish, sync_sqlite, monkeypatch):
+    from background_tasks import _signal_support
+    from celery import current_app
+
+    monkeypatch.setitem(current_app.conf, "sm_track_publish", track_publish)
+    opened = Mock(wraps=_signal_support.sync_session)
+    monkeypatch.setattr(_signal_support, "sync_session", opened)
+    headers = {
+        "id": "worker-first",
+        "sm_owner_id": "owner",
+        "sm_tenant_id": "org",
+        "sm_resource_type": "demo.image",
+        "sm_resource_id": "7",
+    }
+    on_task_publish(
+        sender="demo.images", headers=headers, body=([7], {"x": 1}, {}), routing_key="images"
+    )
+    assert opened.call_count == int(track_publish)
+    request = SimpleNamespace(
+        id="worker-first",
+        headers=headers,
+        args=[7],
+        kwargs={"x": 1},
+        delivery_info={"routing_key": "images"},
+    )
+    task = SimpleNamespace(name="demo.images", request=request)
+    on_task_success(sender=task, result=None)
+    with sync_db.get_sync_session_factory()() as db:
+        row = db.execute(select(TaskExecution)).scalar_one()
+        assert (row.owner_user_id, row.tenant_id, row.queue) == ("owner", "org", "images")
+        assert (row.resource_type, row.resource_id) == ("demo.image", "7")
+        assert (row.args, row.kwargs) == ([7], {"x": 1})
+
+
+def test_resolution_precedence_and_invalid_override(monkeypatch):
+    from background_tasks import SM_TRACKING_HEADER, TrackingMode
+    from background_tasks.tracking import tracking_mode
+    from celery import current_app
+
+    monkeypatch.setitem(current_app.conf, "sm_execution_tracking", "terminal")
+    task = SimpleNamespace(name="demo.policy", request=SimpleNamespace(headers={}))
+    assert tracking_mode(task=task) == TrackingMode.TERMINAL
+    task.sm_tracking = "off"
+    assert tracking_mode(task=task) == TrackingMode.OFF
+    task.request.headers[SM_TRACKING_HEADER] = "failures"
+    assert tracking_mode(task=task) == TrackingMode.FAILURES
+    task.request.headers[SM_TRACKING_HEADER] = "bad-mode"
+    assert tracking_mode(task=task) == TrackingMode.FULL
+
+
+@pytest.mark.parametrize(
+    "mode,expected", [("full", 1), ("terminal", 1), ("failures", 0), ("off", 0)]
+)
+def test_eager_task_uses_task_option_and_message_override(mode, expected, sync_sqlite):
+    from celery import Celery
+
+    app = Celery(
+        "tracking-eager", broker="memory://", backend="cache+memory://", set_as_current=False
+    )
+    app.conf.update(task_always_eager=True, task_eager_propagates=True)
+
+    @app.task(name="demo.eager_modes", sm_tracking="off")
+    def work():
+        return 7
+
+    assert work.apply_async(headers={"sm_tracking": mode}).get() == 7
+    with sync_db.get_sync_session_factory()() as db:
+        assert len(list(db.execute(select(TaskExecution)).scalars())) == expected
+
+
+def test_db_hydration_and_frozen_publish_settings(monkeypatch):
+    from background_tasks import _signal_support, celery_app
+    from background_tasks.settings import BackgroundTasksSettings
+    from celery import current_app
+    from simple_module_core.settings_hydrate import build_settings, env_supplied_fields
+
+    monkeypatch.setenv("SM_BG_TASKS_EXECUTION_TRACKING", "off")
+    monkeypatch.setenv("SM_BG_TASKS_TRACK_PUBLISH", "true")
+    settings = build_settings(
+        BackgroundTasksSettings,
+        {"execution_tracking": ("terminal", "str"), "track_publish": ("false", "bool")},
+    )
+    assert settings.execution_tracking == "terminal"
+    assert settings.track_publish is False
+    assert not {"execution_tracking", "track_publish"} & env_supplied_fields(
+        BackgroundTasksSettings
+    )
+    monkeypatch.setattr(celery_app, "_discover_task_packages", list)
+    previous = current_app._get_current_object()
+    app = celery_app.build_celery(settings)
+    try:
+        settings.execution_tracking = "full"
+        settings.track_publish = True
+        assert app.conf.sm_execution_tracking == "terminal"
+        assert app.conf.sm_track_publish is False
+
+        def forbidden():
+            pytest.fail("publish opened a tracking session")
+
+        monkeypatch.setattr(_signal_support, "sync_session", forbidden)
+
+        @app.task(name="demo.publish_option", sm_tracking="off")
+        def task():
+            pass
+
+        on_task_publish(sender=task.name, headers={"id": "task-option"})
+        app.conf.sm_execution_tracking = "full"
+        on_task_publish(sender="demo.unregistered", headers={"id": "publish-disabled"})
+    finally:
+        previous.set_current()
+        app.close()
+
+
+def test_postrun_always_cleans_context_on_tracking_error(monkeypatch):
+    from background_tasks import signals, task_heartbeat
+    from background_tasks.tenant_context import release_tenant
+    from simple_module_db import current_tenant_id
+
+    task = SimpleNamespace(
+        name="demo.cleanup",
+        request=SimpleNamespace(
+            id="cleanup", headers={"sm_tracking": "off", "sm_tenant_id": "org"}
+        ),
+    )
+    before = current_tenant_id.get()
+    signals.on_task_prerun(sender=task, task=task, task_id="cleanup")
+    task_heartbeat._pending["cleanup"] = {"progress_current": 1}
+
+    def broken(*args, **kwargs):
+        raise RuntimeError("tracking error")
+
+    monkeypatch.setattr(signals, "apply_signal", broken)
+    try:
+        with pytest.raises(RuntimeError):
+            signals.on_task_postrun(sender=task, task=task, task_id="cleanup")
+        assert current_tenant_id.get() == before
+        assert "cleanup" not in task_heartbeat._pending
+    finally:
+        release_tenant(task_id="cleanup")
+        task_heartbeat.forget_heartbeat("cleanup")
diff --git a/modules/background_tasks/tests/test_bg_service.py b/modules/background_tasks/tests/test_bg_service.py
index f0a31beba..1b5228b7f 100644
--- a/modules/background_tasks/tests/test_bg_service.py
+++ b/modules/background_tasks/tests/test_bg_service.py
@@ -201,6 +201,7 @@ async def _on_retried(event: TaskRetried) -> None:
             kwargs={"priority": "high"},
             queue="orders",
             task_id=detail.celery_task_id,
+            headers={"sm_tracking": "full"},
         )
 
         assert len(received) == 1