trucking: compute coupon discounted prices on the fly (true per A/B arm) + fix CTA URL bug

Two correctness fixes that gate enabling the coupon test:

1. On-the-fly pricing. The coupon block hardcoded '$79 $47' (only true at 40%
   off) — a false claim on the 20/30% arms. Now build_trucking_campaigns.py
   reads api/src/service-catalog.ts (same source checkout uses) and computes
   coupon_price_full / coupon_price_deal per recipient as full - round(full*pct/100),
   exactly matching the server. Service-fee-only; non-discountable services
   (boc3-filing passthrough) get NO price and fall back to percent-only copy.
   Quotes the service the email is ABOUT (mcs150 $79, reactivation $149), not the
   bundle the CTA happens to link to. service-catalog.ts now ships in the worker
   image; helper degrades to percent-only if it can't be read.

2. CTA URL bug (likely a big driver of the zero-click problem). Main campaign
   CTAs render '/order/slug&utm_source=...' (no '?') -> HTTP 404, verified live.
   Deficiency CTAs would double-'?' once a coupon added '?code='. lp_link now
   owns the query (?dot=...&code=...) so every template appends with a leading
   '&' and is valid in all 4 states (main/deficiency x coupon on/off), verified
   against live URLs returning 200.

Deficiency _deal_box now shows real was/now prices (percent-only for boc3).
Tests: 7/7 pass (adds URL-wellformed + price-matches-checkout cases).

scripts/Dockerfile

@@ -20,6 +20,9 @@ RUN playwright install chromium && playwright install-deps chromium
 # Copy all scripts
 COPY scripts/ /app/scripts/
 COPY docs/product-facts.md /app/docs/product-facts.md
+# Authoritative service prices: build_trucking_campaigns.py reads this to compute
+# coupon discounted prices that match what checkout charges (service-fee only).
+COPY api/src/service-catalog.ts /app/api/src/service-catalog.ts
 COPY ["docs/MCS-150 Form.pdf", "/app/docs/MCS-150 Form.pdf"]
 COPY ["docs/MCS-150B Form.pdf", "/app/docs/MCS-150B Form.pdf"]
 COPY ["docs/MCS-150C Form.pdf", "/app/docs/MCS-150C Form.pdf"]

scripts/build_trucking_campaigns.py

@@ -127,6 +127,109 @@ _WEIGHT_TAX_LP = {
 }
 
 
