diff --git a/containers/backupchecks/src/backend/app/__init__.py b/containers/backupchecks/src/backend/app/__init__.py
index e46dd29..ca9b7ed 100644
--- a/containers/backupchecks/src/backend/app/__init__.py
+++ b/containers/backupchecks/src/backend/app/__init__.py
@@ -13,6 +13,7 @@ from .main.routes import main_bp
 from .main.routes_documentation import doc_bp
 from .migrations import run_migrations
 from .auto_importer_service import start_auto_importer
+from .cove_importer_service import start_cove_importer
 
 
 def _get_today_ui_date() -> str:
@@ -212,4 +213,7 @@ def create_app():
     # Start automatic mail importer background thread
     start_auto_importer(app)
 
+    # Start Cove Data Protection importer background thread
+    start_cove_importer(app)
+
     return app
diff --git a/containers/backupchecks/src/backend/app/cove_importer.py b/containers/backupchecks/src/backend/app/cove_importer.py
new file mode 100644
index 0000000..caaf6a4
--- /dev/null
+++ b/containers/backupchecks/src/backend/app/cove_importer.py
@@ -0,0 +1,414 @@
+"""Cove Data Protection API importer.
+
+Fetches backup job run data from the Cove (N-able) API and creates
+JobRun records in the local database.
+"""
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+import requests
+from sqlalchemy import text
+
+from .database import db
+
+logger = logging.getLogger(__name__)
+
+COVE_DEFAULT_URL = "https://api.backup.management/jsonapi"
+
+# Columns to request from EnumerateAccountStatistics
+COVE_COLUMNS = [
+    "I1",     # Account name
+    "I18",    # Account ID
+    "I8",     # Customer / partner name
+    "I78",    # Backup type / product
+    "D09F00", # Overall last session status
+    "D09F09", # Last session start timestamp
+    "D09F15", # Last session end timestamp
+    "D09F08", # 28-day colorbar
+    # Datasource-specific status (F00) and last session time (F15)
+    "D1F00",  "D1F15",   # Files & Folders
+    "D10F00", "D10F15",  # VssMsSql
+    "D11F00", "D11F15",  # VssSharePoint
+    "D19F00", "D19F15",  # M365 Exchange
+    "D20F00", "D20F15",  # M365 OneDrive
+    "D5F00",  "D5F15",   # M365 SharePoint
+    "D23F00", "D23F15",  # M365 Teams
+]
+
+# Mapping from Cove status code to Backupchecks status string
+STATUS_MAP: dict[int, str] = {
+    1:  "Warning",  # Running
+    2:  "Error",    # Failed
+    3:  "Error",    # Interrupted
+    5:  "Success",  # Completed
+    6:  "Error",    # Overdue
+    7:  "Warning",  # No recent backup
+    8:  "Warning",  # Idle
+    9:  "Warning",  # Unknown
+    10: "Error",    # Account error
+    11: "Warning",  # Quota exceeded
+    12: "Warning",  # License issue
+}
+
+# Datasource label mapping (prefix → human-readable label)
+DATASOURCE_LABELS: dict[str, str] = {
+    "D1":  "Files & Folders",
+    "D10": "VssMsSql",
+    "D11": "VssSharePoint",
+    "D19": "M365 Exchange",
+    "D20": "M365 OneDrive",
+    "D5":  "M365 SharePoint",
+    "D23": "M365 Teams",
+}
+
+
+class CoveImportError(Exception):
+    """Raised when Cove API interaction fails."""
+
+
+def _cove_login(url: str, username: str, password: str) -> tuple[str, int]:
+    """Login to the Cove API and return (visa, partner_id).
+
+    Raises CoveImportError on failure.
+    """
+    payload = {
+        "jsonrpc": "2.0",
+        "method": "Login",
+        "id": 1,
+        "params": {
+            "Username": username,
+            "Password": password,
+        },
+    }
+    try:
+        resp = requests.post(url, json=payload, timeout=30)
+        resp.raise_for_status()
+        data = resp.json()
+    except requests.RequestException as exc:
+        raise CoveImportError(f"Cove login request failed: {exc}") from exc
+    except ValueError as exc:
+        raise CoveImportError(f"Cove login response is not valid JSON: {exc}") from exc
+
+    result = data.get("result") or {}
+    if not result:
+        error = data.get("error") or {}
+        raise CoveImportError(f"Cove login failed: {error.get('message', 'unknown error')}")
+
+    visa = result.get("Visa") or ""
+    if not visa:
+        raise CoveImportError("Cove login succeeded but no Visa token returned")
+
+    # Partner/account info may be nested
+    account_info = result.get("Accounts", [{}])[0] if result.get("Accounts") else {}
+    partner_id = (
+        account_info.get("PartnerId")
+        or result.get("PartnerID")
+        or result.get("PartnerId")
+        or 0
+    )
+
+    return visa, int(partner_id)
+
+
+def _cove_enumerate(
+    url: str,
+    visa: str,
+    partner_id: int,
+    start: int,
+    count: int,
+) -> list[dict]:
+    """Call EnumerateAccountStatistics and return a list of account dicts.
+
+    Returns empty list when no more results.
+    """
+    payload = {
+        "jsonrpc": "2.0",
+        "method": "EnumerateAccountStatistics",
+        "id": 2,
+        "visa": visa,
+        "params": {
+            "Query": {
+                "PartnerId": partner_id,
+                "DisplayColumns": COVE_COLUMNS,
+                "StartRecordNumber": start,
+                "RecordCount": count,
+                "Columns": COVE_COLUMNS,
+            }
+        },
+    }
+    try:
+        resp = requests.post(url, json=payload, timeout=60)
+        resp.raise_for_status()
+        data = resp.json()
+    except requests.RequestException as exc:
+        raise CoveImportError(f"Cove EnumerateAccountStatistics request failed: {exc}") from exc
+    except ValueError as exc:
+        raise CoveImportError(f"Cove EnumerateAccountStatistics response is not valid JSON: {exc}") from exc
+
+    result = data.get("result") or {}
+    if not result:
+        error = data.get("error") or {}
+        raise CoveImportError(f"Cove EnumerateAccountStatistics failed: {error.get('message', 'unknown error')}")
+
+    return result.get("result", []) or []
+
+
+def _flatten_settings(account: dict) -> dict:
+    """Convert the Settings array in an account dict to a flat key→value dict.
+
+    Cove returns settings as a list of {Key, Value} objects.
+    This flattens them so we can do `flat['D09F00']` instead of iterating.
+    """
+    flat: dict[str, Any] = {}
+    settings_list = account.get("Settings") or []
+    if isinstance(settings_list, list):
+        for item in settings_list:
+            if isinstance(item, dict):
+                key = item.get("Key") or item.get("key")
+                value = item.get("Value") if "Value" in item else item.get("value")
+                if key is not None:
+                    flat[str(key)] = value
+    return flat
+
+
+def _map_status(code: Any) -> str:
+    """Map a Cove status code (int) to a Backupchecks status string."""
+    if code is None:
+        return "Warning"
+    try:
+        return STATUS_MAP.get(int(code), "Warning")
+    except (ValueError, TypeError):
+        return "Warning"
+
+
+def run_cove_import(settings) -> tuple[int, int, int, int]:
+    """Fetch Cove account statistics and create/update JobRun records.
+
+    Args:
+        settings: SystemSettings ORM object with cove_* fields.
+
+    Returns:
+        Tuple of (total_accounts, created_runs, skipped_runs, error_count).
+
+    Raises:
+        CoveImportError if the API login fails.
+    """
+    from .models import Job, JobRun  # lazy import to avoid circular refs
+
+    url = (getattr(settings, "cove_api_url", None) or "").strip() or COVE_DEFAULT_URL
+    username = (getattr(settings, "cove_api_username", None) or "").strip()
+    password = (getattr(settings, "cove_api_password", None) or "").strip()
+
+    if not username or not password:
+        raise CoveImportError("Cove API username or password not configured")
+
+    visa, partner_id = _cove_login(url, username, password)
+
+    # Save partner_id back to settings
+    if partner_id and partner_id != getattr(settings, "cove_partner_id", None):
+        settings.cove_partner_id = partner_id
+        try:
+            db.session.commit()
+        except Exception:
+            db.session.rollback()
+
+    total = 0
+    created = 0
+    skipped = 0
+    errors = 0
+
+    page_size = 250
+    start = 0
+
+    while True:
+        try:
+            accounts = _cove_enumerate(url, visa, partner_id, start, page_size)
+        except CoveImportError:
+            raise
+        except Exception as exc:
+            raise CoveImportError(f"Unexpected error fetching accounts at offset {start}: {exc}") from exc
+
+        if not accounts:
+            break
+
+        for account in accounts:
+            total += 1
+            try:
+                created_this = _process_account(account, settings)
+                if created_this:
+                    created += 1
+                else:
+                    skipped += 1
+            except Exception as exc:
+                errors += 1
+                logger.warning("Cove import: error processing account: %s", exc)
+                try:
+                    db.session.rollback()
+                except Exception:
+                    pass
+
+        if len(accounts) < page_size:
+            break
+        start += page_size
+
+    # Update last import timestamp
+    settings.cove_last_import_at = datetime.utcnow()
+    try:
+        db.session.commit()
+    except Exception:
+        db.session.rollback()
+
+    return total, created, skipped, errors
+
+
+def _process_account(account: dict, settings) -> bool:
+    """Process a single Cove account and create a JobRun if needed.
+
+    Returns True if a new JobRun was created, False if skipped (duplicate or no job match).
+    """
+    from .models import Job, JobRun
+    from sqlalchemy import text
+
+    flat = _flatten_settings(account)
+
+    # AccountId – try both the top-level field and Settings
+    account_id = account.get("AccountId") or account.get("AccountID") or flat.get("I18")
+    if not account_id:
+        return False
+    try:
+        account_id = int(account_id)
+    except (ValueError, TypeError):
+        return False
+
+    # Last session end timestamp (D09F15) – used as run_at
+    run_ts_raw = flat.get("D09F15")
+    if not run_ts_raw:
+        # No last run timestamp – skip this account
+        return False
+
+    try:
+        run_ts = int(run_ts_raw)
+    except (ValueError, TypeError):
+        return False
+
+    if run_ts <= 0:
+        return False
+
+    # Deduplication key
+    external_id = f"cove-{account_id}-{run_ts}"
+
+    # Check for duplicate
+    existing = JobRun.query.filter_by(external_id=external_id).first()
+    if existing:
+        return False
+
+    # Find corresponding Job
+    job = Job.query.filter_by(cove_account_id=account_id).first()
+    if not job:
+        return False
+
+    # Convert Unix timestamp to datetime (UTC)
+    run_at = datetime.fromtimestamp(run_ts, tz=timezone.utc).replace(tzinfo=None)
+
+    # Map overall status
+    status_code = flat.get("D09F00")
+    status = _map_status(status_code)
+
+    # Create JobRun
+    run = JobRun(
+        job_id=job.id,
+        mail_message_id=None,
+        run_at=run_at,
+        status=status,
+        remark=None,
+        missed=False,
+        override_applied=False,
+        source_type="cove_api",
+        external_id=external_id,
+    )
+    db.session.add(run)
+    db.session.flush()  # get run.id
+
+    # Persist per-datasource objects
+    customer_id = job.customer_id
+    if customer_id:
+        _persist_datasource_objects(flat, customer_id, job.id, run.id, run_at)
+
+    db.session.commit()
+    return True
+
+
+def _persist_datasource_objects(
+    flat: dict,
+    customer_id: int,
+    job_id: int,
+    run_id: int,
+    observed_at: datetime,
+) -> None:
+    """Create run_object_links for each active datasource found in the account stats."""
+    engine = db.get_engine()
+
+    with engine.begin() as conn:
+        for ds_prefix, ds_label in DATASOURCE_LABELS.items():
+            status_key = f"{ds_prefix}F00"
+            status_code = flat.get(status_key)
+            if status_code is None:
+                continue
+
+            status = _map_status(status_code)
+
+            # Upsert customer_objects
+            customer_object_id = conn.execute(
+                text(
+                    """
+                    INSERT INTO customer_objects (customer_id, object_name, object_type, first_seen_at, last_seen_at)
+                    VALUES (:customer_id, :object_name, :object_type, NOW(), NOW())
+                    ON CONFLICT (customer_id, object_name)
+                    DO UPDATE SET
+                        last_seen_at = NOW(),
+                        object_type = COALESCE(EXCLUDED.object_type, customer_objects.object_type)
+                    RETURNING id
+                    """
+                ),
+                {
+                    "customer_id": customer_id,
+                    "object_name": ds_label,
+                    "object_type": "cove_datasource",
+                },
+            ).scalar()
+
+            # Upsert job_object_links
+            conn.execute(
+                text(
+                    """
+                    INSERT INTO job_object_links (job_id, customer_object_id, first_seen_at, last_seen_at)
+                    VALUES (:job_id, :customer_object_id, NOW(), NOW())
+                    ON CONFLICT (job_id, customer_object_id)
+                    DO UPDATE SET last_seen_at = NOW()
+                    """
+                ),
+                {"job_id": job_id, "customer_object_id": customer_object_id},
+            )
+
+            # Upsert run_object_links
+            conn.execute(
+                text(
+                    """
+                    INSERT INTO run_object_links (run_id, customer_object_id, status, error_message, observed_at)
+                    VALUES (:run_id, :customer_object_id, :status, NULL, :observed_at)
+                    ON CONFLICT (run_id, customer_object_id)
+                    DO UPDATE SET
+                        status = EXCLUDED.status,
+                        observed_at = EXCLUDED.observed_at
+                    """
+                ),
+                {
+                    "run_id": run_id,
+                    "customer_object_id": customer_object_id,
+                    "status": status,
+                    "observed_at": observed_at,
+                },
+            )
diff --git a/containers/backupchecks/src/backend/app/cove_importer_service.py b/containers/backupchecks/src/backend/app/cove_importer_service.py
new file mode 100644
index 0000000..0e46a43
--- /dev/null
+++ b/containers/backupchecks/src/backend/app/cove_importer_service.py
@@ -0,0 +1,101 @@
+"""Cove Data Protection importer background service.
+
+Runs a background thread that periodically fetches backup job run data
+from the Cove API and creates JobRun records in the local database.
+"""
+from __future__ import annotations
+
+import threading
+import time
+from datetime import datetime
+
+from .admin_logging import log_admin_event
+from .cove_importer import CoveImportError, run_cove_import
+from .models import SystemSettings
+
+
+_COVE_IMPORTER_THREAD_NAME = "cove_importer"
+
+
+def start_cove_importer(app) -> None:
+    """Start the Cove importer background thread.
+
+    The thread checks settings on every loop and only runs imports when
+    enabled and the configured interval has elapsed.
+    """
+
+    # Avoid starting multiple threads if create_app() is called more than once.
+    if any(t.name == _COVE_IMPORTER_THREAD_NAME for t in threading.enumerate()):
+        return
+
+    def _worker() -> None:
+        last_run_at: datetime | None = None
+
+        while True:
+            try:
+                with app.app_context():
+                    settings = SystemSettings.query.first()
+                    if settings is None:
+                        time.sleep(10)
+                        continue
+
+                    enabled = bool(getattr(settings, "cove_import_enabled", False))
+                    try:
+                        interval_minutes = int(getattr(settings, "cove_import_interval_minutes", 30) or 30)
+                    except (TypeError, ValueError):
+                        interval_minutes = 30
+                    if interval_minutes < 1:
+                        interval_minutes = 1
+
+                    now = datetime.utcnow()
+                    due = False
+                    if enabled:
+                        if last_run_at is None:
+                            due = True
+                        else:
+                            due = (now - last_run_at).total_seconds() >= (interval_minutes * 60)
+
+                    if not due:
+                        time.sleep(5)
+                        continue
+
+                    try:
+                        total, created, skipped, errors = run_cove_import(settings)
+                    except CoveImportError as exc:
+                        log_admin_event(
+                            "cove_import_error",
+                            f"Cove import failed: {exc}",
+                        )
+                        last_run_at = now
+                        time.sleep(5)
+                        continue
+                    except Exception as exc:
+                        log_admin_event(
+                            "cove_import_error",
+                            f"Unexpected error during Cove import: {exc}",
+                        )
+                        last_run_at = now
+                        time.sleep(5)
+                        continue
+
+                    log_admin_event(
+                        "cove_import",
+                        f"Cove import finished. accounts={total}, created={created}, skipped={skipped}, errors={errors}",
+                    )
+                    last_run_at = now
+
+            except Exception:
+                # Never let the thread die.
+                try:
+                    with app.app_context():
+                        log_admin_event(
+                            "cove_import_error",
+                            "Cove importer thread recovered from an unexpected exception.",
+                        )
+                except Exception:
+                    pass
+
+            time.sleep(5)
+
+    t = threading.Thread(target=_worker, name=_COVE_IMPORTER_THREAD_NAME, daemon=True)
+    t.start()
diff --git a/containers/backupchecks/src/backend/app/main/routes_jobs.py b/containers/backupchecks/src/backend/app/main/routes_jobs.py
index f9c9d99..16257cd 100644
--- a/containers/backupchecks/src/backend/app/main/routes_jobs.py
+++ b/containers/backupchecks/src/backend/app/main/routes_jobs.py
@@ -187,6 +187,35 @@ def unarchive_job(job_id: int):
     return redirect(url_for("main.archived_jobs"))
 
 
+@main_bp.route("/jobs/<int:job_id>/set-cove-account", methods=["POST"])
+@login_required
+@roles_required("admin", "operator")
+def job_set_cove_account(job_id: int):
+    """Save or clear the Cove Account ID for this job."""
+    job = Job.query.get_or_404(job_id)
+    account_id_raw = (request.form.get("cove_account_id") or "").strip()
+    if account_id_raw:
+        try:
+            job.cove_account_id = int(account_id_raw)
+        except (ValueError, TypeError):
+            flash("Invalid Cove Account ID – must be a number.", "warning")
+            return redirect(url_for("main.job_detail", job_id=job_id))
+    else:
+        job.cove_account_id = None
+
+    db.session.commit()
+    try:
+        log_admin_event(
+            "job_cove_account_set",
+            f"Set Cove Account ID for job {job.id} to {job.cove_account_id!r}",
+            details=f"job_name={job.job_name}",
+        )
+    except Exception:
+        pass
+    flash("Cove Account ID saved.", "success")
+    return redirect(url_for("main.job_detail", job_id=job_id))
+
+
 @main_bp.route("/jobs/<int:job_id>")
 @login_required
 @roles_required("admin", "operator", "viewer")
@@ -491,6 +520,11 @@ def job_detail(job_id: int):
     if job.customer_id:
         customer = Customer.query.get(job.customer_id)
 
+    # Load system settings for Cove integration display
+    from ..models import SystemSettings as _SystemSettings
+    _settings = _SystemSettings.query.first()
+    cove_enabled = bool(getattr(_settings, "cove_enabled", False)) if _settings else False
+
     return render_template(
         "main/job_detail.html",
         job=job,
@@ -507,6 +541,7 @@ def job_detail(job_id: int):
         has_prev=has_prev,
         has_next=has_next,
         can_manage_jobs=can_manage_jobs,
+        cove_enabled=cove_enabled,
     )
 
 
diff --git a/containers/backupchecks/src/backend/app/main/routes_settings.py b/containers/backupchecks/src/backend/app/main/routes_settings.py
index 6d5173a..ba28ee7 100644
--- a/containers/backupchecks/src/backend/app/main/routes_settings.py
+++ b/containers/backupchecks/src/backend/app/main/routes_settings.py
@@ -786,6 +786,7 @@ def settings():
 
     if request.method == "POST":
         autotask_form_touched = any(str(k).startswith("autotask_") for k in (request.form or {}).keys())
+        cove_form_touched = any(str(k).startswith("cove_") for k in (request.form or {}).keys())
         import_form_touched = any(str(k).startswith("auto_import_") or str(k).startswith("manual_import_") or str(k).startswith("ingest_eml_") for k in (request.form or {}).keys())
         general_form_touched = "ui_timezone" in request.form
         mail_form_touched = any(k in request.form for k in ["graph_tenant_id", "graph_client_id", "graph_mailbox", "incoming_folder", "processed_folder"])
