feat(compliance): admin verification gate + durable submission evidence

Per request: after the customer signs but BEFORE we submit to the government, hold
the order for a human to verify the prepared filing is correct.

- MCS-150 handler (mcs150-update + usdot-reactivation): new admin-verification gate
  after the signature gate -- if not admin_approved, set fulfillment_status=
  'ready_to_file', create a HIGH-priority 'VERIFY before filing' admin todo, and
  STOP (no FMCSA submission). job_server injects admin_approved from the dispatch
  payload (mirrors client_approved).
- New admin endpoint POST /api/v1/admin/compliance-orders/:id/approve-submit
  (requireAdmin): verifies status=ready_to_file, re-dispatches the worker with
  admin_approved=true to proceed to actual submission.
- Durable submission EVIDENCE: the web/fax submitters only wrote confirmation
  screenshots to an ephemeral temp dir. Now _upload_submission_evidence copies the
  FMCSA confirmation screenshot + attested PDF + fax_log_id to MinIO under
  filings/<slug>/<order>/evidence/ and records the keys on the order, so we keep
  proof of every government submission.

(state-trucking + the FCC handlers already gate via admin todos / auto_filing.py;
this brings MCS-150 to parity and adds evidence retention.)

api/src/routes/compliance-orders.ts

@@ -13,6 +13,7 @@ import { Router } from "express";
 import { pool } from "../db.js";
 import { randomBytes } from "crypto";
 import { COMPLIANCE_SERVICES } from "../service-catalog.js";
+import { requireAdmin } from "../middleware/admin-auth.js";
 
 const router = Router();
 
@@ -179,6 +180,10 @@ const REQUIRED_FIELDS: Record<string, FieldSpec> = {
   "foreign-qualification-single": { required: ["legal_name", "home_state_code", "entity_type", "target_states"], soft: ["ein"] },
   "foreign-qualification-multi":  { required: ["legal_name", "home_state_code", "entity_type", "target_states"], soft: ["ein"] },
 
+  // ── Business name reservation (binding hold; admin-assisted filing) ──
+  "name-reservation-tx": { required: ["entity_name", "entity_type"], soft: ["entity_name_alt"] },
+  "name-reservation-nv": { required: ["entity_name", "entity_type"], soft: ["entity_name_alt"] },
+
   // ── DOT / FMCSA Motor Carrier Services ───────────────────────────────
   // All collected via the unified dot-intake step (DOTIntakeStep.astro).
   "mcs150-update":   { required: ["dot_number", "legal_name", "address_street", "address_city", "address_state", "address_zip", "phone", "email", "signer_name", "signer_title", "power_units", "drivers", "carrier_operation", "interstate_intrastate", "hazmat"], soft: ["mc_number", "ein", "annual_miles", "cargo_types"] },
@@ -1753,5 +1758,60 @@ router.post("/api/v1/compliance-orders/:id/usac-delegation", async (req, res) =>
   }
 });
 
+// ── POST /api/v1/admin/compliance-orders/:id/approve-submit ──────────────────
+// Admin verification gate: an order that has been prepared + (if needed) signed
+// is held at fulfillment_status='ready_to_file'. The admin reviews the prepared
+// filing and calls this to APPROVE it, which re-dispatches the worker with
+// admin_approved=true so it proceeds to the actual government submission. We
+// never submit to a government system until an admin has cleared this gate.
+router.post("/api/v1/admin/compliance-orders/:id/approve-submit", requireAdmin, async (req, res) => {
+  const id = req.params.id;
+  try {
+    const { rows } = await pool.query(
+      `SELECT order_number, service_slug, fulfillment_status
+         FROM compliance_orders WHERE order_number = $1`,
+      [id],
+    );
+    const order = rows[0];
+    if (!order) {
+      return res.status(404).json({ error: "Order not found" });
+    }
+    if (order.fulfillment_status !== "ready_to_file") {
+      return res.status(409).json({
+        error: `Order is not awaiting submission approval (status=${order.fulfillment_status ?? "none"})`,
+      });
+    }
+
+    await pool.query(
+      `UPDATE compliance_orders SET fulfillment_status = 'authorization_signed',
+              fulfillment_status_at = now() WHERE order_number = $1`,
+      [id],
+    );
+    const workerUrl = process.env.WORKER_URL || "http://workers:8090";
+    let dispatched = false;
+    try {
+      const r = await fetch(`${workerUrl}/jobs`, {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({
+          action: "process_compliance_service",
+          order_name: id,
+          order_number: id,
+          service_slug: order.service_slug,
+          admin_approved: true,
+        }),
+      });
+      dispatched = r.ok;
+    } catch (err) {
+      console.error(`[compliance-orders] approve-submit dispatch failed for ${id}:`, err);
+    }
+    console.log(`[compliance-orders] Admin approved + dispatched submission for ${id} (${order.service_slug})`);
+    res.json({ success: true, order_number: id, dispatched });
+  } catch (err) {
+    console.error(`[compliance-orders] approve-submit error for ${id}:`, err);
+    res.status(500).json({ error: "Approve-submit failed" });
+  }
+});
+
 export { COMPLIANCE_SERVICES, REQUIRED_FIELDS, BUNDLE_COMPONENTS, MUTUALLY_EXCLUSIVE_GROUPS };
 export default router;

