diff --git a/users/management/commands/migrate_and_sync_users.py b/users/management/commands/migrate_and_sync_users.py
new file mode 100644
index 0000000000..e2e62bc054
--- /dev/null
+++ b/users/management/commands/migrate_and_sync_users.py
@@ -0,0 +1,265 @@
+"""
+Orchestrate the full edX-to-Keycloak user migration pipeline: backfill edX
+data, classify each candidate user by how confident we are in their name
+data, sync to Keycloak via SCIM, verify what was actually stored, and report.
+
+This replaces the old two-step manual process (running `migrate_edx_data`,
+then an ad hoc SCIM sync script with no field-level visibility) with one
+auditable command.
+
+Stage 1 delegates to `migrate_edx_data --type users` via `call_command()`
+rather than reimplementing its Trino/bulk_create logic here - that command
+also serves five other unrelated migration types (course_runs, entitlements,
+etc.) and owns the Trino schema, so duplicating its logic would mean two
+places to keep in sync as that schema changes. This delegation has real
+limits worth knowing: `migrate_edx_data`'s "users" type doesn't support
+`--dry-run` (only its course_runs/entitlements types do), so this command's
+own `--dry-run` cannot make Stage 1 a no-op - it always writes unless
+`--skip-edx-migration` is also passed. `--limit` is threaded through, since
+that type does respect it.
+"""
+
+import json
+
+from django.contrib.auth import get_user_model
+from django.core.management import BaseCommand, call_command
+from django.db.models import Q
+from mitol.scim import api as scim_api
+
+from users.adapters import LearnUserAdapter
+
+User = get_user_model()
+
+
+class Command(BaseCommand):
+    """Orchestrate the edX-backfill-then-SCIM-sync user migration pipeline."""
+
+    help = __doc__
+
+    def add_arguments(self, parser):
+        """Define the command's CLI flags."""
+        parser.add_argument(
+            "--skip-edx-migration",
+            action="store_true",
+            help="Skip Stage 1 (call_command('migrate_edx_data', type='users')) "
+            "if it's already been run separately.",
+        )
+        parser.add_argument(
+            "--batch-size",
+            type=int,
+            default=250,
+            help="Number of users per SCIM sync batch (default: 250).",
+        )
+        parser.add_argument(
+            "--limit",
+            type=int,
+            help="Limit the number of users processed (for testing purposes). "
+            "Also passed through to migrate_edx_data's Trino query in Stage 1, "
+            "so a small test run doesn't backfill the entire edX dataset just "
+            "to test-sync a handful of users.",
+        )
+        parser.add_argument(
+            "--force",
+            action="store_true",
+            help="Sync users with no name data anywhere in mitxonline anyway, "
+            "with a blank name.givenName/familyName, instead of blocking them.",
+        )
+        parser.add_argument(
+            "--dry-run",
+            action="store_true",
+            help="Run classification only; do not sync to Keycloak. Note this "
+            "does NOT make Stage 1 a no-op - migrate_edx_data's own 'users' "
+            "migration type doesn't support --dry-run, so it still writes "
+            "User/LegalAddress/UserProfile rows unless --skip-edx-migration "
+            "is also passed.",
+        )
+        parser.add_argument(
+            "--report-path",
+            type=str,
+            help="Write the JSON report to this path instead of stdout.",
+        )
+
+    def handle(self, *args, **options):  # noqa: ARG002
+        """Run the backfill/classify/sync/verify/report pipeline."""
+        batch_size = options.get("batch_size") or 250
+        limit = options.get("limit")
+        force = options.get("force", False)
+        dry_run = options.get("dry_run", False)
+        report_path = options.get("report_path")
+
+        if options.get("skip_edx_migration", False):
+            self.stdout.write("Stage 1: skipped (--skip-edx-migration).")
+        else:
+            self._run_edx_backfill(limit)
+
+        self.stdout.write("Stage 2: classifying sync candidates...")
+        candidates = list(
+            User.objects.filter(is_active=True)
+            .filter(Q(global_id="") | Q(scim_external_id=None))
+            .select_related("legal_address")
+            .order_by("id")
+        )
+        if limit is not None:
+            candidates = candidates[:limit]
+
+        to_sync, blocked = self._classify(candidates, force=force)
+        self.stdout.write(
+            f"  {len(to_sync)} ready to sync, {len(blocked)} blocked "
+            f"(pass --force to sync blocked users anyway with a blank name)."
+        )
+
+        if dry_run:
+            self.stdout.write("Dry run: stopping before Stage 3 (sync).")
+            self._write_report(
+                {
+                    "to_sync": [self._report_row(row) for row in to_sync],
+                    "blocked": [self._report_row(row) for row in blocked],
+                    "verified": [],
+                    "mismatched": [],
+                },
+                report_path,
+            )
+            return
+
+        self.stdout.write(f"Stage 3: syncing {len(to_sync)} users to Keycloak...")
+        verified, mismatched = [], []
+        for start in range(0, len(to_sync), batch_size):
+            batch = to_sync[start : start + batch_size]
+            states = scim_api.sync_users_to_scim_remote([row["user"] for row in batch])
+            self.stdout.write(
+                f"  batch {start // batch_size + 1}: "
+                f"{sum(1 for s in states if s.success)}/{len(states)} succeeded"
+            )
+
+            self.stdout.write("Stage 4: verifying batch...")
+            rows_by_user_id = {row["user"].id: row for row in batch}
+            for state in states:
+                row = rows_by_user_id[state.user.id]
+                if not state.success:
+                    self.stdout.write(
+                        self.style.ERROR(
+                            f"  FAILED user={state.user.email} error={state.error}"
+                        )
+                    )
+                    row["outcome"] = "failed"
+                    row["error"] = state.error
+                    blocked.append(row)
+                    continue
+
+                row["outcome"] = "synced"
+                if state.response_body is None:
+                    # matched an existing remote user via search, not a fresh
+                    # create - nothing to verify against, since nothing new
+                    # was sent
+                    verified.append(row)
+                    continue
+
+                sent_name = row["given_name"], row["family_name"]
+                # Keycloak may omit "name" entirely, or echo it back as an
+                # explicit null, for a user synced with a blank given/family
+                # name (exactly the population this command targets) -
+                # `.get("name", {})` only covers the omitted case; a present
+                # `"name": null` returns None itself, and calling .get() on
+                # that would raise AttributeError. Normalize every case to "".
+                name_body = state.response_body.get("name") or {}
+                got_name = (
+                    name_body.get("givenName") or "",
+                    name_body.get("familyName") or "",
+                )
+                if got_name == sent_name:
+                    verified.append(row)
+                else:
+                    row["got_given_name"], row["got_family_name"] = got_name
+                    mismatched.append(row)
+                    self.stdout.write(
+                        self.style.ERROR(
+                            f"  MISMATCH user={state.user.email} "
+                            f"sent={sent_name} got={got_name}"
+                        )
+                    )
+
+        self.stdout.write("Stage 5: report")
+        self.stdout.write(
+            self.style.SUCCESS(
+                f"  {len(verified)} synced and verified, "
+                f"{len(blocked)} blocked/failed, "
+                f"{len(mismatched)} verified-but-mismatched"
+            )
+        )
+        self._write_report(
+            {
+                "verified": [self._report_row(row) for row in verified],
+                "blocked": [self._report_row(row) for row in blocked],
+                "mismatched": [self._report_row(row) for row in mismatched],
+            },
+            report_path,
+        )
+
+    def _run_edx_backfill(self, limit):
+        """Run Stage 1 (migrate_edx_data's "users" migration type).
+
+        migrate_edx_data's "users" type doesn't support --dry-run (only
+        course_runs/entitlements do) - it always writes, so there's no
+        dry_run to thread through here. --limit is threaded through so a
+        small test run doesn't backfill the entire edX dataset just to
+        test-sync a handful of users.
+        """
+        self.stdout.write("Stage 1: backfilling edX user data...")
+        if limit is not None:
+            call_command("migrate_edx_data", type="users", limit=limit)
+        else:
+            call_command("migrate_edx_data", type="users")
+
+    def _classify(self, candidates, *, force):
+        """Split candidates into (to_sync, blocked) using LearnUserAdapter's
+        _resolve_name() tiers, purely for classification/reporting - this
+        never writes to legal_address.
+        """
+        to_sync, blocked = [], []
+        for user in candidates:
+            adapter = LearnUserAdapter(user)
+            given_name, family_name = adapter._resolve_name()  # noqa: SLF001
+            legal_address_complete = bool(
+                user.legal_address.first_name and user.legal_address.last_name
+            )
+            row = {
+                "user": user,
+                "given_name": given_name,
+                "family_name": family_name,
+                "tier": (
+                    "legal_address"
+                    if legal_address_complete
+                    else ("split_name" if given_name or family_name else "none")
+                ),
+            }
+            if row["tier"] == "none" and not force:
+                row["outcome"] = "blocked"
+                blocked.append(row)
+            else:
+                if row["tier"] == "none":
+                    row["outcome"] = "forced-blank-name"
+                to_sync.append(row)
+        return to_sync, blocked
+
+    @staticmethod
+    def _report_row(row):
+        return {
+            "user_id": row["user"].id,
+            "email": row["user"].email,
+            "tier": row["tier"],
+            "given_name": row["given_name"],
+            "family_name": row["family_name"],
+            "outcome": row.get("outcome"),
+            "error": row.get("error"),
+            "got_given_name": row.get("got_given_name"),
+            "got_family_name": row.get("got_family_name"),
+        }
+
+    def _write_report(self, report, report_path):
+        output = json.dumps(report, indent=2, default=str)
+        if report_path:
+            with open(report_path, "w") as f:  # noqa: PTH123
+                f.write(output)
+            self.stdout.write(f"Report written to {report_path}")
+        else:
+            self.stdout.write(output)
diff --git a/users/management/tests/migrate_and_sync_users_test.py b/users/management/tests/migrate_and_sync_users_test.py
new file mode 100644
index 0000000000..f0f726fe17
--- /dev/null
+++ b/users/management/tests/migrate_and_sync_users_test.py
@@ -0,0 +1,288 @@
+"""Tests for migrate_and_sync_users management command"""
+
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from users.factories import UserFactory
+from users.management.commands import migrate_and_sync_users
+
+COMMAND = migrate_and_sync_users.Command()
+
+
+def _state(user, *, success=True, response_body=None, error=None, external_id="ext"):
+    """Build a fake UserState-like object, decoupled from whatever version of
+    mitol-django-scim happens to be installed - the command only duck-types
+    on .user/.success/.response_body/.error.
+    """
+    return SimpleNamespace(
+        user=user,
+        success=success,
+        external_id=external_id if success else None,
+        response_body=response_body,
+        error=error,
+    )
+
+
+def _run(tmp_path, **options):
+    """Run the command and return its parsed JSON report - reading the report
+    file, rather than scraping stdout, avoids Command()'s stdout reference
+    being grabbed at module-import time (before pytest's capture fixtures are
+    active for the current test).
+    """
+    report_path = tmp_path / "report.json"
+    COMMAND.handle(report_path=str(report_path), **options)
+    return json.loads(report_path.read_text())
+
+
+@pytest.fixture(autouse=True)
+def mock_edx_migration(mocker):
+    """Stage 1 always gets skipped/mocked in these tests - never hit Trino."""
+    return mocker.patch("users.management.commands.migrate_and_sync_users.call_command")
+
+
+@pytest.mark.django_db
+def test_limit_is_threaded_through_to_migrate_edx_data(mock_edx_migration, mocker):
+    """--limit should also limit Stage 1's Trino query, not just which
+    mitxonline candidates get classified/synced - otherwise a small test run
+    still backfills the entire edX dataset before test-syncing a handful
+    of users
+    """
+    mocker.patch(
+        "users.management.commands.migrate_and_sync_users.scim_api.sync_users_to_scim_remote",
+        return_value=[],
+    )
+
+    COMMAND.handle(dry_run=False, skip_edx_migration=False, force=False, limit=5)
+
+    mock_edx_migration.assert_called_once_with(
+        "migrate_edx_data", type="users", limit=5
+    )
+
+
+@pytest.mark.django_db
+def test_no_limit_does_not_pass_limit_kwarg(mock_edx_migration, mocker):
+    """Without --limit, migrate_edx_data should run with its own default
+    (unbounded), not an explicit limit=None which would behave differently
+    if migrate_edx_data ever starts treating limit=None as limit=0
+    """
+    mocker.patch(
+        "users.management.commands.migrate_and_sync_users.scim_api.sync_users_to_scim_remote",
+        return_value=[],
+    )
+
+    COMMAND.handle(dry_run=False, skip_edx_migration=False, force=False, limit=None)
+
+    mock_edx_migration.assert_called_once_with("migrate_edx_data", type="users")
+
+
+@pytest.mark.django_db
+def test_dry_run_classifies_without_syncing(mocker, tmp_path):
+    """Dry run reports classification tiers and never calls the sync API"""
+    mock_sync = mocker.patch(
+        "users.management.commands.migrate_and_sync_users.scim_api.sync_users_to_scim_remote"
+    )
+
+    tier1_user = UserFactory.create(name="Joe Smith", global_id=None)
+    tier1_user.legal_address.first_name = "Joe"
+    tier1_user.legal_address.last_name = "Smith"
+    tier1_user.legal_address.save()
+
+    tier2_user = UserFactory.create(name="Jane Doe", global_id=None)
+    tier2_user.legal_address.first_name = ""
+    tier2_user.legal_address.last_name = ""
+    tier2_user.legal_address.save()
+
+    tier3_user = UserFactory.create(name="", global_id=None)
+    tier3_user.legal_address.first_name = ""
+    tier3_user.legal_address.last_name = ""
+    tier3_user.legal_address.save()
+
+    report = _run(
+        tmp_path, dry_run=True, skip_edx_migration=False, force=False, limit=None
+    )
+
+    mock_sync.assert_not_called()
+
+    to_sync_ids = {row["user_id"] for row in report["to_sync"]}
+    blocked_ids = {row["user_id"] for row in report["blocked"]}
+    assert tier1_user.id in to_sync_ids
+    assert tier2_user.id in to_sync_ids
+    assert tier3_user.id in blocked_ids
+
+
+@pytest.mark.django_db
+def test_tier3_blocked_without_force(mocker):
+    """A user with no name data anywhere is excluded from the sync call by default"""
+    mock_sync = mocker.patch(
+        "users.management.commands.migrate_and_sync_users.scim_api.sync_users_to_scim_remote",
+        return_value=[],
+    )
+
+    tier1_user = UserFactory.create(name="Joe Smith", global_id=None)
+    tier1_user.legal_address.first_name = "Joe"
+    tier1_user.legal_address.last_name = "Smith"
+    tier1_user.legal_address.save()
+
+    tier3_user = UserFactory.create(name="", global_id=None)
+    tier3_user.legal_address.first_name = ""
+    tier3_user.legal_address.last_name = ""
+    tier3_user.legal_address.save()
+
+    mock_sync.side_effect = lambda users: [_state(u) for u in users]
+
+    COMMAND.handle(dry_run=False, skip_edx_migration=False, force=False, limit=None)
+
+    synced_users = mock_sync.call_args[0][0]
+    assert tier1_user in synced_users
+    assert tier3_user not in synced_users
+
+
+@pytest.mark.django_db
+def test_tier3_synced_with_force(mocker):
+    """--force syncs a no-name-data user anyway, with a blank name"""
+    mock_sync = mocker.patch(
+        "users.management.commands.migrate_and_sync_users.scim_api.sync_users_to_scim_remote"
+    )
+
+    tier3_user = UserFactory.create(name="", global_id=None)
+    tier3_user.legal_address.first_name = ""
+    tier3_user.legal_address.last_name = ""
+    tier3_user.legal_address.save()
+
+    mock_sync.side_effect = lambda users: [_state(u) for u in users]
+
+    COMMAND.handle(dry_run=False, skip_edx_migration=False, force=True, limit=None)
+
+    synced_users = mock_sync.call_args[0][0]
+    assert tier3_user in synced_users
+
+
+@pytest.mark.django_db
+def test_verifies_matching_response_body(mocker, tmp_path):
+    """A synced user whose echoed response matches what was sent is verified"""
+    user = UserFactory.create(name="Joe Smith", global_id=None)
+    user.legal_address.first_name = "Joe"
+    user.legal_address.last_name = "Smith"
+    user.legal_address.save()
+
+    mocker.patch(
+        "users.management.commands.migrate_and_sync_users.scim_api.sync_users_to_scim_remote",
+        return_value=[
+            _state(
+                user,
+                response_body={"name": {"givenName": "Joe", "familyName": "Smith"}},
+            )
+        ],
+    )
+
+    report = _run(
+        tmp_path, dry_run=False, skip_edx_migration=False, force=False, limit=None
+    )
+
+    assert [row["user_id"] for row in report["verified"]] == [user.id]
+    assert report["mismatched"] == []
+
+
+@pytest.mark.django_db
+def test_verifies_blank_name_when_response_body_omits_name(mocker, tmp_path):
+    """A tier-3 (forced blank name) user's response body omitting "name"
+    entirely must still verify - sent ("", "") should match an absent name,
+    not be flagged as a mismatch
+    """
+    user = UserFactory.create(name="", global_id=None)
+    user.legal_address.first_name = ""
+    user.legal_address.last_name = ""
+    user.legal_address.save()
+
+    mocker.patch(
+        "users.management.commands.migrate_and_sync_users.scim_api.sync_users_to_scim_remote",
+        return_value=[_state(user, response_body={})],
+    )
+
+    report = _run(
+        tmp_path, dry_run=False, skip_edx_migration=False, force=True, limit=None
+    )
+
+    assert [row["user_id"] for row in report["verified"]] == [user.id]
+    assert report["mismatched"] == []
+
+
+@pytest.mark.django_db
+def test_verifies_blank_name_when_response_body_has_explicit_null_name(
+    mocker, tmp_path
+):
+    """A response body with "name": null (present, not omitted) must not
+    crash - .get("name", {}) returns None itself in that case, and calling
+    .get() on it would raise AttributeError without a guard
+    """
+    user = UserFactory.create(name="", global_id=None)
+    user.legal_address.first_name = ""
+    user.legal_address.last_name = ""
+    user.legal_address.save()
+
+    mocker.patch(
+        "users.management.commands.migrate_and_sync_users.scim_api.sync_users_to_scim_remote",
+        return_value=[_state(user, response_body={"name": None})],
+    )
+
+    report = _run(
+        tmp_path, dry_run=False, skip_edx_migration=False, force=True, limit=None
+    )
+
+    assert [row["user_id"] for row in report["verified"]] == [user.id]
+    assert report["mismatched"] == []
+
+
+@pytest.mark.django_db
+def test_flags_mismatched_response_body(mocker, tmp_path):
+    """A synced user whose echoed response doesn't match what was sent is flagged,
+    not silently counted as a success
+    """
+    user = UserFactory.create(name="Joe Smith", global_id=None)
+    user.legal_address.first_name = "Joe"
+    user.legal_address.last_name = "Smith"
+    user.legal_address.save()
+
+    mocker.patch(
+        "users.management.commands.migrate_and_sync_users.scim_api.sync_users_to_scim_remote",
+        return_value=[
+            _state(
+                user,
+                response_body={"name": {"givenName": "", "familyName": ""}},
+            )
+        ],
+    )
+
+    report = _run(
+        tmp_path, dry_run=False, skip_edx_migration=False, force=False, limit=None
+    )
+
+    assert report["verified"] == []
+    assert [row["user_id"] for row in report["mismatched"]] == [user.id]
+
+
+@pytest.mark.django_db
+def test_failed_sync_is_reported_not_swallowed(mocker, tmp_path):
+    """A failed SCIM operation is reported distinctly, not counted as a success"""
+    user = UserFactory.create(name="Joe Smith", global_id=None)
+    user.legal_address.first_name = "Joe"
+    user.legal_address.last_name = "Smith"
+    user.legal_address.save()
+
+    mocker.patch(
+        "users.management.commands.migrate_and_sync_users.scim_api.sync_users_to_scim_remote",
+        return_value=[
+            _state(user, success=False, error={"status": "409"}),
+        ],
+    )
+
+    report = _run(
+        tmp_path, dry_run=False, skip_edx_migration=False, force=False, limit=None
+    )
+
+    assert report["verified"] == []
+    blocked_row = next(row for row in report["blocked"] if row["user_id"] == user.id)
+    assert blocked_row["outcome"] == "failed"
+    assert blocked_row["error"] == {"status": "409"}
diff --git a/users/management/tests/remediate_keycloak_user_names_test.py b/users/management/tests/remediate_keycloak_user_names_test.py
index 34948da4a9..55ac8aaead 100644
--- a/users/management/tests/remediate_keycloak_user_names_test.py
+++ b/users/management/tests/remediate_keycloak_user_names_test.py
@@ -43,7 +43,8 @@ def test_mitxonline_users_lookup_query_count_is_flat(django_assert_max_num_queri
     """_mitxonline_users_by_scim_id() plus constructing a LearnUserAdapter per
     user (as handle() does) must not issue extra queries per user - the fixed
     query count covers select_related(legal_address, user_profile) plus one
-    bulk prefetch for openedx_users, regardless of how many users there are"""
+    bulk prefetch for openedx_users, regardless of how many users there are
+    """
     for i in range(5):
         user = UserFactory.create(scim_external_id=f"kc-{i}")
         user.legal_address.first_name = "Joe"