@@ -908,6 +909,31 @@ def settings():
             except (ValueError, TypeError):
                 pass
 
+        # Cove Data Protection integration
+        if cove_form_touched:
+            settings.cove_enabled = bool(request.form.get("cove_enabled"))
+            settings.cove_import_enabled = bool(request.form.get("cove_import_enabled"))
+
+        if "cove_api_url" in request.form:
+            settings.cove_api_url = (request.form.get("cove_api_url") or "").strip() or None
+
+        if "cove_api_username" in request.form:
+            settings.cove_api_username = (request.form.get("cove_api_username") or "").strip() or None
+
+        if "cove_api_password" in request.form:
+            pw = (request.form.get("cove_api_password") or "").strip()
+            if pw:
+                settings.cove_api_password = pw
+
+        if "cove_import_interval_minutes" in request.form:
+            try:
+                interval = int(request.form.get("cove_import_interval_minutes") or 30)
+                if interval < 1:
+                    interval = 1
+                settings.cove_import_interval_minutes = interval
+            except (ValueError, TypeError):
+                pass
+
         # Daily Jobs
         if "daily_jobs_start_date" in request.form:
             daily_jobs_start_date_str = (request.form.get("daily_jobs_start_date") or "").strip()
@@ -1119,6 +1145,7 @@ def settings():
 
     has_client_secret = bool(settings.graph_client_secret)
     has_autotask_password = bool(getattr(settings, "autotask_api_password", None))
