update

backend/app/database.py

@@ -267,6 +267,57 @@ class LinkedinCampaign(Base):
     executed_at = Column(DateTime, nullable=True)
 
 
+class FileOutreachExecution(Base):
+    """Server-side orchestration for wizard email+LinkedIn sequences (per uploaded CSV file)."""
+
+    __tablename__ = "file_outreach_executions"
+
+    id = Column(Integer, primary_key=True, index=True)
+    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True)
+    user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
+    file_id = Column(String, index=True, nullable=False)
+    status = Column(String, default="idle")  # idle | running | paused | completed
+    progress_done = Column(Integer, default=0)
+    progress_total = Column(Integer, default=0)
+    stats_json = Column(Text, nullable=True)
+    cursor_json = Column(Text, nullable=True)  # delivery_started_at, etc.
+    last_error = Column(Text, nullable=True)
+    linkedin_followup_interval_hours = Column(Integer, default=72)
+    created_at = Column(DateTime, default=datetime.utcnow)
+    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
+
+    __table_args__ = (UniqueConstraint("tenant_id", "file_id", name="uq_outreach_exec_tenant_file"),)
+
+
+class OutreachSendReceipt(Base):
+    """One row per (file, contact, sequence action slot) send attempt."""
+
+    __tablename__ = "outreach_send_receipts"
+
+    id = Column(Integer, primary_key=True, index=True)
+    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True)
+    file_id = Column(String, index=True, nullable=False)
+    contact_id = Column(Integer, ForeignKey("contacts.id"), nullable=False, index=True)
+    step_slot = Column(Integer, nullable=False)
+    channel = Column(String, nullable=False)
+    action = Column(String, default="")
+    status = Column(String, default="pending")  # sent | failed | skipped
+    detail = Column(Text, nullable=True)
+    unipile_message_id = Column(String, nullable=True)
+    unipile_tracking_id = Column(String, nullable=True, index=True)
+    smtp_message_id = Column(String, nullable=True, index=True)
+    tracking_label = Column(String(220), nullable=True, index=True)
+    opened_at = Column(DateTime, nullable=True)
+    open_count = Column(Integer, default=0)
+    replied_at = Column(DateTime, nullable=True)
+    bounced_at = Column(DateTime, nullable=True)
+    created_at = Column(DateTime, default=datetime.utcnow)
+
+    __table_args__ = (
+        UniqueConstraint("tenant_id", "file_id", "contact_id", "step_slot", name="uq_outreach_receipt_slot"),
+    )
+
+
 class UnipileHostedAuthState(Base):
     __tablename__ = "unipile_hosted_auth_states"
 
@@ -456,6 +507,36 @@ def run_migrations(connection_engine):
             if "linkedin_followup_next_email_number" not in ctcols:
                 conn.execute(text("ALTER TABLE contacts ADD COLUMN linkedin_followup_next_email_number INTEGER"))
 
+        insp = inspect(connection_engine)
+        if insp.has_table("file_outreach_executions"):
+            ecols = [c["name"] for c in insp.get_columns("file_outreach_executions")]
+            if "linkedin_followup_interval_hours" not in ecols:
+                conn.execute(
+                    text(
+                        "ALTER TABLE file_outreach_executions ADD COLUMN linkedin_followup_interval_hours INTEGER DEFAULT 72"
+                    )
+                )
+
+        insp = inspect(connection_engine)
+        if insp.has_table("outreach_send_receipts"):
+            rcols = [c["name"] for c in insp.get_columns("outreach_send_receipts")]
+            if "tracking_label" not in rcols:
+                conn.execute(text("ALTER TABLE outreach_send_receipts ADD COLUMN tracking_label TEXT"))
+            if "unipile_tracking_id" not in rcols:
+                conn.execute(text("ALTER TABLE outreach_send_receipts ADD COLUMN unipile_tracking_id TEXT"))
+            if "smtp_message_id" not in rcols:
+                conn.execute(text("ALTER TABLE outreach_send_receipts ADD COLUMN smtp_message_id TEXT"))
+            if "opened_at" not in rcols:
+                conn.execute(text("ALTER TABLE outreach_send_receipts ADD COLUMN opened_at DATETIME"))
+            if "open_count" not in rcols:
+                conn.execute(
+                    text("ALTER TABLE outreach_send_receipts ADD COLUMN open_count INTEGER DEFAULT 0")
+                )
+            if "replied_at" not in rcols:
+                conn.execute(text("ALTER TABLE outreach_send_receipts ADD COLUMN replied_at DATETIME"))
+            if "bounced_at" not in rcols:
+                conn.execute(text("ALTER TABLE outreach_send_receipts ADD COLUMN bounced_at DATETIME"))
+
 
 # Create tables then migrate legacy SQLite schemas
 Base.metadata.create_all(bind=engine)

backend/app/main.py

@@ -18,6 +18,7 @@ import json
 import asyncio
 import math
 import re
+import logging
 import requests
 from urllib.parse import urlparse, unquote
 from datetime import datetime, timedelta
@@ -39,6 +40,7 @@ from .database import (
     UnipileAccount,
     LinkedinCampaign,
     UnipileHostedAuthState,
+    OutreachSendReceipt,
 )
 from pydantic import ValidationError
 
@@ -72,6 +74,8 @@ from .smartlead_client import SmartleadClient
 from .auth_routes import router as auth_router
 from .tenant_deps import TenantContext, get_tenant_context
 from .tenant_routes import router as tenant_router
+from .outreach_routes import router as outreach_router
+from .outreach_routes import parse_tracking_label, refresh_outreach_execution_stats_for_file, normalize_smtp_message_id
 from .deal_revenue import build_quarterly_board
 
 app = FastAPI()
@@ -105,6 +109,7 @@ app.add_middleware(
 
 app.include_router(auth_router)
 app.include_router(tenant_router)
+app.include_router(outreach_router)
 
 # Create uploads directory
 UPLOAD_DIR = Path("/data/uploads")
@@ -113,6 +118,12 @@ UNIPILE_API_BASE = os.getenv("UNIPILE_API_BASE", "").rstrip("/")
 UNIPILE_API_KEY = os.getenv("UNIPILE_API_KEY", "")
 FRONTEND_ORIGIN = os.getenv("FRONTEND_ORIGIN", "").rstrip("/")
 UNIPILE_WEBHOOK_SECRET = os.getenv("UNIPILE_WEBHOOK_SECRET", "").strip()
+UNIPILE_WEBHOOK_LOG_EVENTS = os.getenv("UNIPILE_WEBHOOK_LOG_EVENTS", "").strip().lower() in (
+    "1",
+    "true",
+    "yes",
+)
+_unipile_webhook_log = logging.getLogger("emailout.unipile_webhook")
 
 # ---- API ----
 def _safe_str(value):
@@ -1682,12 +1693,26 @@ async def unipile_linkedin_hosted_callback(request: Request, db: Session = Depen
     return {"ok": True}
 
 
+def _parse_unipile_webhook_datetime(raw) -> Optional[datetime]:
+    if not raw or not isinstance(raw, str):
+        return None
+    try:
+        s = raw.strip().replace("Z", "")
+        return datetime.fromisoformat(s)
+    except Exception:
+        return None
+
+
 @app.post("/api/webhooks/unipile")
-async def unipile_users_webhook(request: Request, db: Session = Depends(get_db)):
+async def unipile_webhook(request: Request, db: Session = Depends(get_db)):
     """
-    Receive UniPile USERS webhook events (e.g. new_relation when an invitation is accepted).
-    Register `request_url` in UniPile to this path (POST /api/v1/webhooks, source: users).
-    See: https://developer.unipile.com/docs/detecting-accepted-invitations
+    UniPile webhooks (same URL for multiple UniPile registrations).
+
+    Supported ``event`` values:
+
+    * ``new_relation`` — LinkedIn invite accepted (source: users). Updates pending contacts.
+    * ``mail_opened``, ``mail_link_clicked`` — email tracking (source: ``email_tracking``).
+    * ``mail_received`` — mailbox email (source: ``email`` in UniPile API, “Email” / new-mail webhook in the dashboard). Replies matched via ``in_reply_to`` / ``tracking_id`` / ``label`` / RFC ``message_id``.
     """
     if UNIPILE_WEBHOOK_SECRET:
         auth_h = request.headers.get("Unipile-Auth") or request.headers.get("X-Unipile-Auth") or ""
@@ -1700,51 +1725,215 @@ async def unipile_users_webhook(request: Request, db: Session = Depends(get_db))
         raise HTTPException(status_code=400, detail="Expected JSON body")
 
     event = _safe_str((body or {}).get("event"))
-    if event != "new_relation":
-        return {"ok": True, "ignored": True, "reason": "unsupported_event"}
+    if UNIPILE_WEBHOOK_LOG_EVENTS and isinstance(body, dict):
+        ir = body.get("in_reply_to")
+        _unipile_webhook_log.info(
+            "UniPile webhook event=%s role=%s keys=%s in_reply_to_keys=%s",
+            event,
+            body.get("role"),
+            sorted(body.keys()),
+            sorted(ir.keys()) if isinstance(ir, dict) else None,
+        )
+    now = datetime.utcnow()
 
-    account_id = _safe_str((body or {}).get("account_id"))
-    provider_id = _safe_str((body or {}).get("user_provider_id"))
-    if not account_id or not provider_id:
-        return {"ok": True, "ignored": True, "reason": "missing_ids"}
+    if event == "new_relation":
+        account_id = _safe_str((body or {}).get("account_id"))
+        provider_id = _safe_str((body or {}).get("user_provider_id"))
+        if not account_id or not provider_id:
+            return {"ok": True, "ignored": True, "reason": "missing_ids"}
 
-    ua = (
-        db.query(UnipileAccount)
-        .filter(UnipileAccount.unipile_account_id == account_id)
-        .first()
-    )
-    if not ua:
-        return {"ok": True, "ignored": True, "reason": "unknown_account"}
+        ua = (
+            db.query(UnipileAccount)
+            .filter(UnipileAccount.unipile_account_id == account_id)
+            .first()
+        )
+        if not ua:
+            return {"ok": True, "ignored": True, "reason": "unknown_account"}
 
-    now = datetime.utcnow()
-    contacts = (
-        db.query(Contact)
-        .filter(
-            Contact.tenant_id == ua.tenant_id,
-            Contact.unipile_provider_id == provider_id,
-            Contact.source == "linkedin_campaign",
+        contacts = (
+            db.query(Contact)
+            .filter(
+                Contact.tenant_id == ua.tenant_id,
+                Contact.unipile_provider_id == provider_id,
+                Contact.source.in_(("apollo_csv", "linkedin_campaign")),
+            )
+            .all()
         )
-        .all()
-    )
-    updated = 0
-    for ct in contacts:
-        if ct.linkedin_connection_accepted_at:
-            continue
-        ct.linkedin_invite_pending = 0
-        ct.linkedin_connection_accepted_at = now
-        updated += 1
-    db.commit()
-    return {"ok": True, "updated_contacts": updated}
+        updated = 0
+        for ct in contacts:
+            if ct.linkedin_connection_accepted_at:
+                continue
+            ct.linkedin_invite_pending = 0
+            ct.linkedin_connection_accepted_at = now
+            updated += 1
+        db.commit()
+        return {"ok": True, "updated_contacts": updated, "event": event}
+
+    if event in ("mail_opened", "mail_link_clicked"):
+        label = _safe_str((body or {}).get("label"))
+        parsed = parse_tracking_label(label)
+        if not parsed:
+            return {"ok": True, "ignored": True, "reason": "no_tracking_label"}
+        tenant_id, file_id, contact_id, step_slot = parsed
+        rec = (
+            db.query(OutreachSendReceipt)
+            .filter(
+                OutreachSendReceipt.tenant_id == tenant_id,
+                OutreachSendReceipt.file_id == file_id,
+                OutreachSendReceipt.contact_id == contact_id,
+                OutreachSendReceipt.step_slot == step_slot,
+                OutreachSendReceipt.channel == "gmail",
+            )
+            .first()
+        )
+        if not rec:
+            return {"ok": True, "ignored": True, "reason": "receipt_not_found"}
+        when = _parse_unipile_webhook_datetime((body or {}).get("date")) or now
+        if not rec.opened_at:
+            rec.opened_at = when
+        try:
+            rec.open_count = int(rec.open_count or 0) + 1
+        except (TypeError, ValueError):
+            rec.open_count = 1
+        db.commit()
+        refresh_outreach_execution_stats_for_file(db, tenant_id, file_id)
+        db.commit()
+        return {"ok": True, "event": event, "updated_receipt_id": rec.id}
+
+    if event == "mail_received":
+        account_id = _safe_str((body or {}).get("account_id"))
+        role = _safe_str((body or {}).get("role"))
+        if role != "inbox":
+            return {"ok": True, "ignored": True, "reason": "not_inbox"}
+
+        ua = (
+            db.query(UnipileAccount)
+            .filter(UnipileAccount.unipile_account_id == account_id)
+            .first()
+        )
+        if not ua:
+            return {"ok": True, "ignored": True, "reason": "unknown_account"}
+
+        tenant_id = int(ua.tenant_id)
+        in_reply = (body or {}).get("in_reply_to") or {}
+        parent_id = _safe_str(in_reply.get("id"))
+        parent_rfc = normalize_smtp_message_id(_safe_str(in_reply.get("message_id")))
+        trk = _safe_str((body or {}).get("tracking_id"))
+        label = _safe_str((body or {}).get("label"))
+
+        rec = None
+        if parent_id:
+            rec = (
+                db.query(OutreachSendReceipt)
+                .filter(
+                    OutreachSendReceipt.tenant_id == tenant_id,
+                    OutreachSendReceipt.channel == "gmail",
+                    OutreachSendReceipt.unipile_message_id == parent_id,
+                )
+                .first()
+            )
+        if not rec and parent_rfc:
+            rec = (
+                db.query(OutreachSendReceipt)
+                .filter(
+                    OutreachSendReceipt.tenant_id == tenant_id,
+                    OutreachSendReceipt.channel == "gmail",
+                    OutreachSendReceipt.smtp_message_id == parent_rfc,
+                )
+                .first()
+            )
+        if not rec and trk:
+            rec = (
+                db.query(OutreachSendReceipt)
+                .filter(
+                    OutreachSendReceipt.tenant_id == tenant_id,
+                    OutreachSendReceipt.channel == "gmail",
+                    or_(
+                        OutreachSendReceipt.unipile_tracking_id == trk,
+                        OutreachSendReceipt.unipile_message_id == trk,
+                    ),
+                )
+                .first()
+            )
+        if not rec and label:
+            p2 = parse_tracking_label(label)
+            if p2:
+                tid2, fid2, cid2, slot2 = p2
+                if tid2 == tenant_id:
+                    rec = (
+                        db.query(OutreachSendReceipt)
+                        .filter(
+                            OutreachSendReceipt.tenant_id == tid2,
+                            OutreachSendReceipt.file_id == fid2,
+                            OutreachSendReceipt.contact_id == cid2,
+                            OutreachSendReceipt.step_slot == slot2,
+                            OutreachSendReceipt.channel == "gmail",
+                        )
+                        .first()
+                    )
+
+        folders = (body or {}).get("folders") or []
+        subj = (_safe_str((body or {}).get("subject")) or "").lower()
+        bounce_like = any("bounce" in str(f).lower() for f in folders if f) or "undeliverable" in subj
+        has_reply_parent = bool(in_reply and (in_reply.get("id") or in_reply.get("message_id")))
+
+        if rec:
+            when = _parse_unipile_webhook_datetime((body or {}).get("date")) or now
+            if bounce_like and not rec.bounced_at:
+                rec.bounced_at = when
+            elif has_reply_parent and not bounce_like and not rec.replied_at:
+                rec.replied_at = when
+            db.commit()
+            refresh_outreach_execution_stats_for_file(db, rec.tenant_id, rec.file_id)
+            db.commit()
+            return {"ok": True, "event": event, "updated_receipt_id": rec.id}
+
+        return {"ok": True, "ignored": True, "reason": "no_matching_receipt"}
+
+    return {"ok": True, "ignored": True, "reason": "unsupported_event", "event": event or ""}
 
 
 @app.get("/api/unipile/webhook-url-hint")
 async def unipile_webhook_url_hint(request: Request):
-    """Public URL your UniPile webhook should POST to (same host as this API)."""
+    """Public URL your UniPile webhooks should POST to (same host as this API)."""
     root = _public_origin_from_request(request)
+    post_url = f"{root}/api/webhooks/unipile"
     return {
-        "post_url": f"{root}/api/webhooks/unipile",
-        "unipile_dashboard": "Create webhook source `users` → event new_relation (default for USERS webhook).",
-        "optional_auth_header": "Unipile-Auth or X-Unipile-Auth matching UNIPILE_WEBHOOK_SECRET",
+        "post_url": post_url,
+        "auth_header": "Set Unipile-Auth or X-Unipile-Auth to UNIPILE_WEBHOOK_SECRET on each webhook registration.",
+        "registrations": [
+            {
+                "source": "users",
+                "events": ["new_relation"],
+                "doc": "https://developer.unipile.com/docs/detecting-accepted-invitations",
+            },
+            {
+                "source": "email_tracking",
+                "events": ["mail_opened", "mail_link_clicked"],
+                "doc": "https://developer.unipile.com/docs/tracking-email",
+            },
+            {
+                "source": "email",
+                "events": ["mail_received", "mail_sent", "mail_moved"],
+                "doc": "https://developer.unipile.com/docs/new-emails-webhook",
+                "dashboard_hint": "Create an **Email** (mailbox) webhook in the UniPile UI — not the **Messaging** (chat) webhook. Use the same callback URL.",
+                "note": "Reply attribution uses mail_received with role inbox and in_reply_to / tracking_id / label; enable mail_received for connected mail accounts.",
+            },
+            {
+                "source": "messaging",
+                "events": [
+                    "message_received",
+                    "message_read",
+                    "message_reaction",
+                    "message_edited",
+                    "message_deleted",
+                    "message_delivered",
+                ],
+                "doc": "https://developer.unipile.com/docs/new-messages-webhook",
+                "note": "LinkedIn/chat channels — not used for Gmail mail_received.",
+            },
+        ],
+        "debug_logging": "Set UNIPILE_WEBHOOK_LOG_EVENTS=1 to log each webhook's event name, role, and JSON keys (no body content).",
         "latency_note": "UniPile may deliver new_relation many hours after acceptance; LinkedIn has no real-time connection API.",
     }
 

backend/app/outreach_routes.py

@@ -0,0 +1,907 @@
+"""
+Orchestrate sending generated email + LinkedIn sequence content via UniPile for wizard campaigns (per file_id).
+
+Progress and receipts are persisted so the dashboard can poll execution state while a campaign is running.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from datetime import datetime, timedelta
+from typing import Any, Dict, List, Optional, Tuple
+
+import requests
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy import func, or_
+from sqlalchemy.orm import Session
+from urllib.parse import urlparse, unquote
+
+from .database import (
+    Contact,
+    FileOutreachExecution,
+    GeneratedSequence,
+    OutreachSendReceipt,
+    UnipileAccount,
+    UploadedFile,
+    User,
+)
+from .tenant_deps import TenantContext, get_tenant_context
+
+router = APIRouter(prefix="/api/campaign-files", tags=["campaign-files"])
+
+
+def _safe_str(value: Any) -> str:
+    if value is None:
+        return ""
+    return str(value).strip()
+
+
+def _pick_from_raw(raw_data: Dict, aliases: List[str]) -> str:
+    if not isinstance(raw_data, dict):
+        return ""
+    lowered = {str(k).strip().lower(): v for k, v in raw_data.items()}
+    for alias in aliases:
+        if alias in lowered:
+            return _safe_str(lowered[alias])
+    return ""
+
+
+def _pick_linkedin_url(raw_data: Dict) -> str:
+    return _pick_from_raw(
+        raw_data,
+        [
+            "linkedin",
+            "linkedin url",
+            "linkedin profile",
+            "person linkedin url",
+            "person linkedin profile url",
+            "linkedin profile url",
+            "linkedin public url",
+            "linkedin_url",
+            "linkedin profile link",
+        ],
+    )
+
+
+def _normalize_linkedin_url(raw: str) -> str:
+    s = _safe_str(raw)
+    if not s:
+        return ""
+    if s.startswith("http://"):
+        s = "https://" + s[len("http://") :]
+    return s
+
+
+def _linkedin_public_identifier(raw: str) -> str:
+    s = _normalize_linkedin_url(raw).strip()
+    if not s:
+        return ""
+    low = s.lower()
+    if "linkedin.com" in low:
+        try:
+            path = urlparse(s).path or ""
+        except Exception:
+            path = ""
+        path = unquote(path).strip("/")
+        parts = [p for p in path.split("/") if p]
+        for i, seg in enumerate(parts):
+            if seg.lower() == "in" and i + 1 < len(parts):
+                return parts[i + 1].split("?")[0].strip()
+            if seg.lower() == "sales" and i + 2 < len(parts) and parts[i + 1].lower() == "lead":
+                return parts[i + 2].split("?")[0].strip()
+        return ""
+    return s.split("?")[0].strip("/").strip()
+
+
+def _unipile_env() -> Tuple[str, str]:
+    base = os.getenv("UNIPILE_API_BASE", "").rstrip("/")
+    key = os.getenv("UNIPILE_API_KEY", "")
+    return base, key
+
+
+def _unipile_post_json(path: str, payload: dict) -> dict:
+    base, key = _unipile_env()
+    if not base or not key:
+        raise HTTPException(status_code=400, detail="UniPile is not configured (UNIPILE_API_BASE / UNIPILE_API_KEY).")
+    url = f"{base}{path}"
+    resp = requests.post(
+        url,
+        headers={
+            "X-API-KEY": key,
+            "accept": "application/json",
+            "content-type": "application/json",
+        },
+        json=payload,
+        timeout=60,
+    )
+    try:
+        data = resp.json()
+    except Exception:
+        data = {"raw": resp.text}
+    if resp.status_code >= 400:
+        msg = None
+        if isinstance(data, dict):
+            msg = data.get("message") or data.get("detail") or data.get("error")
+        raise HTTPException(
+            status_code=502,
+            detail=msg or f"UniPile error ({resp.status_code})",
+        )
+    return data if isinstance(data, dict) else {}
+
+
+def _unipile_call(method: str, path: str, payload: Optional[dict] = None) -> Tuple[int, Any]:
+    base, key = _unipile_env()
+    if not base or not key:
+        raise HTTPException(status_code=400, detail="UniPile is not configured.")
+    url = f"{base}{path}"
+    try:
+        resp = requests.request(
+            method=method.upper(),
+            url=url,
+            headers={
+                "X-API-KEY": key,
+                "accept": "application/json",
+                "content-type": "application/json",
+            },
+            json=payload,
+            timeout=60,
+        )
+    except requests.RequestException as e:
+        raise HTTPException(status_code=502, detail=f"UniPile request failed: {e}") from e
+    try:
+        data = resp.json()
+    except Exception:
+        data = {"raw": resp.text}
+    return resp.status_code, data
+
+
+def _normalize_plan_steps(raw: Any) -> List[dict]:
+    if raw is None:
+        return []
+    if isinstance(raw, dict) and isinstance(raw.get("steps"), list):
+        return [x for x in raw["steps"] if isinstance(x, dict)]
+    if isinstance(raw, list):
+        return [x for x in raw if isinstance(x, dict)]
+    return []
+
+
+def build_ordered_actions(steps: List[dict]) -> List[dict]:
+    """Flatten sequence plan to sendable actions with global step_order (matches GeneratedSequence.step_order).
+
+    ``wait_days_before`` is the sum of wizard ``wait`` steps immediately preceding this action (calendar days
+    after the previous slot completes before this slot may send).
+    """
+    out: List[dict] = []
+    ord_global = 0
+    wait_accum = 0
+    for s in steps:
+        if s.get("type") == "wait":
+            try:
+                d = int(s.get("days") if s.get("days") is not None else 1)
+            except (TypeError, ValueError):
+                d = 1
+            wait_accum += max(0, min(d, 365))
+            continue
+        if s.get("type") != "action":
+            continue
+        ord_global += 1
+        ch = (s.get("channel") or "").strip()
+        act = (s.get("action") or "").strip()
+        ref = s.get("unipile_account_ref_id")
+        try:
+            ref_i = int(ref) if ref is not None and str(ref).strip() != "" else None
+        except (TypeError, ValueError):
+            ref_i = None
+        out.append(
+            {
+                "order": ord_global,
+                "channel": ch,
+                "action": act,
+                "account_ref_id": ref_i,
+                "wait_days_before": int(wait_accum),
+            }
+        )
+        wait_accum = 0
+    return out
+
+
+def compose_tracking_label(tenant_id: int, file_id: str, contact_id: int, step_slot: int) -> str:
+    """Stable label for UniPile tracking + webhook attribution (pipe-separated, no pipes in file_id assumed)."""
+    raw = f"eo|{tenant_id}|{file_id}|{contact_id}|{step_slot}"
+    return raw[:200]
+
+
+def parse_tracking_label(label: str) -> Optional[Tuple[int, str, int, int]]:
+    if not label or not label.startswith("eo|"):
+        return None
+    parts = label.split("|")
+    if len(parts) != 5:
+        return None
+    try:
+        return int(parts[1]), parts[2], int(parts[3]), int(parts[4])
+    except (TypeError, ValueError):
+        return None
+
+
+def normalize_smtp_message_id(raw: Optional[str]) -> str:
+    """Normalize RFC 5322 Message-ID for comparison (strip angle brackets, lower case)."""
+    s = _safe_str(raw)
+    if not s:
+        return ""
+    s = s.strip()
+    if len(s) >= 2 and s[0] == "<" and s[-1] == ">":
+        s = s[1:-1].strip()
+    return s.lower()
+
+
+def _delivery_started_at_naive(ex: FileOutreachExecution) -> Optional[datetime]:
+    if not ex.cursor_json:
+        return None
+    try:
+        cur = json.loads(ex.cursor_json)
+        raw = cur.get("delivery_started_at")
+        if not raw or not isinstance(raw, str):
+            return None
+        s = raw.strip().replace("Z", "")
+        return datetime.fromisoformat(s)
+    except Exception:
+        return None
+
+
+def touch_delivery_anchor(ex: FileOutreachExecution, *, reset_run: bool) -> None:
+    cur: Dict[str, Any] = {}
+    if ex.cursor_json:
+        try:
+            cur = json.loads(ex.cursor_json)
+        except Exception:
+            cur = {}
+    if reset_run or not cur.get("delivery_started_at"):
+        cur["delivery_started_at"] = datetime.utcnow().replace(microsecond=0).isoformat() + "Z"
+        ex.cursor_json = json.dumps(cur)
+
+
+def _wait_eligible_for_slot(
+    ex: FileOutreachExecution,
+    act: dict,
+    slot: int,
+    prev_rec: Optional[OutreachSendReceipt],
+    now: datetime,
+) -> bool:
+    try:
+        wd = int(act.get("wait_days_before") or 0)
+    except (TypeError, ValueError):
+        wd = 0
+    if wd <= 0:
+        return True
+    if slot <= 0:
+        anchor = _delivery_started_at_naive(ex) or now
+        return now >= anchor + timedelta(days=wd)
+    if not prev_rec or not prev_rec.created_at:
+        return False
+    return now >= prev_rec.created_at + timedelta(days=wd)
+
+
+def _linkedin_dm_interval_eligible(contact: Contact, interval_h: int, now: datetime) -> bool:
+    if contact.linkedin_invite_pending and not contact.linkedin_connection_accepted_at:
+        return False
+    h = max(1, min(int(interval_h or 72), 720))
+    if contact.linkedin_connection_accepted_at:
+        if contact.linkedin_last_followup_sent_at:
+            return now >= contact.linkedin_last_followup_sent_at + timedelta(hours=h)
+        return now >= contact.linkedin_connection_accepted_at + timedelta(hours=h)
+    if contact.linkedin_last_followup_sent_at:
+        return now >= contact.linkedin_last_followup_sent_at + timedelta(hours=h)
+    return True
+
+
+def _resolve_unipile_row(db: Session, tenant_id: int, user_id: int, ref_id: Optional[int]) -> Optional[UnipileAccount]:
+    if ref_id is None:
+        return None
+    return (
+        db.query(UnipileAccount)
+        .filter(
+            UnipileAccount.id == int(ref_id),
+            UnipileAccount.tenant_id == tenant_id,
+            UnipileAccount.user_id == user_id,
+        )
+        .first()
+    )
+
+
+def _send_unipile_email(
+    account_uid: str,
+    to_email: str,
+    display_name: str,
+    subject: str,
+    body: str,
+    tracking_label: str,
+) -> Dict[str, Optional[str]]:
+    payload = {
+        "account_id": account_uid,
+        "subject": subject or "(no subject)",
+        "body": body or "",
+        "to": [{"identifier": to_email, "display_name": display_name or to_email}],
+        "tracking_options": {"opens": True, "links": True, "label": tracking_label[:200]},
+    }
+    data = _unipile_post_json("/api/v1/emails", payload)
+    mid = data.get("id") or data.get("email_id")
+    trk = data.get("tracking_id")
+    rfc = data.get("message_id") or data.get("internet_message_id") or data.get("smtp_message_id")
+    return {
+        "message_id": _safe_str(mid) if mid else None,
+        "tracking_id": _safe_str(trk) if trk else None,
+        "smtp_message_id": _safe_str(rfc) if rfc else None,
+    }
+
+
+def _send_unipile_dm(account_id: str, provider_id: str, text: str) -> None:
+    chat = _unipile_post_json(
+        "/api/v1/chats",
+        {"account_id": account_id, "attendees": [provider_id]},
+    )
+    chat_id = _safe_str(chat.get("id") if isinstance(chat, dict) else "")
+    if not chat_id:
+        raise ValueError("UniPile chat id not found")
+    _unipile_post_json(
+        f"/api/v1/chats/{requests.utils.quote(chat_id, safe='')}/messages",
+        {"text": text or ""},
+    )
+
+
+def _linkedin_invite_note(text: str) -> str:
+    t = _safe_str(text).replace("\r\n", "\n").strip()
+    if len(t) <= 280:
+        return t
+    return t[:277].rstrip() + "..."
+
+
+def _aggregate_stats(db: Session, tenant_id: int, file_id: str) -> Dict[str, Any]:
+    rows = (
+        db.query(OutreachSendReceipt.status, func.count(OutreachSendReceipt.id))
+        .filter(OutreachSendReceipt.tenant_id == tenant_id, OutreachSendReceipt.file_id == file_id)
+        .group_by(OutreachSendReceipt.status)
+        .all()
+    )
+    counts = {st: int(n or 0) for st, n in rows}
+    sent = int(counts.get("sent", 0))
+    failed = int(counts.get("failed", 0))
+    skipped = int(counts.get("skipped", 0))
+    total_done = sent + failed + skipped
+
+    sent_gmail = (
+        db.query(func.count(OutreachSendReceipt.id))
+        .filter(
+            OutreachSendReceipt.tenant_id == tenant_id,
+            OutreachSendReceipt.file_id == file_id,
+            OutreachSendReceipt.channel == "gmail",
+            OutreachSendReceipt.status == "sent",
+        )
+        .scalar()
+        or 0
+    )
+    opens = (
+        db.query(func.count(OutreachSendReceipt.id))
+        .filter(
+            OutreachSendReceipt.tenant_id == tenant_id,
+            OutreachSendReceipt.file_id == file_id,
+            OutreachSendReceipt.channel == "gmail",
+            OutreachSendReceipt.status == "sent",
+            or_(OutreachSendReceipt.opened_at.isnot(None), func.coalesce(OutreachSendReceipt.open_count, 0) > 0),
+        )
+        .scalar()
+        or 0
+    )
+    replies = (
+        db.query(func.count(OutreachSendReceipt.id))
+        .filter(
+            OutreachSendReceipt.tenant_id == tenant_id,
+            OutreachSendReceipt.file_id == file_id,
+            OutreachSendReceipt.channel == "gmail",
+            OutreachSendReceipt.status == "sent",
+            OutreachSendReceipt.replied_at.isnot(None),
+        )
+        .scalar()
+        or 0
+    )
+    bounces = (
+        db.query(func.count(OutreachSendReceipt.id))
+        .filter(
+            OutreachSendReceipt.tenant_id == tenant_id,
+            OutreachSendReceipt.file_id == file_id,
+            OutreachSendReceipt.channel == "gmail",
+            OutreachSendReceipt.status == "sent",
+            OutreachSendReceipt.bounced_at.isnot(None),
+        )
+        .scalar()
+        or 0
+    )
+
+    open_rate = None
+    reply_rate = None
+    if sent_gmail and int(sent_gmail) > 0:
+        open_rate = round(100.0 * float(opens) / float(sent_gmail), 1)
+        reply_rate = round(100.0 * float(replies) / float(sent_gmail), 1)
+
+    return {
+        "sent": sent,
+        "failed": failed,
+        "skipped": skipped,
+        "total_dispatched": total_done,
+        "opens": int(opens),
+        "replies": int(replies),
+        "bounces": int(bounces),
+        "sent_gmail": int(sent_gmail),
+        "open_rate": open_rate,
+        "reply_rate": reply_rate,
+    }
+
+
+def refresh_outreach_execution_stats_for_file(db: Session, tenant_id: int, file_id: str) -> None:
+    """Recompute stats_json on the execution row (e.g. after UniPile webhooks update receipts)."""
+    ex = (
+        db.query(FileOutreachExecution)
+        .filter(FileOutreachExecution.tenant_id == tenant_id, FileOutreachExecution.file_id == file_id)
+        .first()
+    )
+    if not ex:
+        return
+    ex.stats_json = json.dumps(_aggregate_stats(db, tenant_id, file_id))
+    ex.updated_at = datetime.utcnow()
+
+
+def _get_or_create_execution(db: Session, tenant_id: int, user_id: int, file_id: str) -> FileOutreachExecution:
+    row = (
+        db.query(FileOutreachExecution)
+        .filter(FileOutreachExecution.tenant_id == tenant_id, FileOutreachExecution.file_id == file_id)
+        .first()
+    )
+    if row:
+        return row
+    row = FileOutreachExecution(
+        tenant_id=tenant_id,
+        user_id=user_id,
+        file_id=file_id,
+        status="idle",
+        progress_done=0,
+        progress_total=0,
+    )
+    db.add(row)
+    db.commit()
+    db.refresh(row)
+    return row
+
+
+def _receipt_for(db: Session, tenant_id: int, file_id: str, contact_id: int, step_slot: int) -> Optional[OutreachSendReceipt]:
+    return (
+        db.query(OutreachSendReceipt)
+        .filter(
+            OutreachSendReceipt.tenant_id == tenant_id,
+            OutreachSendReceipt.file_id == file_id,
+            OutreachSendReceipt.contact_id == contact_id,
+            OutreachSendReceipt.step_slot == step_slot,
+        )
+        .first()
+    )
+
+
+def _find_next_send(
+    db: Session,
+    tenant_id: int,
+    file_id: str,
+    contacts: List[Contact],
+    actions: List[dict],
+    ex: FileOutreachExecution,
+) -> Optional[Tuple[Contact, int, dict]]:
+    n_actions = len(actions)
+    if not contacts or not actions:
+        return None
+    now = datetime.utcnow()
+    try:
+        li_h = int(ex.linkedin_followup_interval_hours or 72)
+    except (TypeError, ValueError):
+        li_h = 72
+    for c in contacts:
+        for slot in range(n_actions):
+            if _receipt_for(db, tenant_id, file_id, c.id, slot):
+                continue
+            ok = True
+            prev_rec: Optional[OutreachSendReceipt] = None
+            for ps in range(slot):
+                pr = _receipt_for(db, tenant_id, file_id, c.id, ps)
+                if not pr or pr.status not in ("sent", "failed", "skipped"):
+                    ok = False
+                    break
+                if ps == slot - 1:
+                    prev_rec = pr
+            if not ok:
+                continue
+            act = actions[slot]
+            if not _wait_eligible_for_slot(ex, act, slot, prev_rec, now):
+                continue
+            if act.get("channel") == "linkedin" and act.get("action") == "linkedin_dm":
+                if not _linkedin_dm_interval_eligible(c, li_h, now):
+                    continue
+            return (c, slot, act)
+    return None
+
+
+def _fetch_generated(
+    db: Session, tenant_id: int, file_id: str, row_index: int, channel: str, step_order: int
+) -> Optional[GeneratedSequence]:
+    return (
+        db.query(GeneratedSequence)
+        .filter(
+            GeneratedSequence.tenant_id == tenant_id,
+            GeneratedSequence.file_id == file_id,
+            GeneratedSequence.sequence_id == row_index,
+            GeneratedSequence.channel == channel,
+            GeneratedSequence.step_order == step_order,
+        )
+        .first()
+    )
+
+
+def _execute_one_send(
+    db: Session,
+    tc: TenantContext,
+    uf: UploadedFile,
+    user_row: User,
+    contact: Contact,
+    slot: int,
+    act: dict,
+) -> OutreachSendReceipt:
+    tenant_id = int(tc.tenant_id)
+    user_id = int(tc.user_id)
+    file_id = uf.file_id
+    order = int(act["order"])
+    channel = act["channel"]
+    action = act["action"]
+    ref_id = act.get("account_ref_id")
+
+    rec = OutreachSendReceipt(
+        tenant_id=tenant_id,
+        file_id=file_id,
+        contact_id=contact.id,
+        step_slot=slot,
+        channel=channel,
+        action=action,
+        status="pending",
+    )
+    db.add(rec)
+    db.flush()
+
+    gen = _fetch_generated(db, tenant_id, file_id, int(contact.row_index or 0), channel, order)
+    if not gen:
+        rec.status = "failed"
+        rec.detail = "No generated content for this step (run generation first)."
+        return rec
+
+    try:
+        if channel == "gmail" and action == "email":
+            mb_ref = ref_id if ref_id is not None else getattr(user_row, "default_mailbox_unipile_account_ref_id", None)
+            acc = _resolve_unipile_row(db, tenant_id, user_id, int(mb_ref) if mb_ref is not None else None)
+            if not acc or not acc.unipile_account_id:
+                rec.status = "failed"
+                rec.detail = "No connected mailbox for this step."
+                return rec
+            to_email = _safe_str(gen.email or contact.email)
+            if not to_email or "@" not in to_email:
+                rec.status = "skipped"
+                rec.detail = "Missing recipient email."
+                return rec
+            label = compose_tracking_label(tenant_id, file_id, contact.id, slot)
+            rec.tracking_label = label
+            display = " ".join(
+                p for p in [_safe_str(gen.first_name or contact.first_name), _safe_str(gen.last_name or contact.last_name)] if p
+            ).strip()
+            send_ids = _send_unipile_email(
+                acc.unipile_account_id,
+                to_email,
+                display or to_email,
+                gen.subject or "",
+                gen.email_content or "",
+                label,
+            )
+            rec.unipile_message_id = send_ids.get("message_id") or send_ids.get("tracking_id")
+            rec.unipile_tracking_id = send_ids.get("tracking_id") or None
+            smtp_raw = send_ids.get("smtp_message_id")
+            rec.smtp_message_id = normalize_smtp_message_id(smtp_raw) if smtp_raw else None
+            rec.status = "sent"
+            return rec
+
+        if channel == "linkedin" and action in ("linkedin_connect", "linkedin_dm"):
+            li_ref = ref_id if ref_id is not None else getattr(user_row, "default_unipile_account_ref_id", None)
+            li_acc = _resolve_unipile_row(db, tenant_id, user_id, int(li_ref) if li_ref is not None else None)
+            if not li_acc or not li_acc.unipile_account_id:
+                rec.status = "failed"
+                rec.detail = "No connected LinkedIn account for this step."
+                return rec
+            acc_uid = li_acc.unipile_account_id
+            raw_li = _pick_linkedin_url(contact.raw_data or {})
+            slug = _linkedin_public_identifier(raw_li)
+            if not slug:
+                rec.status = "skipped"
+                rec.detail = "No LinkedIn profile URL on contact."
+                return rec
+
+            slug_enc = requests.utils.quote(slug, safe="")
+            acc_enc = requests.utils.quote(acc_uid, safe="")
+            st_prof, profile = _unipile_call("GET", f"/api/v1/users/{slug_enc}?account_id={acc_enc}")
+            if st_prof >= 400 or not isinstance(profile, dict):
+                rec.status = "failed"
+                rec.detail = "UniPile profile lookup failed."
+                return rec
+            provider_id = _safe_str(profile.get("provider_id"))
+            if not provider_id:
+                rec.status = "failed"
+                rec.detail = "UniPile profile had no provider_id."
+                return rec
+
+            if action == "linkedin_connect":
+                note = _linkedin_invite_note(gen.email_content or "")
+                st_inv, inv_res = _unipile_call(
+                    "POST",
+                    "/api/v1/users/invite",
+                    {"provider_id": provider_id, "account_id": acc_uid, "message": note},
+                )
+                if st_inv < 400:
+                    contact.unipile_provider_id = provider_id
+                    contact.linkedin_invite_pending = 1
+                    contact.linkedin_invite_sent_at = datetime.utcnow()
+                    contact.linkedin_followup_next_email_number = 2
+                    rec.status = "sent"
+                    return rec
+                inv_err = ""
+                if isinstance(inv_res, dict):
+                    inv_err = _safe_str(
+                        inv_res.get("message") or inv_res.get("detail") or inv_res.get("error") or ""
+                    )
+                try:
+                    _send_unipile_dm(acc_uid, provider_id, gen.email_content or "")
+                    contact.unipile_provider_id = provider_id
+                    contact.linkedin_invite_pending = 0
+                    contact.linkedin_connection_accepted_at = datetime.utcnow()
+                    contact.linkedin_last_followup_sent_at = datetime.utcnow()
+                    contact.linkedin_followup_next_email_number = 2
+                    rec.status = "sent"
+                    rec.detail = f"invite_failed_dm_sent: {inv_err[:200]}"
+                    return rec
+                except Exception as e:
+                    rec.status = "failed"
+                    rec.detail = f"invite: {inv_err}; dm: {e}"
+                    return rec
+
+            # linkedin_dm
+            pid = _safe_str(contact.unipile_provider_id) or provider_id
+            if not pid:
+                rec.status = "skipped"
+                rec.detail = "No provider_id for DM (connection may not be accepted yet)."
+                return rec
+            _send_unipile_dm(acc_uid, pid, gen.email_content or "")
+            rec.status = "sent"
+            contact.linkedin_last_followup_sent_at = datetime.utcnow()
+            return rec
+
+        rec.status = "skipped"
+        rec.detail = f"Unsupported action {channel}/{action}"
+        return rec
+    except HTTPException as e:
+        rec.status = "failed"
+        rec.detail = _safe_str(e.detail)[:2000]
+        return rec
+    except Exception as e:
+        rec.status = "failed"
+        rec.detail = str(e)[:2000]
+        return rec
+
+
+def _refresh_execution_counters(db: Session, ex: FileOutreachExecution, tenant_id: int, file_id: str, total: int) -> None:
+    stats = _aggregate_stats(db, tenant_id, file_id)
+    done = int(stats["total_dispatched"])
+    ex.progress_done = min(done, total)
+    ex.progress_total = total
+    ex.stats_json = json.dumps(stats)
+    if total > 0 and ex.progress_done >= total:
+        ex.status = "completed"
+    ex.updated_at = datetime.utcnow()
+
+
+@router.get("/{file_id}/execution")
+def get_campaign_execution(file_id: str, tc: TenantContext = Depends(get_tenant_context)):
+    db = tc.db
+    uf = (
+        db.query(UploadedFile)
+        .filter(UploadedFile.tenant_id == tc.tenant_id, UploadedFile.file_id == file_id)
+        .first()
+    )
+    if not uf:
+        raise HTTPException(status_code=404, detail="File not found")
+    ex = _get_or_create_execution(db, int(tc.tenant_id), int(tc.user_id), file_id)
+    stats = json.loads(ex.stats_json) if ex.stats_json else _aggregate_stats(db, int(tc.tenant_id), file_id)
+    pct = 0
+    if ex.progress_total and ex.progress_total > 0:
+        pct = min(100, max(0, round(100 * (ex.progress_done or 0) / ex.progress_total)))
+    return {
+        "file_id": file_id,
+        "status": ex.status,
+        "progress_done": ex.progress_done,
+        "progress_total": ex.progress_total,
+        "progress_percent": pct,
+        "stats": stats,
+        "last_error": ex.last_error,
+    }
+
+
+@router.post("/{file_id}/execution/start")
+def start_campaign_execution(
+    file_id: str,
+    reset: bool = False,
+    linkedin_followup_hours: Optional[int] = Query(None, ge=1, le=720),
+    tc: TenantContext = Depends(get_tenant_context),
+):
+    db = tc.db
+    uf = (
+        db.query(UploadedFile)
+        .filter(UploadedFile.tenant_id == tc.tenant_id, UploadedFile.file_id == file_id)
+        .first()
+    )
+    if not uf:
+        raise HTTPException(status_code=404, detail="File not found")
+
+    raw_plan = getattr(uf, "sequence_plan_json", None)
+    steps: List[dict] = []
+    if raw_plan and str(raw_plan).strip():
+        try:
+            steps = _normalize_plan_steps(json.loads(raw_plan))
+        except Exception:
+            steps = []
+    actions = build_ordered_actions(steps)
+    contacts = (
+        db.query(Contact)
+        .filter(
+            Contact.tenant_id == tc.tenant_id,
+            Contact.file_id == file_id,
+            Contact.source == "apollo_csv",
+        )
+        .order_by(Contact.row_index.asc(), Contact.id.asc())
+        .all()
+    )
+    if not actions:
+        raise HTTPException(status_code=400, detail="No sendable sequence steps on this file. Save a sequence in the campaign wizard.")
+    if not contacts:
+        raise HTTPException(status_code=400, detail="No contacts for this file.")
+
+    gen_n = (
+        db.query(func.count(GeneratedSequence.id))
+        .filter(
+            GeneratedSequence.tenant_id == tc.tenant_id,
+            GeneratedSequence.file_id == file_id,
+        )
+        .scalar()
+        or 0
+    )
+    if int(gen_n) == 0:
+        raise HTTPException(status_code=400, detail="Generate sequences before starting the campaign.")
+
+    tenant_id = int(tc.tenant_id)
+    user_id = int(tc.user_id)
+
+    ex = _get_or_create_execution(db, tenant_id, user_id, file_id)
+
+    if ex.status == "running" and reset:
+        raise HTTPException(status_code=400, detail="Pause the campaign before restarting delivery (reset).")
+
+    if ex.status == "running" and not reset:
+        return get_campaign_execution(file_id, tc)
+    if ex.status == "paused" and not reset:
+        total = ex.progress_total or len(contacts) * len(actions)
+        ex.status = "running"
+        ex.progress_total = total
+        ex.last_error = None
+        if linkedin_followup_hours is not None:
+            ex.linkedin_followup_interval_hours = int(linkedin_followup_hours)
+        ex.updated_at = datetime.utcnow()
+        db.commit()
+        return get_campaign_execution(file_id, tc)
+
+    if ex.status == "completed" and not reset:
+        return get_campaign_execution(file_id, tc)
+
+    if reset:
+        db.query(OutreachSendReceipt).filter(
+            OutreachSendReceipt.tenant_id == tenant_id,
+            OutreachSendReceipt.file_id == file_id,
+        ).delete()
+
+        db.commit()
+
+    total = len(contacts) * len(actions)
+    ex.status = "running"
+    ex.progress_total = total
+    ex.progress_done = 0
+    ex.last_error = None
+    touch_delivery_anchor(ex, reset_run=bool(reset))
+    if linkedin_followup_hours is not None:
+        ex.linkedin_followup_interval_hours = int(linkedin_followup_hours)
+    ex.stats_json = json.dumps(_aggregate_stats(db, tenant_id, file_id))
+    ex.updated_at = datetime.utcnow()
+    db.commit()
+    return get_campaign_execution(file_id, tc)
+
+
+@router.post("/{file_id}/execution/pause")
+def pause_campaign_execution(file_id: str, tc: TenantContext = Depends(get_tenant_context)):
+    db = tc.db
+    ex = (
+        db.query(FileOutreachExecution)
+        .filter(FileOutreachExecution.tenant_id == tc.tenant_id, FileOutreachExecution.file_id == file_id)
+        .first()
+    )
+    if not ex:
+        raise HTTPException(status_code=404, detail="No execution state for this file")
+    if ex.status == "running":
+        ex.status = "paused"
+        ex.updated_at = datetime.utcnow()
+        db.commit()
+    return get_campaign_execution(file_id, tc)
+
+
+@router.post("/{file_id}/execution/tick")
+def tick_campaign_execution(
+    file_id: str,
+    limit: int = Query(3, ge=1, le=25),
+    tc: TenantContext = Depends(get_tenant_context),
+):
+    db = tc.db
+    uf = (
+        db.query(UploadedFile)
+        .filter(UploadedFile.tenant_id == tc.tenant_id, UploadedFile.file_id == file_id)
+        .first()
+    )
+    if not uf:
+        raise HTTPException(status_code=404, detail="File not found")
+
+    ex = _get_or_create_execution(db, int(tc.tenant_id), int(tc.user_id), file_id)
+    if ex.status != "running":
+        return get_campaign_execution(file_id, tc)
+
+    steps = []
+    raw = getattr(uf, "sequence_plan_json", None)
+    if raw and str(raw).strip():
+        try:
+            steps = _normalize_plan_steps(json.loads(raw))
+        except Exception:
+            steps = []
+    actions = build_ordered_actions(steps)
+    contacts = (
+        db.query(Contact)
+        .filter(
+            Contact.tenant_id == tc.tenant_id,
+            Contact.file_id == file_id,
+            Contact.source == "apollo_csv",
+        )
+        .order_by(Contact.row_index.asc(), Contact.id.asc())
+        .all()
+    )
+    user_row = db.query(User).filter(User.id == tc.user_id).first()
+    if not user_row or not actions or not contacts:
+        ex.status = "failed"
+        ex.last_error = "Missing user, actions, or contacts."
+        db.commit()
+        return get_campaign_execution(file_id, tc)
+
+    n = max(1, min(int(limit or 3), 25))
+    for _ in range(n):
+        nxt = _find_next_send(db, int(tc.tenant_id), file_id, contacts, actions, ex)
+        if not nxt:
+            break
+        c, slot, act = nxt
+        _execute_one_send(db, tc, uf, user_row, c, slot, act)
+        db.commit()
+        _refresh_execution_counters(db, ex, int(tc.tenant_id), file_id, ex.progress_total or len(contacts) * len(actions))
+        db.commit()
+
+    _refresh_execution_counters(db, ex, int(tc.tenant_id), file_id, ex.progress_total or len(contacts) * len(actions))
+    db.commit()
+    return get_campaign_execution(file_id, tc)

frontend/src/components/campaigns/CampaignsDashboardTab.jsx

@@ -9,6 +9,7 @@ import {
     ExternalLink,
     Archive,
     ArchiveRestore,
+    RotateCcw,
 } from 'lucide-react';
 import { Button } from '@/components/ui/button';
 import CreateCampaignWizard from '@/components/campaigns/CreateCampaignWizard';
@@ -84,6 +85,12 @@ function normalizeCampaign(c) {
             Math.round(c.generationProgressPercent ?? (genDone ? 100 : hasFile ? 0 : 100))
         ),
         sequence: Array.isArray(c.sequence) ? c.sequence : [],
+        executionProgressPercent: Math.min(
+            100,
+            Math.round(Number(c.executionProgressPercent) || 0)
+        ),
+        executionDeliveryStatus: c.executionDeliveryStatus || 'idle',
+        deliveryStats: c.deliveryStats && typeof c.deliveryStats === 'object' ? c.deliveryStats : null,
     };
 }
 
@@ -151,7 +158,7 @@ function statusLabel(status) {
 }
 
 /** Circular ring; hover shows native tooltip with percent (title). */
-function CampaignGenerationRing({ percent, complete, fileId, compact = false }) {
+function CampaignGenerationRing({ percent, complete, fileId, compact = false, variant = 'generation' }) {
     if (!fileId || complete) return null;
     const p = Math.min(100, Math.max(0, percent));
     const r = compact ? 14 : 17;
@@ -160,10 +167,17 @@ function CampaignGenerationRing({ percent, complete, fileId, compact = false })
     const cx = size / 2;
     const c = 2 * Math.PI * r;
     const offset = c * (1 - p / 100);
+    const strokeColor = variant === 'delivery' ? '#0369a1' : '#7c3aed';
+    const title =
+        variant === 'delivery' ? `Delivery: ${Math.round(p)}%` : `Content generation: ${Math.round(p)}%`;
+    const textClass =
+        variant === 'delivery'
+            ? 'text-sky-900'
+            : 'text-violet-800';
     return (
         <div
             className={cn('relative shrink-0 cursor-default', compact ? 'h-9 w-9' : 'h-11 w-11')}
-            title={`Content generation: ${Math.round(p)}%`}
+            title={title}
         >
             <svg
                 width={size}
@@ -178,7 +192,7 @@ function CampaignGenerationRing({ percent, complete, fileId, compact = false })
                     cy={cx}
                     r={r}
                     fill="none"
-                    stroke="#7c3aed"
+                    stroke={strokeColor}
                     strokeWidth={stroke}
                     strokeLinecap="round"
                     strokeDasharray={c}
@@ -188,7 +202,8 @@ function CampaignGenerationRing({ percent, complete, fileId, compact = false })
             </svg>
             <span
                 className={cn(
-                    'pointer-events-none absolute inset-0 flex items-center justify-center gap-0 px-0.5 text-center font-bold tabular-nums leading-none text-violet-800',
+                    'pointer-events-none absolute inset-0 flex items-center justify-center gap-0 px-0.5 text-center font-bold tabular-nums leading-none',
+                    textClass,
                     compact ? 'text-[9px]' : 'text-[10px]'
                 )}
                 aria-hidden
@@ -277,6 +292,70 @@ export default function CampaignsDashboardTab() {
         return () => clearInterval(iv);
     }, []);
 
+    useEffect(() => {
+        const pollDelivery = async () => {
+            const list = campaignsRef.current.filter(
+                (c) =>
+                    c.fileId &&
+                    c.generationComplete &&
+                    !c.archived &&
+                    (c.status === 'running' ||
+                        c.status === 'paused' ||
+                        c.executionDeliveryStatus === 'completed')
+            );
+            if (!list.length) return;
+            const updates = await Promise.all(
+                list.map(async (c) => {
+                    try {
+                        if (c.status === 'running' && c.executionDeliveryStatus !== 'completed') {
+                            await apiFetch(
+                                `/api/campaign-files/${encodeURIComponent(c.fileId)}/execution/tick?limit=4`,
+                                { method: 'POST' }
+                            );
+                        }
+                        const res = await apiFetch(
+                            `/api/campaign-files/${encodeURIComponent(c.fileId)}/execution`
+                        );
+                        if (!res.ok) return { id: c.id, skip: true };
+                        const d = await res.json();
+                        return {
+                            id: c.id,
+                            skip: false,
+                            executionProgressPercent: d.progress_percent ?? 0,
+                            executionDeliveryStatus: d.status || 'idle',
+                            deliveryStats: d.stats || null,
+                            openRate:
+                                d.stats?.open_rate != null ? Math.round(Number(d.stats.open_rate)) : c.openRate,
+                            replyRate:
+                                d.stats?.reply_rate != null
+                                    ? Math.round(Number(d.stats.reply_rate))
+                                    : c.replyRate,
+                        };
+                    } catch {
+                        return { id: c.id, skip: true };
+                    }
+                })
+            );
+            setCampaigns((prev) =>
+                prev.map((c) => {
+                    const u = updates.find((x) => x.id === c.id);
+                    if (!u || u.skip) return c;
+                    return {
+                        ...c,
+                        executionProgressPercent: u.executionProgressPercent,
+                        executionDeliveryStatus: u.executionDeliveryStatus,
+                        deliveryStats: u.deliveryStats,
+                        openRate: u.openRate ?? c.openRate,
+                        replyRate: u.replyRate ?? c.replyRate,
+                    };
+                })
+            );
+        };
+        const iv = setInterval(pollDelivery, 6000);
+        pollDelivery();
+        return () => clearInterval(iv);
+    }, []);
+
     const tabCounts = useMemo(() => {
         const active = campaigns.filter((c) => !c.archived);
         const archived = campaigns.filter((c) => c.archived);
@@ -347,7 +426,23 @@ export default function CampaignsDashboardTab() {
 
     const closeBulkMenu = useCallback(() => setBulkMenuOpen(false), []);
 
-    const bulkResume = useCallback(() => {
+    const bulkResume = useCallback(async () => {
+        const targets = campaignsRef.current.filter(
+            (c) => selectedSet.has(c.id) && c.status === 'paused' && c.fileId && c.generationComplete
+        );
+        for (const c of targets) {
+            try {
+                await apiFetch(`/api/campaign-files/${encodeURIComponent(c.fileId)}/execution/start`, {
+                    method: 'POST',
+                });
+                await apiFetch(
+                    `/api/campaign-files/${encodeURIComponent(c.fileId)}/execution/tick?limit=5`,
+                    { method: 'POST' }
+                );
+            } catch (e) {
+                console.error(e);
+            }
+        }
         setCampaigns((prev) =>
             prev.map((c) =>
                 selectedSet.has(c.id) && c.status === 'paused' ? { ...c, status: 'running' } : c
@@ -356,7 +451,19 @@ export default function CampaignsDashboardTab() {
         closeBulkMenu();
     }, [selectedSet, closeBulkMenu]);
 
-    const bulkPause = useCallback(() => {
+    const bulkPause = useCallback(async () => {
+        const targets = campaignsRef.current.filter(
+            (c) => selectedSet.has(c.id) && c.status === 'running' && c.fileId && c.generationComplete
+        );
+        for (const c of targets) {
+            try {
+                await apiFetch(`/api/campaign-files/${encodeURIComponent(c.fileId)}/execution/pause`, {
+                    method: 'POST',
+                });
+            } catch (e) {
+                console.error(e);
+            }
+        }
         setCampaigns((prev) =>
             prev.map((c) =>
                 selectedSet.has(c.id) && c.status === 'running' ? { ...c, status: 'paused' } : c
@@ -398,12 +505,68 @@ export default function CampaignsDashboardTab() {
         closeBulkMenu();
     }, [selectedIds, closeBulkMenu]);
 
-    const toggleStatus = useCallback((id) => {
+    const toggleStatus = useCallback(async (id) => {
+        const cur = campaignsRef.current.find((c) => c.id === id);
+        if (!cur) return;
+
+        if (!cur.fileId || !cur.generationComplete) {
+            setCampaigns((prev) =>
+                prev.map((c) => {
+                    if (c.id !== id) return c;
+                    if (c.status === 'running') return { ...c, status: 'paused' };
+                    if (c.status === 'paused') return { ...c, status: 'running' };
+                    if (c.status === 'draft') return { ...c, status: 'running' };
+                    return c;
+                })
+            );
+            return;
+        }
+
+        if (cur.status === 'running') {
+            try {
+                const pr = await apiFetch(
+                    `/api/campaign-files/${encodeURIComponent(cur.fileId)}/execution/pause`,
+                    { method: 'POST' }
+                );
+                if (!pr.ok) {
+                    const err = await pr.json().catch(() => ({}));
+                    alert(typeof err.detail === 'string' ? err.detail : 'Could not pause campaign send');
+                    return;
+                }
+            } catch (e) {
+                alert(String(e));
+                return;
+            }
+        } else {
+            try {
+                const st = await apiFetch(
+                    `/api/campaign-files/${encodeURIComponent(cur.fileId)}/execution/start`,
+                    { method: 'POST' }
+                );
+                if (!st.ok) {
+                    const err = await st.json().catch(() => ({}));
+                    alert(typeof err.detail === 'string' ? err.detail : 'Could not start campaign send');
+                    return;
+                }
+                await apiFetch(
+                    `/api/campaign-files/${encodeURIComponent(cur.fileId)}/execution/tick?limit=5`,
+                    { method: 'POST' }
+                );
+            } catch (e) {
+                alert(String(e));
+                return;
+            }
+        }
+
         setCampaigns((prev) =>
             prev.map((c) => {
                 if (c.id !== id) return c;
-                if (c.status === 'running') return { ...c, status: 'paused' };
-                if (c.status === 'paused') return { ...c, status: 'running' };
+                if (c.status === 'running')
+                    return { ...c, status: 'paused', executionDeliveryStatus: 'paused' };
+                if (c.status === 'paused')
+                    return { ...c, status: 'running', executionDeliveryStatus: 'running' };
+                if (c.status === 'draft')
+                    return { ...c, status: 'running', executionDeliveryStatus: 'running' };
                 return c;
             })
         );
@@ -415,6 +578,68 @@ export default function CampaignsDashboardTab() {
         setSelectedIds((prev) => prev.filter((x) => x !== id));
     }, []);
 
+    const restartDelivery = useCallback(async (c) => {
+        if (!c.fileId || !c.generationComplete) return;
+        if (
+            !window.confirm(
+                'Restart delivery from scratch? All send receipts for this campaign will be cleared and delivery will run again from step one.'
+            )
+        ) {
+            setMenuOpenId(null);
+            return;
+        }
+        setMenuOpenId(null);
+        try {
+            if (c.status === 'running') {
+                const pr = await apiFetch(
+                    `/api/campaign-files/${encodeURIComponent(c.fileId)}/execution/pause`,
+                    { method: 'POST' }
+                );
+                if (!pr.ok) {
+                    const err = await pr.json().catch(() => ({}));
+                    alert(typeof err.detail === 'string' ? err.detail : 'Pause the campaign before restarting.');
+                    return;
+                }
+            }
+            const st = await apiFetch(
+                `/api/campaign-files/${encodeURIComponent(c.fileId)}/execution/start?reset=1`,
+                { method: 'POST' }
+            );
+            if (!st.ok) {
+                const err = await st.json().catch(() => ({}));
+                alert(typeof err.detail === 'string' ? err.detail : 'Could not restart delivery');
+                return;
+            }
+            await apiFetch(
+                `/api/campaign-files/${encodeURIComponent(c.fileId)}/execution/tick?limit=5`,
+                { method: 'POST' }
+            );
+            const r = await apiFetch(`/api/campaign-files/${encodeURIComponent(c.fileId)}/execution`);
+            const d = r.ok ? await r.json().catch(() => ({})) : {};
+            setCampaigns((prev) =>
+                prev.map((row) =>
+                    row.id === c.id
+                        ? normalizeCampaign({
+                              ...row,
+                              status: 'running',
+                              executionProgressPercent: d.progress_percent ?? 0,
+                              executionDeliveryStatus: d.status || 'running',
+                              deliveryStats: d.stats || null,
+                              openRate:
+                                  d.stats?.open_rate != null ? Math.round(Number(d.stats.open_rate)) : row.openRate,
+                              replyRate:
+                                  d.stats?.reply_rate != null
+                                      ? Math.round(Number(d.stats.reply_rate))
+                                      : row.replyRate,
+                          })
+                        : row
+                )
+            );
+        } catch (e) {
+            alert(String(e));
+        }
+    }, []);
+
     const onWizardComplete = useCallback((payload) => {
         const contacts = payload.contacts || 0;
         const id = payload.campaignId || uid();
@@ -458,6 +683,37 @@ export default function CampaignsDashboardTab() {
             return [row, ...prev];
         });
         setWizardCampaign(null);
+        if (payload.fileId && payload.generationComplete) {
+            const fid = payload.fileId;
+            void (async () => {
+                try {
+                    const st = await apiFetch(`/api/campaign-files/${encodeURIComponent(fid)}/execution/start`, {
+                        method: 'POST',
+                    });
+                    if (!st.ok) return;
+                    await apiFetch(`/api/campaign-files/${encodeURIComponent(fid)}/execution/tick?limit=5`, {
+                        method: 'POST',
+                    });
+                    const r = await apiFetch(`/api/campaign-files/${encodeURIComponent(fid)}/execution`);
+                    if (!r.ok) return;
+                    const d = await r.json();
+                    setCampaigns((prev) =>
+                        prev.map((row) =>
+                            row.id === id
+                                ? normalizeCampaign({
+                                      ...row,
+                                      executionProgressPercent: d.progress_percent ?? 0,
+                                      executionDeliveryStatus: d.status || 'running',
+                                      deliveryStats: d.stats || null,
+                                  })
+                                : row
+                        )
+                    );
+                } catch (e) {
+                    console.error(e);
+                }
+            })();
+        }
     }, []);
 
     const onWizardDraftPersist = useCallback((payload) => {
@@ -722,6 +978,24 @@ export default function CampaignsDashboardTab() {
                                         const contentPct = c.generationComplete
                                             ? '100%'
                                             : `${Math.round(c.generationProgressPercent ?? 0)}%`;
+                                        const execStatus = c.executionDeliveryStatus || 'idle';
+                                        const deliveryPct = Math.round(c.executionProgressPercent ?? 0);
+                                        const deliveryDone =
+                                            execStatus === 'completed' ||
+                                            (deliveryPct >= 100 && execStatus === 'running');
+                                        const showDeliveryRing =
+                                            !!c.fileId &&
+                                            c.generationComplete &&
+                                            (c.status === 'running' || c.status === 'paused') &&
+                                            !deliveryDone;
+                                        const sentCount =
+                                            c.deliveryStats && typeof c.deliveryStats.sent === 'number'
+                                                ? c.deliveryStats.sent
+                                                : null;
+                                        const opensCount =
+                                            c.deliveryStats && typeof c.deliveryStats.opens === 'number'
+                                                ? c.deliveryStats.opens
+                                                : null;
                                         return (
                                             <tr
                                                 key={c.id}
@@ -739,26 +1013,42 @@ export default function CampaignsDashboardTab() {
                                                 <td className="max-w-[360px] px-3 py-3 align-middle">
                                                     <div className="flex items-start gap-3">
                                                         <div className="mt-0.5 flex h-9 w-9 shrink-0 items-center justify-center">
-                                                            {c.fileId && !c.generationComplete ? (
+                                                            {!c.fileId ? (
+                                                                <span
+                                                                    className="flex h-9 w-9 items-center justify-center rounded-full bg-slate-100 text-[10px] font-medium text-slate-400"
+                                                                    title="No upload"
+                                                                >
+                                                                    —
+                                                                </span>
+                                                            ) : !c.generationComplete ? (
                                                                 <CampaignGenerationRing
                                                                     percent={c.generationProgressPercent ?? 0}
                                                                     complete={false}
                                                                     fileId={c.fileId}
                                                                     compact
                                                                 />
-                                                            ) : c.fileId && c.generationComplete ? (
+                                                            ) : showDeliveryRing ? (
+                                                                <CampaignGenerationRing
+                                                                    variant="delivery"
+                                                                    percent={deliveryPct}
+                                                                    complete={false}
+                                                                    fileId={c.fileId}
+                                                                    compact
+                                                                />
+                                                            ) : deliveryDone &&
+                                                              (c.status === 'running' || c.status === 'paused') ? (
                                                                 <span
-                                                                    className="flex h-9 w-9 items-center justify-center rounded-full bg-emerald-50 text-[10px] font-bold text-emerald-700 ring-1 ring-emerald-200"
-                                                                    title="Content generation complete"
+                                                                    className="flex h-9 w-9 items-center justify-center rounded-full bg-sky-50 text-[10px] font-bold text-sky-800 ring-1 ring-sky-200"
+                                                                    title="Delivery complete"
                                                                 >
                                                                     100%
                                                                 </span>
                                                             ) : (
                                                                 <span
-                                                                    className="flex h-9 w-9 items-center justify-center rounded-full bg-slate-100 text-[10px] font-medium text-slate-400"
-                                                                    title="No upload"
+                                                                    className="flex h-9 w-9 items-center justify-center rounded-full bg-emerald-50 text-[10px] font-bold text-emerald-700 ring-1 ring-emerald-200"
+                                                                    title="Content generation complete"
                                                                 >
-                                                                    —
+                                                                    100%
                                                                 </span>
                                                             )}
                                                         </div>
@@ -802,10 +1092,14 @@ export default function CampaignsDashboardTab() {
                                                     {contentPct}
                                                 </td>
                                                 <td className="px-3 py-3 text-right tabular-nums text-slate-500">
-                                                    —
+                                                    {sentCount != null ? sentCount.toLocaleString() : '—'}
                                                 </td>
                                                 <td className="px-3 py-3 text-right tabular-nums text-fuchsia-700">
-                                                    {c.openRate != null ? `— / ${c.openRate}%` : '—'}
+                                                    {opensCount != null && opensCount > 0
+                                                        ? `${opensCount}`
+                                                        : c.openRate != null
+                                                          ? `— / ${c.openRate}%`
+                                                          : '—'}
                                                 </td>
                                                 <td className="px-3 py-3 text-right tabular-nums text-amber-700">
                                                     —
@@ -870,6 +1164,16 @@ export default function CampaignsDashboardTab() {
                                                                         >
                                                                             Open in wizard
                                                                         </button>
+                                                                        {c.fileId && c.generationComplete ? (
+                                                                            <button
+                                                                                type="button"
+                                                                                className="flex w-full items-center gap-2 px-3 py-2 text-left hover:bg-slate-50"
+                                                                                onClick={() => restartDelivery(c)}
+                                                                            >
+                                                                                <RotateCcw className="h-3.5 w-3.5 shrink-0 text-slate-500" />
+                                                                                Restart delivery
+                                                                            </button>
+                                                                        ) : null}
                                                                         <button
                                                                             type="button"
                                                                             className="block w-full px-3 py-2 text-left text-red-600 hover:bg-red-50"