feat(ucr): Playwright auto-filing for UCR registration on approval

Adds scripts/workers/services/ucr_playwright.py — a UCR.gov National Registration
System automation that, given a USDOT + fleet size, runs the register/pay flow,
pays the federal UCR fee with the matched PW filing card (Relay/Stripe Issuing),
and captures a confirmation screenshot + number. Conventions match
boc3_playwright / fmcsa_web_submitter: dev-mode dry-run guard, undetected
(patchright) browser, CAPTCHA detection, screenshot evidence, dataclass result.

Safety: verifies the displayed fee against the federal schedule before paying and
refuses to auto-charge a surprising amount (UCR_MAX_AUTO_FEE_USD) — falls back to
manual filing instead.

Wires it into MCS150UpdateHandler: when an approved (admin_approved) order has
slug ucr-registration, _file_ucr_registration runs the automation, uploads the
confirmation screenshot to MinIO, records filing_status + confirmation, and sets
fulfillment_status=completed on success. On CAPTCHA / fee-mismatch / failure it
reverts to ready_to_file with a high-priority 'file manually' todo. This replaces
the old behavior where approving a UCR just sat at authorization_signed.

scripts/workers/services/mcs150_update.py

@@ -338,6 +338,15 @@ class MCS150UpdateHandler:
                      order_number)
             return [minio_path] if minio_path else []
 
+        # Step 4c: ADMIN-ASSISTED AUTO-FILING. Some admin-assisted services have
+        # a real automated filing path even though they don't produce an MCS-150
+        # form. Once an admin has approved (admin_approved=True), run the
+        # automation here instead of just creating a manual todo. Currently:
+        #   - ucr-registration -> ucr.gov National Registration System
+        if slug == "ucr-registration":
+            return self._file_ucr_registration(
+                order_number, entity_name, dot_number, intake, customer_email)
+
         # Step 5: Submit electronically (3x web → fax fallback)
         # GUARD: Skip actual submission in dev/test environments
         is_production = os.environ.get("NODE_ENV") == "production" or os.environ.get("ENV") == "production"
@@ -861,6 +870,169 @@ class MCS150UpdateHandler:
         except Exception as exc:  # noqa: BLE001
             LOG.warning("[%s] Failed to mark intake validated: %s", order_number, exc)
 