scripts/workers/job_server.py

@@ -1240,6 +1240,10 @@ def handle_process_compliance_service(payload: dict) -> dict:
         order["esign_document_type"] = payload["esign_document_type"]
     if payload.get("esign_signer_email"):
         order["esign_signer_email"] = payload["esign_signer_email"]
+    # Inject admin-approval flag (set when an admin clears the post-signature
+    # verification gate and re-dispatches the order for submission).
+    if payload.get("admin_approved"):
+        order["admin_approved"] = True
 
     # Final entity check before dispatch
     ent = order.get("entity", {})

scripts/workers/services/mcs150_update.py

@@ -87,6 +87,7 @@ class MCS150UpdateHandler:
         from scripts.workers.services.dot_esign import requires_signature, request_dot_esign
         needs_signature = requires_signature(slug)
         client_approved = bool(order_data.get("client_approved"))
+        admin_approved = bool(order_data.get("admin_approved"))
 
         # Validate required fields
         if not dot_number:
@@ -172,6 +173,23 @@ class MCS150UpdateHandler:
         # Past this point: either no signature is required for this service, or
         # the client has signed (re-dispatched with client_approved=True).
 
+        # Step 4b: ADMIN VERIFICATION GATE. Before we submit anything to the
+        # government, a human verifies the prepared filing is correct (right
+        # form, right DOT#, right data, signed by the client). We STOP here and
+        # create an admin todo; an admin reviews the generated PDF and, when
+        # satisfied, re-dispatches this order with admin_approved=True (via the
+        # admin "Approve & Submit" action) to proceed to actual submission. This
+        # prevents a wrong/auto-generated filing from being submitted to FMCSA.
+        if not admin_approved:
+            self._set_fulfillment_status(order_number, "ready_to_file")
+            self._create_admin_review_todo(
+                order_number, entity_name, dot_number, slug, minio_path, customer_email,
+                client_signed=(needs_signature and client_approved),
+            )
+            LOG.info("[%s] Prepared + signed; HELD for admin verification before FMCSA submission",
+                     order_number)
+            return [minio_path] if minio_path else []
+
         # 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"
@@ -215,6 +233,14 @@ class MCS150UpdateHandler:
             except Exception as exc:
                 LOG.error("[%s] Filing submission error: %s", order_number, exc)
 
+        # Persist durable submission EVIDENCE to MinIO (confirmation screenshot
+        # for web, attested PDF + fax log for fax) so we keep proof of every
+        # government submission -- the submitters only write to an ephemeral
+        # temp dir otherwise.
+        evidence = {}
+        if filing_result and filing_result.get("success") and filing_result.get("method") != "dev_skip":
+            evidence = self._upload_submission_evidence(order_number, slug, filing_result)
+
         # Step 5: Update order status in database
         try:
             import psycopg2
@@ -229,6 +255,7 @@ class MCS150UpdateHandler:
                 "screenshot_path": filing_result.get("screenshot_path") if filing_result else None,
                 "submitted_at": filing_result.get("submitted_at") if filing_result else None,
                 "attested_pdf": filing_result.get("attested_pdf") if filing_result else None,
