Verify bounces before acting on them; refresh the dashboard GIF

CHANGELOG.md

@@ -7,6 +7,15 @@ minor versions can still change behaviour.
 
 ## [Unreleased]
 
+### Security
+
+- **Bounces are verified before Franco acts on them.** Anyone can send a message that looks like a bounce, and acting on one marks a contact's address invalid and moves the bounce rate toward the kill switch. A delivery report is acted on only when it references the Message-ID of an email Franco actually sent (`In-Reply-To` or `References`) and names that email's recipient. A report that matches only by naming an address Franco emailed in the last 14 days could be forged by anyone who knows whom Franco emails, so it is logged as `bounce_needs_review` for a person to check, and changes nothing. Unmatched reports are logged as `bounce_ignored`.
+- Temporary delivery delays ("delayed", `4.x.x` statuses) are no longer treated as hard bounces, and the same bounce is never counted twice.
+
+### Changed
+
+- The README's dashboard GIF shows the current Outbox: approved-copy campaigns in preview, sign-off cards, and each draft's campaign, owner, sourced observation and compliance footer.
+
 ## [0.5.0] — 2026-10-16
 
 ### Added

franco/agents/handler.py

@@ -500,12 +500,67 @@ async def _run_native(self) -> bool:
             logger.info("Handler: no new replies.")
         return complete
 
+    async def _match_bounce(self, msg) -> tuple[dict | None, str, str]:
+        """The sent email a delivery report is about, how it was matched
+        ("message-id" or "address"), or why it isn't trusted.
+
+        Anyone can send mail that *looks* like a bounce. Acting on it marks a
+        contact's address invalid and moves the bounce rate toward the kill
+        switch, so a report must point at an email Franco really sent and
+        name that email's recipient. Only a Message-ID match is acted on
+        automatically: Franco's Message-IDs reach nobody but the recipient.
+        A match by address alone could be forged by anyone who knows whom
+        Franco emails, so it is left for a person to review.
+        """
+        from franco.integrations.mail_provider import parse_bounce, referenced_ids
+
+        info = parse_bounce(msg)
+        item, matched_by = None, "message-id"
+        for ref in referenced_ids(msg):
+            item = await self.state.find_outbox_by_message_id(ref)
+            if item:
+                break
+        if item is None:
+            matched_by = "address"
+            for address in sorted(info.recipients):
+                item = await self.state.find_recent_sent_to(address)
+                if item:
+                    break
+        if item is None or item.get("status") != "sent":
+            return None, "matches no email Franco sent", ""
+        if info.recipients and item["to_email"].lower() not in info.recipients:
+            return None, "names a different recipient than the email it refers to", ""
+        if (item.get("error") or "") == "bounced":
+            return None, "already counted", ""
+        if not info.permanent:
+            return None, "a temporary delay, not a failure", ""
+        return item, "", matched_by
+
     async def _handle_bounce(self, msg):
-        """A bounce is a data bug AND a reputation threat. Fix both."""
-        outbox_item = await self.state.find_outbox_by_message_id(msg.in_reply_to)
-        prospect = None
-        if outbox_item:
-            prospect = await self.state.get_prospect(outbox_item["prospect_id"])
+        """A bounce is a data bug AND a reputation threat. Fix both, but
+        only for reports that match an email Franco really sent."""
+        outbox_item, why_not, matched_by = await self._match_bounce(msg)
+        if outbox_item is None:
+            logger.info(f"Handler: delivery report not acted on: {why_not}.")
+            await self.state.log_action(
+                action_type="bounce_ignored", agent="handler",
+                details={"reason": why_not, "subject": (msg.subject or "")[:80]},
+            )
+            return
+        if matched_by != "message-id":
+            # Plausible, but forgeable: flag it, change nothing.
+            logger.warning(
+                f"Handler: possible bounce for {outbox_item['to_email']} matched by "
+                "address only; not acted on automatically. Check the mailbox, then "
+                "suppress the address if it really bounced.")
+            await self.state.log_action(
+                action_type="bounce_needs_review", agent="handler",
+                details={"to": outbox_item["to_email"], "outbox_id": outbox_item["id"],
+                         "from": msg.from_email, "subject": (msg.subject or "")[:80]},
+            )
+            return
+        await self.state.update_outbox_item(outbox_item["id"], error="bounced")
+        prospect = await self.state.get_prospect(outbox_item["prospect_id"])
 
         if prospect:
             await self.state.update_prospect_email(

franco/integrations/mail_provider.py

@@ -14,6 +14,7 @@
 """
 
 import logging
+import re
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
 
@@ -53,11 +54,55 @@ class InboundMessage:
 
 
 def looks_like_bounce(from_email: str, subject: str) -> bool:
+    """A *candidate* bounce. Anyone can send a message that looks like one,
+    so the handler only acts on it after matching it to an email Franco
+    actually sent (see parse_bounce and referenced_ids)."""
     f = (from_email or "").lower()
     s = (subject or "").lower()
     return any(m in f for m in BOUNCE_SENDERS) or any(m in s for m in BOUNCE_SUBJECTS)
 
 
+_DSN_RECIPIENT = re.compile(
+    r"^(?:final|original)-recipient:\s*rfc822;\s*<?([^>\s]+@[^>\s]+)>?", re.I | re.M)
+_DSN_STATUS = re.compile(r"^status:\s*([245])\.\d{1,3}\.\d{1,3}", re.I | re.M)
+_ADDRESS = re.compile(r"[\w.+'-]+@[\w-]+(?:\.[\w-]+)+")
+_TRANSIENT = ("delay", "will retry", "'ll retry", "keep trying", "still trying",
+              "temporary problem", "temporarily", "try again later")
+
+
+@dataclass
+class BounceInfo:
+    recipients: set[str]      # addresses the report says failed ("" set if none found)
+    permanent: bool           # False for "delayed, still trying" (4.x.x) notices
+
+
+def parse_bounce(msg: "InboundMessage") -> BounceInfo:
+    """Who a delivery report says failed, and whether it gave up.
+
+    Uses the RFC 3464 fields when the text carries them (Final-Recipient,
+    Status), else the addresses in the text and its wording.
+    """
+    text = f"{msg.subject}\n{msg.body}"
+    recipients = {a.lower() for a in _DSN_RECIPIENT.findall(text)}
+    if not recipients:
+        recipients = {a.lower().strip(".") for a in _ADDRESS.findall(msg.body or "")}
+    statuses = _DSN_STATUS.findall(text)
+    if statuses:
+        permanent = "5" in statuses
+    else:
+        lowered = text.lower()
+        permanent = not any(hint in lowered for hint in _TRANSIENT)
+    return BounceInfo(recipients=recipients, permanent=permanent)
+
+
+def referenced_ids(msg: "InboundMessage") -> list[str]:
+    """Message-IDs the report points back to (In-Reply-To, then References)."""
+    ids = [msg.in_reply_to] if msg.in_reply_to else []
+    refs = (msg.headers or {}).get("references", "")
+    ids += [r for r in re.findall(r"<[^>]+>", refs) if r not in ids]
+    return [i.strip() for i in ids if i.strip()]
+
+
 class MailProvider(ABC):
     """A mailbox Franco can send from and read replies out of."""
 

franco/integrations/smtp_mail.py

@@ -154,6 +154,8 @@ def _parse_rfc822(raw: bytes) -> InboundMessage | None:
             thread_ref=str(msg.get("In-Reply-To", "")).strip(),
             date=str(msg.get("Date", "")),
             is_bounce=looks_like_bounce(from_email, subject),
+            headers={k.lower(): str(msg.get(k, "")) for k in
+                     ("References", "Content-Type", "Auto-Submitted") if msg.get(k)},
         )
 
     async def test_connection(self) -> tuple[bool, str]:

franco/state.py

@@ -12,7 +12,7 @@
 import json
 import uuid
 from contextlib import asynccontextmanager
-from datetime import datetime, date, timezone
+from datetime import datetime, date, timedelta, timezone
 from pathlib import Path
 
 import aiosqlite
@@ -1253,6 +1253,18 @@ async def count_outbox_sent_today(self) -> int:
             ) as cursor:
                 return (await cursor.fetchone())[0]
 
+    async def find_recent_sent_to(self, email: str, days: int = 14) -> dict | None:
+        """The newest email Franco actually sent to this address lately."""
+        async with self._connect() as db:
+            db.row_factory = aiosqlite.Row
+            async with db.execute(
+                "SELECT * FROM outbox WHERE to_email = ? AND status = 'sent' "
+                "AND sent_at >= ? ORDER BY sent_at DESC LIMIT 1",
+                (_norm(email), (_utcnow() - timedelta(days=days)).isoformat()),
+            ) as cursor:
+                row = await cursor.fetchone()
+                return dict(row) if row else None
+
     async def find_outbox_by_message_id(self, message_id: str) -> dict | None:
         if not message_id:
             return None

scripts/record_demo.py

@@ -36,6 +36,7 @@
 REPO = Path(__file__).resolve().parent.parent
 sys.path.insert(0, str(REPO))
 
+from franco.models.campaign import Campaign, EmailStep  # noqa: E402
 from franco.models.company import Company  # noqa: E402
 from franco.models.prospect import Prospect  # noqa: E402
 from franco.signals import seed_signal_catalog  # noqa: E402
@@ -103,27 +104,33 @@
             "part of a deliberate discovery experience, rather than a separate "
             "insertion process.")
 
-# (contact index, subject, observation, pitch, close)
+# (contact index, subject, observation, pitch, close, source page)
 OUTBOX = [
     (0, "Sponsored listings at Tradeloft",
      "Tradeloft's seller help center describes a paid Featured Listing upgrade",
-     M1_PITCH, "Would a short outline of a one-surface pilot be useful?"),
+     M1_PITCH, "Would a short outline of a one-surface pilot be useful?", "/help/featured-listings"),
     (4, "Sponsored posts at Threadwise",
      "Threadwise's advertise page offers sponsored posts in the main developer feed",
-     F1_PITCH, "Would a short native-sponsorship outline be useful?"),
+     F1_PITCH, "Would a short native-sponsorship outline be useful?", "/advertise"),
     (1, "Sponsored listings at Petlane",
      "Petlane is hiring a Head of Monetization",
-     M1_PITCH, "Would a short outline of a one-surface pilot be useful?"),
+     M1_PITCH, "Would a short outline of a one-surface pilot be useful?", "/careers"),
     (5, "Sponsored posts at Climbline",
      "Climbline's partner page lists gear brands sponsoring the route feed",
-     F1_PITCH, "Would a short native-sponsorship outline be useful?"),
+     F1_PITCH, "Would a short native-sponsorship outline be useful?", "/partners"),
 ]
 
+# The compliance footer every cold email carries (sender identity is real;
+# everything else in the demo is invented).
+FOOTER = ("Hemin Joshi | Gortex | 440 N Wolfe Rd, Suite E133, Sunnyvale, CA, 94085\n"
+          "Business outreach about Gortex. Reply \"no thanks\" to stop.")
+
 PROPOSED = {"SERP_RANK", "LIKELY_OWNER", "REGISTRY_VERIFIED"}  # left for the viewer
 
 
 def email_body(first: str, observation: str, pitch: str, close: str) -> str:
-    return (f"Hi {first},\n\nI noticed {observation}.\n\n{pitch}\n\n{close}\n\nHemin")
+    return (f"Hi {first},\n\nI noticed {observation}.\n\n{pitch}\n\n{close}\n\nHemin"
+            f"\n\n{FOOTER}")
 
 
 def demo_config() -> dict:
@@ -146,6 +153,26 @@ def demo_config() -> dict:
                                       "Ads Product Manager"])
     cfg["channels"]["email"]["provider"] = "gmail"
     cfg["channels"]["linkedin"]["enabled"] = False
+    # Approved copy, as Gortex runs it: two campaigns, both still in preview.
+    step = lambda pitch, close: {"day": 1, "subject": "Sponsored listings at {{company}}",
+                                 "body": f"Hi {{{{first_name}}}},\n\nI noticed "
+                                         f"{{{{verified_observation}}}}.\n\n{pitch}\n\n{close}\n\nHemin"}
+    cfg["outreach"] = {
+        "mode": "templates", "timezone": "America/Los_Angeles", "send_hour": 9,
+        "preview_accounts": 10,
+        "campaigns": {
+            "M1": {"name": "Vertical marketplaces",
+                   "steps": [step(M1_PITCH, "Would a short outline of a one-surface pilot be useful?")]},
+            "F1": {"name": "Feed-based communities",
+                   "steps": [step(F1_PITCH, "Would a short native-sponsorship outline be useful?")]},
+        },
+    }
+    cfg["compliance"] = {
+        "require_footer": True,
+        "postal_address": "440 N Wolfe Rd, Suite E133, Sunnyvale, CA, 94085",
+        "footer": "{sender_name} | {company} | {postal_address}\n"
+                  "Business outreach about {company}. Reply \"no thanks\" to stop.",
+    }
     return cfg
 
 
@@ -214,11 +241,30 @@ async def seed(db_path: Path):
             email=emails[-1], email_verified=True, email_status="verified",
             company=name, industry=industry, source="team_page", score=8)))
 
+    # Researched accounts: a campaign, an owner, a score and the sourced,
+    # dated observation behind each first email (what `franco import` records).
+    campaign_ids: dict[str, str] = {}
+    for key, name in (("M1", "Vertical marketplaces"), ("F1", "Feed-based communities")):
+        campaign_ids[key] = await state.add_campaign(Campaign(
+            id="", name=name, channel="email", status="active", campaign_key=key,
+            sequence=[EmailStep(step=1, subject="(approved copy)", body="", day=1)],
+            prospect_ids=[]))
+    checked = (datetime.now() - timedelta(days=3)).strftime("%Y-%m-%d 00:00:00")
     send_at = datetime.now().replace(hour=9, minute=15, second=0) + timedelta(days=1)
-    for n, (pi, subject, observation, pitch, close) in enumerate(OUTBOX):
+    for n, (pi, subject, observation, pitch, close, page) in enumerate(OUTBOX):
+        ci = CONTACTS[pi][0]
+        domain, industry = COMPANIES[ci][1], COMPANIES[ci][2]
+        key = "M1" if pitch == M1_PITCH else "F1"
+        await state.update_company_account(company_ids[ci], campaign_key=key, owner="Hemin",
+                                           segment=industry, qualified_score=8 - n % 2)
+        await state.add_observation(
+            "TRIGGER_OBSERVATION", company_id=company_ids[ci], collector="import",
+            value_text=observation[0].lower() + observation[1:] if observation.startswith("The") else observation,
+            evidence_url=f"https://{domain}{page}", observed_at=checked, confidence=1.0)
         await state.add_outbox_item(
             prospect_id=prospect_ids[pi], to_email=emails[pi], subject=subject,
             body=email_body(CONTACTS[pi][1], observation, pitch, close),
+            campaign_id=campaign_ids[key], step=1, seq_day=1,
             provider="gmail", send_at=(send_at + timedelta(minutes=7 * n)).isoformat())
 
     for action in ("prospect", "write_campaign", "handle_replies", "send_outbox",