@celery_app.task(bind=True, base=SendTask, name="app.tasks.send_tasks.send_one",
                 ignore_result=False)
def send_one(self, campaign_id: int, campaign_recipient_id: int):
    """Process one recipient: build message, send through Gmail, record real outcome."""
    with SyncSession() as db:
        campaign = db.get(Campaign, campaign_id)
        if campaign is None:
            return {"ok": False, "status": "cancelled", "reason": "campaign deleted"}
        if campaign.status in ("paused", "cancelled", "completed"):
            return {"ok": False, "status": "cancelled", "reason": f"campaign {campaign.status}"}

        cr = db.get(CampaignRecipient, campaign_recipient_id)
        if cr is None:
            return {"ok": False, "status": "cancelled", "reason": "campaign_recipient deleted"}
        if cr.status in ("sent", "suppressed", "perm_failed"):
            return {"ok": True, "status": cr.status, "reason": "already finalized"}

        job = (db.execute(select(SendJob)
                          .where(SendJob.campaign_recipient_id == cr.id,
                                 SendJob.campaign_id == campaign.id))).scalar_one_or_none()

        # Re-check suppression at send time (recipient may have unsubscribed since enqueue)
        sup = (db.execute(select(SuppressionEntry)
                          .where(SuppressionEntry.email == cr.email))).scalar_one_or_none()
        if sup:
            cr.status = "suppressed"
            campaign.suppressed_count += 1
            cr.last_error = "Suppressed at send time"
            db.add(EmailEvent(campaign_id=campaign.id, campaign_recipient_id=cr.id,
                              event_type="suppressed", detail="Suppressed at send time"))
            db.commit()
            _publish_progress(campaign.id)
            return {"ok": True, "status": "suppressed", "reason": "suppression hit"}

        account = db.get(GmailAccount, cr.assigned_account_id) if cr.assigned_account_id else None
        if account is None or not account.is_active or account.status != "authorized":
            cr.status = "skipped"
            campaign.skipped_count += 1
            cr.last_error = "Assigned account unavailable/unauthorized"
            db.commit()
            _publish_progress(campaign.id)
            return {"ok": False, "status": "skipped", "reason": "account unavailable"}

        # Per-account daily cap (never exceed provider limits)
        today = now().date()
        if account.last_send_date and account.last_send_date.date() == today \
                and account.sends_today >= account.daily_limit:
            campaign.status = "paused"
            db.commit()
            _publish_control(campaign.id, "pause",
                             f"Account {account.email} reached daily limit ({account.daily_limit})")
            return {"ok": False, "status": "paused", "reason": "daily limit reached"}

        if account.cooldown_until and account.cooldown_until > now():
            _reschedule(db, self, campaign.id, cr, job, account.cooldown_until - now(),
                        "account cooling down")
            return {"ok": False, "status": "rescheduled", "reason": "cooldown"}

        # Per-account rate window
        interval = float(campaign.send_interval_seconds or 1.0)
        ok, wait = _rate_window(f"acct:{account.id}", interval)
        if not ok:
            _reschedule(db, self, campaign.id, cr, job, timedelta(seconds=wait),
                        "rate limit window")
            return {"ok": False, "status": "rescheduled", "reason": "rate window"}

        # ------------------------------------------------------------- build message
        recipient = db.get(Recipient, cr.recipient_id)
        subject_row = db.get(SubjectLine, cr.generated_subject_id) if cr.generated_subject_id else None
        template = db.get(EmailTemplate, cr.generated_template_id) if cr.generated_template_id else None
        subject = subject_row.subject if subject_row else campaign.name
        if template is None:
            cr.status = "perm_failed"
            cr.last_error = "No template assigned"
            campaign.failed_count += 1
            db.commit()
            return {"ok": False, "status": "perm_failed", "reason": "no template"}

        var_cfg = dict(campaign.variable_config or {})
        engine = VariableEngine(var_cfg)
        _today = date.today()
        g = GeneratedValues(
            invoice_number=cr.generated_invoice_number,
            amount=cr.generated_amount,
            shipping=cr.generated_shipping,
            random_name=engine.generate_random_name(),
            invoice_date=var_cfg.get("invoice_date") or _today.isoformat(),
            due_date=var_cfg.get("due_date")
                      or (_effective_invoice_date(var_cfg.get("invoice_date") or _today.isoformat())
                          + timedelta(days=int(var_cfg.get("due_days", 14)))).isoformat(),
            phone=cr.generated_phone or var_cfg.get("company_phone", ""),
        )
        unsub_url = _unsubscribe_url(recipient, campaign)
        rec = _recipient_dict(recipient)
        cf = (recipient.custom_fields if recipient else None) or {}
        result_html = engine.substitute(template.body_html, rec, g, custom_fields=cf,
                                        extra={"unsubscribe_link": unsub_url}).text
        result_text = engine.substitute(template.body_text or _plain_fallback(result_html),
                                        rec, g, custom_fields=cf,
                                        extra={"unsubscribe_link": unsub_url}).text

        attachments, att_names = [], []