+                "evidence": evidence,  # durable MinIO keys for proof of submission
             }
             cur.execute("""
                 UPDATE compliance_orders SET intake_data = jsonb_set(
@@ -363,6 +390,120 @@ class MCS150UpdateHandler:
         except Exception as exc:
             LOG.warning("[%s] Failed to create pending-signature todo: %s", order_number, exc)
 
+    def _set_fulfillment_status(self, order_number, status):
+        """Persist the fulfillment_status on the compliance order (e.g.
+        'ready_to_file' = prepared + signed, held for admin verification)."""
+        try:
+            import psycopg2
+            conn = psycopg2.connect(os.environ.get("DATABASE_URL", ""))
+            with conn.cursor() as cur:
+                cur.execute(
+                    "UPDATE compliance_orders SET fulfillment_status=%s, "
+                    "fulfillment_status_at=now() WHERE order_number=%s",
+                    (status, order_number),
+                )
+            conn.commit()
+            conn.close()
+        except Exception as exc:
+            LOG.warning("[%s] Failed to set fulfillment_status=%s: %s", order_number, status, 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.
+
+        The order is held at fulfillment_status='ready_to_file'. An admin opens
+        the generated PDF (minio_path), confirms it is correct (right form, DOT#,
+        data, signature), and then re-dispatches with admin_approved=True via the
+        admin 'Approve & Submit' action to proceed to actual FMCSA submission.
+        """
+        try:
+            import psycopg2
+            conn = psycopg2.connect(os.environ.get("DATABASE_URL", ""))
+            sig = "client SIGNED" if client_signed else "no signature required"
+            todo_title = f"VERIFY before filing — {entity_name} (DOT {dot_number})"
+            todo_description = (
+                f"{slug} for {entity_name} (DOT {dot_number}).\n"
+                f"Status: READY TO FILE — held for admin verification ({sig}).\n"
+                f"ACTION: review the prepared PDF, confirm it is correct, then\n"
+                f"approve & submit (re-dispatch with admin_approved=true).\n"
+                f"We do NOT submit to FMCSA until you approve.\n"
+                f"PDF: {minio_path or 'not generated'}\n"
+                f"Customer: {customer_email}"
+            )
+            with conn.cursor() as cur:
+                cur.execute(
+                    """
+                    INSERT INTO admin_todos (
+                        title, category, priority, order_number, service_slug,
+                        description, data, status
+                    ) VALUES (%s, %s, %s, %s, %s, %s, %s, 'pending')
+                    """,
+                    (
+                        todo_title,
+                        "filing", "high", order_number, slug,
+                        todo_description,
+                        json.dumps({"order_number": order_number, "dot_number": dot_number,
+                                    "entity_name": entity_name, "awaiting_admin_review": True,
+                                    "client_signed": client_signed, "pdf_minio_path": minio_path}),
+                    ),
+                )
+            conn.commit()
+            notify_fulfillment_todo(
+                title=todo_title,
+                order_number=order_number,
+                service_slug=slug,
+                priority="high",
+                description=todo_description,
+            )
+            conn.close()
+            LOG.info("[%s] Admin-review (pre-submission) todo created", order_number)
+        except Exception as exc:
+            LOG.warning("[%s] Failed to create admin-review todo: %s", order_number, exc)
+
+    @staticmethod
+    def _upload_submission_evidence(order_number, slug, filing_result):
+        """Persist submission proof (confirmation screenshot for web, attested
+        PDF for fax) to MinIO so we keep durable evidence of every government
+        submission. The submitters write screenshots to an ephemeral temp dir;
+        we copy the proof into MinIO under filings/<slug>/<order>/evidence/ and
+        return the MinIO keys to store on the order.
+        """
+        evidence = {}
+        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")
+            base = f"filings/{slug}/{order_number}/evidence"
+
+            def _put(local_path, name, content_type):
+                if local_path and os.path.exists(local_path):
+                    key = f"{base}/{name}"
+                    mc.fput_object(bucket, key, local_path, content_type=content_type)
+                    return key
+                return None
+
+            shot = filing_result.get("screenshot_path") or filing_result.get("pre_submit_screenshot")
+            k = _put(shot, "fmcsa_confirmation.png", "image/png")
+            if k:
+                evidence["confirmation_screenshot"] = k
+            att = filing_result.get("attested_pdf")
+            k = _put(att, "attested_filing.pdf", "application/pdf")
+            if k:
+                evidence["attested_pdf_minio"] = k
+            if filing_result.get("fax_log_id"):
+                evidence["fax_log_id"] = filing_result["fax_log_id"]
+            if evidence:
+                LOG.info("[%s] Submission evidence saved to MinIO: %s",
+                         order_number, list(evidence.keys()))
+        except Exception as exc:
+            LOG.warning("[%s] Failed to persist submission evidence: %s", order_number, exc)
+        return evidence
+
     def _send_status_email(self, order_number, entity_name, dot_number, customer_email):
         """Send client an email that we're working on their update."""
         if not customer_email: