Add worker delivery acceptance probe

docs/DEPLOYMENT_OPERATOR_GUIDE.md

@@ -158,6 +158,7 @@ release evidence.
 | `REDIS_URL` | `redis://redis:6379/0` | Celery broker/result backend when async workers are enabled. |
 | `CELERY_ENABLED` | `false` | Local/dev can send synchronously. Production campaign delivery should run workers and set this to `true`. |
 | `CELERY_QUEUES` | `send_email,append_sent,notifications,mail,calendar,dataflow,workflow,postbox,events,idm,default` | Queue list expected by worker/process manager definitions. Keep this aligned with every enabled module task route; Ops reports missing worker queue consumers. |
+| `CELERY_VISIBILITY_TIMEOUT_SECONDS` | `3600` | Maximum time before Redis may redeliver work left unacknowledged by a lost worker. Set this above the longest supported task duration; changing it requires a worker-loss acceptance drill. |
 | `PLATFORM_EVENT_OUTBOX_MAX_ATTEMPTS` | `8` | Failed durable consumer deliveries are quarantined after this many attempts. |
 | `PLATFORM_EVENT_OUTBOX_TERMINAL_RETENTION_DAYS` | `90` | Successful event envelopes older than this are removed by the daily retention task. Quarantined evidence is retained. |
 
@@ -177,6 +178,13 @@ crashes, and expired worker leases:
 python -m celery -A govoplan_core.celery_app:celery beat --loglevel INFO
 ```
 
+Before promoting a worker composition, run the repository worker-runtime drill
+against the same Redis and Core build. It uses the bounded
+`govoplan.worker.acceptance` task and records publish/consume, retry, warm
+SIGTERM, and worker-loss redelivery evidence without accessing tenant data.
+Production evidence must use the deployed queue configuration and a visibility
+timeout that is longer than every supported business task.
+
 ### Storage
 
 | Setting | Default | Notes |

src/govoplan_core/celery_app.py

@@ -4,6 +4,7 @@ from datetime import datetime, timedelta, timezone
 from importlib.metadata import PackageNotFoundError, version
 import logging
 import os
+import time
 
 from celery import Celery
 from celery.signals import (
@@ -110,6 +111,14 @@ celery.conf.update(
     worker_prefetch_multiplier=1,
     task_acks_late=True,
     task_reject_on_worker_lost=True,
+    task_track_started=True,
+    broker_transport_options={
+        "visibility_timeout": settings.celery_visibility_timeout_seconds,
+    },
+    result_backend_transport_options={
+        "visibility_timeout": settings.celery_visibility_timeout_seconds,
+    },
+    visibility_timeout=settings.celery_visibility_timeout_seconds,
     beat_schedule={
         "calendar-outbox-every-minute": {
             "task": "govoplan.calendar.dispatch_outbox",
@@ -308,6 +317,69 @@ def ping():
     return "pong"
 
 
+@celery.task(
+    bind=True,
+    name="govoplan.worker.acceptance",
+    max_retries=1,
+    acks_late=True,
+    reject_on_worker_lost=True,
+    track_started=True,
+)
+def worker_acceptance_probe(
+    task,
+    probe_id: str,
+    *,
+    mode: str = "complete",
+    delay_seconds: float = 0.0,
+    track_delivery: bool = False,
+):
+    """Exercise broker delivery semantics without touching business data."""
+
+    if mode not in {"complete", "retry_once"}:
+        raise ValueError(f"Unsupported worker acceptance mode: {mode}")
+    normalized_probe_id = str(probe_id).strip()
+    if not normalized_probe_id or len(normalized_probe_id) > 120:
+        raise ValueError("Worker acceptance probe_id must contain 1-120 characters")
+    bounded_delay = max(0.0, min(float(delay_seconds), 120.0))
+    delivery_count = _worker_acceptance_delivery_count(
+        task,
+        normalized_probe_id,
+    ) if track_delivery else None
+    task.update_state(
+        state="PROGRESS",
+        meta={
+            "phase": "started",
+            "probe_id": normalized_probe_id,
+            "retries": int(task.request.retries or 0),
+            "delivery_count": delivery_count,
+        },
+    )
+    if mode == "retry_once" and int(task.request.retries or 0) == 0:
+        raise task.retry(countdown=0.2)
+    if bounded_delay:
+        time.sleep(bounded_delay)
+    delivery_info = task.request.delivery_info or {}
+    return {
+        "probe_id": normalized_probe_id,
+        "mode": mode,
+        "retries": int(task.request.retries or 0),
+        "redelivered": bool(delivery_info.get("redelivered")),
+        "delivery_count": delivery_count,
+    }
+
+
+def _worker_acceptance_delivery_count(task, probe_id: str) -> int:
+    client = getattr(task.backend, "client", None)
+    if client is None:
+        raise RuntimeError(
+            "Worker acceptance delivery tracking requires a Redis result backend"
+        )
+    key = f"govoplan:worker-acceptance:{probe_id}"
+    count = int(client.incr(key))
+    client.expire(key, 60 * 60)
+    return count
+
+
 def _platform_registry() -> PlatformRegistry:
     raw_enabled_modules = load_startup_enabled_modules(settings.enabled_modules)
     candidate_modules = startup_candidate_module_ids(

src/govoplan_core/settings.py

@@ -115,6 +115,12 @@ class Settings(BaseSettings):
     migration_track: str = Field(default="release", alias="GOVOPLAN_MIGRATION_TRACK")
     redis_url: str = Field(default="redis://redis:6379/0", alias="REDIS_URL")
     celery_enabled: bool = Field(default=False, alias="CELERY_ENABLED")
+    celery_visibility_timeout_seconds: int = Field(
+        default=3600,
+        ge=30,
+        le=7 * 24 * 60 * 60,
+        alias="CELERY_VISIBILITY_TIMEOUT_SECONDS",
+    )
 
     s3_endpoint_url: str = Field(default="http://garage:3900", alias="S3_ENDPOINT_URL")
     s3_region: str = Field(default="garage", alias="S3_REGION")

tests/test_celery_queue_contract.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
 
 import unittest
+from unittest.mock import patch
 
-from govoplan_core.celery_app import celery
+from govoplan_core.celery_app import celery, worker_acceptance_probe
 from govoplan_core.settings import settings
 
 
@@ -26,9 +27,33 @@ class CeleryQueueContractTests(unittest.TestCase):
     def test_delivery_tasks_keep_worker_loss_protection(self) -> None:
         self.assertTrue(celery.conf.task_acks_late)
         self.assertTrue(celery.conf.task_reject_on_worker_lost)
+        self.assertTrue(celery.conf.task_track_started)
         self.assertEqual(1, celery.conf.worker_prefetch_multiplier)
+        self.assertEqual(
+            settings.celery_visibility_timeout_seconds,
+            celery.conf.broker_transport_options["visibility_timeout"],
+        )
+
+    def test_worker_acceptance_probe_is_bounded_and_side_effect_free(self) -> None:
+        with patch.object(worker_acceptance_probe, "update_state") as update_state:
+            result = worker_acceptance_probe.run(
+                "probe-1",
+                mode="complete",
+                delay_seconds=0,
+            )
+
+        self.assertEqual(
+            {
+                "probe_id": "probe-1",
+                "mode": "complete",
+                "retries": 0,
+                "redelivered": False,
+                "delivery_count": None,
+            },
+            result,
+        )
+        update_state.assert_called_once()
 
 
 if __name__ == "__main__":
     unittest.main()
-