diff --git a/backend/scripts/run-e2e.sh b/backend/scripts/run-e2e.sh
new file mode 100755
index 0000000..72a7acc
--- /dev/null
+++ b/backend/scripts/run-e2e.sh
@@ -0,0 +1,15 @@
+#!/usr/bin/env bash
+# Run the Playwright e2e suite.
+#
+# Prerequisites:
+#   - Full stack running: backend on :9000, frontend, ngrok (BASE_URL).
+#   - BASE_URL + PLAYWRIGHT_CHROME_PROFILE set in .env / .env.test.
+#   - Your real Chrome CLOSED (the persistent profile can't be shared).
+#   - For the Google Tasks sync test: a Google account connected for tasks
+#     (Settings → re-auth Google, then "Use for tasks").
+#
+# pytest-asyncio's auto mode and pytest-playwright each inject a running event loop
+# that breaks sync Playwright, so we disable both plugins here.
+set -euo pipefail
+cd "$(dirname "$0")/.."
+exec poetry run pytest tests/e2e -m e2e -p no:asyncio -p no:playwright "$@"
diff --git a/backend/tests/e2e/conftest.py b/backend/tests/e2e/conftest.py
index aad5cc8..f4d2b6b 100644
--- a/backend/tests/e2e/conftest.py
+++ b/backend/tests/e2e/conftest.py
@@ -1,17 +1,34 @@
 """Playwright e2e test fixtures — uses real Chrome profile with Google auth.
 
-Prerequisites: Full Stack must be running (backend, frontend, ngrok).
-
-All fixtures use sync Playwright (sync_playwright).
+Prerequisites:
+- Full stack running: backend on :9000, frontend, ngrok (BASE_URL).
+- BASE_URL + PLAYWRIGHT_CHROME_PROFILE set (in .env / .env.test).
+- Your real Chrome must be CLOSED — the persistent profile can't be shared, and a
+  headful launch against a locked profile crashes (SIGTRAP).
+- These tests use SYNC Playwright. pytest-asyncio's auto mode and pytest-playwright
+  both inject a running event loop that breaks sync_playwright, so run via
+  `scripts/run-e2e.sh` (it passes `-p no:asyncio -p no:playwright`).
+
+Some tests need a connected, task-enabled Google account (see `google_tasks_account`).
 """
 
+import asyncio
+import contextlib
 import os
+import uuid
 from collections.abc import Generator
 from pathlib import Path
 
 import pytest
 from dotenv import find_dotenv, load_dotenv
+from google.oauth2.credentials import Credentials
+from googleapiclient.discovery import build
 from playwright.sync_api import Page, sync_playwright
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from underway.config import get_settings
+from underway.models.external_account import ExternalAccount
 
 load_dotenv(find_dotenv())  # BASE_URL etc. — shared with the app
 load_dotenv(find_dotenv(".env.test"))  # test-only creds — kept out of Settings
