async def start_campaign(db: AsyncSession, campaign: Campaign) -> dict:
    """Enqueue one job per pending campaign_recipient. Real Celery tasks."""
    if campaign.status not in ("draft", "queued", "paused"):
        return {"ok": False, "reason": f"campaign is {campaign.status}"}
    rows = (await db.execute(
        select(CampaignRecipient)
        .where(CampaignRecipient.campaign_id == campaign.id,
               CampaignRecipient.status.in_(("pending", "queued", "temp_failed")))
        .order_by(CampaignRecipient.seq))).scalars().all()
    if not rows:
        return {"ok": False, "reason": "no pending recipients"}
    campaign.status = "running"
    campaign.started_at = campaign.started_at or datetime.now(timezone.utc)
    campaign.finished_at = None
    campaign.queued_count = len(rows)
    jobs = []
    for cr in rows:
        job = SendJob(campaign_id=campaign.id, campaign_recipient_id=cr.id,
                      status="queued", max_attempts=campaign.max_retries + 1)
        jobs.append(job)
        cr.status = "queued"
    db.add_all(jobs)
    await db.commit()

    # Enqueue with per-job stagger honoring send interval
    interval = float(campaign.send_interval_seconds or 1.0)
    for i, (job, cr) in enumerate(zip(jobs, rows)):
        send_one.apply_async(args=[campaign.id, cr.id],
                             countdown=int(i * interval) if interval > 0 else 0)
    return {"ok": True, "queued": len(rows)}


async def pause_campaign(db: AsyncSession, campaign: Campaign) -> dict:
    campaign.status = "paused"
    await db.commit()
    return {"ok": True, "status": "paused"}


async def resume_campaign(db: AsyncSession, campaign: Campaign) -> dict:
    return await start_campaign(db, campaign)


async def cancel_campaign(db: AsyncSession, campaign: Campaign) -> dict:
    campaign.status = "cancelled"
    campaign.finished_at = datetime.now(timezone.utc)
    await db.commit()
    return {"ok": True, "status": "cancelled"}