+    def _file_ucr_registration(self, order_number, entity_name, dot_number,
+                               intake, customer_email) -> list:
+        """Run the UCR.gov Playwright automation for an approved UCR order, then
+        persist the result. On success -> completed (+ confirmation + screenshot
+        evidence). On CAPTCHA / fee-mismatch / failure -> ready_to_file with a
+        high-priority 'file manually' todo so a human takes over."""
+        import asyncio
+        LOG.info("[%s] Auto-filing UCR registration via ucr.gov", order_number)
+
+        # Resolve the customer's payment method so we charge the right card.
+        payment_method = "card"
+        try:
+            import psycopg2
+            conn = psycopg2.connect(os.environ.get("DATABASE_URL", ""))
+            with conn.cursor() as cur:
+                cur.execute("SELECT payment_method FROM compliance_orders WHERE order_number=%s",
+                            (order_number,))
+                row = cur.fetchone()
+                if row and row[0]:
+                    payment_method = row[0]
+            conn.close()
+        except Exception as exc:  # noqa: BLE001
+            LOG.warning("[%s] Could not read payment_method (default card): %s", order_number, exc)
+
+        try:
+            from scripts.workers.services.ucr_playwright import UCRRegistration
+        except Exception as exc:  # noqa: BLE001
+            LOG.error("[%s] UCR adapter import failed: %s", order_number, exc)
+            self._set_fulfillment_status(order_number, "ready_to_file")
+            self._create_admin_review_todo(
+                order_number, entity_name, dot_number, "ucr-registration", None,
+                customer_email, client_signed=False)
+            return []
+
+        data = {
+            "dot_number": dot_number,
+            "legal_name": intake.get("legal_name", entity_name),
+            "power_units": intake.get("power_units", ""),
+            "email": intake.get("email") or customer_email,
+            "phone": intake.get("phone", ""),
+            "address_street": intake.get("address_street", ""),
+            "address_city": intake.get("address_city", ""),
+            "address_state": intake.get("address_state", ""),
+            "address_zip": intake.get("address_zip", ""),
+        }
+        try:
+            adapter = UCRRegistration()
+            loop = asyncio.new_event_loop()
+            try:
+                result = loop.run_until_complete(
+                    adapter.file_ucr(data, order_number=order_number, payment_method=payment_method))
+            finally:
+                loop.close()
+        except Exception as exc:  # noqa: BLE001
+            LOG.error("[%s] UCR automation crashed: %s", order_number, exc)
+            self._set_fulfillment_status(order_number, "ready_to_file")
+            self._create_admin_review_todo(
+                order_number, entity_name, dot_number, "ucr-registration", None,
+                customer_email, client_signed=False)
+            return []
+
+        # Upload the confirmation screenshot to MinIO as durable evidence.
+        evidence = {}
+        if result.screenshot_path and os.path.exists(result.screenshot_path):
+            try:
+                from minio import Minio
+                mc = Minio(
+                    f"{os.environ.get('MINIO_ENDPOINT', 'minio')}:{os.environ.get('MINIO_PORT', '9000')}",
+                    access_key=os.environ.get("MINIO_ACCESS_KEY", ""),
+                    secret_key=os.environ.get("MINIO_SECRET_KEY", ""),
+                    secure=os.environ.get("MINIO_SECURE", "false").lower() == "true",
+                )
+                bucket = os.environ.get("MINIO_BUCKET", "performancewest")
+                key = f"filings/ucr-registration/{order_number}/evidence/ucr_confirmation.png"
+                mc.fput_object(bucket, key, result.screenshot_path, content_type="image/png")
+                evidence["confirmation_screenshot"] = key
+            except Exception as exc:  # noqa: BLE001
+                LOG.warning("[%s] Could not upload UCR screenshot: %s", order_number, exc)
+
+        # Persist filing_status on the order.
+        filing_status = {
+            "filing_method": "ucr_gov_web",
+            "filing_success": bool(result.success),
+            "submitted_at": datetime.now(timezone.utc).isoformat() if result.success else None,
+            "manual_confirmation": result.confirmation_number or None,
+            "fee_paid_usd": result.fee_paid_usd or None,
+            "dry_run": bool(result.dry_run),
+            "evidence": evidence,
+            "error": result.error or None,
+        }
+        try:
+            import psycopg2
+            conn = psycopg2.connect(os.environ.get("DATABASE_URL", ""))
+            with conn.cursor() as cur:
+                cur.execute(
+                    "UPDATE compliance_orders SET intake_data = jsonb_set("
+                    "COALESCE(intake_data,'{}'::jsonb), '{filing_status}', %s::jsonb), "
+                    "updated_at=now() WHERE order_number=%s",
+                    (json.dumps(filing_status), order_number),
+                )
+            conn.commit()
+            conn.close()
+        except Exception as exc:  # noqa: BLE001
+            LOG.error("[%s] Failed to persist UCR filing_status: %s", order_number, exc)
+
+        svc_label = self._service_label("ucr-registration")
+        if result.success:
+            self._set_fulfillment_status(order_number, "completed")
+            conf = result.confirmation_number or "(see receipt screenshot)"
+            self._notify_todo(
+                order_number, "ucr-registration",
+                f"{svc_label} FILED — {entity_name} (DOT {dot_number})",
+                "low",
+                (f"{svc_label} for {entity_name} (DOT {dot_number}).\n"
+                 f"Status: FILED on ucr.gov{' (DEV dry-run)' if result.dry_run else ''}.\n"
+                 f"Confirmation: {conf}\n"
+                 f"Fee paid: ${result.fee_paid_usd:.2f}\n"
+                 f"Customer: {customer_email}"),
+            )
+            self._send_status_email(order_number, entity_name, dot_number, customer_email, "ucr-registration")
+            LOG.info("[%s] UCR filed (completed). conf=%s", order_number, conf)
+        else:
+            # Could not auto-file (CAPTCHA, fee mismatch, error) -> back to the
+            # manual-filing queue with a clear, high-priority todo.
+            self._set_fulfillment_status(order_number, "ready_to_file")
+            reason = "CAPTCHA — needs manual filing" if result.captcha_hit else (result.error or "automation could not confirm")
+            self._notify_todo(
+                order_number, "ucr-registration",
+                f"{svc_label} — FILE MANUALLY — {entity_name} (DOT {dot_number})",
+                "high",
+                (f"{svc_label} for {entity_name} (DOT {dot_number}).\n"
+                 f"Status: AUTO-FILE FAILED — {reason}.\n"
+                 f"ACTION: file manually on ucr.gov, then mark the order completed "
+                 f"in the admin dashboard (enter the confirmation #).\n"
+                 f"Customer: {customer_email}"),
+            )
+            LOG.warning("[%s] UCR auto-file failed: %s", order_number, reason)
+        return []
+
+    def _notify_todo(self, order_number, slug, title, priority, description):
+        """Create an admin_todos row + Telegram fulfillment notification."""
+        try:
+            import psycopg2
+            conn = psycopg2.connect(os.environ.get("DATABASE_URL", ""))
+            with conn.cursor() as cur:
+                cur.execute(
+                    """INSERT INTO admin_todos (title, category, priority, order_number,
+                         service_slug, description, data, status)
+                       VALUES (%s, 'filing', %s, %s, %s, %s, %s, 'pending')""",
+                    (title, priority, order_number, slug, description,
+                     json.dumps({"order_number": order_number})),
+                )
+            conn.commit()
+            conn.close()
+        except Exception as exc:  # noqa: BLE001
+            LOG.error("[%s] Failed to create todo: %s", order_number, exc)
+        try:
+            notify_fulfillment_todo(title=title, order_number=order_number,
+                                    service_slug=slug, priority=priority,
+                                    description=description)
+        except Exception as exc:  # noqa: BLE001
+            LOG.warning("[%s] Telegram notify failed: %s", order_number, exc)
+
     def _create_admin_review_todo(self, order_number, entity_name, dot_number,
                                   slug, minio_path, customer_email, client_signed):
         """High-priority admin todo: verify the prepared filing BEFORE submission.

scripts/workers/services/ucr_playwright.py

@@ -0,0 +1,474 @@
+"""UCR Annual Registration — Playwright automation for the National Registration
+System at https://www.ucr.gov.
+
+Unified Carrier Registration (UCR) is an annual registration + fee that every
+interstate for-hire carrier, broker, and freight forwarder must pay. There is no
+public API; registration is done through the official National Registration
+System website. This adapter automates the "Register / Pay" flow:
+
+  1. Start a registration for the given registration year by USDOT number.
+  2. Confirm the carrier's company information (pre-filled from the USDOT record).
+  3. Set / confirm the fleet size (total power units) — this determines the fee
+     bracket per the federal UCR fee schedule.
+  4. Pay the UCR fee with the Performance West filing card (Relay/Stripe Issuing,
+     matched to the customer's payment method).
+  5. Capture the confirmation screenshot + receipt / confirmation number.
+
+reCAPTCHA: the site may use a CAPTCHA on the payment step. We launch with the
+undetected (patchright) browser to avoid triggering it; if one appears we abort
+cleanly and surface captcha_hit so the order falls back to manual filing.
+
+Conventions mirror boc3_playwright.py / fmcsa_web_submitter.py:
+  - dev/test environments NEVER submit (NODE_ENV/ENV != production -> dry run).
+  - screenshots written under SCREENSHOTS_DIR for durable evidence.
+  - returns a dataclass with success / confirmation_number / screenshot_path.
+
+Usage:
+    adapter = UCRRegistration()
+    result = await adapter.file_ucr({
+        "dot_number": "1167703",
+        "legal_name": "COMPOUND TECHNOLOGIES INC",
+        "power_units": "2",
+        "registration_year": 2026,            # optional; defaults to current UCR year
+        "email": "carrier@example.com",
+        "phone": "5551234567",
+        "address_street": "72 RIVERSIDE DRIVE",
+        "address_city": "CARTERSVILLE",
+        "address_state": "GA",
+        "address_zip": "30120",
+    }, order_number="CO-...", payment_method="card")
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import random
+import re
+import time
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+
+LOG = logging.getLogger("workers.services.ucr_playwright")
+SCREENSHOTS_DIR = Path(os.getenv("SCREENSHOTS_DIR", "/tmp/ucr-screenshots"))
+SCREENSHOTS_DIR.mkdir(parents=True, exist_ok=True)
+
+UCR_URL = "https://www.ucr.gov"
+
+# Federal UCR fee schedule (2024/2025 plan year) by total power units.
+# (low, high_inclusive, fee_usd). Used to sanity-check the fee the site shows
+# before we authorize payment, so we never overpay on a mis-read bracket.
+UCR_FEE_SCHEDULE = [
+    (0, 2, 46.00),
+    (3, 5, 138.00),
+    (6, 20, 276.00),
+    (21, 100, 963.00),
+    (101, 1000, 4592.00),
+    (1001, 10**9, 44836.00),
+]
+
+# Hard ceiling: never authorize a UCR charge above this without a human. A
+# small-fleet carrier should be in the $46-$276 range; anything wildly above
+# this for a 0-20 unit fleet means we mis-read the page.
+UCR_MAX_AUTO_FEE_USD = float(os.getenv("UCR_MAX_AUTO_FEE_USD", "300"))
+
+
+def expected_ucr_fee(power_units: int) -> float:
+    """Return the federal UCR fee for a given power-unit count (0 if unknown)."""
+    for lo, hi, fee in UCR_FEE_SCHEDULE:
+        if lo <= power_units <= hi:
+            return fee
+    return 0.0
+
+
+def current_ucr_year() -> int:
+    """UCR registration opens in the fall for the NEXT calendar year. Use the
+    next year once we're past Oct 1, otherwise the current year."""
+    now = datetime.now(timezone.utc)
+    return now.year + 1 if now.month >= 10 else now.year
+
+
+@dataclass
+class UCRFilingResult:
+    success: bool
+    confirmation_number: str = ""
+    fee_paid_usd: float = 0.0
+    error: str = ""
+    screenshot_path: str = ""
+    captcha_hit: bool = False
+    dry_run: bool = False
+
+
+def _load_filing_card(order_number: str, payment_method: str):
+    """Load the PW filing card matching the customer's payment method, mirroring
+    boc3_playwright._load_matching_card. Returns a CardDetails or None."""
+    try:
+        from scripts.workers.relay_integration import load_card_from_erpnext
+        card = load_card_from_erpnext(payment_method or "card")
+        if card:
+            LOG.info("[ucr] Using %s card for %s (customer paid via %s)",
+                     card.masked(), order_number, payment_method)
+        return card
+    except Exception as exc:  # noqa: BLE001
+        LOG.warning("[ucr] Could not load filing card: %s", exc)
+        return None
+
+
+class UCRRegistration:
+    """Automate UCR annual registration + payment through ucr.gov."""
+
+    async def file_ucr(self, data: dict, order_number: str = "",
+                       payment_method: str = "card") -> UCRFilingResult:
+        dot = str(data.get("dot_number", "")).strip()
+        legal_name = data.get("legal_name", "")
+        try:
+            power_units = int(str(data.get("power_units", "")).strip() or "0")
+        except ValueError:
+            power_units = 0
+        reg_year = int(data.get("registration_year") or current_ucr_year())
+
+        if not dot:
+            return UCRFilingResult(success=False, error="DOT number required for UCR")
+        if power_units <= 0:
+            return UCRFilingResult(
+                success=False,
+                error="power_units required to determine UCR fee bracket",
+            )
+
+        expected_fee = expected_ucr_fee(power_units)
+        LOG.info("[ucr] DOT %s — %s power units → expected fee $%.2f (year %s)",
+                 dot, power_units, expected_fee, reg_year)
+
+        # ── Dev/test guard: never touch the live registration system ─────────
+        is_prod = os.getenv("NODE_ENV") == "production" or os.getenv("ENV") == "production"
+        if not is_prod:
+            LOG.info("[ucr] DEV MODE — skipping live UCR registration for DOT %s", dot)
+            return UCRFilingResult(
+                success=True, dry_run=True,
+                confirmation_number="DEV-SKIP",
+                fee_paid_usd=expected_fee,
+                error="",
+            )
+
+        card = _load_filing_card(order_number, payment_method)
+        if not card or not card.card_number:
+            return UCRFilingResult(
+                success=False,
+                error="No filing card configured (RELAY_FILING_CARD / PW card)",
+            )
+
+        timestamp = int(time.time())
+        shot_base = SCREENSHOTS_DIR / f"ucr_{dot}_{timestamp}"
+
+        try:
+            from patchright.async_api import async_playwright
+        except ImportError:
+            from playwright.async_api import async_playwright
+
+        try:
+            from scripts.workers.services.telecom.undetected_browser import launch_context
+        except ImportError:
+            launch_context = None
+
+        browser = context = page = None
+        try:
+            async with async_playwright() as pw:
+                if launch_context:
+                    browser, context = await launch_context(pw, headless=True)
+                else:
+                    browser = await pw.chromium.launch(headless=True)
+                    context = await browser.new_context(
+                        viewport={"width": 1366, "height": 900},
+                        user_agent=(
+                            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                            "AppleWebKit/537.36 (KHTML, like Gecko) "
+                            "Chrome/131.0.0.0 Safari/537.36"
+                        ),
+                        locale="en-US",
+                        timezone_id="America/New_York",
+                    )
+                page = await context.new_page()
+
+                # ── Step 1: Open the registration flow ───────────────────────
+                LOG.info("[ucr] Navigating to %s for DOT %s", UCR_URL, dot)
+                await page.goto(UCR_URL, wait_until="domcontentloaded", timeout=45000)
+                await asyncio.sleep(random.uniform(2, 4))
+                await page.screenshot(path=f"{shot_base}_01_home.png", full_page=True)
+
+                # Click the primary "Register / File Now" CTA. Try several
+                # accessible-name variants the site has used.
+                started = await self._click_first(page, [
+                    'a:has-text("Register Now")', 'a:has-text("File Now")',
+                    'a:has-text("Begin Registration")', 'button:has-text("Register")',
+                    'a:has-text("Start")', 'a[href*="register"]',
+                ])
+                if not started:
+                    return self._fail(page, shot_base, "Could not find the UCR Register/File CTA")
+                await asyncio.sleep(random.uniform(2, 4))
+
+                # ── Step 2: Identify the carrier (USDOT + year) ──────────────
+                await self._fill_first(page, [
+                    'input[name*="dot" i]', 'input[id*="dot" i]',
+                    'input[placeholder*="USDOT" i]', 'input[placeholder*="DOT" i]',
+                ], dot)
+                # Registration year, when the site asks for it.
+                await self._select_first(page, [
+                    'select[name*="year" i]', 'select[id*="year" i]',
+                ], str(reg_year))
+                await page.screenshot(path=f"{shot_base}_02_identify.png", full_page=True)
+                await self._click_first(page, [
+                    'button:has-text("Continue")', 'button:has-text("Next")',
+                    'button:has-text("Search")', 'button:has-text("Lookup")',
+                    'input[type="submit"]',
+                ])
+                await asyncio.sleep(random.uniform(3, 5))
+
+                # ── CAPTCHA gate ─────────────────────────────────────────────
+                if await self._captcha_present(page):
+                    LOG.warning("[ucr] CAPTCHA detected — aborting for manual filing")
+                    await page.screenshot(path=f"{shot_base}_03_captcha.png", full_page=True)
+                    return UCRFilingResult(
+                        success=False, captcha_hit=True,
+                        error="CAPTCHA detected — requires manual filing",
+                        screenshot_path=f"{shot_base}_03_captcha.png",
+                    )
+
+                # ── Step 3: Confirm company info + fleet size ────────────────
+                # The site pre-fills the carrier record from USDOT; we only set
+                # the power-unit count if it asks (it drives the fee bracket).
+                await self._fill_first(page, [
+                    'input[name*="power" i]', 'input[id*="power" i]',
+                    'input[name*="vehicle" i]', 'input[placeholder*="power unit" i]',
+                ], str(power_units), required=False)
+                await self._fill_first(page, [
+                    'input[name*="email" i]', 'input[type="email"]',
+                ], data.get("email", ""), required=False)
+                await page.screenshot(path=f"{shot_base}_04_company.png", full_page=True)
+                await self._click_first(page, [
+                    'button:has-text("Continue")', 'button:has-text("Next")',
+                    'input[type="submit"]',
+                ])
+                await asyncio.sleep(random.uniform(3, 5))
+
+                # ── Step 4: Verify the fee BEFORE paying ─────────────────────
+                shown_fee = await self._read_fee(page)
+                if shown_fee is not None:
+                    LOG.info("[ucr] Site shows fee $%.2f (expected $%.2f)", shown_fee, expected_fee)
+                    # Refuse to pay a surprising amount for a small fleet.
+                    if shown_fee > max(expected_fee * 1.5, UCR_MAX_AUTO_FEE_USD):
+                        await page.screenshot(path=f"{shot_base}_05_fee_mismatch.png", full_page=True)
+                        return UCRFilingResult(
+                            success=False,
+                            error=(f"UCR fee ${shown_fee:.2f} exceeds expected "
+                                   f"${expected_fee:.2f} — held for manual review"),
+                            screenshot_path=f"{shot_base}_05_fee_mismatch.png",
+                        )
+                fee_to_record = shown_fee if shown_fee is not None else expected_fee
+
+                # ── Step 5: Payment ──────────────────────────────────────────
+                if await self._captcha_present(page):
+                    await page.screenshot(path=f"{shot_base}_05_captcha_pay.png", full_page=True)
+                    return UCRFilingResult(
+                        success=False, captcha_hit=True,
+                        error="CAPTCHA on payment step — requires manual filing",
+                        screenshot_path=f"{shot_base}_05_captcha_pay.png",
+                    )
+
+                await self._fill_card(page, card)
+                await page.screenshot(path=f"{shot_base}_06_payment.png", full_page=True)
+
+                await self._click_first(page, [
+                    'button:has-text("Pay")', 'button:has-text("Submit Payment")',
+                    'button:has-text("Place Order")', 'button:has-text("Complete")',
+                    'button[type="submit"]', 'input[type="submit"]',
+                ])
+                LOG.info("[ucr] Payment submitted — waiting for confirmation")
+                await asyncio.sleep(random.uniform(6, 9))
+                try:
+                    await page.wait_for_load_state("networkidle", timeout=20000)
+                except Exception:
+                    pass
+                await page.screenshot(path=f"{shot_base}_07_confirmation.png", full_page=True)
+
+                # ── Step 6: Parse the confirmation ───────────────────────────
+                body = (await page.inner_text("body")).lower()
+                ok_words = ["thank you", "confirmation", "successfully", "receipt",
+                            "registration complete", "payment received", "approved"]
+                err_words = ["declined", "error", "failed", "invalid card", "try again"]
+                success = any(w in body for w in ok_words) and not any(w in body for w in err_words)
+
+                conf = self._extract_confirmation(await page.inner_text("body"))
+                if success:
+                    LOG.info("[ucr] SUCCESS DOT %s — conf=%s fee=$%.2f", dot, conf or "(none)", fee_to_record)
+                    return UCRFilingResult(
+                        success=True, confirmation_number=conf,
+                        fee_paid_usd=fee_to_record,
+                        screenshot_path=f"{shot_base}_07_confirmation.png",
+                    )
+                LOG.error("[ucr] Could not confirm UCR registration for DOT %s", dot)
+                return UCRFilingResult(
+                    success=False,
+                    error="Could not confirm UCR registration — check screenshot",
+                    screenshot_path=f"{shot_base}_07_confirmation.png",
+                )
+        except Exception as exc:  # noqa: BLE001
+            LOG.error("[ucr] UCR automation failed for DOT %s: %s", dot, exc)
+            try:
+                if page:
+                    await page.screenshot(path=f"{shot_base}_99_error.png", full_page=True)
+            except Exception:
+                pass
+            return UCRFilingResult(success=False, error=str(exc),
+                                   screenshot_path=f"{shot_base}_99_error.png")
+        finally:
+            try:
+                if context:
+                    await context.close()
+                if browser:
+                    await browser.close()
+            except Exception:
+                pass
+
+    # ── helpers ──────────────────────────────────────────────────────────────
+    async def _click_first(self, page, selectors: list[str]) -> bool:
+        for sel in selectors:
+            try:
+                el = page.locator(sel).first
+                if await el.count() and await el.is_visible():
+                    await el.click(timeout=4000)
+                    return True
+            except Exception:
+                continue
+        return False
+
+    async def _fill_first(self, page, selectors: list[str], value: str,
+                          required: bool = True) -> bool:
+        if not value and not required:
+            return False
+        for sel in selectors:
+            try:
+                el = page.locator(sel).first
+                if await el.count() and await el.is_visible():
+                    await el.click(timeout=4000)
+                    await el.fill(value, timeout=4000)
+                    await asyncio.sleep(random.uniform(0.2, 0.5))
+                    return True
+            except Exception:
+                continue
+        if required:
+            LOG.warning("[ucr] Could not fill any of: %s", selectors)
+        return False
+
+    async def _select_first(self, page, selectors: list[str], value: str) -> bool:
+        for sel in selectors:
+            try:
+                el = page.locator(sel).first
+                if await el.count() and await el.is_visible():
+                    await el.select_option(value=value, timeout=4000)
+                    return True
+            except Exception:
+                continue
+        return False
+
+    async def _captcha_present(self, page) -> bool:
+        try:
+            return bool(await page.locator(
+                'iframe[src*="recaptcha"], .g-recaptcha, #recaptcha, '
+                'iframe[src*="hcaptcha"], .h-captcha'
+            ).count())
+        except Exception:
+            return False
+
+    async def _read_fee(self, page) -> float | None:
+        """Best-effort scrape of the displayed fee/total dollar amount."""
+        try:
+            text = await page.inner_text("body")
+        except Exception:
+            return None
+        # Prefer an amount near a fee/total label.
+        for pat in (
+            r'(?:total|amount due|fee|registration fee)[^$]{0,40}\$\s*([0-9][0-9,]*\.?[0-9]{0,2})',
+            r'\$\s*([0-9][0-9,]*\.[0-9]{2})',
+        ):
+            m = re.search(pat, text, re.I)
+            if m:
+                try:
+                    return float(m.group(1).replace(",", ""))
+                except ValueError:
+                    continue
+        return None
+
+    def _extract_confirmation(self, text: str) -> str:
+        for pat in (
+            r'(?:confirmation|registration|receipt|transaction)\s*(?:number|no\.?|#)\s*:?\s*([A-Z0-9-]{4,})',
+            r'\b(UCR[-\s]?[A-Z0-9]{4,})\b',
+        ):
+            m = re.search(pat, text, re.I)
+            if m:
+                return m.group(1).strip()
+        return ""
+
+    async def _fill_card(self, page, card) -> None:
+        """Fill the payment fields. Tries common name/autocomplete attributes
+        and handles split or single expiry fields."""
+        await self._fill_first(page, [
+            'input[autocomplete="cc-number"]', 'input[name*="card" i][name*="num" i]',
+            'input[name="number"]', 'input[placeholder*="card number" i]',
+        ], card.card_number, required=False)
+        # Cardholder name
+        await self._fill_first(page, [
+            'input[autocomplete="cc-name"]', 'input[name*="cardholder" i]',
+            'input[name*="name_on_card" i]', 'input[placeholder*="name on card" i]',
+        ], card.cardholder_name, required=False)
+        # CVV
+        await self._fill_first(page, [
+            'input[autocomplete="cc-csc"]', 'input[name*="cvc" i]',
+            'input[name*="cvv" i]', 'input[placeholder*="CVC" i]', 'input[placeholder*="CVV" i]',
+        ], card.cvv, required=False)
+        # Expiry: try a single MM/YY field first, then split selects/inputs.
+        single = await self._fill_first(page, [
+            'input[autocomplete="cc-exp"]', 'input[placeholder*="MM/YY" i]',
+            'input[name*="exp" i][name*="date" i]',
+        ], f"{card.exp_month}/{str(card.exp_year)[-2:]}", required=False)
+        if not single:
+            await self._select_first(page, [
+                'select[autocomplete="cc-exp-month"]', 'select[name*="exp" i][name*="month" i]',
+            ], card.exp_month) or await self._fill_first(page, [
+                'input[name*="exp" i][name*="month" i]',
+            ], card.exp_month, required=False)
+            yr_full = str(card.exp_year) if len(str(card.exp_year)) == 4 else f"20{card.exp_year}"
+            await self._select_first(page, [
+                'select[autocomplete="cc-exp-year"]', 'select[name*="exp" i][name*="year" i]',
+            ], yr_full) or await self._fill_first(page, [
+                'input[name*="exp" i][name*="year" i]',
+            ], yr_full, required=False)
+        # Billing ZIP (often required by the payment processor).
+        if card.billing_zip:
+            await self._fill_first(page, [
+                'input[autocomplete="postal-code"]', 'input[name*="zip" i]',
+                'input[name*="postal" i]',
+            ], card.billing_zip, required=False)
+
+    def _fail(self, page, shot_base, msg: str) -> UCRFilingResult:
+        LOG.error("[ucr] %s", msg)
+        return UCRFilingResult(success=False, error=msg,
+                               screenshot_path=f"{shot_base}_fail.png")
+
+
+async def _selftest():
+    """Dry-run self test (no live submission unless NODE_ENV=production)."""
+    res = await UCRRegistration().file_ucr({
+        "dot_number": "1167703",
+        "legal_name": "COMPOUND TECHNOLOGIES INC",
+        "power_units": "2",
+        "email": "test@performancewest.net",
+        "address_state": "GA",
+    }, order_number="CO-SELFTEST", payment_method="card")
+    print("dry_run:", res.dry_run, "success:", res.success,
+          "fee:", res.fee_paid_usd, "conf:", res.confirmation_number, "err:", res.error)
+
+
+if __name__ == "__main__":
+    asyncio.run(_selftest())