+    has_cove_password = bool(getattr(settings, "cove_api_password", None))
 
     # Common UI timezones (IANA names)
     tz_options = [
@@ -1244,6 +1271,7 @@ def settings():
         free_disk_warning=free_disk_warning,
         has_client_secret=has_client_secret,
         has_autotask_password=has_autotask_password,
+        has_cove_password=has_cove_password,
         tz_options=tz_options,
         users=users,
         admin_users_count=admin_users_count,
@@ -1258,6 +1286,44 @@ def settings():
     )
 
 
+@main_bp.route("/settings/cove/test-connection", methods=["POST"])
+@login_required
+@roles_required("admin")
+def settings_cove_test_connection():
+    """Test the Cove Data Protection API connection and return JSON result."""
+    from flask import jsonify
+    from ..cove_importer import CoveImportError, _cove_login, COVE_DEFAULT_URL
+
+    settings = _get_or_create_settings()
+
+    username = (getattr(settings, "cove_api_username", None) or "").strip()
+    password = (getattr(settings, "cove_api_password", None) or "").strip()
+    url = (getattr(settings, "cove_api_url", None) or "").strip() or COVE_DEFAULT_URL
+
+    if not username or not password:
+        return jsonify({"ok": False, "message": "Cove API username and password must be saved first."})
+
+    try:
+        visa, partner_id = _cove_login(url, username, password)
+        # Store the partner_id
+        settings.cove_partner_id = partner_id
+        db.session.commit()
+        _log_admin_event(
+            "cove_test_connection",
+            f"Cove connection test succeeded. Partner ID: {partner_id}",
+        )
+        return jsonify({
+            "ok": True,
+            "partner_id": partner_id,
+            "message": f"Connected – Partner ID: {partner_id}",
+        })
+    except CoveImportError as exc:
+        db.session.rollback()
+        return jsonify({"ok": False, "message": str(exc)})
+    except Exception as exc:
+        db.session.rollback()
+        return jsonify({"ok": False, "message": f"Unexpected error: {exc}"})
+
 
 @main_bp.route("/settings/news/create", methods=["POST"])
 @login_required
diff --git a/containers/backupchecks/src/backend/app/migrations.py b/containers/backupchecks/src/backend/app/migrations.py
index fb8f4fa..c2ea17c 100644
--- a/containers/backupchecks/src/backend/app/migrations.py
+++ b/containers/backupchecks/src/backend/app/migrations.py
@@ -1070,6 +1070,70 @@ def migrate_rename_admin_logs_to_audit_logs() -> None:
             print("[migrations] audit_logs table will be created by db.create_all()")
 
 
