feat(fulfillment): bundle/exclusion enforcement + REQUIRED_FIELDS + intake wiring (Phases 1/1.5/2)

- compliance-orders: hazmat-phmsa/state-emissions products, full REQUIRED_FIELDS
  table for all DOT/state/hazmat slugs, BUNDLE_COMPONENTS dedup + MUTUALLY_EXCLUSIVE
  enforcement on /batch (single source of truth, exported)
- checkout: empty ADMIN_ASSISTED_SLUGS (state/hazmat now get intake links)
- services/__init__: register HazmatPHMSAHandler + state-emissions handler
- state_trucking: _summarize_intake admin-todo enrichment
- Wizard: wire StateTruckingIntakeStep + step labels

scripts/workers/services/__init__.py

@@ -55,6 +55,8 @@ from .ein_application import EINApplicationHandler
 from .mailbox_setup import MailboxSetupHandler
 # Carrier close-out / trucking wrap-up (shutdown) + entity dissolution
 from .carrier_closeout import CarrierCloseoutHandler
+# PHMSA hazmat registration (admin-assisted, 49 CFR Part 107)
+from .hazmat_phmsa import HazmatPHMSAHandler
 
 SERVICE_HANDLERS: dict[str, type] = {
     "flsa-audit": FLSAAuditHandler,
@@ -134,6 +136,10 @@ SERVICE_HANDLERS: dict[str, type] = {
     "intrastate-authority":      StateTruckingHandler,
     "osow-permit":               StateTruckingHandler,
     "state-trucking-bundle":     StateTruckingHandler,
+    # ── Hazmat / PHMSA Registration ───────────────────────────────────
+    "hazmat-phmsa":              HazmatPHMSAHandler,
+    # ── State emissions / clean-truck (admin-assisted) ────────────────
+    "state-emissions":           StateTruckingHandler,
 }
 
 # Service slugs that operate on a telecom entity — used by job_server.py

scripts/workers/services/state_trucking.py

@@ -184,6 +184,19 @@ SERVICE_INFO = {
             "7. Send all registrations and confirmations to client",
         ],
     },
+    "state-emissions": {
+        "name": "State Clean-Truck / Emissions Compliance",
+        "category": "emissions",
+        "steps": [
+            "1. Identify carrier's base/operating states with emissions programs "
+            "(NY, CO, MD, NJ, MA, etc. — Advanced Clean Trucks / Clean Truck Check)",
+            "2. Review fleet engine model-years against state emissions thresholds",
+            "3. Register/report fleet in the applicable state emissions portal",
+            "4. File any required compliance certification or fee",
+            "5. Set up annual renewal/reporting reminders",
+            "6. Send compliance confirmation + next-steps to client",
+        ],
+    },
 }
 
 
@@ -220,6 +233,9 @@ class StateTruckingHandler:
         base_state = intake.get("base_state", intake.get("phy_state", ""))
         operating_states = intake.get("operating_states", [])
 
+        # Slug-specific intake fields collected by StateTruckingIntakeStep.
+        intake_summary = self._summarize_intake(service_slug, intake)
+
         # Look up state requirements if we have a base state
         state_reqs = None
         if base_state:
@@ -244,10 +260,18 @@ class StateTruckingHandler:
             "base_state": base_state,
             "operating_states": operating_states,
             "intake_data": intake,
+            "intake_summary": intake_summary,
             "state_requirements": state_reqs,
             "steps": steps,
         }
 
+        # Render the slug-specific intake fields into the description.
+        intake_lines = ""
+        if intake_summary:
+            intake_lines = "\nFiling details:\n" + "\n".join(
+                f"  - {k}: {v}" for k, v in intake_summary.items()
+            ) + "\n"
+
         try:
             import psycopg2
             conn = psycopg2.connect(os.environ.get("DATABASE_URL", ""))
@@ -269,8 +293,9 @@ class StateTruckingHandler:
                         f"DOT: {dot_number}\n"
                         f"Base state: {base_state}\n"
                         f"Operating states: {', '.join(operating_states) if operating_states else 'N/A'}\n"
-                        f"Customer: {customer_email}\n\n"
-                        f"Steps:\n" + "\n".join(steps),
+                        f"Customer: {customer_email}\n"
+                        + intake_lines +
+                        f"\nSteps:\n" + "\n".join(steps),
                         json.dumps(todo_data),
                     ))
                 conn.commit()
@@ -287,6 +312,70 @@ class StateTruckingHandler:
 
         return []
 
+    # Map slug -> list of (intake_data key, human label) to surface in the todo.
+    _INTAKE_FIELD_MAP = {
+        "_common": [
+            ("power_units", "Power units"),
+            ("mc_number", "MC/MX/FF #"),
+        ],
+        "irp-ifta": [
+            ("fuel_type", "Fuel type"),
+            ("gross_weight_bracket", "Gross weight bracket"),
+        ],
+        "emissions": [
+            ("ca_number", "CA #"),
+            ("engine_model_years", "Oldest engine model year"),
+        ],
+        "intrastate": [
+            ("authority_type", "Authority type"),
+            ("boc3_on_file", "BOC-3 on file"),
+            ("insurance_carrier", "Insurance carrier"),
+            ("insurance_policy", "Insurance policy #"),
+        ],
+        "osow": [
+            ("load_dimensions", "Load dimensions"),
+            ("load_weight", "Load weight (lbs)"),
+        ],
+        "hazmat": [
+            ("hazmat_classes", "Hazmat classes"),
+            ("bulk_packaging", "Bulk packaging"),
+            ("small_business", "Small business"),
+        ],
+    }
+
+    # Which field groups apply to each slug (mirrors the front-end ST_SECTIONS).
+    _SLUG_FIELD_GROUPS = {
+        "irp-registration": ["irp-ifta"],
+        "ifta-application": ["irp-ifta"],
+        "ifta-quarterly": ["irp-ifta"],
+        "or-weight-mile-tax": ["irp-ifta"],
+        "ny-hut-registration": ["irp-ifta"],
+        "ky-kyu-registration": ["irp-ifta"],
+        "nm-weight-distance": ["irp-ifta"],
+        "ct-highway-use-fee": ["irp-ifta"],
+        "ca-mcp-carb": ["emissions"],
+        "state-emissions": ["emissions"],
+        "state-dot-registration": [],
+        "intrastate-authority": ["intrastate"],
+        "osow-permit": ["osow"],
+        "state-trucking-bundle": ["irp-ifta", "emissions", "intrastate"],
+        "hazmat-phmsa": ["hazmat"],
+    }
+
+    def _summarize_intake(self, service_slug: str, intake: dict) -> dict:
+        """Extract the slug-relevant intake fields into a flat label->value dict."""
+        groups = ["_common"] + self._SLUG_FIELD_GROUPS.get(service_slug, [])
+        summary: dict = {}
+        for group in groups:
+            for key, label in self._INTAKE_FIELD_MAP.get(group, []):
+                val = intake.get(key)
+                if val in (None, "", [], {}):
+                    continue
+                if isinstance(val, (list, tuple)):
+                    val = ", ".join(str(v) for v in val)
+                summary[label] = val
+        return summary
+
     def _resolve_slug(self, order_number: str) -> str:
         """Look up the service_slug from compliance_orders by order_number."""
         try: