feat(intrastate): automate state PUC/PSC authority filing (email + invoice + auto-bill)

Intrastate operating authority is state-specific + application-based like IRP, so
it reuses the same email/POA + invoice-reconciliation flow:
  - intrastate_filing.send_intrastate_submission: emails the state PSC/PUC the
    authority application with the signed POA attached (subject tag [PW-ISA CO-..]),
    reusing irp_filing's MinIO download + census enrich helpers.
  - The shared poller (irp_invoice_poller) now matches BOTH [PW-IRP] and [PW-ISA]
    tags, parses the fee, Telegram-alerts, and bills the customer the exact amount
    with the correct service slug.
  - state_trucking gov-fee gate routes intrastate-authority to the PSC/PUC email
    path; if no submission email is configured for the base state it falls back
    to a manual todo (safe default — no emailing guessed agency addresses).

Per-state ISA_<ST>_EMAIL env (blank until the exact agency address is verified).
SC/GA/TX scaffolded. Customer still only sees an exact-fee payment link; you only
approve the final filing.

infra/ansible/roles/app/templates/app.env.j2

@@ -118,6 +118,12 @@ IRP_FILINGS_IMAP_PASS={{ vault_irp_filings_imap_pass | default('') }}
 IRP_FILINGS_IMAP_FOLDER={{ vault_irp_filings_imap_folder | default('INBOX') }}
 IRP_FILINGS_FROM={{ vault_irp_filings_from | default('filings@performancewest.net') }}
 IRP_SC_EMAIL={{ vault_irp_sc_email | default('MCS@scdmv.net') }}
+# Intrastate operating-authority (PSC/PUC) submission emails per base state.
+# Leave blank until the exact agency submission address is confirmed — the
+# worker then creates a manual todo instead of emailing a guessed address.
+ISA_SC_EMAIL={{ vault_isa_sc_email | default('') }}
+ISA_GA_EMAIL={{ vault_isa_ga_email | default('') }}
+ISA_TX_EMAIL={{ vault_isa_tx_email | default('') }}
 
 # ── Porkbun (.ca domain registration) ────────────────────────────────────────
 PORKBUN_API_KEY={{ vault_porkbun_api_key | default('') }}

scripts/workers/services/intrastate_filing.py