+def migrate_cove_integration() -> None:
+    """Add Cove Data Protection integration columns if missing.
+
+    Adds to system_settings:
+    - cove_enabled (BOOLEAN NOT NULL DEFAULT FALSE)
+    - cove_api_url (VARCHAR(255) NULL)
+    - cove_api_username (VARCHAR(255) NULL)
+    - cove_api_password (VARCHAR(255) NULL)
+    - cove_import_enabled (BOOLEAN NOT NULL DEFAULT FALSE)
+    - cove_import_interval_minutes (INTEGER NOT NULL DEFAULT 30)
+    - cove_partner_id (INTEGER NULL)
+    - cove_last_import_at (TIMESTAMP NULL)
+
+    Adds to jobs:
+    - cove_account_id (INTEGER NULL)
+
+    Adds to job_runs:
+    - source_type (VARCHAR(20) NULL)
+    - external_id (VARCHAR(100) NULL)
+    """
+    try:
+        engine = db.get_engine()
+    except Exception as exc:
+        print(f"[migrations] Could not get engine for Cove integration migration: {exc}")
+        return
+
+    try:
+        with engine.begin() as conn:
+            # system_settings columns
+            ss_columns = [
+                ("cove_enabled", "BOOLEAN NOT NULL DEFAULT FALSE"),
+                ("cove_api_url", "VARCHAR(255) NULL"),
+                ("cove_api_username", "VARCHAR(255) NULL"),
+                ("cove_api_password", "VARCHAR(255) NULL"),
+                ("cove_import_enabled", "BOOLEAN NOT NULL DEFAULT FALSE"),
+                ("cove_import_interval_minutes", "INTEGER NOT NULL DEFAULT 30"),
+                ("cove_partner_id", "INTEGER NULL"),
+                ("cove_last_import_at", "TIMESTAMP NULL"),
+            ]
+            for column, ddl in ss_columns:
+                if _column_exists_on_conn(conn, "system_settings", column):
+                    continue
+                conn.execute(text(f'ALTER TABLE "system_settings" ADD COLUMN {column} {ddl}'))
+
+            # jobs column
+            if not _column_exists_on_conn(conn, "jobs", "cove_account_id"):
+                conn.execute(text('ALTER TABLE "jobs" ADD COLUMN cove_account_id INTEGER NULL'))
+
+            # job_runs columns
+            if not _column_exists_on_conn(conn, "job_runs", "source_type"):
+                conn.execute(text('ALTER TABLE "job_runs" ADD COLUMN source_type VARCHAR(20) NULL'))
+            if not _column_exists_on_conn(conn, "job_runs", "external_id"):
+                conn.execute(text('ALTER TABLE "job_runs" ADD COLUMN external_id VARCHAR(100) NULL'))
+
+            # Index for deduplication lookups
+            conn.execute(text(
+                'CREATE INDEX IF NOT EXISTS idx_job_runs_external_id ON "job_runs" (external_id)'
+            ))
+
+        print("[migrations] migrate_cove_integration completed.")
+    except Exception as exc:
+        print(f"[migrations] Failed to migrate Cove integration columns: {exc}")
+
+
 def run_migrations() -> None:
     print("[migrations] Starting migrations...")
     migrate_add_username_to_users()
@@ -1112,6 +1176,7 @@ def run_migrations() -> None:
     migrate_performance_indexes()
     migrate_system_settings_require_daily_dashboard_visit()
     migrate_rename_admin_logs_to_audit_logs()
+    migrate_cove_integration()
     print("[migrations] All migrations completed.")
 
 
diff --git a/containers/backupchecks/src/backend/app/models.py b/containers/backupchecks/src/backend/app/models.py
index c4a4703..27e9bd1 100644
--- a/containers/backupchecks/src/backend/app/models.py
+++ b/containers/backupchecks/src/backend/app/models.py
@@ -117,6 +117,16 @@ class SystemSettings(db.Model):
     # this is not a production environment.
     is_sandbox_environment = db.Column(db.Boolean, nullable=False, default=False)
 
+    # Cove Data Protection integration settings
+    cove_enabled = db.Column(db.Boolean, nullable=False, default=False)
+    cove_api_url = db.Column(db.String(255), nullable=True)        # default: https://api.backup.management/jsonapi
+    cove_api_username = db.Column(db.String(255), nullable=True)
+    cove_api_password = db.Column(db.String(255), nullable=True)
+    cove_import_enabled = db.Column(db.Boolean, nullable=False, default=False)
+    cove_import_interval_minutes = db.Column(db.Integer, nullable=False, default=30)
+    cove_partner_id = db.Column(db.Integer, nullable=True)         # stored after successful login
+    cove_last_import_at = db.Column(db.DateTime, nullable=True)
+
     # Autotask integration settings
     autotask_enabled = db.Column(db.Boolean, nullable=False, default=False)
     autotask_environment = db.Column(db.String(32), nullable=True)  # sandbox | production
@@ -242,6 +252,9 @@ class Job(db.Model):
     auto_approve = db.Column(db.Boolean, nullable=False, default=True)
     active = db.Column(db.Boolean, nullable=False, default=True)
 
+    # Cove Data Protection integration
+    cove_account_id = db.Column(db.Integer, nullable=True)  # Cove AccountId mapping
+
     # Archived jobs are excluded from Daily Jobs and Run Checks.
     # JobRuns remain in the database and are still included in reporting.
     archived = db.Column(db.Boolean, nullable=False, default=False)
@@ -290,6 +303,10 @@ class JobRun(db.Model):
     reviewed_at = db.Column(db.DateTime, nullable=True)
     reviewed_by_user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True)
 
+    # Import source tracking
+    source_type = db.Column(db.String(20), nullable=True)   # NULL = email (backwards compat), "cove_api"
+    external_id = db.Column(db.String(100), nullable=True)  # e.g. "cove-{account_id}-{run_ts}" for deduplication
+
     # Autotask integration (Phase 4: ticket creation from Run Checks)
     autotask_ticket_id = db.Column(db.Integer, nullable=True)
     autotask_ticket_number = db.Column(db.String(64), nullable=True)
diff --git a/containers/backupchecks/src/templates/main/job_detail.html b/containers/backupchecks/src/templates/main/job_detail.html
index 86bbc6c..098cfa7 100644
--- a/containers/backupchecks/src/templates/main/job_detail.html
+++ b/containers/backupchecks/src/templates/main/job_detail.html
@@ -59,6 +59,34 @@
 </div>
 {% endif %}
 
+{% if cove_enabled and can_manage_jobs %}
+<div class="card mb-3">
+  <div class="card-header">Cove Integration</div>
+  <div class="card-body">
+    <form method="post" action="{{ url_for('main.job_set_cove_account', job_id=job.id) }}" class="row g-2 align-items-end mb-0">
+      <div class="col-auto">
+        <label for="cove_account_id" class="form-label mb-1">Cove Account ID</label>
+        <input type="number" class="form-control form-control-sm" id="cove_account_id" name="cove_account_id"
+          value="{{ job.cove_account_id or '' }}" placeholder="e.g. 4504627" style="width: 180px;" />
+      </div>
+      <div class="col-auto">
+        <button type="submit" class="btn btn-sm btn-primary">Save</button>
+        {% if job.cove_account_id %}
+        <button type="submit" name="cove_account_id" value="" class="btn btn-sm btn-outline-secondary ms-1">Clear</button>
+        {% endif %}
+      </div>
+      <div class="col-auto text-muted small">
+        {% if job.cove_account_id %}
+        Linked to Cove account <strong>{{ job.cove_account_id }}</strong>
+        {% else %}
+        Not linked to a Cove account – runs will not be imported automatically.
+        {% endif %}
+      </div>
+    </form>
+  </div>
+</div>
+{% endif %}
+
 <h3 class="mt-4 mb-3">Job history</h3>
 
 <div class="table-responsive">
diff --git a/containers/backupchecks/src/templates/main/settings.html b/containers/backupchecks/src/templates/main/settings.html
index f9920a3..b9cdec2 100644
--- a/containers/backupchecks/src/templates/main/settings.html
+++ b/containers/backupchecks/src/templates/main/settings.html
@@ -504,6 +504,106 @@
 </div>
 {% endif %}
 
