Staging

apps/worker/tests/contract/test_worker_shutdown_contract.py

@@ -3,6 +3,9 @@
 import json
 from pathlib import Path
 
+import pytest
+from pydantic import ValidationError
+
 
 def test_should_preserve_fargate_worker_sigterm_shutdown_contract(
     worker_contract_environment: None,
@@ -33,3 +36,40 @@ def test_should_preserve_fargate_worker_sigterm_shutdown_contract(
     assert celery_app.conf.worker_enable_soft_shutdown_on_idle is True
     assert "REMAP_SIGTERM" not in environment_values
     assert worker_container["stopTimeout"] == 120
+
+
+def test_should_redeliver_interrupted_tasks_before_processing_jobs_expire(
+    worker_contract_environment: None,
+) -> None:
+    from shared.core.celery_app import celery_app
+    from shared.core.config.job import JobConfig
+
+    task_time_limit_seconds: int = celery_app.conf.task_time_limit
+    visibility_timeout_seconds: int = celery_app.conf.broker_transport_options[
+        "visibility_timeout"
+    ]
+    processing_expiry_seconds: int = JobConfig().JOB_PROCESSING_EXPIRE_SECONDS
+
+    assert task_time_limit_seconds == 3600
+    assert visibility_timeout_seconds == 4500
+    assert processing_expiry_seconds == 14400
+    assert (
+        task_time_limit_seconds
+        < visibility_timeout_seconds
+        < processing_expiry_seconds
+    )
+
+
+def test_should_reject_processing_expiry_before_interrupted_task_redelivery(
+    monkeypatch: pytest.MonkeyPatch,
+    worker_contract_environment: None,
+) -> None:
+    from shared.core.config import AppConfig
+
+    monkeypatch.setenv("JOB_PROCESSING_EXPIRE_SECONDS", "4500")
+
+    with pytest.raises(
+        ValidationError,
+        match="TASK_TIME_LIMIT_SECONDS < BROKER_VISIBILITY_TIMEOUT_SECONDS",
+    ):
+        AppConfig()

packages/shared-python/shared/core/celery_app.py

@@ -55,7 +55,7 @@ def get_unique_node_name() -> str:
     enable_utc=True,
     # Task execution
     task_track_started=True,
-    task_time_limit=3600,  # 1 hour hard limit
+    task_time_limit=app_config.TASK_TIME_LIMIT_SECONDS,
     task_soft_time_limit=3300,  # 55 minutes soft limit
     worker_prefetch_multiplier=1,
     task_acks_late=True,
@@ -77,7 +77,7 @@ def get_unique_node_name() -> str:
     broker_pool_limit=app_config.BROKER_POOL_LIMIT,
     # Redis transport — visibility_timeout must exceed task_time_limit
     broker_transport_options={
-        "visibility_timeout": 43200,  # 12 hours
+        "visibility_timeout": app_config.BROKER_VISIBILITY_TIMEOUT_SECONDS,
         "retry_on_timeout": True,
         # Keep all Kombu Redis broker keys in one Redis Cluster hash slot.
         "global_keyprefix": "{celery}",

packages/shared-python/shared/core/config/app.py

@@ -2,6 +2,7 @@
 
 from typing import Callable, cast
 
+from pydantic import model_validator
 from pydantic_settings import SettingsConfigDict
 
 from .ai import AIConfig
@@ -32,6 +33,22 @@ class AppConfig(
 ):
     """Application configuration — all config components merged."""
 
+    @model_validator(mode="after")
+    def validate_worker_redelivery_window(self) -> "AppConfig":
+        """Ensure interrupted tasks are redelivered before stale-job expiry."""
+        if not (
+            self.TASK_TIME_LIMIT_SECONDS
+            < self.BROKER_VISIBILITY_TIMEOUT_SECONDS
+            < self.JOB_PROCESSING_EXPIRE_SECONDS
+        ):
+            raise ValueError(
+                "Worker timing must satisfy TASK_TIME_LIMIT_SECONDS "
+                "< BROKER_VISIBILITY_TIMEOUT_SECONDS "
+                "< JOB_PROCESSING_EXPIRE_SECONDS"
+            )
+
+        return self
+
     def validate_all(self) -> bool:
         """Validate the combined application configuration."""
         validations = [

packages/shared-python/shared/core/config/celery.py

@@ -2,14 +2,17 @@
 Celery configuration — Redis-backed broker and result backend.
 """
 
-from typing import Dict
+from typing import ClassVar, Dict
 
 from pydantic import BaseModel, Field
 
 
 class CeleryConfig(BaseModel):
     """Celery configuration backed by a dedicated Redis instance."""
 
+    TASK_TIME_LIMIT_SECONDS: ClassVar[int] = 3600
+    BROKER_VISIBILITY_TIMEOUT_SECONDS: ClassVar[int] = 4500
+
     # Dedicated Redis instance for Celery broker / result backend / RedBeat.
     # Separate from the application Redis (REDIS_*) to isolate connection pools.
     CELERY_REDIS_URL: str = Field(