diff --git a/cli/nao_core/commands/init.py b/cli/nao_core/commands/init.py
index c479aab34..56d36d13c 100644
--- a/cli/nao_core/commands/init.py
+++ b/cli/nao_core/commands/init.py
@@ -210,28 +210,6 @@ def _cleanup_partial_project(project_path: Path) -> None:
         UI.warn(f"Could not remove incomplete project folder [dim]{project_path}[/dim]: {cleanup_error}")
 
 
-def _install_with_progress(extras: list[str]) -> bool:
-    """Run the extras install with a Rich spinner. Returns True on success."""
-    from rich.console import Console
-    from rich.status import Status
-
-    from nao_core.deps import install_extras
-
-    console = Console()
-
-    with Status("[bold cyan]Installing dependencies…[/bold cyan]", console=console, spinner="dots"):
-        success = install_extras(extras)
-
-    if success:
-        UI.success("Dependencies installed successfully.")
-        return True
-
-    extras_str = ",".join(extras)
-    UI.error("Automatic installation failed.")
-    UI.print(f"Install manually with: [bold cyan]pip install 'nao-core[{extras_str}]'[/bold cyan]")
-    return False
-
-
 def _build_no_tty_config(project_name: str, existing_config: NaoConfig | None) -> NaoConfig:
     """Return a config to save in non-interactive mode.
 
@@ -300,26 +278,10 @@ def init(
         UI.print()
 
         # Install missing optional dependencies inline
-        from nao_core.deps import get_missing_extras
+        from nao_core.deps import ensure_extras_installed, get_missing_extras
 
         missing = get_missing_extras(config)
-        deps_ready = not missing
-        if missing:
-            extras_label = ", ".join(missing)
-            UI.title("Installing provider dependencies")
-            UI.print(f"[dim]Extras: {extras_label}[/dim]\n")
-
-            should_install = yes or ask_confirm("Install the required provider dependencies now?", default=True)
-            if should_install:
-                UI.print()
-                deps_ready = _install_with_progress(missing)
-            else:
-                extras_str = ",".join(missing)
-                UI.print()
-                UI.warn("Skipped dependency installation.")
-                UI.print(
-                    f"You can install them later with: [bold cyan]pip install 'nao-core[{extras_str}]'[/bold cyan]"
-                )
+        deps_ready = ensure_extras_installed(missing, assume_yes=yes)
 
         UI.print()
         UI.print("[bold green]Done![/bold green] Your nao project is ready. 🎉")
diff --git a/cli/nao_core/commands/sync/__init__.py b/cli/nao_core/commands/sync/__init__.py
index a51b5b709..68e6cfe37 100644
--- a/cli/nao_core/commands/sync/__init__.py
+++ b/cli/nao_core/commands/sync/__init__.py
@@ -1,6 +1,7 @@
 """Sync command for synchronizing repositories and database schemas."""
 
 import sys
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Annotated, Any
 
@@ -23,6 +24,65 @@
 console = Console()
 
 
+@dataclass
+class PlannedSync:
+    """A selection's resolved items, or the error raised while resolving them."""
+
+    selection: ProviderSelection
+    items: list[Any]
+    error: Exception | None = None
+
+
+def _resolve_planned_syncs(active_providers: list[ProviderSelection], config: NaoConfig) -> list[PlannedSync]:
+    """Resolve each selection's items once, isolating per-selection resolution failures.
+
+    Resolving up front keeps a single source of truth for "which items are being
+    synced" (reused for driver detection and execution), while capturing any
+    failure per selection so one broken provider cannot abort the whole command.
+    """
+    planned: list[PlannedSync] = []
+    for selection in active_providers:
+        try:
+            planned.append(PlannedSync(selection, _filter_items_for_selection(selection, config)))
+        except Exception as error:
+            planned.append(PlannedSync(selection, [], error))
+    return planned
+
+
+def _filter_items_for_selection(selection: ProviderSelection, config: NaoConfig) -> list[Any]:
+    """Return the items a selection will sync, applying its connection filter.
+
+    Single source of truth for "which items are being synced" so dependency
+    detection and provider execution can never target different connections.
+    """
+    items = selection.provider.get_items(config)
+    if selection.connection_name:
+        items = [item for item in items if getattr(item, "name", None) == selection.connection_name]
+    return items
+
+
+def _ensure_database_drivers(planned_syncs: list[PlannedSync]) -> None:
+    """Auto-install missing database driver extras before connecting, like `nao init`.
+
+    Prompts interactively and installs without prompting in non-interactive
+    sessions (CI) so the sync does not hang. If installation is skipped or fails,
+    the sync proceeds and the connection surfaces the usual missing-extra error.
+    """
+    from nao_core.deps import ensure_extras_installed, get_missing_extras_for_databases
+
+    databases = [
+        item
+        for planned in planned_syncs
+        if planned.error is None and isinstance(planned.selection.provider, DatabaseSyncProvider)
+        for item in planned.items
+    ]
+    missing = get_missing_extras_for_databases(databases)
+    if not missing:
+        return
+
+    ensure_extras_installed(missing, assume_yes=not sys.stdin.isatty())
+
+
 @track_command("sync")
 def sync(
     *,
@@ -100,11 +160,19 @@ def sync(
     if select and not any(isinstance(s.provider, DatabaseSyncProvider) for s in active_providers):
         console.print("[yellow]Warning:[/yellow] --select only applies to the databases provider; ignoring it here.")
 
+    # Resolve the items each selection will sync once, then reuse for both
+    # driver-extra detection and provider execution.
+    planned_syncs = _resolve_planned_syncs(active_providers, config)
+
+    _ensure_database_drivers(planned_syncs)
+
     output_dirs = output_dirs or {}
 
     # Run each provider
     results: list[SyncResult] = []
-    for selection in active_providers:
+    for planned in planned_syncs:
+        selection = planned.selection
+        items = planned.items
         sync_provider = selection.provider
         connection_filter = selection.connection_name
 
@@ -113,20 +181,19 @@ def sync(
         output_path = Path(output_dir)
 
         try:
+            if planned.error is not None:
+                raise planned.error
+
             sync_provider.pre_sync(config, output_path)
 
             if not sync_provider.should_sync(config):
                 continue
 
-            # Get items and filter by connection name if specified
-            items = sync_provider.get_items(config)
-            if connection_filter:
-                items = [item for item in items if getattr(item, "name", None) == connection_filter]
-                if not items:
-                    console.print(
-                        f"[yellow]Warning:[/yellow] No connection named '{connection_filter}' found for {sync_provider.name}"
-                    )
-                    continue
+            if connection_filter and not items:
+                console.print(
+                    f"[yellow]Warning:[/yellow] No connection named '{connection_filter}' found for {sync_provider.name}"
+                )
+                continue
 
             sync_kwargs: dict[str, Any] = {"project_path": project_path, "threads": resolved_threads}
             if isinstance(sync_provider, DatabaseSyncProvider):
diff --git a/cli/nao_core/deps.py b/cli/nao_core/deps.py
index 35e09ad6a..abcebbc5b 100644
--- a/cli/nao_core/deps.py
+++ b/cli/nao_core/deps.py
@@ -8,7 +8,7 @@
 from __future__ import annotations
 
 import importlib
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
     from nao_core.config.base import NaoConfig
@@ -137,6 +137,47 @@ def get_install_command(config: NaoConfig) -> str | None:
     return f"pip install 'nao-core[{extras_str}]'"
 
 
+def ensure_extras_installed(extras: list[str], *, assume_yes: bool = False) -> bool:
+    """Install missing extras, prompting first unless *assume_yes*.
+
+    Returns True when the extras are ready (nothing missing or install
+    succeeded), False when installation was skipped or failed.
+    """
+    if not extras:
+        return True
+
+    from nao_core.ui import UI, ask_confirm
+
+    extras_label = ", ".join(extras)
+    UI.title("Installing provider dependencies")
+    UI.print(f"[dim]Extras: {extras_label}[/dim]\n")
+
+    should_install = assume_yes or ask_confirm("Install the required provider dependencies now?", default=True)
+    if not should_install:
+        extras_str = ",".join(extras)
+        UI.print()
+        UI.warn("Skipped dependency installation.")
+        UI.print(f"You can install them later with: [bold cyan]pip install 'nao-core[{extras_str}]'[/bold cyan]")
+        return False
+
+    UI.print()
+    return _install_with_progress(extras)
+
+
+def get_missing_extras_for_databases(databases: list[Any]) -> list[str]:
+    """Return the extras needed by the given databases that are not installed yet."""
+    missing: list[str] = []
+    seen: set[str] = set()
+
+    for db in databases:
+        extra = _resolve_extra(db.type)
+        if extra and extra not in seen and not _is_extra_installed(extra):
+            missing.append(extra)
+            seen.add(extra)
+
+    return missing
+
+
 def install_extras(extras: list[str]) -> bool:
     """Install the given nao-core extras using pip or uv.
 
@@ -170,6 +211,28 @@ def install_extras(extras: list[str]) -> bool:
 # ---------------------------------------------------------------------------
 
 
+def _install_with_progress(extras: list[str]) -> bool:
+    """Run the extras install with a Rich spinner. Returns True on success."""
+    from rich.console import Console
+    from rich.status import Status
+
+    from nao_core.ui import UI
+
+    console = Console()
+
+    with Status("[bold cyan]Installing dependencies…[/bold cyan]", console=console, spinner="dots"):
+        success = install_extras(extras)
+
+    if success:
+        UI.success("Dependencies installed successfully.")
+        return True
+
+    extras_str = ",".join(extras)
+    UI.error("Automatic installation failed.")
+    UI.print(f"Install manually with: [bold cyan]pip install 'nao-core[{extras_str}]'[/bold cyan]")
+    return False
+
+
 def _resolve_extra(provider_or_type: str) -> str | None:
     """Map a config provider/database type to its extra name."""
     name = _PROVIDER_ALIASES.get(provider_or_type, provider_or_type)
diff --git a/cli/tests/nao_core/commands/sync/test_sync.py b/cli/tests/nao_core/commands/sync/test_sync.py
index 13dc11fba..879830445 100644
--- a/cli/tests/nao_core/commands/sync/test_sync.py
+++ b/cli/tests/nao_core/commands/sync/test_sync.py
@@ -1,6 +1,7 @@
 """Unit tests for the main sync command function."""
 
 from pathlib import Path
+from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -192,6 +193,81 @@ def test_sync_shows_partial_success_when_some_providers_fail(self, create_config
         # Should show error details
         assert any("API error" in call for call in calls)
 
+    def test_sync_auto_installs_missing_database_drivers(self, create_config):
+        """Missing driver extras are installed before providers run, like `nao init`."""
+        create_config()
+        provider = MagicMock(spec=DatabaseSyncProvider)
+        provider.should_sync.return_value = True
+        provider.name = "Databases"
+        provider.default_output_dir = "databases"
+        db_item = SimpleNamespace(type="snowflake", name="my-db")
+        provider.get_items.return_value = [db_item]
+        provider.sync.return_value = SyncResult(provider_name="Databases", items_synced=0)
+        selection = ProviderSelection(provider)
+
+        with patch("nao_core.commands.sync.console"):
+            with patch("nao_core.deps.get_missing_extras_for_databases", return_value=["snowflake"]) as mock_missing:
+                with patch("nao_core.deps.ensure_extras_installed", return_value=True) as mock_install:
+                    sync(_providers=[selection], render_templates=False)
+
+        assert mock_missing.call_args[0][0] == [db_item]
+        assert mock_install.call_args[0][0] == ["snowflake"]
+
+    def test_sync_installs_without_prompt_when_non_interactive(self, create_config):
+        """In a non-interactive session the install runs without prompting (assume_yes)."""
+        create_config()
+        provider = MagicMock(spec=DatabaseSyncProvider)
+        provider.should_sync.return_value = True
+        provider.name = "Databases"
+        provider.default_output_dir = "databases"
+        provider.get_items.return_value = [SimpleNamespace(type="snowflake", name="my-db")]
+        provider.sync.return_value = SyncResult(provider_name="Databases", items_synced=0)
+        selection = ProviderSelection(provider)
+
+        with patch("nao_core.commands.sync.console"):
+            with patch("nao_core.deps.get_missing_extras_for_databases", return_value=["snowflake"]):
+                with patch("nao_core.deps.ensure_extras_installed", return_value=True) as mock_install:
+                    with patch("nao_core.commands.sync.sys.stdin") as mock_stdin:
+                        mock_stdin.isatty.return_value = False
+                        sync(_providers=[selection], render_templates=False)
+
+        assert mock_install.call_args.kwargs["assume_yes"] is True
+
+    def test_sync_does_not_install_for_non_database_providers(self, create_config):
+        create_config()
+        selection = _make_provider(items=["item1"], items_synced=1)
+
+        with patch("nao_core.commands.sync.console"):
+            with patch("nao_core.deps.ensure_extras_installed") as mock_install:
+                sync(_providers=[selection], render_templates=False)
+
+        mock_install.assert_not_called()
+
+    def test_sync_isolates_provider_item_resolution_failure(self, create_config):
+        """A provider whose item resolution raises is recorded as a failure without
+        aborting the command; other providers still sync and the summary prints."""
+        create_config()
+        failing = MagicMock(spec=SyncProvider)
+        failing.name = "FailingResolution"
+        failing.emoji = "❌"
+        failing.default_output_dir = "failing-output"
+        failing.get_items.side_effect = Exception("cannot resolve items")
+        failing_selection = ProviderSelection(failing)
+
+        working = _make_provider(
+            name="WorkingProvider", emoji="✅", output_dir="working-output", items=["item1"], items_synced=1
+        )
+
+        with patch("nao_core.commands.sync.console") as mock_console:
+            with pytest.raises(SystemExit) as exc_info:
+                sync(_providers=[failing_selection, working], render_templates=False)
+
+        assert exc_info.value.code == 1
+        working.provider.sync.assert_called_once()
+        calls = [str(call) for call in mock_console.print.call_args_list]
+        assert any("cannot resolve items" in call for call in calls)
+        assert any("Sync Completed with Errors" in call for call in calls)
+
     def test_sync_shows_failure_when_all_providers_fail(self, create_config):
         """Test that sync shows failure status when all providers fail."""
         create_config()
diff --git a/cli/tests/nao_core/test_deps.py b/cli/tests/nao_core/test_deps.py
index 69936bdcc..445443c97 100644
--- a/cli/tests/nao_core/test_deps.py
+++ b/cli/tests/nao_core/test_deps.py
@@ -1,6 +1,13 @@
+from types import SimpleNamespace
+
 import pytest
 
-from nao_core.deps import MissingDependencyError, require_database_backend
+from nao_core.deps import (
+    MissingDependencyError,
+    ensure_extras_installed,
+    get_missing_extras_for_databases,
+    require_database_backend,
+)
 
 
 def test_require_database_backend_uses_public_extra_for_shared_ibis_backend(monkeypatch):
@@ -17,3 +24,43 @@ def raise_missing_backend(module_name: str):
     assert "to connect to redshift databases" in message
     assert "pip install 'nao-core[redshift]'" in message
     assert "uv pip install 'nao-core[redshift]'" in message
+
+
+def test_get_missing_extras_for_databases_returns_uninstalled_deduped(monkeypatch):
+    monkeypatch.setattr("nao_core.deps._is_extra_installed", lambda extra: extra == "duckdb")
+
+    databases = [
+        SimpleNamespace(type="snowflake"),
+        SimpleNamespace(type="snowflake"),
+        SimpleNamespace(type="duckdb"),
+        SimpleNamespace(type="bigquery"),
+    ]
+
+    assert get_missing_extras_for_databases(databases) == ["snowflake", "bigquery"]
+
+
+def test_ensure_extras_installed_returns_true_when_nothing_missing():
+    assert ensure_extras_installed([]) is True
+
+
+def test_ensure_extras_installed_auto_installs_when_assume_yes(monkeypatch):
+    installed: list[list[str]] = []
+    monkeypatch.setattr("nao_core.deps._install_with_progress", lambda extras: installed.append(extras) or True)
+
+    def fail_if_prompted(*args, **kwargs):
+        raise AssertionError("should not prompt when assume_yes is True")
+
+    monkeypatch.setattr("nao_core.ui.ask_confirm", fail_if_prompted)
+
+    assert ensure_extras_installed(["snowflake"], assume_yes=True) is True
+    assert installed == [["snowflake"]]
+
+
+def test_ensure_extras_installed_prompts_and_skips_when_declined(monkeypatch):
+    monkeypatch.setattr("nao_core.ui.ask_confirm", lambda *args, **kwargs: False)
+    monkeypatch.setattr(
+        "nao_core.deps._install_with_progress",
+        lambda extras: (_ for _ in ()).throw(AssertionError("should not install when declined")),
+    )
+
+    assert ensure_extras_installed(["snowflake"], assume_yes=False) is False