+{% if section == 'integrations' %}
+<form method="post" class="mb-4" id="cove-settings-form">
+  <div class="card mb-3">
+    <div class="card-header">Cove Data Protection (N-able)</div>
+    <div class="card-body">
+      <div class="form-check form-switch mb-3">
+        <input class="form-check-input" type="checkbox" id="cove_enabled" name="cove_enabled" {% if settings.cove_enabled %}checked{% endif %} />
+        <label class="form-check-label" for="cove_enabled">Enable Cove integration</label>
+      </div>
+
+      <div class="row g-3">
+        <div class="col-md-12">
+          <label for="cove_api_url" class="form-label">API URL</label>
+          <input type="url" class="form-control" id="cove_api_url" name="cove_api_url"
+            value="{{ settings.cove_api_url or '' }}"
+            placeholder="https://api.backup.management/jsonapi" />
+          <div class="form-text">Leave empty to use the default Cove API endpoint.</div>
+        </div>
+
+        <div class="col-md-6">
+          <label for="cove_api_username" class="form-label">API Username <span class="text-danger">*</span></label>
+          <input type="text" class="form-control" id="cove_api_username" name="cove_api_username"
+            value="{{ settings.cove_api_username or '' }}" />
+        </div>
+
+        <div class="col-md-6">
+          <label for="cove_api_password" class="form-label">API Password {% if not has_cove_password %}<span class="text-danger">*</span>{% endif %}</label>
+          <input type="password" class="form-control" id="cove_api_password" name="cove_api_password"
+            placeholder="{% if has_cove_password %}******** (stored){% else %}enter password{% endif %}" />
+          <div class="form-text">Leave empty to keep the existing password.</div>
+        </div>
+
+        <div class="col-md-6">
+          <div class="form-check form-switch mt-2">
+            <input class="form-check-input" type="checkbox" id="cove_import_enabled" name="cove_import_enabled" {% if settings.cove_import_enabled %}checked{% endif %} />
+            <label class="form-check-label" for="cove_import_enabled">Enable automatic import</label>
+          </div>
+        </div>
+
+        <div class="col-md-6">
+          <label for="cove_import_interval_minutes" class="form-label">Import interval (minutes)</label>
+          <input type="number" class="form-control" id="cove_import_interval_minutes" name="cove_import_interval_minutes"
+            value="{{ settings.cove_import_interval_minutes or 30 }}" min="1" max="1440" />
+          <div class="form-text">How often (in minutes) to fetch new data from the Cove API.</div>
+        </div>
+      </div>
+
+      <div class="d-flex justify-content-between align-items-center mt-3">
+        <div id="cove-test-result" class="small"></div>
+        <div class="d-flex gap-2">
+          <button type="button" class="btn btn-outline-secondary" id="cove-test-btn">Test connection</button>
+          <button type="submit" class="btn btn-primary">Save Cove Settings</button>
+        </div>
+      </div>
+
+      {% if settings.cove_partner_id %}
+      <div class="mt-2 text-muted small">
+        Connected – Partner ID: <strong>{{ settings.cove_partner_id }}</strong>
+        {% if settings.cove_last_import_at %}
+        &nbsp;·&nbsp; Last import: {{ settings.cove_last_import_at|local_datetime }}
+        {% endif %}
+      </div>
+      {% endif %}
+    </div>
+  </div>
+</form>
+
+<script>
+(function () {
+  var btn = document.getElementById('cove-test-btn');
+  var resultDiv = document.getElementById('cove-test-result');
+  if (!btn) return;
+  btn.addEventListener('click', function () {
+    btn.disabled = true;
+    resultDiv.textContent = 'Testing…';
+    resultDiv.className = 'small text-muted';
+    fetch('{{ url_for("main.settings_cove_test_connection") }}', {
+      method: 'POST',
+      headers: { 'X-CSRFToken': document.querySelector('meta[name="csrf-token"]') ? document.querySelector('meta[name="csrf-token"]').content : '' },
+      credentials: 'same-origin',
+    })
+    .then(function (r) { return r.json(); })
+    .then(function (data) {
+      if (data.ok) {
+        resultDiv.textContent = data.message;
+        resultDiv.className = 'small text-success';
+      } else {
+        resultDiv.textContent = data.message;
+        resultDiv.className = 'small text-danger';
+      }
+    })
+    .catch(function (err) {
+      resultDiv.textContent = 'Request failed: ' + err;
+      resultDiv.className = 'small text-danger';
+    })
+    .finally(function () { btn.disabled = false; });
+  });
+})();
+</script>
+{% endif %}
 
 {% if section == 'maintenance' %}
 <div class="row g-3 mb-4">
diff --git a/docs/changelog-claude.md b/docs/changelog-claude.md
index 9ca468e..4ff9054 100644
--- a/docs/changelog-claude.md
+++ b/docs/changelog-claude.md
@@ -5,6 +5,17 @@ This file documents all changes made to this project via Claude Code.
 ## [2026-02-23]
 
 ### Added
+- Cove Data Protection full integration into Backupchecks:
+  - `app/cove_importer.py` – Cove API client: login, paginated EnumerateAccountStatistics, status mapping, deduplication, per-datasource object persistence
+  - `app/cove_importer_service.py` – background thread that polls Cove API on configurable interval
+  - `SystemSettings` model: 8 new Cove fields (`cove_enabled`, `cove_api_url`, `cove_api_username`, `cove_api_password`, `cove_import_enabled`, `cove_import_interval_minutes`, `cove_partner_id`, `cove_last_import_at`)
+  - `Job` model: `cove_account_id` column to link a job to a Cove account
+  - `JobRun` model: `source_type` (NULL = email, "cove_api") and `external_id` (deduplication key) columns
+  - DB migration `migrate_cove_integration()` for all new columns + deduplication index
+  - Settings > Integrations tab: new Cove section with enable toggle, API URL/username/password, import interval, and Test Connection button (AJAX → JSON response with partner ID)
+  - Job Detail page: Cove Integration card showing Account ID input (only when `cove_enabled`)
+  - Route `POST /settings/cove/test-connection` – verifies Cove credentials and stores partner ID
+  - Route `POST /jobs/<id>/set-cove-account` – saves or clears Cove Account ID on a job
 - `cove_api_test.py` – standalone Python test script to verify Cove Data Protection API column codes
   - Tests D9Fxx (Total), D10Fxx (VssMsSql), D11Fxx (VssSharePoint), and D1Fxx (Files&Folders)
   - Displays backup status (F00), timestamps (F09/F15/F18), error counts (F06) per account