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.

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