feat(campaigns): implement Sandbox Mode for safe campaign testing (#397)

backend/campaigns/ai.py

@@ -217,12 +217,29 @@ def personalize_email(template_subject, template_body, lead):
     """
     Uses Gemini to personalize the given email template for a specific lead.
     """
+    logger.debug("personalize_email() called")
+
     api_key = _get_gemini_api_key()
     merged_subject = _apply_merge_tags(template_subject, lead)
     merged_body = _apply_merge_tags(template_body, lead)
-    if not api_key or not template_body:
+
+    # Resolve the effective key BEFORE the missing-key early return, so a
+    # tenant-level key can still enable personalization even when the global
+    # GEMINI_API_KEY env var is unset. Checking only `api_key` here previously
+    # skipped personalization for orgs with their own key configured.
+    active_key = None
+    if hasattr(lead, 'organization') and lead.organization:
+        if not getattr(lead.organization, 'enable_ai_personalization', True):
+            logger.debug("AI Personalization is explicitly disabled for this organization workspace.")
+            return merged_subject, merged_body
+        active_key = getattr(lead.organization, 'gemini_api_key', None)
+
+    final_api_key = active_key if active_key else api_key
+    logger.debug("Effective Gemini API key present: %s", bool(final_api_key))
+
+    if not final_api_key or not template_body:
         return merged_subject, merged_body
-        
+
     prompt = f"""
 You are an expert sales representative. Personalize the following email template for a lead.
 Lead details:
@@ -241,23 +258,21 @@ def personalize_email(template_subject, template_body, lead):
 
     try:
         import google.generativeai as genai
-        # 1. Check if organization has personal tracking tokens and personalization toggled on
-        active_key = None
-        if hasattr(lead, 'organization') and lead.organization:
-            # If the user explicitly disabled personalization, trigger an early exit exception to drop back to standard templates
-            if not getattr(lead.organization, 'enable_ai_personalization', True):
-                raise Exception("AI Personalization is explicitly disabled for this organization workspace.")
-
-            active_key = getattr(lead.organization, 'gemini_api_key', None)
-
-        # 2. Fall back to the default system environment variable token if no tenant-level key exists
-        final_api_key = active_key if active_key else api_key
+        from google.generativeai.types import RequestOptions
 
         genai.configure(api_key=final_api_key)
-
-        # 3. Upgrade the deprecated engine version string to the current 2.0 version
-        model = genai.GenerativeModel('gemini-2.0-flash')
-        response = model.generate_content(prompt)
+        logger.debug("Using Gemini model: gemini-2.5-flash")
+
+        # Use the current Gemini Flash model
+        model = genai.GenerativeModel("gemini-2.5-flash")
+        logger.debug("Sending request to Gemini...")
+        # send_email_step runs this inline, so an unbounded request could stall
+        # the worker and hold up campaign progression — cap it at 30s.
+        response = model.generate_content(
+            prompt,
+            request_options=RequestOptions(timeout=30),
+        )
+        logger.debug("Gemini response received")
         # Parse basic JSON from response...
         # For MVP we will just do simple replacement if JSON parsing fails
         text = response.text.strip()
@@ -270,4 +285,4 @@ def personalize_email(template_subject, template_body, lead):
         return subject, body
     except Exception as e:
         logger.error(f"Gemini Personalization Error: {e}")
-        return merged_subject, merged_body
+        return merged_subject, merged_body

backend/campaigns/migrations/0011_campaign_created.py

@@ -0,0 +1,26 @@
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('campaigns', '0010_custom_mailbox_security_updates'),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='campaign',
+            name='created_by',
+            field=models.ForeignKey(
+                blank=True,
+                help_text='The user who created this campaign. Used as the Sandbox Mode test-email recipient.',
+                null=True,
+                on_delete=django.db.models.deletion.SET_NULL,
+                related_name='created_campaigns',
+                to=settings.AUTH_USER_MODEL,
+            ),
+        ),
+    ]

backend/campaigns/models.py

@@ -64,6 +64,14 @@ class Campaign(TenantModel):
     id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
     name = models.CharField(max_length=255)
     status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='DRAFT')
+    created_by = models.ForeignKey(
+        settings.AUTH_USER_MODEL,
+        on_delete=models.SET_NULL,
+        null=True,
+        blank=True,
+        related_name='created_campaigns',
+        help_text="The user who created this campaign. Used as the Sandbox Mode test-email recipient.",
+    )
     settings = models.JSONField(default=dict, blank=True)
     connected_account = models.ForeignKey(ConnectedEmailAccount, on_delete=models.SET_NULL, null=True, blank=True, related_name='campaigns')
 
@@ -146,4 +154,4 @@ class Meta:
         ]
 
     def __str__(self):
-        return f"{self.lead.email} in {self.campaign.name}"
+        return f"{self.lead.email} in {self.campaign.name}"

backend/campaigns/tasks.py

@@ -541,14 +541,56 @@ def rewrite_email_links(html_body, campaign_lead_id, step_id):
 # -----------------------------------
 
 
+def _resolve_sandbox_recipient(campaign):
+    """
+    Determines where Sandbox Mode emails should be redirected, per the spec:
+    "send the email to the logged-in user's address instead."
+
+    Priority:
+      1. An explicit `sandbox_email` override stored in campaign.settings —
+         this must win over the creator default, since it's the whole point
+         of exposing an override field: letting a rep deliberately test-deliver
+         to a different inbox than their own.
+      2. `campaign.created_by.email` — the actual logged-in user who created
+         the campaign. This is the literal "logged-in user" from the spec,
+         captured at creation time since Celery tasks have no request/session
+         of their own to read a "logged-in user" from directly.
+      3. The email of whoever connected the campaign's sending account, as a
+         last-resort fallback for campaigns created before this field existed
+         (created_by will be null on those).
+
+    Returns None if no safe destination can be resolved, so callers can
+    decide how to fail safely instead of silently emailing a real lead.
+    """
+    settings_dict = campaign.settings if isinstance(campaign.settings, dict) else {}
+
+    override = str(settings_dict.get('sandbox_email') or '').strip()
+    if override:
+        return override
+
+    if campaign.created_by_id and campaign.created_by.email:
+        return campaign.created_by.email
+
+    account = campaign.connected_account
+    if account and account.connected_by_id and account.connected_by.email:
+        return account.connected_by.email
+
+    return None
+
+
 @shared_task
 def send_email_step(campaign_lead_id, step_id):
     """
     Dispatch an email through the selected connected account or fall back to mock logging.
     """
 
     try:
-        clead = CampaignLead.objects.select_related('lead', 'campaign__connected_account').get(id=campaign_lead_id)
+        clead = CampaignLead.objects.select_related(
+            'lead',
+            'campaign__connected_account',
+            'campaign__created_by',
+            'campaign__connected_account__connected_by',
+        ).get(id=campaign_lead_id)
         step = SequenceStep.objects.get(id=step_id)
 
         if clead.lead.global_unsubscribe:
@@ -587,40 +629,88 @@ def send_email_step(campaign_lead_id, step_id):
         body = rewrite_email_links(body, campaign_lead_id, step_id)
         # -------------------------------------------
 
+        # --- Sandbox Mode: redirect delivery without touching personalization,
+        # analytics, delays, or workflow progression below. ---------------------
+        original_recipient = clead.lead.email
+        recipient_email = original_recipient
+        sandbox_mode = bool((clead.campaign.settings or {}).get('sandbox_mode'))
+
+        if sandbox_mode:
+            sandbox_recipient = _resolve_sandbox_recipient(clead.campaign)
+            if sandbox_recipient:
+                recipient_email = sandbox_recipient
+                subject = f"[TEST SANDBOX] {subject}"
+                logger.info(
+                    "Sandbox Email Queued | Original Recipient: %s | Delivered To: %s | Campaign: %s | Step: %s",
+                    original_recipient, recipient_email, clead.campaign.name, step.step_order,
+                )
+            else:
+                # Fail safe: if we can't resolve a test inbox, do NOT risk emailing
+                # the real lead under a sandboxed campaign. Retry later instead.
+                logger.warning(
+                    "Sandbox Mode is enabled for campaign '%s' but no sandbox recipient could be "
+                    "resolved (no sandbox_email override and no connected account owner). "
+                    "Holding send and retrying in 15 minutes.",
+                    clead.campaign.name,
+                )
+                clead.next_execution_time = timezone.now() + timedelta(minutes=15)
+                clead.save(update_fields=['next_execution_time'])
+                return
+        # -------------------------------------------------------------------------
+
         account = clead.campaign.connected_account
         if account:
             try:
                 if account.provider == 'GOOGLE':
                     message_id = send_gmail(
                         account,
-                        clead.lead.email,
+                        recipient_email,
                         subject,
                         body,
                         unsubscribe_url=build_unsubscribe_url(clead.lead),
                     )
-                    logger.info(f"Gmail SENT to {clead.lead.email} | msg_id={message_id}")
+                    logger.info(f"Gmail SENT to {recipient_email} | msg_id={message_id}")
                 elif account.provider == 'CUSTOM':
                     message_id = send_smtp_email(
                         account,
-                        clead.lead.email,
+                        recipient_email,
                         subject,
                         body,
                         unsubscribe_url=build_unsubscribe_url(clead.lead),
                     )
-                    logger.info(f"SMTP SENT to {clead.lead.email} | msg_id={message_id}")
+                    logger.info(f"SMTP SENT to {recipient_email} | msg_id={message_id}")
                 else:
                     raise RuntimeError(f"Unsupported email provider: {account.provider}")
                 clead.last_sent_message_id = message_id
                 clead.save(update_fields=['last_sent_message_id'])
+
+                if sandbox_mode:
+                    logger.info(
+                        "Sandbox Email Sent\n"
+                        "  Original Recipient: %s\n"
+                        "  Delivered To:       %s\n"
+                        "  Subject:            %s\n"
+                        "  Campaign:           %s (step %s)",
+                        original_recipient, recipient_email, subject,
+                        clead.campaign.name, step.step_order,
+                    )
             except Exception as send_err:
-                logger.error(f"Email send failed for {clead.lead.email}: {send_err}")
+                logger.error(f"Email send failed for {recipient_email}: {send_err}")
                 # Restore next_execution_time so the lead can be retried later.
                 clead.next_execution_time = timezone.now() + timedelta(minutes=15)
                 clead.save(update_fields=['next_execution_time'])
                 return
         else:
-            logger.info(f"Mock SENDING EMAIL to {clead.lead.email} | Subject: {subject}")
+            if sandbox_mode:
+                logger.info(
+                    f"Mock SANDBOX SENDING EMAIL to {recipient_email} "
+                    f"(would have gone to {original_recipient}) | Subject: {subject}"
+                )
+            else:
+                logger.info(f"Mock SENDING EMAIL to {recipient_email} | Subject: {subject}")
 
+        # Sandbox sends still advance the lead through the workflow normally,
+        # so reps see the exact same progression/timing a real send would produce.
         _advance_to_next_step(clead, step)
 
     except Exception as e:
@@ -861,4 +951,4 @@ def check_imap_bounces():
                         f"Failed to mark bounce email {message_id} as read for {account.email_address}: {exc}"
                     )
 
-    return f"Processed {scanned_messages} bounce emails and marked {total_bounced} campaign leads as BOUNCED."
+    return f"Processed {scanned_messages} bounce emails and marked {total_bounced} campaign leads as BOUNCED."

backend/campaigns/views.py

@@ -47,7 +47,7 @@ def get_queryset(self):
         )
 
     def perform_create(self, serializer):
-        serializer.save(organization=self.request.user.organization)
+        serializer.save(organization=self.request.user.organization, created_by=self.request.user)
 
     @action(detail=True, methods=['post'])
     def enroll(self, request, pk=None):
@@ -812,4 +812,4 @@ def get(self, request, *args, **kwargs):
         # Original Destination par redirect karna
         decoded_dest = urllib.parse.unquote(dest_url)
         return HttpResponseRedirect(decoded_dest)
-# ------------------------------------------
+# ------------------------------------------