@@ -0,0 +1,156 @@
+"""Intrastate operating-authority correspondence + invoice reconciliation.
+
+Intrastate operating authority (PSC/PUC/state-DOT) is state-specific and
+application-based, much like IRP: there's no universal portal/API and the fee
+varies by state. We use the email/POA path (most convenient for the carrier):
+
+  send_intrastate_submission(...)
+      Email the state PUC/PSC/DOT an authority application request with the
+      signed POA + BOC-3 evidence, Reply-To the filings mailbox, tagged
+      [PW-ISA CO-XXXX] for reply matching.
+
+  Replies are handled by the shared poller (scripts.workers.irp_invoice_poller),
+  which now scans for BOTH [PW-IRP ...] and [PW-ISA ...] tags, parses the fee,
+  Telegram-alerts the operator, and bills the customer the exact amount.
+
+Reuses the generic helpers in irp_filing.py (MinIO download, census enrich,
+fee parse) so the two state-agency flows stay consistent.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import smtplib
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+from email.mime.application import MIMEApplication
+
+from scripts.workers.telegram_notify import send_telegram
+from scripts.workers.services.irp_filing import (
+    _download_minio, _enrich_address_from_census, FILINGS_FROM,
+    SMTP_HOST, SMTP_PORT, SMTP_USER, SMTP_PASS,
+)
+
+LOG = logging.getLogger("workers.services.intrastate_filing")
+
+SUBJECT_TAG = "PW-ISA"  # intrastate-authority reply tag
+
+# Per-state intrastate-authority agency contacts. email is what we submit to;
+# left BLANK until the exact PSC/PUC submission address is confirmed, so the
+# handler safely falls back to a manual todo instead of emailing a guessed
+# address. Set ISA_<ST>_EMAIL in env once verified.
+ISA_STATE_CONTACTS = {
+    "SC": {"agency": "South Carolina Public Service Commission",
+           "authority": "Certificate of Authority",
+           "email": os.getenv("ISA_SC_EMAIL", ""),
+           "portal": "https://www.psc.sc.gov/"},
+    "GA": {"agency": "Georgia Public Service Commission",
+           "authority": "Georgia Intrastate Motor Carrier (GIMC)",
+           "email": os.getenv("ISA_GA_EMAIL", ""),
+           "portal": "https://psc.ga.gov/"},
+    "TX": {"agency": "Texas Department of Motor Vehicles",
+           "authority": "Intrastate Operating Authority",
+           "email": os.getenv("ISA_TX_EMAIL", ""),
+           "portal": "https://www.txdmv.gov/motor-carriers"},
+}
+
+
+def state_isa_contact(base_state: str) -> dict | None:
+    """Return the intrastate contact ONLY if a submission email is configured;
+    otherwise None so the caller falls back to a manual todo."""
+    c = ISA_STATE_CONTACTS.get((base_state or "").upper())
+    return c if (c and c.get("email")) else None
+
+
+def _order_boc3_key(order_number: str, dot_number: str) -> str | None:
+    """Best-effort: find a BOC-3 evidence PDF for this carrier in MinIO. The
+    BOC-3 service stores filings under filings/boc3/<order>/...; we can't always
+    map carrier->boc3 order, so this is optional and degrades gracefully."""
+    # Intentionally light: most carriers tell us BOC-3 is on file. If we later
+    # store a per-carrier BOC-3 key we can resolve it here.
+    return None
+
+
+def send_intrastate_submission(order_number: str, entity_name: str, dot_number: str,
+                               base_state: str, intake: dict,
+                               signed_auth_key: str = "") -> bool:
+    """Email the state PUC/PSC/DOT an intrastate-authority application with the
+    signed POA attached. Returns True if sent (False -> caller falls back to a
+    manual todo, e.g. when the state has no email submission path)."""
+    contact = state_isa_contact(base_state)
+    if not contact or not contact.get("email"):
+        LOG.warning("[%s] No intrastate email contact for %s (portal-only) — manual todo",
+                    order_number, base_state)
+        return False
+
+    poa_bytes = _download_minio(signed_auth_key)
+    if not poa_bytes:
+        LOG.warning("[%s] No signed POA (key=%s) — not emailing PSC/PUC",
+                    order_number, signed_auth_key or "(none)")
+        return False
+
+    intake = _enrich_address_from_census(dot_number, intake)
+    entity_name = entity_name or intake.get("legal_name", "")
+    authority = contact.get("authority", "intrastate operating authority")
+    addr = ", ".join(p for p in [
+        intake.get("address_street", ""),
+        intake.get("address_city", ""),
+        f"{intake.get('address_state','')} {intake.get('address_zip','')}".strip(),
+    ] if p.strip())
+    boc3 = "yes" if str(intake.get("boc3_on_file", "")).lower() in ("yes", "true", "1") else "to be filed"
+
+    subject = (f"{authority} Application — {entity_name} (USDOT {dot_number}) "
+               f"[{SUBJECT_TAG} {order_number}]")
+    body = (
+        f"To {contact['agency']},\n\n"
+        f"On behalf of our client, and under the signed Power of Attorney attached, "
+        f"we request {authority} for the following intrastate for-hire motor "
+        f"carrier. Please reply with the application requirements and the filing "
+        f"fee invoice so we can submit supporting documents and remit payment.\n\n"
+        f"Carrier: {entity_name}\n"
+        f"USDOT: {dot_number}\n"
+        f"MC/MX/FF: {intake.get('mc_number','')}\n"
+        f"State: {base_state}\n"
+        f"Authority type: {intake.get('authority_type','common')}\n"
+        f"Power units: {intake.get('power_units','')}\n"
+        f"Registered address: {addr or '(see attached)'}\n"
+        f"BOC-3 process agent: {boc3}\n"
+        f"Insurance carrier: {intake.get('insurance_carrier','(to be provided on request)')}\n\n"
+        f"Attached: signed Power of Attorney authorizing Performance West Inc. to "
+        f"file and remit fees on the carrier's behalf.\n\n"
+        f"Please reply to {FILINGS_FROM} with the fee total and any required forms, "
+        f"keeping the subject reference [{SUBJECT_TAG} {order_number}].\n\n"
+        f"Thank you,\n"
+        f"Performance West Inc. — DOT / State Motor Carrier Compliance\n"
+        f"(888) 411-0383 · {FILINGS_FROM}\n"
+    )
+    try:
+        msg = MIMEMultipart()
+        msg["From"] = SMTP_USER
+        msg["To"] = contact["email"]
+        msg["Reply-To"] = FILINGS_FROM
+        msg["Subject"] = subject
+        msg.attach(MIMEText(body, "plain"))
+        poa = MIMEApplication(poa_bytes, _subtype="pdf")
+        poa.add_header("Content-Disposition", "attachment",
+                       filename=f"POA_{entity_name.replace(' ','_')}_{dot_number}.pdf")
+        msg.attach(poa)
+        with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=30) as s:
+            s.starttls()
+            if SMTP_USER and SMTP_PASS:
+                s.login(SMTP_USER, SMTP_PASS)
+            s.sendmail(SMTP_USER, [contact["email"], FILINGS_FROM], msg.as_string())
+        LOG.info("[%s] Intrastate authority application emailed to %s (%s) with POA",
+                 order_number, contact["email"], base_state)
+        send_telegram(
+            f"📤 Intrastate authority application sent (POA attached)\n"
+            f"{entity_name} (DOT {dot_number})\n"
+            f"{base_state} {authority} → {contact['email']}\n"
+            f"Order: {order_number}\nAwaiting the agency's requirements + fee."
+        )
+        return True
+    except Exception as exc:  # noqa: BLE001
+        LOG.error("[%s] Failed to send intrastate submission: %s", order_number, exc)
+        return False

scripts/workers/services/irp_filing.py

@@ -71,7 +71,10 @@ IRP_STATE_CONTACTS = {
 }
 
 SUBJECT_TAG = "PW-IRP"   # [PW-IRP CO-XXXXXXXX] in subject for reply matching
-TAG_RE = re.compile(r"\[PW-IRP\s+(C[OG]-[A-Z0-9]+)\]", re.I)
+# Match either state-agency tag: IRP or intrastate authority (ISA). Capture the
+# tag kind + the order number so the poller knows which service to bill.
+TAG_RE = re.compile(r"\[PW-(IRP|ISA)\s+(C[OG]-[A-Z0-9]+)\]", re.I)
+TAG_SLUG = {"IRP": "irp-registration", "ISA": "intrastate-authority"}
 # Fee patterns commonly seen on IRP invoices / replies.
 FEE_RE = [
     re.compile(r"(?:total\s+(?:fees?\s+)?due|amount\s+due|total\s+amount|apportioned\s+fees?)[^$]{0,40}\$\s*([0-9][0-9,]*\.?\d{0,2})", re.I),
@@ -265,12 +268,19 @@ def poll_irp_invoices() -> int:
         m = imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT, timeout=30)
         m.login(IMAP_USER, IMAP_PASS)
         m.select(IMAP_FOLDER)
-        # Unseen mail only. On a shared mailbox, restrict to our subject tag.
+        # Unseen mail only. On a shared mailbox, restrict to our subject tags.
         if DEDICATED_MAILBOX:
             typ, data = m.search(None, "UNSEEN")
+            ids = data[0].split() if data and data[0] else []
         else:
-            typ, data = m.search(None, "UNSEEN", "SUBJECT", SUBJECT_TAG)
+            ids = []
-        ids = data[0].split() if data and data[0] else []
+            seen_ids = set()
+            for tagword in ("PW-IRP", "PW-ISA"):
+                typ, data = m.search(None, "UNSEEN", "SUBJECT", tagword)
+                for i in (data[0].split() if data and data[0] else []):
+                    if i not in seen_ids:
+                        seen_ids.add(i)
+                        ids.append(i)
         for mid in ids:
             typ, md = m.fetch(mid, "(RFC822)")
             if typ != "OK" or not md or not md[0]:
@@ -279,11 +289,14 @@ def poll_irp_invoices() -> int:
             subject = str(email.header.make_header(email.header.decode_header(msg.get("Subject", ""))))
             tag = TAG_RE.search(subject)
             if not tag:
-                # Not an IRP reply we can match; leave it unseen on shared mailbox.
+                # Not a state-agency reply we can match; leave unseen on shared mailbox.
                 if not DEDICATED_MAILBOX:
                     m.store(mid, "-FLAGS", "(\\Seen)")
                 continue
-            parent_order = tag.group(1).upper()
+            tag_kind = tag.group(1).upper()
+            parent_order = tag.group(2).upper()
+            slug = TAG_SLUG.get(tag_kind, "irp-registration")
+            kind_label = "IRP" if tag_kind == "IRP" else "Intrastate authority"
             body = _body_text(msg)
             fee_cents = _parse_fee_cents(subject + "\n" + body)
             sender = msg.get("From", "")
@@ -291,7 +304,7 @@ def poll_irp_invoices() -> int:
             if not fee_cents:
                 # Could not parse a fee — alert the operator to read it manually.
                 send_telegram(
-                    f"📬 IRP reply received (no fee auto-parsed)\n"
+                    f"📬 {kind_label} reply received (no fee auto-parsed)\n"
                     f"Order: {parent_order}\nFrom: {sender}\n"
                     f"Subject: {subject}\nOpen the {IMAP_USER} mailbox to review + enter the fee."
                 )
@@ -299,10 +312,10 @@ def poll_irp_invoices() -> int:
                 processed += 1
                 continue
 
-            ok = _bill_parent_irp_fee(parent_order, fee_cents, sender,
+            ok = _bill_parent_fee(parent_order, slug, kind_label, fee_cents, sender,
-                                      create_gov_fee_order, send_gov_fee_payment_email, GovFeeEstimate)
+                                  create_gov_fee_order, send_gov_fee_payment_email, GovFeeEstimate)
             send_telegram(
-                f"💵 IRP invoice received → customer billed\n"
+                f"💵 {kind_label} invoice received → customer billed\n"
                 f"Order: {parent_order}\nState fee: ${fee_cents/100:,.2f}\n"
                 f"From: {sender}\n"
                 + ("Payment link emailed to the customer." if ok else "⚠️ Could not auto-bill — check logs.")
@@ -312,12 +325,12 @@ def poll_irp_invoices() -> int:
         m.logout()
     except Exception as exc:  # noqa: BLE001
         LOG.error("[irp-poll] IMAP error: %s", exc)
-    LOG.info("[irp-poll] processed %s IRP invoice reply(ies)", processed)
+    LOG.info("[irp-poll] processed %s state-agency invoice reply(ies)", processed)
     return processed
 
 
-def _bill_parent_irp_fee(parent_order, fee_cents, sender,
+def _bill_parent_fee(parent_order, slug, kind_label, fee_cents, sender,
-                         create_gov_fee_order, send_gov_fee_payment_email, GovFeeEstimate) -> bool:
+                     create_gov_fee_order, send_gov_fee_payment_email, GovFeeEstimate) -> bool:
     """Create the exact-amount gov-fee child + email the customer the payment link."""
     try:
         import psycopg2
@@ -334,22 +347,22 @@ def _bill_parent_irp_fee(parent_order, fee_cents, sender,
         LOG.error("[irp-poll] DB lookup failed for %s: %s", parent_order, exc)
         return False
     if not row:
-        LOG.warning("[irp-poll] No parent order %s for IRP invoice", parent_order)
+        LOG.warning("[irp-poll] No parent order %s for %s invoice", parent_order, kind_label)
         return False
     customer_email, customer_name, customer_phone, service_name = row
 
     est = GovFeeEstimate(
         cents=fee_cents,
-        label=f"IRP apportioned registration fee (state invoice) — {sender}",
+        label=f"{kind_label} fee (state invoice) — {sender}",
         exact=True,
-        breakdown=[f"State IRP invoice: ${fee_cents/100:,.2f}"],
+        breakdown=[f"State {kind_label} invoice: ${fee_cents/100:,.2f}"],
     )
-    child = create_gov_fee_order(parent_order, "irp-registration", est,
+    child = create_gov_fee_order(parent_order, slug, est,
                                  customer_email, customer_name or "", customer_phone or "")
     if not child:
         return False
     send_gov_fee_payment_email(customer_email, customer_name or "",
-                               service_name or "IRP Registration",
+                               service_name or kind_label,
                                customer_name or "", est, child)
     return True
 

scripts/workers/services/state_trucking.py

@@ -751,6 +751,39 @@ class StateTruckingHandler:
                     pass
             return sent
 
+        # ── Intrastate authority: email the PSC/PUC, wait for the fee invoice ──
+        # (Falls through to manual todo when the state has no email submission
+        # path — handled by returning False below.)
+        if service_slug == "intrastate-authority":
+            try:
+                from scripts.workers.services.intrastate_filing import (
+                    send_intrastate_submission, state_isa_contact,
+                )
+            except Exception as exc:  # noqa: BLE001
+                LOG.error("[%s] intrastate_filing import failed: %s", order_number, exc)
+                return False
+            base_state = (intake.get("base_state") or intake.get("address_state") or "").upper()
+            if not state_isa_contact(base_state):
+                LOG.info("[%s] No intrastate email contact for %s — manual todo", order_number, base_state)
+                return False
+            sent = send_intrastate_submission(order_number, entity_name,
+                                              intake.get("dot_number", ""), base_state, intake,
+                                              signed_auth_key=signed_auth_key)
+            if sent:
+                try:
+                    notify_fulfillment_todo(
+                        title=f"Intrastate authority submitted to {base_state} PSC — awaiting fee — {entity_name}",
+                        order_number=order_number,
+                        service_slug=service_slug,
+                        priority="normal",
+                        description=(f"Intrastate authority application emailed to the {base_state} "
+                                     f"PSC/PUC with the signed POA.\nWaiting on their requirements + "
+                                     f"fee invoice; when it arrives we auto-bill the customer the exact "
+                                     f"amount and you'll get a Telegram alert.\nCustomer: {customer_email}"),
+                    )
+                except Exception:
+                    pass
+            return sent
         # ── IFTA / intrastate: published fee, bill the estimate now ───────────
         try:
             from scripts.workers.services.gov_fee import (