@@ -91,3 +108,91 @@ def authenticated_page(
 
     context.close()
     pw.stop()
+
+
+async def _load_google_tasks_account() -> tuple[str, str | None, str | None] | None:
+    """Read a connected, task-enabled Google account from the app DB, or None."""
+    settings = get_settings()
+    engine = create_async_engine(settings.database_url)
+    try:
+        async with async_sessionmaker(engine, expire_on_commit=False)() as session:
+            result = await session.execute(
+                select(ExternalAccount).where(
+                    ExternalAccount.provider == "google",
+                    ExternalAccount.use_for_tasks.is_(True),
+                    ExternalAccount.needs_reauth.is_(False),
+                )
+            )
+            account = result.scalars().first()
+            if account is None:
+                return None
+            return account.external_email, account.token, account.refresh_token
+    finally:
+        await engine.dispose()
+
+
+class GoogleTasksSandbox:
+    """Creates sentinel tasks in a real Google Tasks account and cleans them all up."""
+
+    def __init__(self, email: str, service: object) -> None:
+        self.email = email
+        self.service = service
+        self._created_ids: list[str] = []
+
+    def create(self, title: str) -> str:
+        task = self.service.tasks().insert(tasklist="@default", body={"title": title}).execute()
+        task_id: str = task["id"]
+        self._created_ids.append(task_id)
+        return task_id
+
+    def update_title(self, task_id: str, title: str) -> None:
+        self.service.tasks().patch(tasklist="@default", task=task_id, body={"title": title}).execute()
+
+    def delete(self, task_id: str) -> None:
+        self.service.tasks().delete(tasklist="@default", task=task_id).execute()
+        if task_id in self._created_ids:
+            self._created_ids.remove(task_id)
+        else:
+            pass
+
+    def cleanup(self) -> None:
+        for task_id in list(self._created_ids):
+            with contextlib.suppress(Exception):
+                self.service.tasks().delete(tasklist="@default", task=task_id).execute()
+
+
+@pytest.fixture
+def google_tasks(base_url: str) -> Generator[GoogleTasksSandbox]:
+    """A real Google Tasks sandbox for the connected, task-enabled Google account.
+
+    Skips if no such account is connected — set one up once via Settings (re-auth Google,
+    then click "Use for tasks"). Any sentinel tasks created are deleted on teardown.
+    """
+    account = asyncio.run(_load_google_tasks_account())
+    if account is None:
+        pytest.skip(
+            "No task-enabled Google account connected. In Settings, re-auth Google "
+            "(grants the tasks scope) then click 'Use for tasks' before running this test."
+        )
+    else:
+        pass
+
+    email, token, refresh_token = account
+    settings = get_settings()
+    creds = Credentials(
+        token=token,
+        refresh_token=refresh_token,
+        token_uri="https://oauth2.googleapis.com/token",
+        client_id=settings.google_client_id,
+        client_secret=settings.google_client_secret,
+        scopes=["https://www.googleapis.com/auth/tasks"],
+    )
+    service = build("tasks", "v1", credentials=creds)
+    sandbox = GoogleTasksSandbox(email, service)
+    yield sandbox
+    sandbox.cleanup()
+
+
+def unique_sentinel() -> str:
+    """A unique task title so stale rows can never cause a false pass."""
+    return f"E2E-SYNC-{uuid.uuid4()}"
diff --git a/backend/tests/e2e/test_tasks.py b/backend/tests/e2e/test_tasks.py
new file mode 100644
index 0000000..4e30969
--- /dev/null
+++ b/backend/tests/e2e/test_tasks.py
@@ -0,0 +1,74 @@
+"""Real e2e test: Google Tasks → Underway combined-list sync.
+
+Exercises the actual linkages end-to-end against a real Google Tasks account and the
+running app (via ngrok): a task created in Google must appear in Underway after a sync,
+an upstream edit must be reflected, and an upstream delete must be reconciled away. It
+fails loudly if any of those don't happen.
+
+Drives BOTH the UI (clicking Sync on /tasks) and the API (POST /api/tasks/sync,
+GET /api/tasks). See conftest.py for prerequisites; run via scripts/run-e2e.sh.
+"""
+
+from typing import Any
+
+import pytest
+from playwright.sync_api import APIResponse, Page
+
+from tests.e2e.conftest import GoogleTasksSandbox, unique_sentinel
+
+
+def _jwt(page: Page) -> str:
+    token = page.evaluate("() => localStorage.getItem('token')")
+    assert token, "expected a JWT in localStorage after authentication"
+    return str(token)
+
+
+def _api_get_tasks(page: Page, base_url: str, headers: dict[str, str]) -> list[dict[str, Any]]:
+    resp: APIResponse = page.request.get(f"{base_url}/api/tasks", headers=headers)
+    assert resp.ok, f"GET /api/tasks failed: {resp.status}"
+    tasks: list[dict[str, Any]] = resp.json()
+    return tasks
+
+
+def _api_sync(page: Page, base_url: str, headers: dict[str, str]) -> None:
+    resp: APIResponse = page.request.post(f"{base_url}/api/tasks/sync", headers=headers)
+    assert resp.ok, f"POST /api/tasks/sync failed: {resp.status}"
+
+
+@pytest.mark.e2e
+def test_google_task_syncs_into_list_and_reconciles(
+    base_url: str,
+    google_tasks: GoogleTasksSandbox,
+    authenticated_page: Page,
+) -> None:
+    page = authenticated_page
+    headers = {"Authorization": f"Bearer {_jwt(page)}"}
+
+    sentinel = unique_sentinel()
+    task_id = google_tasks.create(sentinel)
+
+    # --- UI: clicking Sync pulls the new Google task onto the board ---
+    page.goto(f"{base_url}/tasks")
+    page.locator("h1", has_text="Tasks").wait_for(state="visible")
+    page.get_by_role("button", name="Sync", exact=True).click()
+    page.locator(".task-card", has_text=sentinel).first.wait_for(state="visible")
+    assert page.locator(".error-banner").count() == 0
+
+    # --- API: it's in the combined list, tagged as a Google Tasks item ---
+    matches = [t for t in _api_get_tasks(page, base_url, headers) if t["title"] == sentinel]
+    assert len(matches) == 1, f"sentinel {sentinel!r} did not sync into /api/tasks"
+    assert matches[0]["provider"] == "google_tasks"
+
+    # --- upstream edit is reflected after sync (content-hash update path) ---
+    updated = f"{sentinel}-UPDATED"
+    google_tasks.update_title(task_id, updated)
+    _api_sync(page, base_url, headers)
+    titles_after_update = {t["title"] for t in _api_get_tasks(page, base_url, headers)}
+    assert updated in titles_after_update, "upstream title change was not synced"
+    assert sentinel not in titles_after_update, "stale pre-edit title lingered after sync"
+
+    # --- upstream delete is reconciled away (sync_task_deletions) ---
+    google_tasks.delete(task_id)
+    _api_sync(page, base_url, headers)
+    titles_after_delete = {t["title"] for t in _api_get_tasks(page, base_url, headers)}
+    assert updated not in titles_after_delete, "task remained after upstream deletion"
diff --git a/backend/tests/integration/test_external_accounts.py b/backend/tests/integration/test_external_accounts.py
new file mode 100644
index 0000000..acf90df
--- /dev/null
+++ b/backend/tests/integration/test_external_accounts.py
@@ -0,0 +1,79 @@
+"""Integration tests for the external-accounts task-enablement action."""
+
+import uuid
+
+from httpx import AsyncClient
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from underway.auth.jwt import create_access_token
+from underway.models.external_account import ExternalAccount
+from underway.models.user import User
+
+SECRET = "test-secret-key-at-least-32-chars!"
+
+
+async def _create_user(db_session: AsyncSession, email: str = "acct@example.com") -> User:
+    user = User(app_login=email)
+    user.id = uuid.uuid4()
+    db_session.add(user)
+    await db_session.commit()
+    return user
+
+
+def _auth_headers(user: User) -> dict[str, str]:
+    token = create_access_token(user.id, user.app_login, SECRET)
+    return {"Authorization": f"Bearer {token}"}
+
+
+async def _create_account(db_session: AsyncSession, user: User, provider: str = "google") -> ExternalAccount:
+    account = ExternalAccount(
+        user_id=user.id,
+        external_email=user.app_login,
+        provider=provider,
+        token="tok",
+        use_for_calendar=True,
+        use_for_tasks=False,
+    )
+    db_session.add(account)
+    await db_session.commit()
+    await db_session.refresh(account)
+    return account
+
+
+class TestUseForTasks:
+    async def test_enables_use_for_tasks(self, client: AsyncClient, db_session: AsyncSession) -> None:
+        user = await _create_user(db_session)
+        account = await _create_account(db_session, user)
+
+        response = await client.post(
+            f"/api/external-accounts/{account.id}/use-for-tasks",
+            headers=_auth_headers(user),
+        )
+        assert response.status_code == 200
+        assert response.json()["status"] == "ok"
+
+        # Verify via the API (the request commits in its own session, separate from db_session).
+        listed = await client.get("/api/external-accounts", headers=_auth_headers(user))
+        accounts = listed.json()
+        assert len(accounts) == 1
+        assert accounts[0]["use_for_tasks"] is True
+        assert accounts[0]["is_primary_tasks"] is True
+
+    async def test_other_users_account_returns_404(self, client: AsyncClient, db_session: AsyncSession) -> None:
+        owner = await _create_user(db_session, "owner@example.com")
+        intruder = await _create_user(db_session, "intruder@example.com")
+        account = await _create_account(db_session, owner)
+
+        response = await client.post(
+            f"/api/external-accounts/{account.id}/use-for-tasks",
+            headers=_auth_headers(intruder),
+        )
+        assert response.status_code == 404
+
+    async def test_invalid_id_returns_400(self, client: AsyncClient, db_session: AsyncSession) -> None:
+        user = await _create_user(db_session)
+        response = await client.post(
+            "/api/external-accounts/not-a-uuid/use-for-tasks",
+            headers=_auth_headers(user),
+        )
+        assert response.status_code == 400
diff --git a/backend/tests/integration/test_tasks.py b/backend/tests/integration/test_tasks.py
index 0ebc002..6cbd940 100644
--- a/backend/tests/integration/test_tasks.py
+++ b/backend/tests/integration/test_tasks.py
@@ -2,12 +2,16 @@
 
 import uuid
 
+import pytest
 from httpx import AsyncClient
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from underway.auth.jwt import create_access_token
+from underway.models.external_account import ExternalAccount
 from underway.models.task import Task
 from underway.models.user import User
+from underway.providers.task_manager import TaskManager
+from underway.providers.task_provider import ProviderTask
 
 SECRET = "test-secret-key-at-least-32-chars!"
 
@@ -259,3 +263,58 @@ async def test_sync_returns_ok(self, client: AsyncClient, db_session: AsyncSessi
         user = await _create_user(db_session)
         response = await client.post("/api/tasks/sync", headers=_auth_headers(user))
         assert response.status_code == 200
+
+    async def test_sync_imports_provider_tasks_and_is_idempotent(
+        self,
+        client: AsyncClient,
+        db_session: AsyncSession,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        user = await _create_user(db_session, "synctest@example.com")
+        account = ExternalAccount(
+            user_id=user.id,
+            external_email="synctest@example.com",
+            provider="todoist",
+            use_for_tasks=True,
+            needs_reauth=False,
+            api_key="key",
+        )
+        db_session.add(account)
+        await db_session.commit()
+
+        async def fake_get_tasks(
+            self: TaskManager,
+            session: AsyncSession,
+            user_id: uuid.UUID,
+            task_user_email: str,
+            provider_name: str,
+        ) -> list[ProviderTask]:
+            return [
+                ProviderTask(
+                    id="t1",
+                    title="Imported from Todoist",
+                    project_id="proj1",
+                    priority=1,
+                    due_date=None,
+                    status="active",
+                    provider_task_id="t1",
+                )
+            ]
+
+        monkeypatch.setattr(TaskManager, "get_tasks", fake_get_tasks)
+
+        response = await client.post("/api/tasks/sync", headers=_auth_headers(user))
+        assert response.status_code == 200
+        body = response.json()
+        assert body["status"] == "ok"
+        assert body["upserted"] == 1
+
+        listed = await client.get("/api/tasks", headers=_auth_headers(user))
+        tasks = listed.json()
+        assert len(tasks) == 1
+        assert tasks[0]["title"] == "Imported from Todoist"
+        assert tasks[0]["provider"] == "todoist"
+
+        # Second sync with identical content is a no-op (content-hash unchanged).
+        response2 = await client.post("/api/tasks/sync", headers=_auth_headers(user))
+        assert response2.json()["upserted"] == 0
diff --git a/backend/tests/unit/test_task_sync.py b/backend/tests/unit/test_task_sync.py
index 719ab4d..fee983c 100644
--- a/backend/tests/unit/test_task_sync.py
+++ b/backend/tests/unit/test_task_sync.py
@@ -1,14 +1,22 @@
 """Unit tests for task sync service."""
 
 import uuid
+from uuid import UUID
 
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from underway.models.external_account import ExternalAccount
 from underway.models.task import Task
 from underway.models.user import User
+from underway.providers.task_manager import TaskManager
 from underway.providers.task_provider import ProviderTask
-from underway.services.task_sync import _compute_hash, sync_provider_tasks, sync_task_deletions
+from underway.services.task_sync import (
+    _compute_hash,
+    sync_all_task_accounts,
+    sync_provider_tasks,
+    sync_task_deletions,
+)
 
 
 async def _create_user(session: AsyncSession, email: str = "sync@example.com") -> User:
@@ -83,6 +91,109 @@ async def test_skips_unchanged_tasks(self, db_session: AsyncSession) -> None:
         assert count == 0
 
 
+class _FakeTaskManager(TaskManager):
+    """TaskManager whose get_tasks returns canned tasks per account email (no network)."""
+
+    def __init__(
+        self,
+        tasks_by_email: dict[str, list[ProviderTask]],
+        fail_emails: set[str] | None = None,
+    ) -> None:
+        super().__init__()
+        self._tasks_by_email = tasks_by_email
+        self._fail_emails = fail_emails or set()
+        self.calls: list[tuple[str, str]] = []
+
+    async def get_tasks(
+        self,
+        session: AsyncSession,
+        user_id: UUID,
+        task_user_email: str,
+        provider_name: str,
+    ) -> list[ProviderTask]:
+        self.calls.append((task_user_email, provider_name))
+        if task_user_email in self._fail_emails:
+            raise RuntimeError("provider boom")
+        else:
+            return self._tasks_by_email.get(task_user_email, [])
+
+
+async def _create_task_account(
+    session: AsyncSession,
+    user: User,
+    provider: str,
+    email: str,
+) -> ExternalAccount:
+    account = ExternalAccount(
+        user_id=user.id,
+        external_email=email,
+        provider=provider,
+        use_for_tasks=True,
+        needs_reauth=False,
+        api_key="key" if provider == "todoist" else None,
+        token=None if provider == "todoist" else "tok",
+    )
+    session.add(account)
+    await session.flush()
+    return account
+
+
+class TestSyncAllTaskAccounts:
+    async def test_maps_db_provider_to_task_provider(self, db_session: AsyncSession) -> None:
+        user = await _create_user(db_session)
+        await _create_task_account(db_session, user, "todoist", "td@example.com")
+        await _create_task_account(db_session, user, "google", "g@example.com")
+        await _create_task_account(db_session, user, "o365", "o@example.com")
+
+        manager = _FakeTaskManager(tasks_by_email={})
+        summary = await sync_all_task_accounts(db_session, user.id, manager)
+
+        assert summary == {"accounts": 3, "upserted": 0, "failed": 0}
+        # Accounts come back ordered by provider name: google, o365, todoist.
+        assert manager.calls == [
+            ("g@example.com", "google_tasks"),
+            ("o@example.com", "outlook"),
+            ("td@example.com", "todoist"),
+        ]
+
+    async def test_syncs_provider_tasks_into_db(self, db_session: AsyncSession) -> None:
+        user = await _create_user(db_session)
+        await _create_task_account(db_session, user, "todoist", "td@example.com")
+        pt = _make_provider_task(title="Imported")
+
+        manager = _FakeTaskManager(tasks_by_email={"td@example.com": [pt]})
+        summary = await sync_all_task_accounts(db_session, user.id, manager)
+
+        assert summary == {"accounts": 1, "upserted": 1, "failed": 0}
+        result = await db_session.execute(select(Task).where(Task.user_id == user.id))
+        task = result.scalar_one()
+        assert task.title == "Imported"
+        assert task.provider == "todoist"
+
+    async def test_one_failing_account_does_not_abort_others(self, db_session: AsyncSession) -> None:
+        user = await _create_user(db_session)
+        await _create_task_account(db_session, user, "todoist", "ok@example.com")
+        await _create_task_account(db_session, user, "google", "boom@example.com")
+        pt = _make_provider_task(title="Survives")
+
+        manager = _FakeTaskManager(
+            tasks_by_email={"ok@example.com": [pt]},
+            fail_emails={"boom@example.com"},
+        )
+        summary = await sync_all_task_accounts(db_session, user.id, manager)
+
+        assert summary == {"accounts": 2, "upserted": 1, "failed": 1}
+        result = await db_session.execute(select(Task).where(Task.user_id == user.id))
+        task = result.scalar_one()
+        assert task.title == "Survives"
+
+    async def test_no_accounts_returns_zero_summary(self, db_session: AsyncSession) -> None:
+        user = await _create_user(db_session)
+        manager = _FakeTaskManager(tasks_by_email={})
+        summary = await sync_all_task_accounts(db_session, user.id, manager)
+        assert summary == {"accounts": 0, "upserted": 0, "failed": 0}
+
+
 class TestSyncTaskDeletions:
     async def test_deletes_tasks_not_in_provider(self, db_session: AsyncSession) -> None:
         user = await _create_user(db_session)
diff --git a/backend/underway/app.py b/backend/underway/app.py
index c0e35eb..a5483e3 100644
--- a/backend/underway/app.py
+++ b/backend/underway/app.py
@@ -16,6 +16,7 @@
 from underway.auth.jwt import create_token_auth
 from underway.chat.streaming import router as chat_router
 from underway.config import Settings, get_settings
+from underway.providers.task_sync_loop import task_sync_loop
 from underway.providers.token_refresh import token_refresh_loop
 from underway.routes.auth import router as auth_router
 from underway.routes.auth import test_router as auth_test_router
@@ -61,14 +62,16 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         """Start background tasks on startup, cancel cleanly on shutdown."""
         factory = _get_or_create_factory()
         refresh_task = asyncio.create_task(token_refresh_loop(factory))
-        logger.info("Token refresh background task started")
+        sync_task = asyncio.create_task(task_sync_loop(factory))
+        logger.info("Token refresh and task sync background tasks started")
         try:
             yield
         finally:
-            refresh_task.cancel()
-            with contextlib.suppress(asyncio.CancelledError):
-                await refresh_task
-            logger.info("Token refresh background task stopped")
+            for background_task in (refresh_task, sync_task):
+                background_task.cancel()
+                with contextlib.suppress(asyncio.CancelledError):
+                    await background_task
+            logger.info("Token refresh and task sync background tasks stopped")
 
     app = FastAPI(
         title="Underway",
diff --git a/backend/underway/models/external_account.py b/backend/underway/models/external_account.py
index 3886146..9172ca5 100644
--- a/backend/underway/models/external_account.py
+++ b/backend/underway/models/external_account.py
@@ -132,6 +132,13 @@ async def get_task_accounts_for_user(cls, session: AsyncSession, user_id: uuid.U
         result = await session.execute(stmt)
         return list(result.scalars().all())
 
+    @classmethod
+    async def get_user_ids_with_task_accounts(cls, session: AsyncSession) -> list[uuid.UUID]:
+        """Return distinct user ids that have at least one task-enabled account."""
+        stmt = select(cls.user_id).where(cls.use_for_tasks.is_(True)).distinct()
+        result = await session.execute(stmt)
+        return list(result.scalars().all())
+
     @classmethod
     async def get_task_account(
         cls,
diff --git a/backend/underway/providers/google_tasks.py b/backend/underway/providers/google_tasks.py
index 33d9d22..14f4277 100644
--- a/backend/underway/providers/google_tasks.py
+++ b/backend/underway/providers/google_tasks.py
@@ -6,11 +6,11 @@
 import logging
 import uuid
 from datetime import datetime
+from typing import TYPE_CHECKING
 from uuid import UUID
 
 from google.auth.transport.requests import Request
 from google.oauth2.credentials import Credentials
-from googleapiclient._apis.tasks.v1 import TasksResource
 from googleapiclient.discovery import build
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -18,6 +18,9 @@
 from underway.models.external_account import ExternalAccount
 from underway.providers.task_provider import ProviderTask, TaskProvider
 
+if TYPE_CHECKING:
+    from googleapiclient._apis.tasks.v1 import TasksResource
+
 logger = logging.getLogger(__name__)
 
 INSTRUCTION_TASK_TITLE = "AI Instructions"
diff --git a/backend/underway/providers/task_sync_loop.py b/backend/underway/providers/task_sync_loop.py
new file mode 100644
index 0000000..1bfb093
--- /dev/null
+++ b/backend/underway/providers/task_sync_loop.py
@@ -0,0 +1,46 @@
+"""Background loop that periodically syncs provider tasks into the combined list."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from underway.models.external_account import ExternalAccount
+from underway.providers.task_manager import TaskManager
+from underway.services.task_sync import sync_all_task_accounts
+
+logger = logging.getLogger(__name__)
+
+TASK_SYNC_INTERVAL_SECONDS = 15 * 60  # 15 minutes
+
+
+async def sync_all_users(
+    session_factory: async_sessionmaker[AsyncSession],
+    task_manager: TaskManager,
+) -> None:
+    """Sync every user that has task-enabled accounts. One pass; resilient per user."""
+    async with session_factory() as session:
+        user_ids = await ExternalAccount.get_user_ids_with_task_accounts(session)
+
+    for user_id in user_ids:
+        try:
+            async with session_factory() as session:
+                summary = await sync_all_task_accounts(session, user_id, task_manager)
+                await session.commit()
+            logger.info("Task sync for user_id=%s: %s", user_id, summary)
+        except Exception:
+            logger.exception("Task sync loop failed for user_id=%s", user_id)
+
+
+async def task_sync_loop(session_factory: async_sessionmaker[AsyncSession]) -> None:
+    """Background loop that syncs provider tasks for all users every interval."""
+    logger.info("Starting task sync background loop")
+    task_manager = TaskManager()
+    while True:
+        try:
+            await sync_all_users(session_factory, task_manager)
+        except Exception:
+            logger.exception("Error in task sync loop")
+        await asyncio.sleep(TASK_SYNC_INTERVAL_SECONDS)
diff --git a/backend/underway/services/task_sync.py b/backend/underway/services/task_sync.py
index 417ccaa..5415550 100644
--- a/backend/underway/services/task_sync.py
+++ b/backend/underway/services/task_sync.py
@@ -11,12 +11,57 @@
 from sqlalchemy import delete, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from underway.models.external_account import PROVIDER_TO_TASK, ExternalAccount
 from underway.models.task import Task
+from underway.providers.task_manager import TaskManager
 from underway.providers.task_provider import ProviderTask
 
 logger = logging.getLogger(__name__)
 
 
+async def sync_all_task_accounts(
+    session: AsyncSession,
+    user_id: UUID,
+    task_manager: TaskManager,
+) -> dict[str, int]:
+    """Sync every task-enabled external account for a user into the combined task list.
+
+    Resilient: a failure on one account is logged and counted but does not abort the rest.
+    Returns a summary with the number of accounts processed, tasks upserted, and failures.
+    """
+    accounts = await ExternalAccount.get_task_accounts_for_user(session, user_id)
+    upserted = 0
+    failed = 0
+
+    for account in accounts:
+        # account.provider is the DB provider ("todoist"/"google"/"o365"); map it to the
+        # task-provider key ("todoist"/"google_tasks"/"outlook") used by TaskManager and the
+        # Task.provider column so reads and write-backs stay consistent.
+        task_provider = PROVIDER_TO_TASK.get(account.provider)
+        if task_provider is None:
+            logger.warning(
+                "Account %s has provider %r not usable for tasks; skipping",
+                account.external_email,
+                account.provider,
+            )
+            failed += 1
+            continue
+        else:
+            # Provider maps to a known task provider — proceed with sync.
+            pass
+
+        try:
+            provider_tasks = await task_manager.get_tasks(session, user_id, account.external_email, task_provider)
+            upserted += await sync_provider_tasks(
+                session, user_id, account.external_email, task_provider, provider_tasks
+            )
+        except Exception:
+            logger.exception("Sync failed for account %s (%s)", account.external_email, task_provider)
+            failed += 1
+
+    return {"accounts": len(accounts), "upserted": upserted, "failed": failed}
+
+
 async def sync_provider_tasks(
     session: AsyncSession,
     user_id: UUID,
diff --git a/backend/underway/viewsets/external_accounts.py b/backend/underway/viewsets/external_accounts.py
index f85e5ff..9d10002 100644
--- a/backend/underway/viewsets/external_accounts.py
+++ b/backend/underway/viewsets/external_accounts.py
@@ -1,7 +1,12 @@
-"""ExternalAccount viewset — read-only, filtered to current user."""
+"""ExternalAccount viewset — read-only list/detail, plus a task-enablement action."""
 
 from __future__ import annotations
 
+import uuid
+
+from fastapi import HTTPException
+from fastrest.decorators import action
+from fastrest.request import Request
 from fastrest.viewsets import ReadOnlyModelViewSet
 from sqlalchemy import select
 
@@ -21,3 +26,36 @@ async def get_queryset(self) -> list[ExternalAccount]:
         stmt = select(ExternalAccount).where(ExternalAccount.user_id == user.id)
         result = await self._session.execute(stmt)
         return list(result.scalars().all())
+
+    @action(methods=["post"], detail=True, url_path="use-for-tasks")
+    async def use_for_tasks(self, request: Request, pk: str = "", **kwargs: str) -> dict[str, str]:
+        """POST /api/external-accounts/{pk}/use-for-tasks — enable an account as a task source."""
+        user = request.user
+        session = self._session
+
+        try:
+            parsed_id = uuid.UUID(pk)
+        except ValueError as exc:
+            raise HTTPException(status_code=400, detail="Invalid account id.") from exc
+
+        result = await session.execute(
+            select(ExternalAccount).where(
+                ExternalAccount.id == parsed_id,
+                ExternalAccount.user_id == user.id,
+            )
+        )
+        account = result.scalar_one_or_none()
+        if account is None:
+            raise HTTPException(status_code=404, detail="Account not found.")
+        else:
+            account.use_for_tasks = True
+            await ExternalAccount.set_as_primary(
+                session,
+                external_email=account.external_email,
+                provider=account.provider,
+                user_id=user.id,
+                account_type="tasks",
+            )
+            await session.flush()
+
+        return {"status": "ok"}
diff --git a/backend/underway/viewsets/tasks.py b/backend/underway/viewsets/tasks.py
index b4637a7..08a435c 100644
--- a/backend/underway/viewsets/tasks.py
+++ b/backend/underway/viewsets/tasks.py
@@ -11,11 +11,13 @@
 from sqlalchemy import select, update
 
 from underway.models.task import Task
+from underway.providers.task_manager import TaskManager
 from underway.serializers.task import (
     TaskMoveSerializer,
     TaskOrderSerializer,
     TaskSerializer,
 )
+from underway.services.task_sync import sync_all_task_accounts
 from underway.viewsets.base import SessionMixin
 
 
@@ -127,10 +129,13 @@ async def reorder(self, request: Request, **kwargs: str) -> dict[str, str]:
         return {"status": "ok"}
 
     @action(methods=["post"], detail=False, url_path="sync")
-    async def sync(self, request: Request, **kwargs: str) -> dict[str, str]:
-        """POST /api/tasks/sync — trigger sync from all providers."""
-        # Provider sync will be implemented in step 2.3/2.6
-        return {"status": "ok", "message": "Sync not yet implemented."}
+    async def sync(self, request: Request, **kwargs: str) -> dict[str, object]:
+        """POST /api/tasks/sync — pull tasks from all connected providers into the list."""
+        user = request.user
+        session = self._session
+        summary = await sync_all_task_accounts(session, user.id, TaskManager())
+        await session.flush()
+        return {"status": "ok", **summary}
 
     @action(methods=["post"], detail=True, url_path="update-status")
     async def update_status(self, request: Request, pk: str = "", **kwargs: str) -> dict[str, str]:
diff --git a/frontend/src/views/SettingsView.vue b/frontend/src/views/SettingsView.vue
index b5933a1..7feee0b 100644
--- a/frontend/src/views/SettingsView.vue
+++ b/frontend/src/views/SettingsView.vue
@@ -58,6 +58,7 @@ interface ModelTestResult {
 const testResult = ref<ModelTestResult | null>(null)
 const testing = ref(false)
 const connecting = ref<'google' | 'o365' | null>(null)
+const enablingTasks = ref<string | null>(null)
 
 async function loadSettings() {
   try {
@@ -79,6 +80,20 @@ async function loadAccounts() {
   }
 }
 
+async function enableTasks(accountId: string) {
+  if (enablingTasks.value) return
+  error.value = null
+  enablingTasks.value = accountId
+  try {
+    await api.post(`/external-accounts/${accountId}/use-for-tasks`)
+    await loadAccounts()
+  } catch (e: unknown) {
+    error.value = e instanceof Error ? e.message : 'Failed to enable account for tasks'
+  } finally {
+    enablingTasks.value = null
+  }
+}
+
 async function saveSettings() {
   saving.value = true
   saved.value = false
@@ -285,6 +300,14 @@ onMounted(() => {
           <span v-if="account.needs_reauth" class="badge warning">Needs re-auth</span>
           <span v-if="account.use_for_calendar" class="badge">Calendar</span>
           <span v-if="account.use_for_tasks" class="badge">Tasks</span>
+          <button
+            v-if="!account.use_for_tasks && (account.provider === 'google' || account.provider === 'o365')"
+            class="btn-enable-tasks"
+            :disabled="enablingTasks === account.id"
+            @click="enableTasks(account.id)"
+          >
+            {{ enablingTasks === account.id ? 'Enabling…' : 'Use for tasks' }}
+          </button>
         </div>
       </div>
 
@@ -394,6 +417,26 @@ button[type='submit']:disabled {
 .account-status {
   display: flex;
   gap: 0.4rem;
+  align-items: center;
+}
+
+.btn-enable-tasks {
+  font-size: 0.75rem;
+  padding: 0.2rem 0.6rem;
+  border-radius: 4px;
+  background: var(--color-background-soft);
+  border: 1px solid var(--color-border);
+  color: var(--color-text);
+  cursor: pointer;
+}
+
+.btn-enable-tasks:hover:not(:disabled) {
+  border-color: var(--color-heading);
+}
+
+.btn-enable-tasks:disabled {
+  opacity: 0.6;
+  cursor: default;
 }
 
 .badge {