    fr.company_phone = "555-0100"
    fr.customer_email = str(payload.to_email)
    fr.custom_fields = {}

    engine = VariableEngine({})
    g = GeneratedValues()
    if payload.use_random_variables:
        g.invoice_number = engine.generate_invoice_number(set())
        g.amount = engine.generate_amount()
        g.shipping = engine.generate_shipping()
        g.invoice_date = "2025-01-15"
        g.due_date = "2025-01-30"
    unsub_url = f"{cfg.BASE_URL}/unsubscribe/test-token"
    res_html = engine.substitute(body_html or "", fr, g, custom_fields={},
                                 extra={"unsubscribe_link": unsub_url,
                                        "company_phone": fr.company_phone}).text
    res_text = engine.substitute(body_text or "", fr, g, custom_fields={},
                                 extra={"unsubscribe_link": unsub_url,
                                        "company_phone": fr.company_phone}).text
    subject = engine.substitute(payload.subject, fr, g).text

    attachments = []
    if payload.include_pdf and body_html:
        pdf = await asyncio.get_event_loop().run_in_executor(
            None, message_builder.html_to_pdf, res_html)
        if pdf:
            attachments.append({"filename": "Test_Preview.pdf", "content": pdf,
                                "mime_type": "application/pdf"})

    mime = message_builder.build_mime_message(
        from_email=account.email,
        to_email=str(payload.to_email),
        subject=subject,
        html_body=res_html,
        text_body=res_text,
        attachments=attachments,
        unsubscribe_url=unsub_url,
    )

    # REAL send in a worker thread with a sync DB session (token refresh writes)
    def _real_send():
        with SyncSession() as s:
            acc = s.get(GmailAccount, account.id)
            token = s.query(OAuthToken).filter_by(account_id=account.id).first()
            if not token:
                return {"ok": False, "error": "No OAuth token stored for this account",
                        "error_class": "auth"}
            client = build_gmail_client_from_account(acc, token)
            if client.method == "smtp":
                return client.send_mime_smtp(mime)
            return client.send_mime_api(mime)

    try:
        result = await asyncio.get_event_loop().run_in_executor(None, _real_send)
    except Exception as e:
        await log_audit(db, user.id, "test_send.error", "gmail_account", account.id,
                        {"error": str(e)[:200]})
        return {"ok": False, "stage": "temp_failed", "detail": str(e)[:400],
                "error_class": "unknown", "elapsed_ms": int((time.monotonic() - t0) * 1000)}

    elapsed = int((time.monotonic() - t0) * 1000)
    if result.get("ok"):
        await log_audit(db, user.id, "test_send.success", "gmail_account", account.id,
                        {"to": str(payload.to_email), "message_id": result.get("message_id")})
        return {"ok": True, "stage": "accepted", "detail": "Gmail accepted the message",
                "message_id": result.get("message_id"),
                "thread_id": result.get("thread_id"),
                "method": client_method(account),
                "provider_response": _clip(str(result.get("raw_response", ""))),
                "elapsed_ms": elapsed}
    # provider refused — surface the REAL error, never fake success
    err = str(result.get("error", "") or result.get("detail", ""))
    await log_audit(db, user.id, "test_send.failed", "gmail_account", account.id,
                    {"to": str(payload.to_email), "error": err[:200]})
    return {"ok": False, "stage": "perm_failed", "detail": err[:400],
            "error_class": result.get("error_class", "unknown"),
            "method": client_method(account),
            "provider_response": _clip(str(result.get("raw_response", ""))),
            "elapsed_ms": elapsed}


def _clip(s: str) -> str:
    return s[:300]


def client_method(account: GmailAccount) -> str:
    return account.sending_method or "api"


@router.get("/accounts")
async def test_accounts(db: AsyncSession = Depends(get_db),
                        user: User = Depends(get_current_user)):
    """Authorized accounts available for test sends."""
    rows = (await db.execute(
        select(GmailAccount).where(GmailAccount.user_id == user.id,
                                   GmailAccount.status == "authorized",
                                   GmailAccount.is_active == True))).scalars().all()  # noqa: E712
    return {"accounts": [{"id": a.id, "email": a.email,
                          "sending_method": a.sending_method} for a in rows]}


@router.get("/templates")
async def test_templates(db: AsyncSession = Depends(get_db),
                         user: User = Depends(get_current_user)):
    rows = (await db.execute(
        select(EmailTemplate).where(EmailTemplate.user_id == user.id))).scalars().all()
    return {"templates": [{"id": t.id, "name": t.name} for t in rows]}
=====CAMP_LIST=====
@router.get("/list")
async def list_campaigns(status: str = Query(""),
                         search: str = Query(""),
                         page: int = Query(1, ge=1),
                         page_size: int = Query(20, ge=1, le=100),
                         db: AsyncSession = Depends(get_db),
                         user: User = Depends(get_current_user)):
    q = select(Campaign).where(Campaign.user_id == user.id)
    if status:
        q = q.where(Campaign.status == status)
    if search:
        q = q.where(Campaign.name.ilike(f"%{search}%"))
    total = (await db.execute(select(func.count()).select_from(q.subquery()))).scalar() or 0
    rows = (await db.execute(q.order_by(Campaign.id.desc())
                             .offset((page - 1) * page_size).limit(page_size))).scalars().all()
    return {"campaigns": [_ser(c) for c in rows], "total": total,
            "page": page, "page_size": page_size}


@router.post("/create")
=====STREAM=====
@router.get("/{cid}/progress")
async def get_progress(cid: int, user: User = Depends(get_current_user),
                       db: AsyncSession = Depends(get_db)):
    c = await _get_own(db, user, cid)
    return await _progress_payload(db, c)


@router.get("/{cid}/stream")
async def campaign_stream(cid: int, request: Request,
                          user: User = Depends(get_current_user),
                          db: AsyncSession = Depends(get_db)):
    """SSE live progress stream. Manual implementation (no sse-starlette): a Redis
    pub/sub subscription on campaign:{id}:progress plus periodic DB polling fallback.
    Emits event: progress with counters; event: done when campaign finishes."""
    c = await _get_own(db, user, cid)

    import redis.asyncio as aioredis
    from app.db.session import AsyncSessionLocal

    async def event_gen():
        r = aioredis.Redis.from_url(cfg.REDIS_URL, decode_responses=True)
        pubsub = r.pubsub()
        channel = f"campaign:{cid}:progress"
        await pubsub.subscribe(channel)
        heartbeat = 2.0
        poll_interval = 5.0
        last_payload = None
        last_poll = 0.0
        import time
        try:
            yield "event: connected\ndata: {\"ok\": true}\n\n"
            while True:
                if await request.is_disconnected():
                    break
                # check for pub/sub tick (non-blocking)
                msg = await pubsub.get_message(ignore_subscribe_messages=True, timeout=0.1)
                now = time.monotonic()
                payload = None
                if msg is not None or now - last_poll >= poll_interval:
                    # refresh counters from DB with a fresh session
                    async with AsyncSessionLocal() as s:
                        camp = await s.get(Campaign, cid)
                        if camp:
                            payload = await _progress_payload(s, camp)
                    last_poll = now
                if payload is None and msg is None and now - last_poll < poll_interval:
                    await asyncio.sleep(heartbeat)
                    yield ": keepalive\n\n"
                    continue
                if payload:
                    data = json.dumps(payload)
                    if data != last_payload or msg is not None:
                        yield f"event: progress\ndata: {data}\n\n"
                        last_payload = data
                    if payload.get("status") in ("completed", "cancelled", "failed"):
                        yield f"event: done\ndata: {json.dumps(payload)}\n\n"
                        break
                await asyncio.sleep(0.2)
        finally:
            try:
                await pubsub.unsubscribe(channel)
                await pubsub.close()
                await r.aclose()
            except Exception:
                pass

    return StreamingResponse(event_gen(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache",
                                      "X-Accel-Buffering": "no",
                                      "Connection": "keep-alive"})


@router.get("/{cid}/recipients")
async def campaign_recipients(cid: int, status: str = Query(""),
                              search: str = Query(""),
                              page: int = Query(1, ge=1),
                              page_size: int = Query(50, ge=1, le=200),
                              db: AsyncSession = Depends(get_db),
                              user: User = Depends(get_current_user)):
    """Monitor table: per-recipient status, assigned account, subject/template, errors."""
    c = await _get_own(db, user, cid)
    q = select(CampaignRecipient).where(CampaignRecipient.campaign_id == cid)
    if status:
        q = q.where(CampaignRecipient.status == status)
    if search:
        q = q.where(CampaignRecipient.email.ilike(f"%{search}%"))
    total = (await db.execute(select(func.count()).select_from(q.subquery()))).scalar() or 0
    rows = (await db.execute(q.order_by(CampaignRecipient.seq)
                             .offset((page - 1) * page_size).limit(page_size))).scalars().all()

    async def names():
        acc_map, subj_map, tpl_map = {}, {}, {}
        for cr in rows:
            if cr.assigned_account_id and cr.assigned_account_id not in acc_map:
                a = await db.get(GmailAccount, cr.assigned_account_id)
                acc_map[cr.assigned_account_id] = a.email if a else "?"
            if cr.generated_subject_id and cr.generated_subject_id not in subj_map:
                s = await db.get(SubjectLine, cr.generated_subject_id)
                subj_map[cr.generated_subject_id] = (s.subject[:80] if s else "?")
            if cr.generated_template_id and cr.generated_template_id not in tpl_map:
                t = await db.get(EmailTemplate, cr.generated_template_id)
                tpl_map[cr.generated_template_id] = (t.name if t else "?")
        return acc_map, subj_map, tpl_map

    acc_map, subj_map, tpl_map = await names()
    out = []
    for cr in rows:
        out.append({
            "id": cr.id, "seq": cr.seq, "email": cr.email, "status": cr.status,
            "account": acc_map.get(cr.assigned_account_id, ""),
            "subject": subj_map.get(cr.generated_subject_id, ""),
            "template": tpl_map.get(cr.generated_template_id, ""),
            "invoice_number": cr.generated_invoice_number,
            "amount": cr.generated_amount, "shipping": cr.generated_shipping,
            "attempts": cr.attempt_count, "message_id": cr.msg_id,
            "thread_id": cr.thread_id,
            "sent_at": cr.sent_at.isoformat() if cr.sent_at else None,
            "last_error": (cr.last_error or "")[:200],
        })
    return {"recipients": out, "total": total, "page": page, "page_size": page_size}


@router.get("/{cid}/export")
async def export_campaign(cid: int, db: AsyncSession = Depends(get_db),
                          user: User = Depends(get_current_user)):
    """CSV export of campaign results (one row per campaign recipient)."""
    c = await _get_own(db, user, cid)
