diff --git a/MULTI_TENANT_PLAN.md b/MULTI_TENANT_PLAN.md
index 1dc6e41..4f8c9e6 100644
--- a/MULTI_TENANT_PLAN.md
+++ b/MULTI_TENANT_PLAN.md
@@ -164,7 +164,7 @@ Each phase additive; existing tenant-zero traffic keeps working throughout.
 >
 > **Operational note:** any *fresh* database (including a new dev DB) must use the bootstrap flow, not a naive `flask db upgrade` from empty, because the historical phase replay still hits the unguarded migrations. Cross-check the baseline against `mysqldump --no-data` of LT and test `bootstrap` on a scratch MySQL before going live.
 
-**MT-3 — Provisioning service.** Create DB → create **per-tenant MySQL user + password** + grant scoped to that DB only → upgrade to head → seed first tenant-admin → invite email. Creds encrypted into the `tenants` row. Idempotent, `provisioning_jobs`-logged.
+**MT-3 — Provisioning service. ✅ DONE.** `control/provision.py`: `create_tenant()` (admin account from `PROVISION_DB_URL` creates per-tenant DB + least-privilege user + grant → `bootstrap_tenant` builds schema + stamps head → `seed_admin` inserts the first admin with a set-password token → registers subdomain + optional custom domain → status active; best-effort rollback drops the DB/user + tenant row on failure). `register_tenant_zero()` adopts an existing DB in place (no DB/user creation, no bootstrap; reads its current head, maps `lts.<base>` + LT's live domain as a verified custom domain — zero data movement). `delete_tenant()` removes control records, optional guarded DB drop (refuses non-provisioner-named DBs, protecting tenant-zero). CLI: `python -m control.provision create-tenant|register-tenant-zero|delete-tenant`. Admin setup link uses `/customers/set-password/<token>` (the route that sets `password_set=True`). Creds Fernet-encrypted; `provisioning_jobs`-logged.
 
 **MT-4 — Superadmin control panel.** Tenant CRUD, plan assign, suspend/resume, domain mgmt, migration status, impersonation (scoped login into a tenant for support). At `admin.jqc.app`, separate blueprint, superadmin-gated.
 
diff --git a/control/provision.py b/control/provision.py
new file mode 100644
index 0000000..9339dbf
--- /dev/null
+++ b/control/provision.py
@@ -0,0 +1,445 @@
+"""
+control/provision.py
+--------------------
+Tenant provisioning service (MT-3).
+
+Two entry points:
+
+  create_tenant(...)         — provision a brand-new tenant:
+      1. CREATE the per-tenant MySQL database + least-privilege user + grant
+         (using a dedicated admin account from PROVISION_DB_URL)
+      2. insert the control Tenant row (Fernet-encrypted credentials)
+      3. bootstrap the schema (baseline + stamp head)  [control.tenant_migrate]
+      4. seed the first tenant admin (set-password token + link)
+      5. register the subdomain (+ optional custom domain)
+      On failure after the database is created, best-effort rollback drops the
+      database/user and removes the tenant row so a retry starts clean.
+
+  register_tenant_zero(...)  — adopt an EXISTING database (e.g. LT) in place:
+      no DB/user creation, no bootstrap, no admin seeding — just register the
+      tenant row pointing at the live DB, read its current alembic head, and map
+      its domains. Zero data movement, zero disruption.
+
+Environment:
+    PROVISION_DB_URL   admin connection that can CREATE DATABASE / CREATE USER /
+                       GRANT, e.g. mysql+pymysql://jqc_provisioner:pw@127.0.0.1/
+    TENANT_BASE_DOMAIN subdomain suffix for tenants, e.g. jqc.app
+    CONTROL_DATABASE_URL, CONTROL_FERNET_KEY  (as for the rest of the control plane)
+
+Required grants for the provisioner account (least privilege):
+    GRANT CREATE, CREATE USER, GRANT OPTION, ... ON *.*  (or scoped jqc_% dbs)
+"""
+
+import argparse
+import os
+import re
+import secrets
+import sys
+
+from sqlalchemy import create_engine, text
+from sqlalchemy.engine import make_url
+from werkzeug.security import generate_password_hash
+
+from control.base import control_session
+from control.models import Plan, Tenant, TenantDomain, ProvisioningJob
+from control.time_utils import now_eastern
+from control.tenant_migrate import bootstrap_tenant, current_revision
+
+SET_PASSWORD_PATH = '/customers/set-password'   # role-agnostic; sets password_set=True
+_SLUG_RE = re.compile(r'^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$')   # DNS label
+
+
+# ── helpers ─────────────────────────────────────────────────────────────────
+
+def _require(name):
+    v = os.environ.get(name)
+    if not v:
+        raise RuntimeError(f"{name} is not set.")
+    return v
+
+
+def _sanitize_ident(slug):
+    """MySQL-safe identifier fragment from a slug (alnum + underscore)."""
+    return re.sub(r'[^a-z0-9]', '_', slug.lower())
+
+
+def db_name_for(slug):
+    return f'jqc_{_sanitize_ident(slug)}'[:64]
+
+
+def db_user_for(slug):
+    return f'jqc_{_sanitize_ident(slug)}_u'[:32]
+
+
+def _provision_engine():
+    return create_engine(_require('PROVISION_DB_URL'), future=True)
+
+
+def _default_db_host():
+    try:
+        return make_url(_require('PROVISION_DB_URL')).host or '127.0.0.1'
+    except Exception:
+        return '127.0.0.1'
+
+
+def _base_domain(override=None):
+    return override or _require('TENANT_BASE_DOMAIN')
+
+
+def setup_link(primary_domain, token, scheme='https'):
+    return f'{scheme}://{primary_domain}{SET_PASSWORD_PATH}/{token}'
+
+
+# ── MySQL admin operations ───────────────────────────────────────────────────
+
+def create_mysql_db_and_user(db_name, db_user, password, user_host='%'):
+    """Idempotently create the tenant database, user, and scoped grant."""
+    eng = _provision_engine()
+    try:
+        with eng.connect() as c:
+            c.execution_options(isolation_level='AUTOCOMMIT')
+            c.execute(text(
+                f"CREATE DATABASE IF NOT EXISTS `{db_name}` "
+                f"CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"))
+            c.execute(text(
+                f"CREATE USER IF NOT EXISTS '{db_user}'@'{user_host}' "
+                f"IDENTIFIED BY :pw"), {'pw': password})
+            # Ensure the stored password matches (in case the user pre-existed).
+            c.execute(text(
+                f"ALTER USER '{db_user}'@'{user_host}' IDENTIFIED BY :pw"),
+                {'pw': password})
+            c.execute(text(
+                f"GRANT ALL PRIVILEGES ON `{db_name}`.* "
+                f"TO '{db_user}'@'{user_host}'"))
+            c.execute(text("FLUSH PRIVILEGES"))
+    finally:
+        eng.dispose()
+
+
+def drop_mysql_db_and_user(db_name, db_user, user_host='%'):
+    """Best-effort teardown (used for rollback on failed provisioning)."""
+    eng = _provision_engine()
+    try:
+        with eng.connect() as c:
+            c.execution_options(isolation_level='AUTOCOMMIT')
+            c.execute(text(f"DROP DATABASE IF EXISTS `{db_name}`"))
+            c.execute(text(f"DROP USER IF EXISTS '{db_user}'@'{user_host}'"))
+            c.execute(text("FLUSH PRIVILEGES"))
+    finally:
+        eng.dispose()
+
+
+# ── admin seeding (writes into the tenant DB) ────────────────────────────────
+
+def seed_admin(db_uri, email, username=None, full_name=None, expires_hours=72):
+    """Insert the first admin into a tenant DB with a set-password token.
+
+    Returns (username, token). The account is created with password_set=0 and an
+    unusable placeholder hash; the admin completes setup via the returned link.
+    """
+    username = username or re.sub(r'[^a-zA-Z0-9_.-]', '', email.split('@')[0]) or 'admin'
+    full_name = full_name or username
+    token = secrets.token_hex(32)
+    placeholder = generate_password_hash(secrets.token_urlsafe(32))
+    now = now_eastern()
+    from datetime import timedelta
+    expires = now + timedelta(hours=expires_hours)
+
+    eng = create_engine(db_uri, future=True)
+    try:
+        with eng.begin() as c:
+            c.execute(text("""
+                INSERT INTO users
+                    (username, full_name, email, password_hash, role,
+                     created_at, active, password_set,
+                     set_password_token, set_password_token_expires)
+                VALUES
+                    (:u, :fn, :em, :ph, 'admin',
+                     :ca, 1, 0, :tok, :exp)
+            """), {'u': username, 'fn': full_name, 'em': email, 'ph': placeholder,
+                   'ca': now, 'tok': token, 'exp': expires})
+    finally:
+        eng.dispose()
+    return username, token
+
+
+# ── domain registration ──────────────────────────────────────────────────────
+
+def _add_domains(session, tenant_id, slug, base_domain, custom_domain=None):
+    primary = f'{slug}.{base_domain}'.lower()
+    session.add(TenantDomain(
+        tenant_id=tenant_id, domain=primary, kind='subdomain',
+        is_primary=True, verified=True, tls_status='pending',
+        created_at=now_eastern()))
+    if custom_domain:
+        session.add(TenantDomain(
+            tenant_id=tenant_id, domain=custom_domain.lower(), kind='custom',
+            is_primary=False, verified=False,
+            verification_token=secrets.token_hex(16),
+            tls_status='pending', created_at=now_eastern()))
+    return primary
+
+
+# ── public: create a brand-new tenant ────────────────────────────────────────
+
+def create_tenant(slug, name, plan_code, admin_email, admin_username=None,
+                  admin_full_name=None, custom_domain=None, base_domain=None,
+                  db_host=None, user_host='%'):
+    if not _SLUG_RE.match(slug or ''):
+        raise ValueError(f"Invalid slug '{slug}' (must be a DNS label).")
+    base = _base_domain(base_domain)
+    host = db_host or _default_db_host()
+    dbname = db_name_for(slug)
+    dbuser = db_user_for(slug)
+    password = secrets.token_urlsafe(24)
+
+    with control_session() as s:
+        if s.query(Tenant).filter_by(slug=slug).first():
+            raise ValueError(f"Tenant slug '{slug}' already exists.")
+        plan = s.query(Plan).filter_by(code=plan_code).first()
+        if plan is None:
+            raise ValueError(f"Unknown plan '{plan_code}'.")
+        plan_id = plan.id
+
+    job_id = None
+    tenant_id = None
+    db_created = False
+    try:
+        with control_session() as s:
+            job = ProvisioningJob(action='create_db', status='running',
+                                  created_at=now_eastern())
+            s.add(job); s.flush(); job_id = job.id
+
+        create_mysql_db_and_user(dbname, dbuser, password, user_host)
+        db_created = True
+
+        with control_session() as s:
+            t = Tenant(slug=slug, name=name, plan_id=plan_id, status='provisioning',
+                       db_host=host, db_port=3306, db_name=dbname, db_user=dbuser,
+                       created_at=now_eastern())
+            t.set_db_password(password)
+            s.add(t); s.flush()
+            tenant_id = t.id
+            db_uri = t.db_uri
+            j = s.get(ProvisioningJob, job_id)
+            if j:
+                j.tenant_id = tenant_id
+
+        # schema: baseline + stamp head (records its own migrate job + alembic_head)
+        class _Ref:
+            pass
+        ref = _Ref(); ref.id = tenant_id; ref.db_uri = db_uri
+        bootstrap_tenant(ref)
+
+        # first admin + setup link
+        username, token = seed_admin(db_uri, admin_email, admin_username, admin_full_name)
+
+        with control_session() as s:
+            primary = _add_domains(s, tenant_id, slug, base, custom_domain)
+            t = s.get(Tenant, tenant_id)
+            t.status = 'active'
+            j = s.get(ProvisioningJob, job_id)
+            if j:
+                j.status = 'ok'; j.finished_at = now_eastern()
+                j.log = f'provisioned tenant {slug} (db={dbname})'
+
+        return {
+            'tenant_id': tenant_id, 'slug': slug, 'db_name': dbname,
+            'db_user': dbuser, 'primary_domain': primary,
+            'custom_domain': custom_domain, 'admin_username': username,
+            'setup_link': setup_link(primary, token),
+        }
+
+    except Exception as e:
+        # best-effort rollback so a retry starts clean
+        if job_id is not None:
+            with control_session() as s:
+                j = s.get(ProvisioningJob, job_id)
+                if j:
+                    j.status = 'failed'; j.finished_at = now_eastern()
+                    j.log = f'{type(e).__name__}: {e}'
+        if tenant_id is not None:
+            with control_session() as s:
+                s.query(TenantDomain).filter_by(tenant_id=tenant_id).delete()
+                t = s.get(Tenant, tenant_id)
+                if t:
+                    s.delete(t)
+        if db_created:
+            try:
+                drop_mysql_db_and_user(dbname, dbuser, user_host)
+            except Exception:
+                pass
+        raise
+
+
+# ── public: adopt an existing DB as tenant-zero ──────────────────────────────
+
+def register_tenant_zero(slug, name, plan_code, db_host, db_name, db_user,
+                         db_password, custom_domain=None, base_domain=None,
+                         db_port=3306):
+    if not _SLUG_RE.match(slug or ''):
+        raise ValueError(f"Invalid slug '{slug}' (must be a DNS label).")
+    base = _base_domain(base_domain)
+
+    with control_session() as s:
+        if s.query(Tenant).filter_by(slug=slug).first():
+            raise ValueError(f"Tenant slug '{slug}' already exists.")
+        plan = s.query(Plan).filter_by(code=plan_code).first()
+        if plan is None:
+            raise ValueError(f"Unknown plan '{plan_code}'.")
+        plan_id = plan.id
+
+    # build a temporary tenant to read the existing DB's current head
+    with control_session() as s:
+        t = Tenant(slug=slug, name=name, plan_id=plan_id, status='active',
+                   db_host=db_host, db_port=db_port, db_name=db_name,
+                   db_user=db_user, created_at=now_eastern())
+        t.set_db_password(db_password)
+        s.add(t); s.flush()
+        tenant_id = t.id
+        db_uri = t.db_uri
+
+    head = current_revision(db_uri)   # read existing alembic_version (no changes)
+
+    with control_session() as s:
+        t = s.get(Tenant, tenant_id)
+        t.alembic_head = head
+        primary = _add_domains(s, tenant_id, slug, base, None)
+        if custom_domain:
+            # tenant-zero's existing domain is trusted/verified (already live)
+            s.add(TenantDomain(
+                tenant_id=tenant_id, domain=custom_domain.lower(), kind='custom',
+                is_primary=False, verified=True, tls_status='active',
+                created_at=now_eastern()))
+        s.add(ProvisioningJob(tenant_id=tenant_id, action='create_db', status='ok',
+                              created_at=now_eastern(), finished_at=now_eastern(),
+                              log=f'registered tenant-zero {slug} in place (head={head})'))
+
+    return {'tenant_id': tenant_id, 'slug': slug, 'db_name': db_name,
+            'primary_domain': primary, 'custom_domain': custom_domain,
+            'alembic_head': head}
+
+
+# ── public: delete / deregister a tenant ─────────────────────────────────────
+
+def delete_tenant(slug, drop_db=False, user_host='%'):
+    """Remove a tenant's control records. Optionally drop its MySQL database.
+
+    Safety: the MySQL drop only proceeds when the stored db_name matches the
+    provisioner's naming convention (db_name_for(slug)). This refuses to drop an
+    externally-named database such as tenant-zero's live LT database.
+    Returns a summary dict.
+    """
+    with control_session() as s:
+        t = s.query(Tenant).filter_by(slug=slug).first()
+        if t is None:
+            raise ValueError(f"No tenant with slug '{slug}'.")
+        tenant_id = t.id
+        db_name = t.db_name
+        db_user = t.db_user
+
+    dropped = False
+    if drop_db:
+        if db_name != db_name_for(slug):
+            raise ValueError(
+                f"Refusing to drop database '{db_name}': it does not match the "
+                f"provisioner naming ('{db_name_for(slug)}'). Drop it manually if "
+                f"you are certain (this guard protects adopted DBs like tenant-zero).")
+        drop_mysql_db_and_user(db_name, db_user, user_host)
+        dropped = True
+
+    with control_session() as s:
+        s.query(TenantDomain).filter_by(tenant_id=tenant_id).delete()
+        s.query(ProvisioningJob).filter_by(tenant_id=tenant_id).delete()
+        t = s.get(Tenant, tenant_id)
+        if t:
+            s.delete(t)
+
+    return {'slug': slug, 'tenant_id': tenant_id, 'db_dropped': dropped,
+            'db_name': db_name}
+
+
+# ── CLI ──────────────────────────────────────────────────────────────────────
+
+def _cmd_create(args):
+    info = create_tenant(
+        slug=args.slug, name=args.name, plan_code=args.plan,
+        admin_email=args.admin_email, admin_username=args.admin_username,
+        admin_full_name=args.admin_name, custom_domain=args.custom_domain,
+        base_domain=args.base_domain, db_host=args.db_host, user_host=args.user_host)
+    print('Tenant provisioned:')
+    for k in ('tenant_id', 'slug', 'db_name', 'db_user', 'primary_domain',
+              'custom_domain', 'admin_username'):
+        print(f'  {k}: {info[k]}')
+    print(f'\n  Admin setup link (send to {args.admin_email}):\n  {info["setup_link"]}')
+    return 0
+
+
+def _cmd_register_zero(args):
+    info = register_tenant_zero(
+        slug=args.slug, name=args.name, plan_code=args.plan,
+        db_host=args.db_host, db_name=args.db_name, db_user=args.db_user,
+        db_password=args.db_password, custom_domain=args.custom_domain,
+        base_domain=args.base_domain, db_port=args.db_port)
+    print('Tenant-zero registered (no data moved):')
+    for k, v in info.items():
+        print(f'  {k}: {v}')
+    return 0
+
+
+def _cmd_delete(args):
+    if not args.yes:
+        print('Refusing to delete without --yes.')
+        return 1
+    info = delete_tenant(args.slug, drop_db=args.drop_db, user_host=args.user_host)
+    print('Tenant deleted:')
+    for k, v in info.items():
+        print(f'  {k}: {v}')
+    return 0
+
+
+def main(argv=None):
+    p = argparse.ArgumentParser(prog='control.provision',
+                                description='Tenant provisioning')
+    sub = p.add_subparsers(dest='command', required=True)
+
+    c = sub.add_parser('create-tenant', help='Provision a brand-new tenant')
+    c.add_argument('--slug', required=True)
+    c.add_argument('--name', required=True)
+    c.add_argument('--plan', required=True, help='plan code, e.g. pro')
+    c.add_argument('--admin-email', required=True)
+    c.add_argument('--admin-username', default=None)
+    c.add_argument('--admin-name', default=None)
+    c.add_argument('--custom-domain', default=None)
+    c.add_argument('--base-domain', default=None, help='overrides TENANT_BASE_DOMAIN')
+    c.add_argument('--db-host', default=None, help='host the app uses to reach the tenant DB')
+    c.add_argument('--user-host', default='%')
+    c.set_defaults(func=_cmd_create)
+
+    z = sub.add_parser('register-tenant-zero', help='Adopt an existing DB in place')
+    z.add_argument('--slug', required=True)
+    z.add_argument('--name', required=True)
+    z.add_argument('--plan', required=True)
+    z.add_argument('--db-host', required=True)
+    z.add_argument('--db-name', required=True)
+    z.add_argument('--db-user', required=True)
+    z.add_argument('--db-password', required=True)
+    z.add_argument('--db-port', type=int, default=3306)
+    z.add_argument('--custom-domain', default=None, help="tenant-zero's existing live domain")
+    z.add_argument('--base-domain', default=None)
+    z.set_defaults(func=_cmd_register_zero)
+
+    d = sub.add_parser('delete-tenant', help='Remove a tenant (control records; optional DB drop)')
+    d.add_argument('--slug', required=True)
+    d.add_argument('--drop-db', action='store_true',
+                   help='also DROP the MySQL database + user (guarded to provisioner-named DBs)')
+    d.add_argument('--user-host', default='%')
+    d.add_argument('--yes', action='store_true', help='confirm deletion')
+    d.set_defaults(func=_cmd_delete)
+
+    args = p.parse_args(argv)
+    sys.exit(args.func(args))
+
+
+if __name__ == '__main__':
+    main()