+# ── Authoritative service prices (single source of truth) ───────────────────
+# The discounted prices shown in the coupon email MUST match what checkout
+# actually charges, so we read them from the same catalog the API uses
+# (api/src/service-catalog.ts). We parse it directly (the prod box has python3
+# but not node) and cache per-process. Each entry: {price_cents, discountable}.
+# If the catalog can't be read (file missing in an image), the price helpers
+# degrade to percent-only copy rather than guessing a number.
+_CATALOG_PATH = os.getenv(
+    "SERVICE_CATALOG_TS", os.path.join(ROOT, "api", "src", "service-catalog.ts")
+)
+_CATALOG_CACHE: dict | None = None
+
+
+def _load_service_catalog() -> dict:
+    """Parse api/src/service-catalog.ts -> {slug: {price_cents, discountable}}.
+
+    discountable defaults to True (the catalog only marks the exceptions with
+    `discountable: false`), matching the TS object's own semantics.
+    """
+    global _CATALOG_CACHE
+    if _CATALOG_CACHE is not None:
+        return _CATALOG_CACHE
+    catalog: dict = {}
+    try:
+        import re as _re
+        ts = open(_CATALOG_PATH, encoding="utf-8").read()
+        m = _re.search(r"export const COMPLIANCE_SERVICES[^=]*=\s*\{(.*)\n\};", ts, _re.S)
+        body = m.group(1) if m else ""
+        for em in _re.finditer(r'"([a-z0-9\-]+)":\s*\{(.*?)\}', body, _re.S):
+            slug, inner = em.group(1), em.group(2)
+            pm = _re.search(r"price_cents:\s*(\d+)", inner)
+            if not pm:
+                continue
+            discountable = _re.search(r"discountable:\s*false", inner) is None
+            catalog[slug] = {
+                "price_cents": int(pm.group(1)),
+                "discountable": discountable,
+            }
+    except Exception as exc:  # noqa: BLE001
+        LOG.warning("[coupon] could not load service catalog (%s); coupon copy "
+                    "will be percent-only", exc)
+    _CATALOG_CACHE = catalog
+    return catalog
+
+
+# The service whose price the coupon copy quotes. This is the specific service
+# each campaign's body is *about*, which is NOT always the slug the CTA links to
+# (e.g. the MCS-150 email talks about the $79 MCS-150 update but the button opens
+# the $399 full-compliance bundle). Pricing from the wrong slug would advertise a
+# discount the landing page doesn't show, so the price slug is explicit here.
+PRICE_SLUG_BY_CAMPAIGN = {
+    "mcs150": "mcs150-update",
+    "inactive": "usdot-reactivation",
+}
+
+
+def price_slug_for(campaign_type: str, phy_state: str | None = None) -> str:
+    """The catalog slug whose price the coupon copy should quote for a segment.
+
+    Main campaigns quote their specific service (MCS-150, reactivation); the
+    deficiency segments quote the same slug their CTA links to (resolved, incl.
+    per-state overrides, by lp_slug_for)."""
+    return PRICE_SLUG_BY_CAMPAIGN.get(campaign_type) or lp_slug_for(campaign_type, phy_state)
+
+
+def discounted_price_attribs(campaign_type: str, phy_state: str | None,
+                             coupon_pct: str | None) -> dict:
+    """Per-recipient price merge fields, computed on the fly to match checkout.
+
+    Mirrors the server exactly: percent discount on the SERVICE fee only
+    (discount_cents = round(fee * pct / 100)); non-discountable services (e.g.
+    boc3-filing, a $25 passthrough) get NO price discount. Returns:
+      coupon_price_full   "$79"     (service list price)
+      coupon_price_deal   "$47"     (after discount)
+      coupon_priceable    "1"/""    (whether a real discounted number is available)
+    Blank when no coupon, no catalog entry, or the service isn't discountable, so
+    the templates fall back to percent-only copy and never print a false number.
+    """
+    blank = {"coupon_price_full": "", "coupon_price_deal": "", "coupon_priceable": ""}
+    if not coupon_pct:
+        return blank
+    try:
+        pct = int(coupon_pct)
+    except (TypeError, ValueError):
+        return blank
+    slug = price_slug_for(campaign_type, phy_state)
+    entry = _load_service_catalog().get(slug)
+    if not entry or not entry.get("discountable") or entry["price_cents"] <= 0:
+        return blank
+    full = entry["price_cents"]
+    discount = round(full * pct / 100)          # same formula as the API
+    deal = full - discount
+
+    def _fmt(cents: int) -> str:
+        return f"${cents // 100}" if cents % 100 == 0 else f"${cents / 100:.2f}"
+
+    return {
+        "coupon_price_full": _fmt(full),
+        "coupon_price_deal": _fmt(deal),
+        "coupon_priceable": "1",
+    }
+
+
 def build_lp_link(campaign_type: str, phy_state: str | None) -> str:
     """Return the order landing-page URL for a (segment, state)."""
     seg = DEFICIENCY_SEGMENTS.get(campaign_type)
@@ -384,12 +487,27 @@ def coupon_attribs(coupon_code: str | None, coupon_pct: str | None = None) -> di
 
 
 def lp_link_with_coupon(campaign_type: str, phy_state: str | None,
-                        coupon_code: str | None) -> str:
+                        coupon_code: str | None, dot: str | None = None) -> str:
-    """build_lp_link + a ?code= query param so the LP pre-applies the deal."""
+    """Order landing-page URL as the email's `lp_link` attrib.
+
+    Always emits a `?`-started query (carrying the carrier's DOT, plus the daily
+    `code=` when a coupon is active) so that EVERY template can safely append its
+    own params with a leading `&`. This eliminates a class of broken-CTA bugs:
+    previously `build_lp_link()` returned a bare path, so a template that wrote
+    `{{ lp_link }}&utm_source=...` produced `/order/slug&utm_source=...` (an
+    invalid URL that 404s), and one that wrote `{{ lp_link }}?dot=...` double-`?`d
+    once a coupon added its own `?code=`. With the query owned here, both template
+    styles converge on `{{ lp_link }}&utm_source=...` and are correct whether or
+    not the coupon is on.
+    """
     url = build_lp_link(campaign_type, phy_state)
+    params = []
+    if dot:
+        params.append(f"dot={dot}")
     if coupon_code:
-        sep = "&" if "?" in url else "?"
+        params.append(f"code={coupon_code}")
-        url = f"{url}{sep}code={coupon_code}"
+    if params:
+        url = f"{url}?" + "&".join(params)
     return url
 
 # ── TZ config: tz_key -> (states, send_hour_utc) ─────────────────────────────
@@ -1309,8 +1427,9 @@ def run(send_date: date, dry_run: bool = False, preview: bool = False,
                     "email": TEST_EMAIL,
                     "name": r0[2] or "Sample Carrier",
                     "attribs": {"dot_number": r0[0], "company": r0[2] or "", "state": r0[3] or "",
-                                "lp_link": lp_link_with_coupon(campaign_type, r0[4], p_code),
+                                "lp_link": lp_link_with_coupon(campaign_type, r0[4], p_code, dot=str(r0[0])),
-                                **coupon_attribs(p_code, p_pct)},
+                                **coupon_attribs(p_code, p_pct),
+                                **discounted_price_attribs(campaign_type, r0[4], p_pct)},
                 }]
             else:
                 subscribers = []
@@ -1320,8 +1439,9 @@ def run(send_date: date, dry_run: bool = False, preview: bool = False,
                         "email": row[1],
                         "name": row[2] or row[1],
                         "attribs": {"dot_number": row[0], "company": row[2] or "", "state": row[3] or "",
-                                    "lp_link": lp_link_with_coupon(campaign_type, row[4], c_code),
+                                    "lp_link": lp_link_with_coupon(campaign_type, row[4], c_code, dot=str(row[0])),
-                                    **coupon_attribs(c_code, c_pct)},
+                                    **coupon_attribs(c_code, c_pct),
+                                    **discounted_price_attribs(campaign_type, row[4], c_pct)},
                     })
 
             # Create list + add subscribers

scripts/create_deficiency_source_campaigns.py

@@ -78,12 +78,59 @@ def _price_box(headline, sub):
     )
 
 
+def _deal_box(headline, sub):
+    """Coupon-aware offer box.
+
+    When the daily same-day coupon is active (coupon_code attrib set), show the
+    deal. If the build step computed a real discounted price for this service
+    (coupon_priceable set — true for discountable services), show the exact
+    "was $X, now $Y" numbers, which are calculated on the fly to match what
+    checkout charges (service fee only) for the current A/B arm. If the service
+    isn't discountable (e.g. BOC-3, a $25 passthrough) the price attribs are
+    blank and we fall back to percent-only copy so we never print a false price.
+    With no coupon at all it shows the normal price box.
+    """
+    priced = (
+        f'<p style="font-size:18px;font-weight:700;color:#9a3412;margin:0 0 6px">{headline} '
+        'for <span style="text-decoration:line-through;color:#c2410c;font-weight:600">'
+        '{{ .Subscriber.Attribs.coupon_price_full }}</span> '
+        '<span style="color:#15803d">{{ .Subscriber.Attribs.coupon_price_deal }}</span>.</p>'
+    )
+    unpriced = (
+        f'<p style="font-size:18px;font-weight:700;color:#9a3412;margin:0 0 6px">{headline}</p>'
+    )
+    deal = (
+        '<div style="background:#fff7ed;border:2px solid #f97316;border-radius:10px;'
+        'padding:20px;margin:20px 0;text-align:center">'
+        '<p style="font-size:13px;font-weight:700;color:#9a3412;letter-spacing:.04em;'
+        'margin:0 0 6px">TODAY ONLY - {{ .Subscriber.Attribs.coupon_pct }}% OFF OUR SERVICE FEE</p>'
+        '{{ if .Subscriber.Attribs.coupon_priceable }}' + priced + '{{ else }}' + unpriced + '{{ end }}'
+        '<p style="font-size:14px;color:#9a3412;margin:0 0 4px">Use code '
+        '<strong style="font-size:16px;letter-spacing:.08em">{{ .Subscriber.Attribs.coupon_code }}</strong> '
+        '(already applied when you click below).</p>'
+        '<p style="font-size:12px;color:#b91c1c;font-weight:700;margin:0">'
+        'Expires {{ .Subscriber.Attribs.coupon_expires }}. Discount applies to our '
+        'service fee; government filing fees are billed at cost.</p></div>'
+    )
+    return (
+        '{{ if .Subscriber.Attribs.coupon_code }}'
+        + deal
+        + '{{ else }}'
+        + _price_box(headline, sub)
+        + '{{ end }}'
+    )
+
+
 def _cta(label):
     # Clicks land on the per-subscriber order page (resolved per row, incl.
-    # per-state overrides) via the lp_link attrib, with UTM tracking.
+    # per-state overrides) via the lp_link attrib. lp_link already carries the
+    # carrier's `?dot=` query (and the daily `?code=` when a coupon is active),
+    # so the template appends its own params with a leading `&` — correct whether
+    # or not the coupon is on. (Previously this used `?dot=`, which double-`?`d
+    # the URL once the coupon added its own query.)
     return (
         '<div style="text-align:center;margin:24px 0">'
-        '<a href="{{ .Subscriber.Attribs.lp_link }}?dot={{ .Subscriber.Attribs.dot_number }}'
+        '<a href="{{ .Subscriber.Attribs.lp_link }}'
         '&utm_source=listmonk&utm_medium=email&utm_campaign=deficiency@TrackLink" '
         'style="display:inline-block;padding:14px 36px;background:#f97316;color:#fff;'
         f'font-weight:700;border-radius:8px;text-decoration:none;font-size:16px">{label} &rarr;</a></div>'
@@ -124,7 +171,7 @@ def _body(headline, intro, bullets, price_headline, price_sub, cta_label):
         'computer. No Login.gov, no government portals, no hours on hold. We handle the paperwork so you can get '
         'back to trucking and making money.</p></div>'
     )
-    return (_HEADER + alert + _price_box(price_headline, price_sub) + reassure
+    return (_HEADER + alert + _deal_box(price_headline, price_sub) + reassure
             + _cta(cta_label) + _dot_check_cta() + _FOOTER)
 
 

scripts/tests/test_coupon_ab.py

@@ -68,6 +68,52 @@ def test_coupon_attribs_reflects_pct():
     }
 
 
+def test_lp_link_well_formed_in_all_states():
+    """The CTA URL must be valid whether or not a coupon is active. lp_link owns
+    the `?`-query so templates append their own params with a leading `&`."""
+    import urllib.parse
+
+    def cta(lp, lead):  # mimic a template appending its own params
+        return f"{lp}{lead}utm_source=listmonk&utm_campaign=x"
+
+    for code in (None, "KQMTN"):
+        for ct, st in (("mcs150", None), ("irp_ifta", "CT")):
+            lp = btc.lp_link_with_coupon(ct, st, code, dot="1228791")
+            url = cta(lp, "&")
+            parts = urllib.parse.urlsplit(url)
+            assert "&" not in parts.path, url          # no stray & in path
+            assert parts.query.count("?") == 0, url    # no double ?
+            assert "dot=1228791" in parts.query, url
+            if code:
+                assert f"code={code}" in parts.query, url
+
+
+def test_discounted_price_matches_checkout_formula():
+    """coupon_price_deal must equal full - round(full*pct/100) for discountable
+    services, and be blank for non-discountable ones (e.g. boc3-filing)."""
+    cat = btc._load_service_catalog()
+    assert cat, "service catalog failed to load"
+    for ct, st in (("mcs150", None), ("inactive", None), ("irp_ifta", "CT"),
+                   ("hazmat", None), ("state_weight_tax", "NY"), ("state_emissions", "CA")):
+        slug = btc.price_slug_for(ct, st)
+        entry = cat[slug]
+        for pct in ("20", "30", "40"):
+            a = btc.discounted_price_attribs(ct, st, pct)
+            full = entry["price_cents"]
+            deal = full - round(full * int(pct) / 100)
+            exp_full = f"${full // 100}" if full % 100 == 0 else f"${full / 100:.2f}"
+            exp_deal = f"${deal // 100}" if deal % 100 == 0 else f"${deal / 100:.2f}"
+            assert a["coupon_price_full"] == exp_full, (ct, pct, a)
+            assert a["coupon_price_deal"] == exp_deal, (ct, pct, a)
+            assert a["coupon_priceable"] == "1"
+
+    # Non-discountable passthrough -> no price, percent-only fallback.
+    boc = btc.discounted_price_attribs("for_hire_boc3", None, "40")
+    assert boc == {"coupon_price_full": "", "coupon_price_deal": "", "coupon_priceable": ""}
+    # No coupon -> blank.
+    assert btc.discounted_price_attribs("mcs150", None, "")["coupon_priceable"] == ""
+
+
 if __name__ == "__main__":
     fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
     for fn in fns: