services/api/
├── core/
│   ├── algorithms/                    ← Phase 0: Shared Primitives
│   │   ├── __init__.py
│   │   ├── sliding_window.py          (10 KB) Redis sorted-set based
│   │   ├── circuit_breaker.py         (16 KB) 3-state machine
│   │   └── retry_budget.py            (12 KB) Lua-scripted tokens
│   └── health/                        ← Phase 4: ROLLBACKX
│       ├── __init__.py
│       ├── service_monitor.py         (12 KB) Multi-layer health pings
│       └── blast_radius.py            (8 KB)  BFS + topological sort
├── scheduler/
│   └── services/
│       ├── cortex_retry.py            ← Phase 1: CORTEX (24 KB)
│       ├── cascade_engine.py          ← Phase 2: CASCADE (19 KB)
│       └── reflux_recovery.py         ← Phase 3: REFLUX (14 KB)
├── campaigns/
│   └── services/
│       └── allocation_engine.py       ← Phase 5: RAPID (18 KB)
└── tests/                             ← Phase 6: 47 Tests
    ├── conftest.py                    Shared fixtures (mock_redis, mock_job)
    ├── algorithms/                    Shared primitives tests
    │   ├── test_sliding_window.py     (6 tests)
    │   └── test_circuit_breaker.py    (8 tests)
    ├── engines/                       Algorithm engine tests
    │   ├── test_cortex_retry.py       (10 tests)
    │   ├── test_cascade_engine.py     (8 tests)
    │   ├── test_reflux_recovery.py    (5 tests)
    │   └── test_allocation_engine.py  (5 tests)
    └── health/                        Health monitoring tests
        └── test_service_monitor.py    (5 tests)

CPS = 0.30·F + 0.20·L + 0.15·Q + 0.15·B + 0.20·C

F = failure_rate        (SlidingWindow)     0.0 - 1.0
L = latency_deviation   (SlidingWindow)     0.0 - 1.0
Q = queue_pressure      (DB query)          0.0 - 1.0
B = budget_burn_rate    (RetryBudget)       0.0 - 1.0
C = circuit_state       (CircuitBreaker)    0.0 / 0.5 / 1.0

from scheduler.services.cortex_retry import CortexRetryEngine

engine = CortexRetryEngine(redis_client=redis_conn)
decision = engine.should_retry(job, attempt=2)
# decision.action: 'RETRY' | 'REJECT'
# decision.delay_seconds: CPS-modulated delay
# decision.reason: human-readable explanation

Step 1: _is_duplicate()     → Redis SISMEMBER (24h TTL)
Step 2: _check_quota()      → Tenant monthly_message_limit
Step 3: _check_causality()  → Version counter comparison
Step 3b: _resolve_conflict() → Trust-scored source reliability
Step 4: _apply()            → Mark seen + increment version

from scheduler.services.cascade_engine import CascadeEngine, DeliveryEvent

cascade = CascadeEngine(redis_client=redis_conn)
event = DeliveryEvent(
    event_id=f'{job_id}:{recipient_id}:{attempt}',
    recipient_id=str(recipient.id),
    tenant_id=str(tenant.id),
    action='SEND',
    source='CELERY_TASK',
)
result = cascade.process(event)
# result.action: 'APPLIED' | 'DUPLICATE_REJECTED' | 'QUOTA_EXCEEDED' | ...
# result.accepted: bool

1. Classify recipients → SENT (skip) vs FAILED/PENDING (recover)
2. Cross-reference webhook delivery receipts
3. Build recovery plan with reduction metrics
4. Execute: reset only needed recipients, re-queue job

from scheduler.services.reflux_recovery import SelectiveRecoveryEngine

reflux = SelectiveRecoveryEngine()
plan = reflux.recover(job)
# plan.total_recipients: 100
# plan.skip_count: 70       ← already sent
# plan.recovery_count: 30   ← needs retry
# plan.reduction_percent: 70.0

if plan.has_work:
    reflux.execute_recovery(plan)

POST /api/scheduler/jobs/{id}/smart-retry/

Response:
{
    "message": "REFLUX selective retry scheduled — 70% reduction",
    "plan": {
        "total_recipients": 100,
        "skip_count": 70,
        "recovery_count": 30,
        "reduction_percent": 70.0,
        "webhook_confirmed": 5
    }
}

api → [redis, postgres]
celery_worker → [redis, postgres]
celery_beat → [redis]
websocket → [redis]
whatsapp_webhook → [api, redis]

Failed: {redis}
Blast radius: {api, celery_worker, celery_beat, websocket, whatsapp_webhook}
Recovery order: [redis, celery_beat, websocket, celery_worker, api, whatsapp_webhook]

from core.health import ServiceHealthMonitor

monitor = ServiceHealthMonitor()
report = monitor.check_health()
# report.overall_status: 'HEALTHY' | 'DEGRADED' | 'CRITICAL'
# report.blast_radius: ['api', 'worker', ...]
# report.recovery_order: ['redis', 'beat', ...]

GET /health/?deep=true

Response:
{
    "overall_status": "HEALTHY",
    "services": [...],
    "blast_radius": [],
    "recovery_order": []
}

Phase 1: Propagation  → eliminate workers above 85% load
Phase 1b: Detection   → identify forced assignments (single worker)
Phase 2: Solve        → greedy least-loaded assignment
Phase 3: Evaluate     → compute load variance + metrics

from campaigns.services.allocation_engine import CampaignAllocator

allocator = CampaignAllocator()
result = allocator.allocate(total_recipients=50000, batch_size=1000)
# result.total_batches: 50
# result.workers_used: 4
# result.forced_count: 0
# result.load_variance: 0.0012
# result.solve_time_ms: 2.5

for assignment in result.assignments:
    print(f"Batch {assignment.batch_id} → {assignment.worker}")