diff --git a/src/borgitory/alembic/versions/b2c3d4e5f6a7_add_source_paths_json_column.py b/src/borgitory/alembic/versions/b2c3d4e5f6a7_add_source_paths_json_column.py
new file mode 100644
index 00000000..6d0817e9
--- /dev/null
+++ b/src/borgitory/alembic/versions/b2c3d4e5f6a7_add_source_paths_json_column.py
@@ -0,0 +1,65 @@
+"""Add source_paths_json column; backfill from source_path (string). source_path kept as deprecated.
+
+Revision ID: b2c3d4e5f6a7
+Revises: 2628b151e709
+Create Date: 2026-03-07 00:00:00.000000
+
+"""
+
+import json
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "b2c3d4e5f6a7"
+down_revision: Union[str, Sequence[str], None] = "2628b151e709"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def _string_to_paths_list(value: str | None) -> list[str]:
+    if value is None or not value or not str(value).strip():
+        return []
+    s = str(value).strip()
+    if s.startswith("["):
+        try:
+            parsed = json.loads(s)
+            if isinstance(parsed, list):
+                return [p for p in parsed if isinstance(p, str) and p.strip()]
+            return []
+        except json.JSONDecodeError, ValueError:
+            return []
+    return [s]
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("schedules", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("source_paths_json", sa.JSON(), nullable=True))
+
+    connection = op.get_bind()
+    rows = connection.execute(
+        sa.text("SELECT id, source_path FROM schedules")
+    ).fetchall()
+
+    for row in rows:
+        row_id, value = row[0], row[1]
+        paths = _string_to_paths_list(value)
+        conn_val = json.dumps(paths) if paths else "[]"
+        connection.execute(
+            sa.text("UPDATE schedules SET source_paths_json = :val WHERE id = :id"),
+            {"val": conn_val, "id": row_id},
+        )
+
+    with op.batch_alter_table("schedules", schema=None) as batch_op:
+        batch_op.alter_column(
+            "source_paths_json",
+            existing_type=sa.JSON(),
+            nullable=False,
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("schedules", schema=None) as batch_op:
+        batch_op.drop_column("source_paths_json")
diff --git a/src/borgitory/api/repositories.py b/src/borgitory/api/repositories.py
index 1303c600..aef7c44e 100644
--- a/src/borgitory/api/repositories.py
+++ b/src/borgitory/api/repositories.py
@@ -133,7 +133,7 @@ async def list_directories_autocomplete(
     for param_name in form_data.keys():
         if param_name in [
             "path",
-            "source_path",
+            "source_paths",
             "create-path",
             "import-path",
             "backup-source-path",
diff --git a/src/borgitory/api/schedules.py b/src/borgitory/api/schedules.py
index b4e1469e..f4df0090 100644
--- a/src/borgitory/api/schedules.py
+++ b/src/borgitory/api/schedules.py
@@ -23,6 +23,10 @@
 from borgitory.services.scheduling.hook_service import HookService
 from borgitory.api.auth import get_current_user
 from borgitory.models.database import User
+from borgitory.utils.source_paths import (
+    parse_source_paths,
+    serialize_source_paths,
+)
 
 router = APIRouter()
 
@@ -93,7 +97,7 @@ async def create_schedule(
         name=schedule.name,
         repository_id=schedule.repository_id,
         cron_expression=schedule.cron_expression,
-        source_path=schedule.source_path or "",
+        source_paths=schedule.source_paths,
         cloud_sync_config_id=schedule.cloud_sync_config_id,
         prune_config_id=schedule.prune_config_id,
         notification_config_id=schedule.notification_config_id,
@@ -242,7 +246,16 @@ async def get_schedule_edit_form(
             raise HTTPException(status_code=404, detail="Schedule not found")
 
         form_data = await config_service.get_schedule_form_data(db)
-        context = {**form_data, "schedule": schedule, "is_edit_mode": True}
+        source_paths_list = list(schedule.source_paths) if schedule.source_paths else []
+        context = {
+            **form_data,
+            "schedule": schedule,
+            "is_edit_mode": True,
+            "source_paths": source_paths_list,
+            "source_paths_json": json.dumps(schedule.source_paths)
+            if schedule.source_paths
+            else "[]",
+        }
 
         return templates.TemplateResponse(
             request, "partials/schedules/edit_form.html", context
@@ -603,6 +616,132 @@ async def close_modal() -> HTMLResponse:
     return HTMLResponse(content='<div id="modal-container"></div>', status_code=200)
 
 
+# Source Paths API endpoints
+
+
+def _extract_source_paths_from_json(data: Dict[str, Any]) -> list[str]:
+    """Extract source_paths from JSON request data.
+
+    json-enc sends multiple inputs with the same name as an array,
+    or a single input as a bare string.
+    """
+    raw = data.get("source_paths", [])
+    if isinstance(raw, list):
+        return [str(p) for p in raw]
+    if isinstance(raw, str):
+        return [raw]
+    return [""]
+
+
+@router.post("/source-paths/source-paths-modal", response_class=HTMLResponse)
+async def get_source_paths_modal(
+    request: Request,
+    templates: TemplatesDep,
+) -> HTMLResponse:
+    """Open source paths configuration modal with current path data from parent."""
+    try:
+        json_data = await request.json()
+        source_paths_value = str(json_data.get("source_paths", "[]"))
+    except ValueError, TypeError, KeyError:
+        source_paths_value = "[]"
+
+    paths = parse_source_paths(source_paths_value)
+    if not paths:
+        paths = [""]
+
+    return templates.TemplateResponse(
+        request,
+        "partials/shared/source_paths_modal.html",
+        {"source_paths": paths},
+    )
+
+
+@router.post("/source-paths/save-source-paths", response_class=HTMLResponse)
+async def save_source_paths(
+    request: Request,
+    templates: TemplatesDep,
+) -> HTMLResponse:
+    """Save source paths and update parent form via OOB swap."""
+    json_data = await request.json()
+    raw_paths = _extract_source_paths_from_json(json_data)
+    filtered = [p.strip() for p in raw_paths if p.strip()]
+
+    non_absolute = [p for p in filtered if not p.startswith("/")]
+    if non_absolute:
+        return templates.TemplateResponse(
+            request,
+            "partials/shared/source_paths_validation_error.html",
+            {
+                "error_message": (
+                    f"All source paths must be absolute (start with /). "
+                    f"Invalid: {', '.join(non_absolute)}"
+                )
+            },
+            status_code=400,
+        )
+
+    source_paths_json = serialize_source_paths(filtered)
+    total_count = len(filtered)
+
+    return templates.TemplateResponse(
+        request,
+        "partials/shared/source_paths_save_response.html",
+        {
+            "source_paths_json": source_paths_json,
+            "total_count": total_count,
+        },
+    )
+
+
+@router.get("/source-paths/close-modal", response_class=HTMLResponse)
+async def close_source_paths_modal() -> HTMLResponse:
+    """Close source paths modal without saving."""
+    return HTMLResponse(content='<div id="modal-container"></div>', status_code=200)
+
+
+@router.post("/source-paths/add-field", response_class=HTMLResponse)
+async def add_source_path_field(
+    request: Request,
+    templates: TemplatesDep,
+) -> HTMLResponse:
+    """Add a new source path field row via HTMX."""
+    json_data = await request.json()
+    current_paths = _extract_source_paths_from_json(json_data)
+    current_paths.append("")
+
+    return templates.TemplateResponse(
+        request,
+        "partials/shared/source_paths_container.html",
+        {"source_paths": current_paths},
+    )
+
+
+@router.post("/source-paths/remove-field", response_class=HTMLResponse)
+async def remove_source_path_field(
+    request: Request,
+    templates: TemplatesDep,
+) -> HTMLResponse:
+    """Remove a source path field row via HTMX."""
+    json_data = await request.json()
+    current_paths = _extract_source_paths_from_json(json_data)
+
+    try:
+        remove_index = int(str(json_data.get("remove_index", 0)))
+        if 0 <= remove_index < len(current_paths):
+            current_paths.pop(remove_index)
+    except ValueError, TypeError:
+        pass
+
+    if not current_paths:
+        current_paths = [""]
+
+    return templates.TemplateResponse(
+        request,
+        "partials/shared/source_paths_container.html",
+        {"source_paths": current_paths},
+    )
+
+
 # Pattern API endpoints
 @router.post("/patterns/add-pattern-field", response_class=HTMLResponse)
 async def add_pattern_field(
diff --git a/src/borgitory/models/database.py b/src/borgitory/models/database.py
index d8ef6312..f65dbde0 100644
--- a/src/borgitory/models/database.py
+++ b/src/borgitory/models/database.py
@@ -28,6 +28,7 @@
     Text,
     ForeignKey,
     Uuid,
+    JSON,
 )
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlalchemy.orm import Mapped, declarative_base, mapped_column
@@ -232,7 +233,12 @@ class Schedule(Base):
     )
     name: Mapped[str] = mapped_column(String, nullable=False)
     cron_expression: Mapped[str] = mapped_column(String, nullable=False)
-    source_path: Mapped[str] = mapped_column(String, nullable=False, default="/data")
+    source_paths: Mapped[List[str]] = mapped_column(
+        "source_paths_json", JSON, nullable=False, default=lambda: []
+    )
+    source_paths_legacy: Mapped[str] = mapped_column(
+        "source_path", String, nullable=False, default="[]"
+    )  # deprecated: use source_paths (JSON) instead
     enabled: Mapped[bool] = mapped_column(Boolean, default=True)
     last_run: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
     next_run: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
diff --git a/src/borgitory/models/schemas.py b/src/borgitory/models/schemas.py
index 9e34b23a..86da5c23 100644
--- a/src/borgitory/models/schemas.py
+++ b/src/borgitory/models/schemas.py
@@ -8,6 +8,9 @@
 from borgitory.custom_types import ConfigDict
 from borgitory.services.hooks.hook_config import validate_hooks_json
 from borgitory.models.enums import EncryptionType
+from borgitory.utils.source_paths import (
+    parse_source_paths,
+)
 
 
 def validate_patterns_json(patterns_json: str) -> tuple[bool, Optional[str]]:
@@ -91,6 +94,31 @@ def validate_patterns_json(patterns_json: str) -> tuple[bool, Optional[str]]:
 ABSOLUTE_PATH_PATTERN = r"^/.*"
 
 
+def _coerce_source_paths_to_list(v: object) -> list[str]:
+    """Coerce source_paths input to a list of path strings.
+
+    Accepts a JSON array string, a plain list of strings, None, or empty string.
+    """
+    if v is None or (isinstance(v, str) and not v.strip()):
+        return []
+    if isinstance(v, list):
+        return [p.strip() for p in v if isinstance(p, str) and p.strip()]
+    if isinstance(v, str):
+        return parse_source_paths(v)
+    return []
+
+
+def _validate_source_paths_absolute_list(paths: list[str]) -> list[str]:
+    """Validate that every path is absolute."""
+    non_absolute = [p for p in paths if not p.startswith("/")]
+    if non_absolute:
+        raise ValueError(
+            f"All source paths must be absolute (start with /). "
+            f"Invalid: {', '.join(non_absolute)}"
+        )
+    return paths
+
+
 # Enums for type safety and validation
 class JobStatus(str, Enum):
     PENDING = "pending"
@@ -305,7 +333,10 @@ def validate_cron_expression(cls, v: str) -> str:
 
 class ScheduleCreate(ScheduleBase):
     repository_id: int
-    source_path: Optional[str] = "/data"
+    source_paths: list[str] = Field(
+        default_factory=list,
+        description="List of absolute source paths",
+    )
     cloud_sync_config_id: Optional[int] = None
     prune_config_id: Optional[int] = None
     check_config_id: Optional[int] = None
@@ -314,6 +345,16 @@ class ScheduleCreate(ScheduleBase):
     post_job_hooks: Optional[str] = None
     patterns: Optional[str] = None
 
+    @field_validator("source_paths", mode="before")
+    @classmethod
+    def coerce_source_paths(cls, v: object) -> list[str]:
+        return _coerce_source_paths_to_list(v)
+
+    @field_validator("source_paths", mode="after")
+    @classmethod
+    def validate_source_paths(cls, v: list[str]) -> list[str]:
+        return _validate_source_paths_absolute_list(v)
+
     @field_validator("cloud_sync_config_id", mode="before")
     @classmethod
     def validate_cloud_sync_config_id(cls, v: Union[str, int, None]) -> Optional[int]:
@@ -388,7 +429,7 @@ class ScheduleUpdate(BaseModel):
     name: Optional[str] = Field(None, min_length=1, max_length=128)
     cron_expression: Optional[str] = Field(None, min_length=5)
     repository_id: Optional[int] = None
-    source_path: Optional[str] = None
+    source_paths: Optional[list[str]] = None
     cloud_sync_config_id: Optional[int] = None
     prune_config_id: Optional[int] = None
     check_config_id: Optional[int] = None
@@ -398,6 +439,20 @@ class ScheduleUpdate(BaseModel):
     post_job_hooks: Optional[str] = None
     patterns: Optional[str] = None
 
+    @field_validator("source_paths", mode="before")
+    @classmethod
+    def coerce_source_paths(cls, v: object) -> Optional[list[str]]:
+        if v is None:
+            return None
+        return _coerce_source_paths_to_list(v)
+
+    @field_validator("source_paths", mode="after")
+    @classmethod
+    def validate_source_paths(cls, v: Optional[list[str]]) -> Optional[list[str]]:
+        if v is None:
+            return None
+        return _validate_source_paths_absolute_list(v)
+
     @field_validator("pre_job_hooks", mode="before")
     @classmethod
     def validate_pre_job_hooks_update(cls, v: Optional[str]) -> Optional[str]:
@@ -486,7 +541,7 @@ def validate_notification_config_id(cls, v: Union[str, int, None]) -> Optional[i
 class Schedule(ScheduleBase):
     id: int = Field(gt=0)
     repository_id: int = Field(gt=0)
-    source_path: str = Field(default="/", pattern=ABSOLUTE_PATH_PATTERN)
+    source_paths: list[str] = Field(default_factory=list)
     enabled: bool
     last_run: Optional[datetime] = None
     next_run: Optional[datetime] = None
@@ -604,10 +659,9 @@ class NotificationConfig(NotificationConfigBase):
 
 class BackupRequest(BaseModel):
     repository_id: int = Field(gt=0)
-    source_path: str = Field(
-        default="/",
-        pattern=ABSOLUTE_PATH_PATTERN,
-        description="Absolute path to source directory",
+    source_paths: list[str] = Field(
+        default_factory=list,
+        description="List of absolute source paths to backup.",
     )
     compression: CompressionType = CompressionType.ZSTD
     dry_run: bool = False
@@ -620,6 +674,16 @@ class BackupRequest(BaseModel):
     post_job_hooks: Optional[str] = None
     patterns: Optional[str] = None
 
+    @field_validator("source_paths", mode="before")
+    @classmethod
+    def coerce_source_paths(cls, v: object) -> list[str]:
+        return _coerce_source_paths_to_list(v)
+
+    @field_validator("source_paths", mode="after")
+    @classmethod
+    def validate_source_paths(cls, v: list[str]) -> list[str]:
+        return _validate_source_paths_absolute_list(v)
+
     @field_validator("dry_run", mode="before")
     @classmethod
     def validate_dry_run(cls, v: Union[str, bool, int]) -> bool:
diff --git a/src/borgitory/services/jobs/job_service.py b/src/borgitory/services/jobs/job_service.py
index be359f00..6bc22229 100644
--- a/src/borgitory/services/jobs/job_service.py
+++ b/src/borgitory/services/jobs/job_service.py
@@ -37,7 +37,7 @@
 class BackupParams:
     """Parameters for backup tasks."""
 
-    source_path: str
+    source_paths: list[str]
     compression: str
     dry_run: bool
     ignore_lock: bool
@@ -124,7 +124,7 @@ async def create_backup_job(
                 logger.warning(f"Failed to parse patterns: {str(e)}")
 
         backup_params: ConfigDict = {
-            "source_path": backup_request.source_path,
+            "source_paths": backup_request.source_paths,
             "compression": backup_request.compression,
             "dry_run": backup_request.dry_run,
             "ignore_lock": backup_request.ignore_lock,
diff --git a/src/borgitory/services/jobs/task_executors/backup_task_executor.py b/src/borgitory/services/jobs/task_executors/backup_task_executor.py
index 40dcd2b6..fad27919 100644
--- a/src/borgitory/services/jobs/task_executors/backup_task_executor.py
+++ b/src/borgitory/services/jobs/task_executors/backup_task_executor.py
@@ -82,14 +82,13 @@ def task_output_callback(line: str) -> None:
                     },
                 )
 
-            # Build backup command
-            source_path = params.get("source_path")
+            source_paths = params.get("source_paths")
             archive_name = params.get(
                 "archive_name", f"backup-{now_utc().strftime('%Y%m%d-%H%M%S')}"
             )
 
             logger.info(
-                f"Backup task parameters - source_path: {source_path}, archive_name: {archive_name}"
+                f"Backup task parameters - source_paths: {source_paths}, archive_name: {archive_name}"
             )
             logger.info(f"All task parameters: {params}")
 
@@ -111,8 +110,10 @@ def task_output_callback(line: str) -> None:
 
             additional_args.append(f"{repository_path}::{archive_name}")
 
-            if source_path:
-                additional_args.append(str(source_path))
+            paths = source_paths if isinstance(source_paths, list) else []
+            for path in paths:
+                if path and str(path).strip():
+                    additional_args.append(str(path).strip())
 
             logger.info(f"Final additional_args for Borg command: {additional_args}")
 
diff --git a/src/borgitory/services/scheduling/schedule_service.py b/src/borgitory/services/scheduling/schedule_service.py
index a79bbca1..b3224717 100644
--- a/src/borgitory/services/scheduling/schedule_service.py
+++ b/src/borgitory/services/scheduling/schedule_service.py
@@ -138,7 +138,7 @@ async def create_schedule(
         name: str,
         repository_id: int,
         cron_expression: str,
-        source_path: str,
+        source_paths: list[str],
         cloud_sync_config_id: Optional[int] = None,
         prune_config_id: Optional[int] = None,
         notification_config_id: Optional[int] = None,
@@ -175,7 +175,7 @@ async def create_schedule(
             db_schedule.name = name
             db_schedule.repository_id = repository_id
             db_schedule.cron_expression = cron_expression
-            db_schedule.source_path = source_path
+            db_schedule.source_paths = source_paths
             db_schedule.enabled = True
             db_schedule.cloud_sync_config_id = cloud_sync_config_id
             db_schedule.prune_config_id = prune_config_id
@@ -427,7 +427,7 @@ def safe_json_string(value: Any) -> Optional[str]:
                 "name": name,
                 "repository_id": repository_id,
                 "cron_expression": cron_expression,
-                "source_path": json_data.get("source_path", ""),
+                "source_paths": json_data.get("source_paths", []),
                 "cloud_sync_config_id": safe_int(json_data.get("cloud_sync_config_id")),
                 "prune_config_id": safe_int(json_data.get("prune_config_id")),
                 "notification_config_id": safe_int(
diff --git a/src/borgitory/services/scheduling/scheduler_service.py b/src/borgitory/services/scheduling/scheduler_service.py
index 31bdf89b..73d48c35 100644
--- a/src/borgitory/services/scheduling/scheduler_service.py
+++ b/src/borgitory/services/scheduling/scheduler_service.py
@@ -79,12 +79,12 @@ async def execute_scheduled_backup(schedule_id: int) -> None:
             logger.info("SCHEDULER: Creating scheduled backup via JobService")
             logger.info(f"  - repository: {repository.name}")
             logger.info(f"  - schedule: {schedule.name}")
-            logger.info(f"  - source_path: {schedule.source_path}")
+            logger.info(f"  - source_paths: {schedule.source_paths}")
             logger.info(f"  - cloud_sync_config_id: {schedule.cloud_sync_config_id}")
 
             backup_request = BackupRequest(
                 repository_id=repository.id,
-                source_path=schedule.source_path,
+                source_paths=schedule.source_paths,
                 compression=CompressionType.ZSTD,
                 dry_run=False,
                 prune_config_id=schedule.prune_config_id,
diff --git a/src/borgitory/services/task_definition_builder.py b/src/borgitory/services/task_definition_builder.py
index f18b4538..05d06bca 100644
--- a/src/borgitory/services/task_definition_builder.py
+++ b/src/borgitory/services/task_definition_builder.py
@@ -38,7 +38,7 @@ def __init__(self) -> None:
     def build_backup_task(
         self,
         repository_name: str,
-        source_path: str = "/data",
+        source_paths: list[str] | None = None,
         compression: str = "zstd",
         dry_run: bool = False,
         ignore_lock: bool = False,
@@ -49,7 +49,7 @@ def build_backup_task(
 
         Args:
             repository_name: Name of the repository for display
-            source_path: Path to backup from
+            source_paths: List of paths to backup
             compression: Compression algorithm to use
             dry_run: Whether this is a dry run
             ignore_lock: Whether to run 'borg break-lock' before backup
@@ -58,7 +58,7 @@ def build_backup_task(
             Task definition dictionary
         """
         parameters: ConfigDict = {
-            "source_path": source_path,
+            "source_paths": source_paths or [],
             "compression": compression,
             "dry_run": dry_run,
             "ignore_lock": ignore_lock,
@@ -419,7 +419,11 @@ async def build_task_list(
 
         if include_backup:
             if backup_params:
-                source_path = str(backup_params.get("source_path", "/data"))
+                source_paths_value = backup_params.get("source_paths") or []
+                if isinstance(source_paths_value, list):
+                    source_paths = source_paths_value
+                else:
+                    source_paths = []
                 compression = str(backup_params.get("compression", "zstd"))
                 dry_run = bool(backup_params.get("dry_run", False))
                 ignore_lock = bool(backup_params.get("ignore_lock", False))
@@ -431,8 +435,7 @@ async def build_task_list(
                 else:
                     patterns = []
             else:
-                # Use defaults when no backup_params provided
-                source_path = "/data"
+                source_paths = []
                 compression = "zstd"
                 dry_run = False
                 ignore_lock = False
@@ -441,7 +444,7 @@ async def build_task_list(
             tasks.append(
                 self.build_backup_task(
                     repository_name,
-                    source_path,
+                    source_paths,
                     compression,
                     dry_run,
                     ignore_lock,
diff --git a/src/borgitory/templates/partials/backups/manual_form.html b/src/borgitory/templates/partials/backups/manual_form.html
index ddbdf24c..60a4bc1a 100644
--- a/src/borgitory/templates/partials/backups/manual_form.html
+++ b/src/borgitory/templates/partials/backups/manual_form.html
@@ -19,13 +19,18 @@ <h3 class="font-medium text-gray-900 dark:text-gray-100 mb-3">Create Backup</h3>
             </div>
             <div>
                 <label class="block text-sm font-medium text-gray-900 dark:text-gray-100">
-                    Source Path
+                    Source Paths
                 </label>
-                {% set input_id = "backup-source-path" %}
-                {% set input_name = "source_path" %}
-                {% set placeholder = "/pictures/" %}
-                {% set required = false %}
-                {% include "partials/shared/path_autocomplete.html" %}
+                <button type="button"
+                        class="flex items-stretch w-full text-sm font-medium text-gray-700 dark:text-gray-300 bg-white dark:bg-gray-800 border border-gray-300 dark:border-gray-600 rounded-md hover:bg-gray-50 dark:hover:bg-gray-700 focus:ring-2 focus:ring-blue-500 transition-colors overflow-hidden mt-1"
+                        hx-post="/api/schedules/source-paths/source-paths-modal"
+                        hx-include="#source-paths-data"
+                        hx-target="#modal-container">
+                    <span class="flex-1 px-4 py-2 text-left" id="source-paths-button-text">Source Paths</span>
+                    <span class="flex items-center justify-center px-3 bg-gray-100 dark:bg-gray-700 border-l border-gray-300 dark:border-gray-600 text-xs font-medium min-w-[2rem]"
+                          id="source-paths-count-badge">0</span>
+                </button>
+                <input type="hidden" name="source_paths" id="source-paths-data" value="[]">
             </div>
             <div>
                 <label class="block text-sm font-medium text-gray-900 dark:text-gray-100">
@@ -121,3 +126,4 @@ <h3 class="font-medium text-gray-900 dark:text-gray-100 mb-3">Create Backup</h3>
     </form>
     <div id="backup-status" class="mt-4"></div>
 </div>
+<div id="modal-container"></div>
diff --git a/src/borgitory/templates/partials/schedules/create_form.html b/src/borgitory/templates/partials/schedules/create_form.html
index 92abd75c..2352da44 100644
--- a/src/borgitory/templates/partials/schedules/create_form.html
+++ b/src/borgitory/templates/partials/schedules/create_form.html
@@ -26,13 +26,18 @@ <h3 class="font-medium text-gray-900 dark:text-gray-100 mb-3">Create Schedule</h
             </div>
             <div>
                 <label class="block text-sm font-medium text-gray-900 dark:text-gray-100">
-                    Source Path
+                    Source Paths
                 </label>
-                {% set input_id = "schedule-source-path" %}
-                {% set input_name = "source_path" %}
-                {% set placeholder = "/path/to/data" %}
-                {% set required = false %}
-                {% include "partials/shared/path_autocomplete.html" %}
+                <button type="button"
+                        class="flex items-stretch w-full text-sm font-medium text-gray-700 dark:text-gray-300 bg-white dark:bg-gray-800 border border-gray-300 dark:border-gray-600 rounded-md hover:bg-gray-50 dark:hover:bg-gray-700 focus:ring-2 focus:ring-blue-500 transition-colors overflow-hidden mt-1"
+                        hx-post="/api/schedules/source-paths/source-paths-modal"
+                        hx-include="#source-paths-data"
+                        hx-target="#modal-container">
+                    <span class="flex-1 px-4 py-2 text-left" id="source-paths-button-text">Source Paths</span>
+                    <span class="flex items-center justify-center px-3 bg-gray-100 dark:bg-gray-700 border-l border-gray-300 dark:border-gray-600 text-xs font-medium min-w-[2rem]"
+                          id="source-paths-count-badge">0</span>
+                </button>
+                <input type="hidden" name="source_paths" id="source-paths-data" value="[]">
             </div>
             <div>
                 <label class="block text-sm font-medium text-gray-900 dark:text-gray-100">
diff --git a/src/borgitory/templates/partials/schedules/create_success.html b/src/borgitory/templates/partials/schedules/create_success.html
index 1ef52cea..1965db99 100644
--- a/src/borgitory/templates/partials/schedules/create_success.html
+++ b/src/borgitory/templates/partials/schedules/create_success.html
@@ -8,23 +8,8 @@
      hx-swap="none"
      hx-on::load="document.body.dispatchEvent(new CustomEvent('scheduleUpdate'))">
 </div>
-<script>
-// Reset form fields (keeping this JavaScript for complex form logic)
-(function() {
-    const form = document.querySelector('form[hx-post="/api/schedules/"]');
-    if (form) {
-        form.reset();
-        
-        // Clear custom form elements
-        const preset = document.getElementById('schedule-preset');
-        const cronExpression = document.getElementById('cron-expression');
-        const customCron = document.getElementById('custom-cron');
-        const cronDescription = document.getElementById('cron-description');
-        
-        if (preset) preset.value = '';
-        if (cronExpression) cronExpression.value = '';
-        if (customCron) customCron.classList.add('hidden');
-        if (cronDescription) cronDescription.classList.add('hidden');
-    }
-})();
-</script>
+<!-- Restore fresh create form -->
+<div hx-get="/api/schedules/form"
+     hx-target="#schedule-form-container"
+     hx-swap="innerHTML"
+     hx-trigger="load delay:100ms"></div>
diff --git a/src/borgitory/templates/partials/schedules/edit_form.html b/src/borgitory/templates/partials/schedules/edit_form.html
index a26aec3f..a4a2f4d4 100644
--- a/src/borgitory/templates/partials/schedules/edit_form.html
+++ b/src/borgitory/templates/partials/schedules/edit_form.html
@@ -29,14 +29,27 @@ <h3 class="font-medium text-gray-900 dark:text-gray-100 mb-3">Edit Schedule</h3>
             </div>
             <div>
                 <label class="block text-sm font-medium text-gray-900 dark:text-gray-100">
-                    Source Path
+                    Source Paths
                 </label>
-                {% set input_id = "schedule-edit-source-path" %}
-                {% set input_name = "source_path" %}
-                {% set input_value = schedule.source_path %}
-                {% set placeholder = "Enter path for scheduled backups" %}
-                {% set required = false %}
-                {% include "partials/shared/path_autocomplete.html" %}
+                {% set sp_count = source_paths | length if source_paths is defined and source_paths else 0 %}
+                <button type="button"
+                        class="flex items-stretch w-full text-sm font-medium text-gray-700 dark:text-gray-300 bg-white dark:bg-gray-800 border border-gray-300 dark:border-gray-600 rounded-md hover:bg-gray-50 dark:hover:bg-gray-700 focus:ring-2 focus:ring-blue-500 transition-colors overflow-hidden mt-1"
+                        hx-post="/api/schedules/source-paths/source-paths-modal"
+                        hx-include="#source-paths-data"
+                        hx-target="#modal-container">
+                    <span class="flex-1 px-4 py-2 text-left" id="source-paths-button-text">Source Paths</span>
+                    {% if sp_count > 0 %}
+                        <span class="flex items-center justify-center px-3 bg-blue-100 text-blue-800 dark:bg-blue-900 dark:text-blue-200 border-l border-gray-300 dark:border-gray-600 text-xs font-medium min-w-[2rem]"
+                              id="source-paths-count-badge">{{ sp_count }}</span>
+                    {% else %}
+                        <span class="flex items-center justify-center px-3 bg-gray-100 dark:bg-gray-700 border-l border-gray-300 dark:border-gray-600 text-xs font-medium min-w-[2rem]"
+                              id="source-paths-count-badge">{{ sp_count }}</span>
+                    {% endif %}
+                </button>
+                <input type="hidden"
+                       name="source_paths"
+                       id="source-paths-data"
+                       value="{{ source_paths_json if source_paths_json is defined else schedule.source_paths }}">
             </div>
             <div>
                 <label class="block text-sm font-medium text-gray-900 dark:text-gray-100">
diff --git a/src/borgitory/templates/partials/shared/path_autocomplete.html b/src/borgitory/templates/partials/shared/path_autocomplete.html
index 7d517ab1..e033c802 100644
--- a/src/borgitory/templates/partials/shared/path_autocomplete.html
+++ b/src/borgitory/templates/partials/shared/path_autocomplete.html
@@ -15,6 +15,7 @@
            hx-get="/api/repositories/directories/autocomplete"
            hx-trigger="input changed delay:300ms, focus delay:100ms, autocomplete-immediate"
            hx-target="#{{ input_id }}-dropdown"
+           hx-swap="innerHTML"
            hx-include="this"
            hx-headers='{"hx-target-input": "{{ input_id }}"}'
            autocomplete="off"
diff --git a/src/borgitory/templates/partials/shared/source_path_field.html b/src/borgitory/templates/partials/shared/source_path_field.html
new file mode 100644
index 00000000..427142bf
--- /dev/null
+++ b/src/borgitory/templates/partials/shared/source_path_field.html
@@ -0,0 +1,26 @@
+<!-- Individual Source Path Field -->
+<div class="flex items-start gap-2 source-path-entry">
+    <div class="flex-1">
+        {% set input_id = "source-path-" ~ index %}
+        {% set input_name = "source_paths" %}
+        {% set input_value = path_value if path_value is defined else "" %}
+        {% set placeholder = "/path/to/data" %}
+        {% set required = false %}
+        {% include "partials/shared/path_autocomplete.html" %}
+    </div>
+    {% if show_remove %}
+        <button type="button"
+                class="mt-1 text-red-600 hover:text-red-800 dark:text-red-400 dark:hover:text-red-300 p-2 rounded-md hover:bg-red-50 dark:hover:bg-red-900/20 transition-colors flex-shrink-0"
+                hx-post="/api/schedules/source-paths/remove-field"
+                hx-include="form"
+                hx-vals='{"remove_index": {{ index }} }'
+                hx-target="#source-paths-container"
+                hx-swap="innerHTML"
+                title="Remove source path">
+            <svg class="w-4 h-4" fill="none" stroke="currentColor" viewBox="0 0 24 24">
+                <path stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M19 7l-.867 12.142A2 2 0 0116.138 21H7.862a2 2 0 01-1.995-1.858L5 7m5 4v6m4-6v6m1-10V4a1 1 0 00-1-1h-4a1 1 0 00-1 1v3M4 7h16">
+                </path>
+            </svg>
+        </button>
+    {% endif %}
+</div>
diff --git a/src/borgitory/templates/partials/shared/source_paths_container.html b/src/borgitory/templates/partials/shared/source_paths_container.html
new file mode 100644
index 00000000..425a0a80
--- /dev/null
+++ b/src/borgitory/templates/partials/shared/source_paths_container.html
@@ -0,0 +1,21 @@
+<!-- Source Paths Container (rendered inside modal) -->
+<div class="space-y-2" id="source-paths-fields">
+    {% for path_value in source_paths %}
+        {% set index = loop.index0 %}
+        {% set show_remove = source_paths | length > 1 %}
+        {% include "partials/shared/source_path_field.html" with context %}
+    {% endfor %}
+    {% if not source_paths %}
+        <div class="text-center py-8 text-gray-500 dark:text-gray-400 border-2 border-dashed border-gray-300 dark:border-gray-600 rounded-lg">
+            <svg class="w-8 h-8 mx-auto mb-2"
+                 fill="none"
+                 stroke="currentColor"
+                 viewBox="0 0 24 24">
+                <path stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M12 6v6m0 0v6m0-6h6m-6 0H6">
+                </path>
+            </svg>
+            <p class="text-sm">No source paths configured</p>
+            <p class="text-xs">Click "+ Add Path" to get started</p>
+        </div>
+    {% endif %}
+</div>
diff --git a/src/borgitory/templates/partials/shared/source_paths_modal.html b/src/borgitory/templates/partials/shared/source_paths_modal.html
new file mode 100644
index 00000000..15462e94
--- /dev/null
+++ b/src/borgitory/templates/partials/shared/source_paths_modal.html
@@ -0,0 +1,85 @@
+<!-- Source Paths Configuration Modal -->
+<div class="fixed inset-0 bg-black bg-opacity-50 z-50 overflow-y-auto"
+     id="modal-container">
+    <div class="flex items-start justify-center min-h-full py-8">
+        <div class="bg-white dark:bg-gray-800 rounded-lg shadow-xl max-w-2xl w-full mx-4">
+            <!-- Modal Header -->
+            <div class="border-b border-gray-200 dark:border-gray-700 px-6 py-4">
+                <div class="flex items-center justify-between">
+                    <h3 class="text-lg font-semibold text-gray-900 dark:text-gray-100">
+                        Source Paths Configuration
+                    </h3>
+                    <button type="button"
+                            class="text-gray-400 hover:text-gray-600 dark:hover:text-gray-300"
+                            hx-get="/api/schedules/source-paths/close-modal"
+                            hx-target="#modal-container"
+                            hx-swap="outerHTML">
+                        <svg class="w-6 h-6" fill="none" stroke="currentColor" viewBox="0 0 24 24">
+                            <path stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M6 18L18 6M6 6l12 12">
+                            </path>
+                        </svg>
+                    </button>
+                </div>
+            </div>
+            <!-- Modal Content -->
+            <div>
+            <form hx-post="/api/schedules/source-paths/save-source-paths"
+                  hx-ext="json-enc"
+                  hx-target="body"
+                  hx-swap="none"
+                  hx-target-error="#source-paths-error-container">
+                <div id="source-paths-error-container" class="p-6 pb-0"></div>
+                <div class="p-6">
+                    <div class="bg-blue-50 dark:bg-blue-900/20 border border-blue-200 dark:border-blue-800 rounded-md p-4 mb-6">
+                        <div class="flex">
+                            <svg class="w-5 h-5 text-blue-400 mt-0.5"
+                                 fill="currentColor"
+                                 viewBox="0 0 20 20">
+                                <path fill-rule="evenodd" d="M18 10a8 8 0 11-16 0 8 8 0 0116 0zm-7-4a1 1 0 11-2 0 1 1 0 012 0zM9 9a1 1 0 000 2v3a1 1 0 001 1h1a1 1 0 100-2v-3a1 1 0 00-1-1H9z" clip-rule="evenodd">
+                                </path>
+                            </svg>
+                            <div class="ml-3">
+                                <h4 class="text-sm font-medium text-blue-800 dark:text-blue-200">
+                                    Source Paths
+                                </h4>
+                                <p class="mt-1 text-sm text-blue-700 dark:text-blue-300">
+                                    Add one or more absolute paths to include in this backup.
+                                    Each path must start with <code class="font-mono">/</code>.
+                                </p>
+                            </div>
+                        </div>
+                    </div>
+                    <div class="space-y-4">
+                        <div class="flex items-center justify-between">
+                            <h4 class="text-lg font-semibold text-gray-900 dark:text-gray-100">Paths</h4>
+                            <button type="button"
+                                    class="bg-green-600 text-white px-3 py-1 rounded-md text-sm hover:bg-green-700 focus:ring-2 focus:ring-green-500"
+                                    hx-post="/api/schedules/source-paths/add-field"
+                                    hx-include="form"
+                                    hx-target="#source-paths-container"
+                                    hx-swap="innerHTML">+ Add Path</button>
+                        </div>
+                        <div id="source-paths-container">
+                            {% include "partials/shared/source_paths_container.html" %}
+                        </div>
+                    </div>
+                </div>
+                <!-- Modal Footer -->
+                <div class="border-t border-gray-200 dark:border-gray-700 px-6 py-4 bg-gray-50 dark:bg-gray-900">
+                    <div class="flex justify-end space-x-3">
+                        <button type="button"
+                                class="px-4 py-2 text-sm font-medium text-gray-700 dark:text-gray-300 bg-white dark:bg-gray-800 border border-gray-300 dark:border-gray-600 rounded-md hover:bg-gray-50 dark:hover:bg-gray-700 focus:ring-2 focus:ring-blue-500"
+                                hx-get="/api/schedules/source-paths/close-modal"
+                                hx-target="#modal-container"
+                                hx-swap="outerHTML">Cancel</button>
+                        <button type="submit"
+                                class="px-4 py-2 text-sm font-medium text-white bg-blue-600 border border-transparent rounded-md hover:bg-blue-700 focus:ring-2 focus:ring-blue-500">
+                            Save & Close
+                        </button>
+                    </div>
+                </div>
+            </form>
+            </div>
+        </div>
+    </div>
+</div>
diff --git a/src/borgitory/templates/partials/shared/source_paths_save_response.html b/src/borgitory/templates/partials/shared/source_paths_save_response.html
new file mode 100644
index 00000000..0d283cbc
--- /dev/null
+++ b/src/borgitory/templates/partials/shared/source_paths_save_response.html
@@ -0,0 +1,25 @@
+<!-- Source Paths Save Response - Updates parent form and closes modal -->
+<!-- Close the modal -->
+<div id="modal-container" hx-swap-oob="innerHTML"></div>
+<!-- Update parent form hidden field via OOB swap -->
+<input type="hidden"
+       name="source_paths"
+       id="source-paths-data"
+       value="{{ source_paths_json if source_paths_json else '[]' }}"
+       hx-swap-oob="true">
+<!-- Update the entire button to ensure proper layout -->
+<button type="button"
+        class="flex items-stretch w-full text-sm font-medium text-gray-700 dark:text-gray-300 bg-white dark:bg-gray-800 border border-gray-300 dark:border-gray-600 rounded-md hover:bg-gray-50 dark:hover:bg-gray-700 focus:ring-2 focus:ring-blue-500 transition-colors overflow-hidden"
+        hx-post="/api/schedules/source-paths/source-paths-modal"
+        hx-include="#source-paths-data"
+        hx-target="#modal-container"
+        hx-swap-oob="outerHTML:[hx-post*='source-paths-modal']">
+    <span class="flex-1 px-4 py-2 text-left" id="source-paths-button-text">Source Paths</span>
+    {% if total_count > 0 %}
+        <span class="flex items-center justify-center px-3 bg-blue-100 text-blue-800 dark:bg-blue-900 dark:text-blue-200 border-l border-gray-300 dark:border-gray-600 text-xs font-medium min-w-[2rem]"
+              id="source-paths-count-badge">{{ total_count }}</span>
+    {% else %}
+        <span class="flex items-center justify-center px-3 bg-gray-100 dark:bg-gray-700 border-l border-gray-300 dark:border-gray-600 text-xs font-medium min-w-[2rem]"
+              id="source-paths-count-badge">{{ total_count }}</span>
+    {% endif %}
+</button>
diff --git a/src/borgitory/templates/partials/shared/source_paths_validation_error.html b/src/borgitory/templates/partials/shared/source_paths_validation_error.html
new file mode 100644
index 00000000..14e112c9
--- /dev/null
+++ b/src/borgitory/templates/partials/shared/source_paths_validation_error.html
@@ -0,0 +1,14 @@
+<div class="bg-red-50 dark:bg-red-900/20 border border-red-200 dark:border-red-800 rounded-md p-4 mb-4">
+    <div class="flex">
+        <svg class="w-5 h-5 text-red-400 mt-0.5"
+             fill="currentColor"
+             viewBox="0 0 20 20">
+            <path fill-rule="evenodd" d="M10 18a8 8 0 100-16 8 8 0 000 16zM8.707 7.293a1 1 0 00-1.414 1.414L8.586 10l-1.293 1.293a1 1 0 101.414 1.414L10 11.414l1.293 1.293a1 1 0 001.414-1.414L11.414 10l1.293-1.293a1 1 0 00-1.414-1.414L10 8.586 8.707 7.293z" clip-rule="evenodd">
+            </path>
+        </svg>
+        <div class="ml-3">
+            <h3 class="text-sm font-medium text-red-800 dark:text-red-200">Validation Error</h3>
+            <p class="mt-1 text-sm text-red-700 dark:text-red-300">{{ error_message }}</p>
+        </div>
+    </div>
+</div>
diff --git a/src/borgitory/utils/source_paths.py b/src/borgitory/utils/source_paths.py
new file mode 100644
index 00000000..67de542c
--- /dev/null
+++ b/src/borgitory/utils/source_paths.py
@@ -0,0 +1,30 @@
+"""Utilities for handling multiple source paths stored as JSON array strings."""
+
+import json
+import logging
+from typing import List
+
+logger = logging.getLogger(__name__)
+
+
+def parse_source_paths(source_paths: str) -> List[str]:
+    """Parse a source_paths JSON array string into a list of path strings.
+
+    Returns only non-empty strings. Invalid JSON returns an empty list.
+    """
+    if not source_paths or not source_paths.strip():
+        return []
+
+    try:
+        parsed = json.loads(source_paths)
+        if isinstance(parsed, list):
+            return [p for p in parsed if isinstance(p, str) and p.strip()]
+        return []
+    except json.JSONDecodeError, ValueError:
+        return []
+
+
+def serialize_source_paths(paths: List[str]) -> str:
+    """Convert a list of paths to a JSON array string for storage."""
+    cleaned = [p.strip() for p in paths if p and p.strip()]
+    return json.dumps(cleaned) if cleaned else "[]"
diff --git a/tests/e2e/repositories/test_repositories.py b/tests/e2e/repositories/test_repositories.py
new file mode 100644
index 00000000..a71927bb
--- /dev/null
+++ b/tests/e2e/repositories/test_repositories.py
@@ -0,0 +1,46 @@
+"""E2E tests for repository management."""
+
+import uuid
+
+import pytest
+from playwright.sync_api import Page, expect
+
+from tests.e2e.conftest import wait_for_htmx
+
+pytestmark = pytest.mark.e2e
+
+
+def _navigate_to_repositories(page: Page) -> None:
+    page.click("#nav-repositories")
+    page.wait_for_load_state("networkidle")
+    page.wait_for_url("**/repositories", timeout=10000)
+    wait_for_htmx(page)
+
+
+def create_repository(page: Page) -> tuple[str, str]:
+    """Create a new repository via the UI. Returns (repo_name, repo_path)."""
+    _navigate_to_repositories(page)
+
+    form = page.locator("form[hx-post='/api/repositories/']")
+    expect(form).to_be_visible(timeout=15000)
+
+    repo_name = f"e2e-repo-{uuid.uuid4().hex[:8]}"
+    repo_path = f"/tmp/borgitory-e2e-{uuid.uuid4().hex[:8]}"
+
+    form.locator("input[name='name']").fill(repo_name)
+    form.locator("input[name='path']").fill(repo_path)
+    form.locator("select[name='encryption_type']").select_option("none")
+    form.locator("button[type='submit']").click()
+    wait_for_htmx(page)
+
+    list_region = page.locator("#repository-list")
+    expect(list_region).to_contain_text(repo_name, timeout=15000)
+    expect(list_region).to_contain_text(repo_path, timeout=5000)
+
+    return repo_name, repo_path
+
+
+def test_create_new_repository(authenticated_page: Page) -> None:
+    """Create a repository via the UI; new repo appears in the list."""
+    page = authenticated_page
+    create_repository(page)
diff --git a/tests/e2e/schedules/test_schedules.py b/tests/e2e/schedules/test_schedules.py
new file mode 100644
index 00000000..75dc9876
--- /dev/null
+++ b/tests/e2e/schedules/test_schedules.py
@@ -0,0 +1,70 @@
+"""E2E tests for schedule management."""
+
+import uuid
+
+import pytest
+from playwright.sync_api import Page, expect
+
+from tests.e2e.conftest import wait_for_htmx
+from tests.e2e.repositories.test_repositories import create_repository
+
+pytestmark = pytest.mark.e2e
+
+
+def _navigate_to_schedules(page: Page) -> None:
+    page.click("#nav-schedules")
+    page.wait_for_load_state("networkidle")
+    page.wait_for_url("**/schedules", timeout=10000)
+    wait_for_htmx(page)
+
+
+def test_create_schedule_with_multiple_source_paths(authenticated_page: Page) -> None:
+    """Create a schedule using a new repository and multiple source paths."""
+    page = authenticated_page
+    repo_name, _ = create_repository(page)
+
+    _navigate_to_schedules(page)
+
+    schedule_form = page.locator("form[hx-post='/api/schedules/']")
+    expect(schedule_form).to_be_visible(timeout=15000)
+
+    schedule_name = f"e2e-schedule-{uuid.uuid4().hex[:8]}"
+    schedule_form.locator("input[name='name']").fill(schedule_name)
+    schedule_form.locator("select[name='repository_id']").select_option(label=repo_name)
+
+    schedule_form.locator("button[hx-post*='source-paths-modal']").click()
+    wait_for_htmx(page)
+
+    source_paths_form = page.locator(
+        "form[hx-post='/api/schedules/source-paths/save-source-paths']"
+    )
+    expect(source_paths_form).to_be_visible(timeout=10000)
+
+    path_inputs = source_paths_form.locator("input[name='source_paths']")
+    expect(path_inputs.first).to_be_visible(timeout=5000)
+    path_inputs.nth(0).fill("/tmp/e2e-src1")
+
+    source_paths_form.locator("button:has-text('+ Add Path')").click()
+    wait_for_htmx(page)
+
+    path_inputs = source_paths_form.locator("input[name='source_paths']")
+    expect(path_inputs.nth(1)).to_be_visible(timeout=5000)
+    path_inputs.nth(1).fill("/tmp/e2e-src2")
+
+    source_paths_form.locator("button[type='submit']:has-text('Save & Close')").click()
+    wait_for_htmx(page)
+
+    expect(page.get_by_text("Source Paths Configuration")).not_to_be_visible(timeout=10000)
+
+    schedule_form = page.locator("form[hx-post='/api/schedules/']")
+    schedule_form.locator("#schedule-preset").select_option("0 2 * * *")
+    wait_for_htmx(page)
+
+    expect(schedule_form.locator("#cron-expression")).to_have_value("0 2 * * *", timeout=5000)
+
+    schedule_form.locator("button[type='submit']:has-text('Create Schedule')").click()
+    wait_for_htmx(page)
+
+    schedules_list = page.locator("#schedules-list")
+    expect(schedules_list).to_contain_text(schedule_name, timeout=15000)
+    expect(schedules_list).to_contain_text(repo_name, timeout=5000)
diff --git a/tests/fixtures/job_fixtures.py b/tests/fixtures/job_fixtures.py
index fbf974d6..aecfb3ed 100644
--- a/tests/fixtures/job_fixtures.py
+++ b/tests/fixtures/job_fixtures.py
@@ -84,7 +84,7 @@ def sample_composite_job() -> BorgJob:
         task_type=TaskTypeEnum.BACKUP,
         task_name="Backup Task",
         status=TaskStatusEnum.COMPLETED,
-        parameters={"source_path": "/data"},
+        parameters={"source_paths": '["/data"]'},
     )
     task2 = BorgJobTask(
         task_type=TaskTypeEnum.PRUNE,
diff --git a/tests/integration/schedules/test_manual_run_apscheduler.py b/tests/integration/schedules/test_manual_run_apscheduler.py
index b53ec07b..7c35339a 100644
--- a/tests/integration/schedules/test_manual_run_apscheduler.py
+++ b/tests/integration/schedules/test_manual_run_apscheduler.py
@@ -100,7 +100,7 @@ async def test_schedule(
         schedule.name = "Test Schedule"
         schedule.repository_id = test_repository.id
         schedule.cron_expression = "0 2 * * *"
-        schedule.source_path = "/test/source"
+        schedule.source_paths = '["/test/source"]'
         schedule.enabled = True
         test_db.add(schedule)
         await test_db.commit()
@@ -448,7 +448,7 @@ def mock_schedule(self) -> Mock:
         schedule.id = 1
         schedule.name = "Test Schedule"
         schedule.repository_id = 10
-        schedule.source_path = "/data/backup"
+        schedule.source_paths = ["/data/backup"]
         schedule.cloud_sync_config_id = None
         schedule.prune_config_id = 2
         schedule.check_config_id = 3
@@ -522,7 +522,7 @@ async def test_execute_scheduled_backup_success(
         assert call_args.args[0] is mock_db
         backup_request = call_args.args[1]
         assert backup_request.repository_id == 10
-        assert backup_request.source_path == "/data/backup"
+        assert backup_request.source_paths == ["/data/backup"]
         assert backup_request.prune_config_id == 2
         assert backup_request.check_config_id == 3
         assert call_args.args[2] == JobType.SCHEDULED_BACKUP
diff --git a/tests/integration/schedules/test_schedule_creation_api.py b/tests/integration/schedules/test_schedule_creation_api.py
index 2109d533..190b8805 100644
--- a/tests/integration/schedules/test_schedule_creation_api.py
+++ b/tests/integration/schedules/test_schedule_creation_api.py
@@ -83,7 +83,7 @@ async def test_create_schedule_valid_data(
             "name": "Daily Backup",
             "repository_id": 1,
             "cron_expression": "0 2 * * *",
-            "source_path": "/data",
+            "source_paths": '["/data"]',
             "cloud_sync_config_id": None,
             "prune_config_id": None,
             "notification_config_id": None,
@@ -108,7 +108,7 @@ async def test_create_schedule_missing_name(
             "name": "",
             "repository_id": 1,
             "cron_expression": "0 2 * * *",
-            "source_path": "/data",
+            "source_paths": '["/data"]',
         }
 
         response = await async_client.post("/api/schedules/", json=invalid_data)
@@ -126,7 +126,7 @@ async def test_create_schedule_missing_repository(
             "name": "Test Schedule",
             "repository_id": "",
             "cron_expression": "0 2 * * *",
-            "source_path": "/data",
+            "source_paths": '["/data"]',
         }
 
         response = await async_client.post("/api/schedules/", json=invalid_data)
@@ -144,7 +144,7 @@ async def test_create_schedule_invalid_repository_id(
             "name": "Test Schedule",
             "repository_id": "not-a-number",
             "cron_expression": "0 2 * * *",
-            "source_path": "/data",
+            "source_paths": '["/data"]',
         }
 
         response = await async_client.post("/api/schedules/", json=invalid_data)
@@ -162,7 +162,7 @@ async def test_create_schedule_missing_cron_expression(
             "name": "Test Schedule",
             "repository_id": 1,
             "cron_expression": "",
-            "source_path": "/data",
+            "source_paths": '["/data"]',
         }
 
         response = await async_client.post("/api/schedules/", json=invalid_data)
@@ -180,7 +180,7 @@ async def test_create_schedule_invalid_cron_expression_too_few_parts(
             "name": "Test Schedule",
             "repository_id": 1,
             "cron_expression": "0 2 * *",  # Only 4 parts
-            "source_path": "/data",
+            "source_paths": '["/data"]',
         }
 
         response = await async_client.post("/api/schedules/", json=invalid_data)
@@ -198,7 +198,7 @@ async def test_create_schedule_invalid_cron_expression_too_many_parts(
             "name": "Test Schedule",
             "repository_id": 1,
             "cron_expression": "0 2 * * * *",  # 6 parts
-            "source_path": "/data",
+            "source_paths": '["/data"]',
         }
 
         response = await async_client.post("/api/schedules/", json=invalid_data)
@@ -225,7 +225,7 @@ async def test_create_schedule_complex_valid_cron_expressions(
                 "name": f"Test Schedule - {description}",
                 "repository_id": 1,
                 "cron_expression": cron_expr,
-                "source_path": "/data",
+                "source_paths": '["/data"]',
             }
 
             response = await async_client.post("/api/schedules/", json=valid_data)
@@ -242,12 +242,12 @@ async def test_create_schedule_complex_valid_cron_expressions(
     async def test_create_schedule_whitespace_handling(
         self, async_client: AsyncClient, setup_dependencies: Dict[str, Any]
     ) -> None:
-        """Test creating a schedule with whitespace in inputs."""
+        """Test creating a schedule with whitespace in inputs (name, cron)."""
         data_with_whitespace = {
             "name": "  Test Schedule  ",
             "repository_id": 1,
             "cron_expression": "  0 2 * * *  ",
-            "source_path": "  /data  ",
+            "source_paths": '["/data"]',
         }
 
         response = await async_client.post("/api/schedules/", json=data_with_whitespace)
@@ -276,7 +276,7 @@ async def test_create_schedule_optional_fields_handling(
                 "name": "Test Schedule",
                 "repository_id": 1,
                 "cron_expression": "0 2 * * *",
-                "source_path": "/data",
+                "source_paths": '["/data"]',
                 **optional_fields,
             }
 
@@ -294,7 +294,7 @@ async def test_create_schedule_nonexistent_repository(
             "name": "Test Schedule",
             "repository_id": 999,  # Non-existent repository
             "cron_expression": "0 2 * * *",
-            "source_path": "/data",
+            "source_paths": '["/data"]',
         }
 
         response = await async_client.post("/api/schedules/", json=invalid_data)
@@ -316,7 +316,7 @@ async def test_create_schedule_scheduler_service_failure(
             "name": "Test Schedule",
             "repository_id": 1,
             "cron_expression": "0 2 * * *",
-            "source_path": "/data",
+            "source_paths": '["/data"]',
         }
 
         response = await async_client.post("/api/schedules/", json=valid_data)
@@ -353,7 +353,7 @@ async def test_create_schedule_htmx_headers(
             "name": "Test Schedule",
             "repository_id": 1,
             "cron_expression": "0 2 * * *",
-            "source_path": "/data",
+            "source_paths": '["/data"]',
         }
 
         response = await async_client.post("/api/schedules/", json=valid_data)
@@ -371,7 +371,7 @@ async def test_create_schedule_response_format(
             "name": "Test Schedule",
             "repository_id": 1,
             "cron_expression": "0 2 * * *",
-            "source_path": "/data",
+            "source_paths": '["/data"]',
         }
 
         response = await async_client.post("/api/schedules/", json=valid_data)
diff --git a/tests/integration/schedules/test_schedules_api.py b/tests/integration/schedules/test_schedules_api.py
index fe3bb697..955d4d32 100644
--- a/tests/integration/schedules/test_schedules_api.py
+++ b/tests/integration/schedules/test_schedules_api.py
@@ -135,7 +135,7 @@ async def test_create_schedule_success(
             "name": "Test Schedule",
             "repository_id": sample_repository.id,
             "cron_expression": "0 2 * * *",
-            "source_path": "/data",
+            "source_paths": '["/data"]',
         }
 
         response = await async_client.post("/api/schedules/", json=schedule_data)
@@ -165,7 +165,7 @@ async def test_create_schedule_repository_not_found(
             "name": "Test Schedule",
             "repository_id": 999,  # Non-existent repository
             "cron_expression": "0 2 * * *",
-            "source_path": "/data",
+            "source_paths": '["/data"]',
         }
 
         response = await async_client.post("/api/schedules/", json=schedule_data)
@@ -187,7 +187,7 @@ async def test_create_schedule_invalid_cron(
             "name": "Test Schedule",
             "repository_id": sample_repository.id,
             "cron_expression": "invalid cron",
-            "source_path": "/data",
+            "source_paths": '["/data"]',
         }
 
         response = await async_client.post("/api/schedules/", json=schedule_data)
@@ -210,13 +210,13 @@ async def test_get_schedules_html(
         schedule1.name = "schedule-1"
         schedule1.repository_id = sample_repository.id
         schedule1.cron_expression = "0 2 * * *"
-        schedule1.source_path = "/data1"
+        schedule1.source_paths = '["/data1"]'
 
         schedule2 = Schedule()
         schedule2.name = "schedule-2"
         schedule2.repository_id = sample_repository.id
         schedule2.cron_expression = "0 3 * * *"
-        schedule2.source_path = "/data2"
+        schedule2.source_paths = '["/data2"]'
         test_db.add_all([schedule1, schedule2])
         await test_db.commit()
 
@@ -242,7 +242,7 @@ async def test_get_schedules_html_pagination(
             schedule.name = f"schedule-{i}"
             schedule.repository_id = sample_repository.id
             schedule.cron_expression = "0 2 * * *"
-            schedule.source_path = f"/data{i}"
+            schedule.source_paths = f'["/data{i}"]'
             test_db.add(schedule)
         await test_db.commit()
 
@@ -316,7 +316,7 @@ async def test_list_schedules(
         schedule1.name = "schedule-1"
         schedule1.repository_id = sample_repository.id
         schedule1.cron_expression = "0 2 * * *"
-        schedule1.source_path = "/data1"
+        schedule1.source_paths = '["/data1"]'
         test_db.add(schedule1)
         await test_db.commit()
 
@@ -357,7 +357,7 @@ async def test_get_schedule_edit_form(
         schedule.name = "test-schedule"
         schedule.repository_id = sample_repository.id
         schedule.cron_expression = "0 2 * * *"
-        schedule.source_path = "/data"
+        schedule.source_paths = '["/data"]'
         test_db.add(schedule)
         await test_db.commit()
         await test_db.refresh(schedule)
@@ -381,7 +381,7 @@ async def test_update_schedule_success(
         schedule.name = "original-schedule"
         schedule.repository_id = sample_repository.id
         schedule.cron_expression = "0 2 * * *"
-        schedule.source_path = "/data"
+        schedule.source_paths = '["/data"]'
         test_db.add(schedule)
         await test_db.commit()
         await test_db.refresh(schedule)
@@ -415,7 +415,7 @@ async def test_toggle_schedule_success(
         schedule.name = "test-schedule"
         schedule.repository_id = sample_repository.id
         schedule.cron_expression = "0 2 * * *"
-        schedule.source_path = "/data"
+        schedule.source_paths = '["/data"]'
         schedule.enabled = False
         test_db.add(schedule)
         await test_db.commit()
@@ -454,7 +454,7 @@ async def test_delete_schedule_success(
         schedule.name = "test-schedule"
         schedule.repository_id = sample_repository.id
         schedule.cron_expression = "0 2 * * *"
-        schedule.source_path = "/data"
+        schedule.source_paths = '["/data"]'
         test_db.add(schedule)
         await test_db.commit()
         await test_db.refresh(schedule)
diff --git a/tests/integration/schedules/test_schedules_source_paths_api.py b/tests/integration/schedules/test_schedules_source_paths_api.py
new file mode 100644
index 00000000..cb47f412
--- /dev/null
+++ b/tests/integration/schedules/test_schedules_source_paths_api.py
@@ -0,0 +1,482 @@
+"""
+Tests for schedule source paths API endpoints in schedules.py
+
+Tests all source-path-related endpoints using proper DI with mocks, no patches.
+"""
+
+import json
+import pytest
+from typing import Any, Dict
+from unittest.mock import AsyncMock, Mock
+from httpx import AsyncClient
+from fastapi.responses import HTMLResponse
+
+from borgitory.main import app
+from borgitory.dependencies import (
+    get_schedule_service,
+    get_templates,
+    get_configuration_service,
+)
+from borgitory.services.scheduling.schedule_service import ScheduleService
+from borgitory.services.configuration_service import ConfigurationService
+
+
+class TestScheduleSourcePathsAPI:
+    """Test the Schedule Source Paths API endpoints - HTMX/HTTP behavior"""
+
+    @pytest.fixture(scope="function")
+    def setup_test_dependencies(self) -> Dict[str, Any]:
+        """Setup dependency overrides for each test."""
+        mock_scheduler_service = AsyncMock()
+        mock_scheduler_service.add_schedule.return_value = None
+        mock_scheduler_service.update_schedule.return_value = None
+        mock_scheduler_service.remove_schedule.return_value = None
+
+        schedule_service = ScheduleService(mock_scheduler_service)
+        configuration_service = ConfigurationService()
+
+        mock_templates = Mock()
+        captured_contexts: list[Dict[str, Any]] = []
+
+        def mock_template_response(
+            request: Any,
+            template_name: str,
+            context: Any = None,
+            status_code: int = 200,
+        ) -> HTMLResponse:
+            captured_contexts.append(
+                {"template": template_name, "context": context or {}}
+            )
+            return HTMLResponse(
+                content=f"<div>Mock response for {template_name}</div>",
+                status_code=status_code,
+            )
+
+        mock_templates.TemplateResponse = mock_template_response
+
+        app.dependency_overrides[get_schedule_service] = lambda: schedule_service
+        app.dependency_overrides[get_configuration_service] = (
+            lambda: configuration_service
+        )
+        app.dependency_overrides[get_templates] = lambda: mock_templates
+
+        return {
+            "schedule_service": schedule_service,
+            "configuration_service": configuration_service,
+            "templates": mock_templates,
+            "captured_contexts": captured_contexts,
+        }
+
+    def teardown_method(self) -> None:
+        app.dependency_overrides.clear()
+
+    # ── source-paths-modal endpoint ────────────────────────────────────
+
+    async def test_modal_opens_with_json_array(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/source-paths-modal",
+            json={"source_paths": '["/data", "/backup"]'},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["source_paths"] == ["/data", "/backup"]
+        assert captured[-1]["template"] == "partials/shared/source_paths_modal.html"
+
+    async def test_modal_opens_with_empty_array(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/source-paths-modal",
+            json={"source_paths": "[]"},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["source_paths"] == [""]
+
+    async def test_modal_opens_with_single_path_json_array(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/source-paths-modal",
+            json={"source_paths": '["/data"]'},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["source_paths"] == ["/data"]
+
+    async def test_modal_opens_with_missing_source_paths(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/source-paths-modal",
+            json={},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["source_paths"] == [""]
+
+    # ── save-source-paths endpoint ─────────────────────────────────────
+
+    async def test_save_valid_paths(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/save-source-paths",
+            json={"source_paths": ["/data", "/home/user/docs"]},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["total_count"] == 2
+        parsed = json.loads(ctx["source_paths_json"])
+        assert parsed == ["/data", "/home/user/docs"]
+        assert (
+            captured[-1]["template"]
+            == "partials/shared/source_paths_save_response.html"
+        )
+
+    async def test_save_single_path(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/save-source-paths",
+            json={"source_paths": ["/data"]},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["total_count"] == 1
+
+    async def test_save_filters_empty_paths(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/save-source-paths",
+            json={"source_paths": ["/data", "", "  ", "/backup"]},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["total_count"] == 2
+        parsed = json.loads(ctx["source_paths_json"])
+        assert parsed == ["/data", "/backup"]
+
+    async def test_save_empty_paths_returns_zero_count(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/save-source-paths",
+            json={"source_paths": ["", "  "]},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["total_count"] == 0
+
+    async def test_save_rejects_relative_paths(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/save-source-paths",
+            json={"source_paths": ["/valid", "relative/path"]},
+        )
+
+        assert response.status_code == 400
+        ctx = captured[-1]["context"]
+        assert "relative/path" in ctx["error_message"]
+        assert (
+            captured[-1]["template"]
+            == "partials/shared/source_paths_validation_error.html"
+        )
+
+    async def test_save_rejects_all_relative_paths(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        _ = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/save-source-paths",
+            json={"source_paths": ["no-slash", "also-bad"]},
+        )
+
+        assert response.status_code == 400
+
+    # ── close-modal endpoint ───────────────────────────────────────────
+
+    async def test_close_modal(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        response = await async_client.get(
+            "/api/schedules/source-paths/close-modal"
+        )
+
+        assert response.status_code == 200
+        assert '<div id="modal-container"></div>' in response.text
+
+    # ── add-field endpoint ─────────────────────────────────────────────
+
+    async def test_add_field_appends_empty_path(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/add-field",
+            json={"source_paths": ["/data"]},
+        )
+
+        assert response.status_code == 200
+        assert "text/html" in response.headers.get("content-type", "")
+        ctx = captured[-1]["context"]
+        assert ctx["source_paths"] == ["/data", ""]
+
+    async def test_add_field_with_multiple_existing_paths(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/add-field",
+            json={"source_paths": ["/src", "/Documents", "/Pictures"]},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["source_paths"] == ["/src", "/Documents", "/Pictures", ""]
+
+    async def test_add_field_empty_form(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/add-field",
+            json={},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["source_paths"] == [""]
+
+    async def test_add_field_single_string_source_paths(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        """json-enc sends a bare string when there's only one input with that name."""
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/add-field",
+            json={"source_paths": "/only-one"},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["source_paths"] == ["/only-one", ""]
+
+    async def test_add_field_preserves_entered_text(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/add-field",
+            json={"source_paths": ["/my/typed/path"]},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert "/my/typed/path" in ctx["source_paths"]
+
+    async def test_add_field_renders_correct_template(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        captured = setup_test_dependencies["captured_contexts"]
+
+        await async_client.post(
+            "/api/schedules/source-paths/add-field",
+            json={"source_paths": ["/data"]},
+        )
+
+        assert (
+            captured[-1]["template"]
+            == "partials/shared/source_paths_container.html"
+        )
+
+    # ── remove-field endpoint ──────────────────────────────────────────
+
+    async def test_remove_field_removes_by_index(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/remove-field",
+            json={
+                "source_paths": ["/keep", "/remove", "/also-keep"],
+                "remove_index": 1,
+            },
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["source_paths"] == ["/keep", "/also-keep"]
+
+    async def test_remove_field_first_index(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/remove-field",
+            json={"source_paths": ["/first", "/second"], "remove_index": 0},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["source_paths"] == ["/second"]
+
+    async def test_remove_field_last_index(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/remove-field",
+            json={
+                "source_paths": ["/first", "/second", "/third"],
+                "remove_index": 2,
+            },
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["source_paths"] == ["/first", "/second"]
+
+    async def test_remove_last_remaining_path_leaves_empty(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        """Removing the only path should leave one empty path so the UI isn't blank."""
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/remove-field",
+            json={"source_paths": ["/only-path"], "remove_index": 0},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["source_paths"] == [""]
+
+    async def test_remove_field_out_of_range_index(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        """Out-of-range index should leave the list unchanged."""
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/remove-field",
+            json={"source_paths": ["/data"], "remove_index": 99},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["source_paths"] == ["/data"]
+
+    async def test_remove_field_negative_index(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        """Negative index should leave the list unchanged."""
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/remove-field",
+            json={"source_paths": ["/data"], "remove_index": -1},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["source_paths"] == ["/data"]
+
+    async def test_remove_field_invalid_index_string(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        """Non-integer index should be handled gracefully."""
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/remove-field",
+            json={"source_paths": ["/data", "/backup"], "remove_index": "abc"},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["source_paths"] == ["/data", "/backup"]
+
+    async def test_remove_field_missing_index_defaults_to_zero(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        """Missing remove_index should default to 0."""
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/remove-field",
+            json={"source_paths": ["/first", "/second"]},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["source_paths"] == ["/second"]
+
+    async def test_remove_field_empty_form(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        """Empty JSON body should result in one empty path."""
+        captured = setup_test_dependencies["captured_contexts"]
+
+        response = await async_client.post(
+            "/api/schedules/source-paths/remove-field",
+            json={},
+        )
+
+        assert response.status_code == 200
+        ctx = captured[-1]["context"]
+        assert ctx["source_paths"] == [""]
+
+    async def test_remove_field_renders_correct_template(
+        self, setup_test_dependencies: Dict[str, Any], async_client: AsyncClient
+    ) -> None:
+        captured = setup_test_dependencies["captured_contexts"]
+
+        await async_client.post(
+            "/api/schedules/source-paths/remove-field",
+            json={"source_paths": ["/a", "/b"], "remove_index": 0},
+        )
+
+        assert (
+            captured[-1]["template"]
+            == "partials/shared/source_paths_container.html"
+        )
diff --git a/tests/integration/test_jobs_api.py b/tests/integration/test_jobs_api.py
index b06b1386..da1da1da 100644
--- a/tests/integration/test_jobs_api.py
+++ b/tests/integration/test_jobs_api.py
@@ -175,7 +175,7 @@ async def test_create_backup_success(
 
         backup_request = {
             "repository_id": sample_repository.id,
-            "source_path": "/test/path",
+            "source_paths": '["/test/path"]',
             "compression": "zstd",
             "dry_run": False,
         }
@@ -198,7 +198,7 @@ async def test_create_backup_repository_not_found(
 
         backup_request = {
             "repository_id": 999,
-            "source_path": "/test/path",
+            "source_paths": '["/test/path"]',
             "compression": "zstd",
             "dry_run": False,
         }
@@ -224,7 +224,7 @@ async def test_create_backup_general_error(
 
         backup_request = {
             "repository_id": sample_repository.id,
-            "source_path": "/test/path",
+            "source_paths": '["/test/path"]',
             "compression": "zstd",
             "dry_run": False,
         }
@@ -690,7 +690,7 @@ async def test_backup_request_validation(
         """Test backup request validation."""
         # Test missing repository_id
         invalid_request = {
-            "source_path": "/test/path",
+            "source_paths": '["/test/path"]',
             "compression": "zstd",
         }
 
@@ -700,7 +700,7 @@ async def test_backup_request_validation(
         # Test invalid repository_id (must be > 0)
         invalid_request = {
             "repository_id": "0",
-            "source_path": "/test/path",
+            "source_paths": '["/test/path"]',
             "compression": "zstd",
         }
 
diff --git a/tests/schemas/test_source_paths_validation.py b/tests/schemas/test_source_paths_validation.py
new file mode 100644
index 00000000..fdbb585b
--- /dev/null
+++ b/tests/schemas/test_source_paths_validation.py
@@ -0,0 +1,195 @@
+"""Tests for source_paths validation in Pydantic schemas."""
+
+import pytest
+from pydantic import ValidationError
+
+from borgitory.models.schemas import (
+    ScheduleCreate,
+    ScheduleUpdate,
+    BackupRequest,
+)
+
+
+class TestScheduleCreateSourcePaths:
+    def test_absolute_paths_accepted(self) -> None:
+        s = ScheduleCreate(
+            name="test",
+            cron_expression="0 2 * * *",
+            repository_id=1,
+            source_paths=["/data", "/backup"],
+        )
+        paths = s.source_paths
+        assert paths == ['/data', '/backup']
+
+    def test_relative_path_rejected(self) -> None:
+        with pytest.raises(ValidationError, match="must be absolute"):
+            ScheduleCreate(
+                name="test",
+                cron_expression="0 2 * * *",
+                repository_id=1,
+                source_paths=["relative/path"],
+            )
+
+    def test_mixed_absolute_and_relative_rejected(self) -> None:
+        with pytest.raises(ValidationError, match="Invalid: data"):
+            ScheduleCreate(
+                name="test",
+                cron_expression="0 2 * * *",
+                repository_id=1,
+                source_paths=["/valid", "data"],
+            )
+
+    def test_empty_strings_filtered_before_validation(self) -> None:
+        s = ScheduleCreate(
+            name="test",
+            cron_expression="0 2 * * *",
+            repository_id=1,
+            source_paths=["/data", "", "  "],
+        )
+        paths = s.source_paths
+        assert paths == ['/data']
+
+    def test_all_empty_strings_becomes_empty_array(self) -> None:
+        s = ScheduleCreate(
+            name="test",
+            cron_expression="0 2 * * *",
+            repository_id=1,
+            source_paths=["", "  "],
+        )
+        assert s.source_paths == []
+
+    def test_no_source_paths_uses_default(self) -> None:
+        s = ScheduleCreate(
+            name="test",
+            cron_expression="0 2 * * *",
+            repository_id=1,
+        )
+        assert s.source_paths == []
+
+    def test_json_array_string_accepted(self) -> None:
+        s = ScheduleCreate(
+            name="test",
+            cron_expression="0 2 * * *",
+            repository_id=1,
+            source_paths=["/single"],
+        )
+        paths = s.source_paths
+        assert paths == ["/single"]
+
+    def test_json_array_string_relative_rejected(self) -> None:
+        with pytest.raises(ValidationError, match="must be absolute"):
+            ScheduleCreate(
+                name="test",
+                cron_expression="0 2 * * *",
+                repository_id=1,
+                source_paths=["/ok", "bad"],
+            )
+
+    def test_json_array_string_absolute_accepted(self) -> None:
+        s = ScheduleCreate(
+            name="test",
+            cron_expression="0 2 * * *",
+            repository_id=1,
+            source_paths=["/a", "/b"],
+        )
+        assert s.source_paths == ["/a", "/b"]
+
+    def test_list_serialized_to_json_string(self) -> None:
+        s = ScheduleCreate(
+            name="test",
+            cron_expression="0 2 * * *",
+            repository_id=1,
+            source_paths=["/x", "/y"],
+        )
+        assert s.source_paths == ["/x", "/y"]
+
+
+class TestScheduleUpdateSourcePaths:
+    def test_absolute_paths_accepted(self) -> None:
+        s = ScheduleUpdate(name="test", source_paths=["/data", "/backup"], cron_expression="0 2 * * *")
+        paths = s.source_paths
+        assert paths == ["/data", "/backup"]
+
+    def test_relative_path_rejected(self) -> None:
+        with pytest.raises(ValidationError, match="must be absolute"):
+            ScheduleUpdate(name="test", source_paths=["not/absolute"], cron_expression="0 2 * * *")
+
+    def test_no_source_paths_leaves_none(self) -> None:
+        s = ScheduleUpdate(name="updated", cron_expression="0 2 * * *")
+        assert s.source_paths is None
+
+    def test_json_array_string_relative_rejected(self) -> None:
+        with pytest.raises(ValidationError, match="must be absolute"):
+            ScheduleUpdate(name="test", source_paths=["relative"], cron_expression="0 2 * * *")
+
+    def test_json_array_string_absolute_accepted(self) -> None:
+        s = ScheduleUpdate(name="test", source_paths=["/valid"], cron_expression="0 2 * * *")
+        assert s.source_paths == ["/valid"]
+
+
+class TestBackupRequestSourcePaths:
+    def test_absolute_paths_accepted(self) -> None:
+        r = BackupRequest(
+            cloud_sync_config_id=1,
+            prune_config_id=1,
+            check_config_id=1,
+            notification_config_id=1,
+            repository_id=1,
+            source_paths=["/home/user/src", "/home/user/docs"],
+        )
+        paths = r.source_paths
+        assert paths == ['/home/user/src', '/home/user/docs']
+
+    def test_relative_path_rejected(self) -> None:
+        with pytest.raises(ValidationError, match="must be absolute"):
+            BackupRequest(
+                cloud_sync_config_id=1,
+                prune_config_id=1,
+                check_config_id=1,
+                notification_config_id=1,
+                repository_id=1,
+                source_paths=["relative"],
+            )
+
+    def test_no_source_paths_uses_default(self) -> None:
+        r = BackupRequest(
+            cloud_sync_config_id=1,
+            prune_config_id=1,
+            check_config_id=1,
+            notification_config_id=1,
+            repository_id=1
+        )
+        assert r.source_paths == []
+
+    def test_json_array_string_relative_rejected(self) -> None:
+        with pytest.raises(ValidationError, match="must be absolute"):
+            BackupRequest(
+                cloud_sync_config_id=1,
+                prune_config_id=1,
+                check_config_id=1,
+                notification_config_id=1,
+                repository_id=1,
+                source_paths=["/ok", "nope"]
+            )
+
+    def test_json_array_string_absolute_accepted(self) -> None:
+        r = BackupRequest(
+            cloud_sync_config_id=1,
+            prune_config_id=1,
+            check_config_id=1,
+            notification_config_id=1,
+            repository_id=1,
+            source_paths=["/explicit"]
+        )
+        assert r.source_paths == ["/explicit"]
+
+    def test_list_serialized_to_json_string(self) -> None:
+        r = BackupRequest(
+            cloud_sync_config_id=1,
+            prune_config_id=1,
+            check_config_id=1,
+            notification_config_id=1,
+            repository_id=1,
+            source_paths=["/a", "/b"],
+        )
+        assert r.source_paths == ["/a", "/b"]
diff --git a/tests/unit/hooks/test_composite_job_critical_failure.py b/tests/unit/hooks/test_composite_job_critical_failure.py
index 21c51698..12e96765 100644
--- a/tests/unit/hooks/test_composite_job_critical_failure.py
+++ b/tests/unit/hooks/test_composite_job_critical_failure.py
@@ -71,7 +71,7 @@ def create_backup_task(self) -> BorgJobTask:
         return BorgJobTask(
             task_type=TaskTypeEnum.BACKUP,
             task_name="Backup repository",
-            parameters={"source_path": "/data"},
+            parameters={"source_paths": '["/data"]'},
         )
 
     def create_notification_task(self) -> BorgJobTask:
diff --git a/tests/unit/hooks/test_task_definition_integration.py b/tests/unit/hooks/test_task_definition_integration.py
index cdf7ea95..396c9a1c 100644
--- a/tests/unit/hooks/test_task_definition_integration.py
+++ b/tests/unit/hooks/test_task_definition_integration.py
@@ -67,7 +67,7 @@ def test_build_hooks_from_json_valid(self) -> None:
                 "command": "echo starting"
             },
             {
-                "name": "Pre Hook 2", 
+                "name": "Pre Hook 2",
                 "command": "mkdir -p /tmp/backup"
             }
         ]"""
@@ -118,7 +118,7 @@ async def test_build_task_list_with_hooks(self, mock_db: AsyncSession) -> None:
             repository_name="test-repo",
             include_backup=True,
             backup_params={
-                "source_path": "/data",
+                "source_paths": '["/data"]',
                 "compression": "zstd",
                 "dry_run": False,
                 "ignore_lock": False,
diff --git a/tests/unit/jobs/test_backup_task_executor.py b/tests/unit/jobs/test_backup_task_executor.py
new file mode 100644
index 00000000..02c9a2d5
--- /dev/null
+++ b/tests/unit/jobs/test_backup_task_executor.py
@@ -0,0 +1,298 @@
+"""
+Tests for BackupTaskExecutor - specifically multi-source-path handling.
+
+Verifies that single paths, JSON array paths, and legacy plain strings
+are all correctly expanded into borg create command arguments.
+"""
+
+import uuid
+from unittest.mock import Mock, AsyncMock, patch
+
+from borgitory.services.jobs.task_executors.backup_task_executor import (
+    BackupTaskExecutor,
+)
+from borgitory.services.jobs.job_models import (
+    BorgJob,
+    BorgJobTask,
+    TaskTypeEnum,
+)
+from borgitory.models.job_results import JobStatusEnum
+from borgitory.protocols.command_protocols import ProcessResult
+from borgitory.utils.datetime_utils import now_utc
+from borgitory.utils.security import BorgCommandResult
+
+
+def _make_executor() -> tuple[BackupTaskExecutor, Mock, Mock, Mock, Mock]:
+    job_executor = Mock()
+    job_executor.start_process = AsyncMock()
+    job_executor.monitor_process_output = AsyncMock()
+
+    output_manager = Mock()
+    output_manager.add_output_line = AsyncMock()
+
+    event_broadcaster = Mock()
+    event_broadcaster.broadcast_event = Mock()
+
+    database_manager = Mock()
+    database_manager.get_repository_data = AsyncMock()
+
+    executor = BackupTaskExecutor(
+        job_executor=job_executor,
+        output_manager=output_manager,
+        event_broadcaster=event_broadcaster,
+        database_manager=database_manager,
+    )
+    return executor, job_executor, output_manager, event_broadcaster, database_manager
+
+
+def _make_job_and_task(
+    source_paths: list[str],
+    archive_name: str = "test-archive",
+    repo_id: int = 1,
+) -> tuple[BorgJob, BorgJobTask]:
+    task = BorgJobTask(
+        task_type=TaskTypeEnum.BACKUP,
+        task_name="Test Backup",
+        parameters={
+            "source_paths": source_paths,
+            "archive_name": archive_name,
+        },
+    )
+    job = BorgJob(
+        id=uuid.uuid4(),
+        job_type="composite",
+        status=JobStatusEnum.RUNNING,
+        started_at=now_utc(),
+        tasks=[task],
+        repository_id=repo_id,
+    )
+    return job, task
+
+
+def _setup_successful_run(
+    job_executor: Mock,
+    database_manager: Mock,
+) -> None:
+    database_manager.get_repository_data.return_value = {
+        "id": 1,
+        "path": "/tmp/test-repo",
+        "passphrase": "test-pass",
+    }
+
+    mock_process = AsyncMock()
+    mock_process.pid = 12345
+    job_executor.start_process.return_value = mock_process
+
+    job_executor.monitor_process_output.return_value = ProcessResult(
+        return_code=0,
+        stdout=b"Archive created successfully",
+        stderr=b"",
+        error=None,
+    )
+
+
+class TestBackupTaskExecutorSourcePaths:
+    """Verify that source_path values are correctly expanded in the borg command."""
+
+    async def test_multiple_paths_json_array(self) -> None:
+        executor, job_executor, _, _, database_manager = _make_executor()
+        _setup_successful_run(job_executor, database_manager)
+        job, task = _make_job_and_task(
+            source_paths=["/home/user/src", "/home/user/Documents"]
+        )
+        captured_args: list[list[str]] = []
+
+        with patch(
+            "borgitory.services.jobs.task_executors.backup_task_executor.create_borg_command"
+        ) as mock_cmd:
+            mock_cmd.side_effect = lambda **kwargs: (
+                captured_args.append(kwargs.get("additional_args", [])),
+                BorgCommandResult(command=["borg", "create"], environment={}),
+            )[1]
+
+            result = await executor.execute_backup_task(job, task)
+
+        assert result is True
+        args = captured_args[0]
+        repo_archive_idx = next(
+            i for i, a in enumerate(args) if "test-repo::test-archive" in a
+        )
+        path_args = args[repo_archive_idx + 1 :]
+        assert path_args == ["/home/user/src", "/home/user/Documents"]
+
+    async def test_three_paths(self) -> None:
+        executor, job_executor, _, _, database_manager = _make_executor()
+        _setup_successful_run(job_executor, database_manager)
+        job, task = _make_job_and_task(
+            source_paths=["/appdata/app1", "/appdata/app2", "/appdata/app3"]
+        )
+        captured_args: list[list[str]] = []
+
+        with patch(
+            "borgitory.services.jobs.task_executors.backup_task_executor.create_borg_command"
+        ) as mock_cmd:
+            mock_cmd.side_effect = lambda **kwargs: (
+                captured_args.append(kwargs.get("additional_args", [])),
+                BorgCommandResult(command=["borg", "create"], environment={}),
+            )[1]
+
+            await executor.execute_backup_task(job, task)
+
+        args = captured_args[0]
+        repo_archive_idx = next(
+            i for i, a in enumerate(args) if "test-repo::test-archive" in a
+        )
+        path_args = args[repo_archive_idx + 1 :]
+        assert path_args == ["/appdata/app1", "/appdata/app2", "/appdata/app3"]
+
+    async def test_single_path_json_array(self) -> None:
+        executor, job_executor, _, _, database_manager = _make_executor()
+        _setup_successful_run(job_executor, database_manager)
+        job, task = _make_job_and_task(source_paths=["/data"])
+        captured_args: list[list[str]] = []
+
+        with patch(
+            "borgitory.services.jobs.task_executors.backup_task_executor.create_borg_command"
+        ) as mock_cmd:
+            mock_cmd.side_effect = lambda **kwargs: (
+                captured_args.append(kwargs.get("additional_args", [])),
+                BorgCommandResult(command=["borg", "create"], environment={}),
+            )[1]
+
+            await executor.execute_backup_task(job, task)
+
+        args = captured_args[0]
+        repo_archive_idx = next(
+            i for i, a in enumerate(args) if "test-repo::test-archive" in a
+        )
+        path_args = args[repo_archive_idx + 1 :]
+        assert path_args == ["/data"]
+
+    async def test_empty_source_path_appends_nothing(self) -> None:
+        executor, job_executor, _, _, database_manager = _make_executor()
+        _setup_successful_run(job_executor, database_manager)
+        job, task = _make_job_and_task(source_paths=[])
+        task.parameters["source_paths"] = []
+        captured_args: list[list[str]] = []
+
+        with patch(
+            "borgitory.services.jobs.task_executors.backup_task_executor.create_borg_command"
+        ) as mock_cmd:
+            mock_cmd.side_effect = lambda **kwargs: (
+                captured_args.append(kwargs.get("additional_args", [])),
+                BorgCommandResult(command=["borg", "create"], environment={}),
+            )[1]
+
+            await executor.execute_backup_task(job, task)
+
+        args = captured_args[0]
+        repo_archive_idx = next(
+            i for i, a in enumerate(args) if "test-repo::test-archive" in a
+        )
+        path_args = args[repo_archive_idx + 1 :]
+        assert path_args == []
+
+    async def test_no_source_path_key_appends_nothing(self) -> None:
+        executor, job_executor, _, _, database_manager = _make_executor()
+        _setup_successful_run(job_executor, database_manager)
+        task = BorgJobTask(
+            task_type=TaskTypeEnum.BACKUP,
+            task_name="Test Backup",
+            parameters={"archive_name": "test-archive"},
+        )
+        job = BorgJob(
+            id=uuid.uuid4(),
+            job_type="composite",
+            status=JobStatusEnum.RUNNING,
+            started_at=now_utc(),
+            tasks=[task],
+            repository_id=1,
+        )
+        captured_args: list[list[str]] = []
+
+        with patch(
+            "borgitory.services.jobs.task_executors.backup_task_executor.create_borg_command"
+        ) as mock_cmd:
+            mock_cmd.side_effect = lambda **kwargs: (
+                captured_args.append(kwargs.get("additional_args", [])),
+                BorgCommandResult(command=["borg", "create"], environment={}),
+            )[1]
+
+            await executor.execute_backup_task(job, task)
+
+        args = captured_args[0]
+        repo_archive_idx = next(
+            i for i, a in enumerate(args) if "test-repo::test-archive" in a
+        )
+        path_args = args[repo_archive_idx + 1 :]
+        assert path_args == []
+
+    async def test_paths_appear_after_archive_argument(self) -> None:
+        """borg create requires paths AFTER repo::archive."""
+        executor, job_executor, _, _, database_manager = _make_executor()
+        _setup_successful_run(job_executor, database_manager)
+        job, task = _make_job_and_task(
+            source_paths=["/src", "/docs"]
+        )
+        captured_args: list[list[str]] = []
+
+        with patch(
+            "borgitory.services.jobs.task_executors.backup_task_executor.create_borg_command"
+        ) as mock_cmd:
+            mock_cmd.side_effect = lambda **kwargs: (
+                captured_args.append(kwargs.get("additional_args", [])),
+                BorgCommandResult(command=["borg", "create"], environment={}),
+            )[1]
+
+            await executor.execute_backup_task(job, task)
+
+        args = captured_args[0]
+        archive_idx = next(
+            i for i, a in enumerate(args) if "::" in a
+        )
+        assert args[archive_idx + 1] == "/src"
+        assert args[archive_idx + 2] == "/docs"
+        assert len(args) == archive_idx + 3
+
+    async def test_paths_with_patterns_and_dry_run(self) -> None:
+        """Multiple paths should coexist with --pattern and --dry-run flags."""
+        executor, job_executor, _, _, database_manager = _make_executor()
+        _setup_successful_run(job_executor, database_manager)
+        task = BorgJobTask(
+            task_type=TaskTypeEnum.BACKUP,
+            task_name="Test Backup",
+            parameters={
+                "source_paths": ["/data", "/backup"],
+                "archive_name": "test-archive",
+                "patterns": ["+*.txt", "-*.log"],
+                "dry_run": True,
+            },
+        )
+        job = BorgJob(
+            id=uuid.uuid4(),
+            job_type="composite",
+            status=JobStatusEnum.RUNNING,
+            started_at=now_utc(),
+            tasks=[task],
+            repository_id=1,
+        )
+        captured_args: list[list[str]] = []
+
+        with patch(
+            "borgitory.services.jobs.task_executors.backup_task_executor.create_borg_command"
+        ) as mock_cmd:
+            mock_cmd.side_effect = lambda **kwargs: (
+                captured_args.append(kwargs.get("additional_args", [])),
+                BorgCommandResult(command=["borg", "create"], environment={}),
+            )[1]
+
+            await executor.execute_backup_task(job, task)
+
+        args = captured_args[0]
+        assert "--pattern=+*.txt" in args
+        assert "--pattern=-*.log" in args
+        assert "--dry-run" in args
+
+        archive_idx = next(i for i, a in enumerate(args) if "::" in a)
+        path_args = args[archive_idx + 1 :]
+        assert path_args == ["/data", "/backup"]
diff --git a/tests/unit/jobs/test_ignore_lock_functionality.py b/tests/unit/jobs/test_ignore_lock_functionality.py
index df370302..4a92263d 100644
--- a/tests/unit/jobs/test_ignore_lock_functionality.py
+++ b/tests/unit/jobs/test_ignore_lock_functionality.py
@@ -82,7 +82,7 @@ def mock_backup_task_with_ignore_lock(self) -> BorgJobTask:
             task_type=TaskTypeEnum.BACKUP,
             task_name="Test Backup with Ignore Lock",
             parameters={
-                "source_path": "/test/source",
+                "source_paths": '["/test/source"]',
                 "compression": "zstd",
                 "dry_run": False,
                 "ignore_lock": True,
@@ -99,7 +99,7 @@ def mock_backup_task_without_ignore_lock(self) -> BorgJobTask:
             task_type=TaskTypeEnum.BACKUP,
             task_name="Test Backup without Ignore Lock",
             parameters={
-                "source_path": "/test/source",
+                "source_paths": '["/test/source"]',
                 "compression": "zstd",
                 "dry_run": False,
                 "ignore_lock": False,
diff --git a/tests/unit/jobs/test_job_manager.py b/tests/unit/jobs/test_job_manager.py
index d808c23a..d4aa4c49 100644
--- a/tests/unit/jobs/test_job_manager.py
+++ b/tests/unit/jobs/test_job_manager.py
@@ -235,12 +235,12 @@ def test_create_job_task(self, job_manager: JobManager) -> None:
         task = BorgJobTask(
             task_type=TaskTypeEnum.BACKUP,
             task_name="Test Backup",
-            parameters={"source_path": "/data"},
+            parameters={"source_paths": '["/data"]'},
         )
 
         assert task.task_type == "backup"
         assert task.task_name == "Test Backup"
-        assert task.parameters["source_path"] == "/data"
+        assert task.parameters["source_paths"] == '["/data"]'
 
     def test_create_job(self, job_manager: JobManager) -> None:
         """Test job creation"""
diff --git a/tests/unit/jobs/test_job_manager_comprehensive.py b/tests/unit/jobs/test_job_manager_comprehensive.py
index 87db3f54..b88f7748 100644
--- a/tests/unit/jobs/test_job_manager_comprehensive.py
+++ b/tests/unit/jobs/test_job_manager_comprehensive.py
@@ -393,7 +393,7 @@ async def test_execute_composite_job_critical_failure(
                 type=TaskTypeEnum.BACKUP,
                 name="Test Backup",
                 parameters={
-                    "source_path": "/tmp/test",
+                    "source_paths": '["/tmp/test"]',
                     "compression": "lz4",
                     "dry_run": False,
                 },
@@ -698,10 +698,10 @@ async def test_execute_backup_task_with_dry_run(
             task_type=TaskTypeEnum.BACKUP,
             task_name="Test Backup Dry Run",
             parameters={
-                "source_path": "/tmp",
+                "source_paths": '["/tmp"]',
                 "excludes": ["*.log"],
                 "archive_name": "test-archive-dry",
-                "dry_run": True,  # This is the key parameter we're testing
+                "dry_run": True,
             },
         )
 
diff --git a/tests/unit/jobs/test_job_manager_task_execution.py b/tests/unit/jobs/test_job_manager_task_execution.py
index 0a0db801..76ce66ca 100644
--- a/tests/unit/jobs/test_job_manager_task_execution.py
+++ b/tests/unit/jobs/test_job_manager_task_execution.py
@@ -288,7 +288,7 @@ async def test_execute_composite_job_critical_failure(
                 type=TaskTypeEnum.BACKUP,
                 name="Test Backup",
                 parameters={
-                    "source_path": "/tmp/test",
+                    "source_paths": '["/tmp/test"]',
                     "compression": "lz4",
                     "dry_run": False,
                 },
@@ -592,7 +592,7 @@ async def test_execute_backup_task_with_dry_run(
             task_type=TaskTypeEnum.BACKUP,
             task_name="Test Backup Dry Run",
             parameters={
-                "source_path": "/tmp",
+                "source_paths": '["/tmp"]',
                 "excludes": ["*.log"],
                 "archive_name": "test-archive-dry",
                 "dry_run": True,  # This is the key parameter we're testing
diff --git a/tests/unit/jobs/test_job_service.py b/tests/unit/jobs/test_job_service.py
index b555f09b..9ddbb436 100644
--- a/tests/unit/jobs/test_job_service.py
+++ b/tests/unit/jobs/test_job_service.py
@@ -61,7 +61,7 @@ async def test_create_backup_job_simple(self, test_db: AsyncSession) -> None:
 
         backup_request = BackupRequest(
             repository_id=repository.id,
-            source_path="/data",
+            source_paths='["/data"]',
             compression=CompressionType.LZ4,
             dry_run=False,
             cloud_sync_config_id=None,
@@ -108,7 +108,7 @@ async def test_create_backup_job_with_prune(self, test_db: AsyncSession) -> None
 
         backup_request = BackupRequest(
             repository_id=repository.id,
-            source_path="/data",
+            source_paths='["/data"]',
             compression=CompressionType.LZ4,
             dry_run=False,
             prune_config_id=prune_config.id,
@@ -139,7 +139,7 @@ async def test_create_backup_job_repository_not_found(
         """Test backup job creation with non-existent repository."""
         backup_request = BackupRequest(
             repository_id=999,
-            source_path="/data",
+            source_paths='["/data"]',
             compression=CompressionType.LZ4,
             dry_run=False,
             cloud_sync_config_id=None,
diff --git a/tests/unit/schedules/test_next_run_update.py b/tests/unit/schedules/test_next_run_update.py
index 603911c5..c823d708 100644
--- a/tests/unit/schedules/test_next_run_update.py
+++ b/tests/unit/schedules/test_next_run_update.py
@@ -22,7 +22,7 @@ def mock_schedule(self) -> Mock:
         schedule.id = 1
         schedule.name = "Test Schedule"
         schedule.repository_id = 10
-        schedule.source_path = "/data/backup"
+        schedule.source_paths = ["/data/backup"]
         schedule.cloud_sync_config_id = None
         schedule.prune_config_id = None
         schedule.check_config_id = None
diff --git a/tests/unit/schedules/test_schedule_service.py b/tests/unit/schedules/test_schedule_service.py
index 070f73d5..ebcf8071 100644
--- a/tests/unit/schedules/test_schedule_service.py
+++ b/tests/unit/schedules/test_schedule_service.py
@@ -69,7 +69,7 @@ async def test_get_schedule_by_id_success(
         schedule.name = "test-schedule"
         schedule.repository_id = sample_repository.id
         schedule.cron_expression = "0 2 * * *"
-        schedule.source_path = "/data"
+        schedule.source_paths = '["/data"]'
         test_db.add(schedule)
         await test_db.commit()
         await test_db.refresh(schedule)
@@ -104,13 +104,13 @@ async def test_get_schedules_with_data(
         schedule1.name = "schedule-1"
         schedule1.repository_id = sample_repository.id
         schedule1.cron_expression = "0 2 * * *"
-        schedule1.source_path = "/data1"
+        schedule1.source_paths = '["/data1"]'
 
         schedule2 = Schedule()
         schedule2.name = "schedule-2"
         schedule2.repository_id = sample_repository.id
         schedule2.cron_expression = "0 3 * * *"
-        schedule2.source_path = "/data2"
+        schedule2.source_paths = '["/data2"]'
         test_db.add(schedule1)
         test_db.add(schedule2)
         await test_db.commit()
@@ -133,7 +133,7 @@ async def test_get_schedules_with_pagination(
             schedule.name = f"schedule-{i}"
             schedule.repository_id = sample_repository.id
             schedule.cron_expression = "0 2 * * *"
-            schedule.source_path = f"/data{i}"
+            schedule.source_paths = f'["/data{i}"]'
             test_db.add(schedule)
         await test_db.commit()
 
@@ -151,7 +151,7 @@ async def test_get_all_schedules(
         schedule.name = "test-schedule"
         schedule.repository_id = sample_repository.id
         schedule.cron_expression = "0 2 * * *"
-        schedule.source_path = "/data"
+        schedule.source_paths = '["/data"]'
         test_db.add(schedule)
         await test_db.commit()
 
@@ -172,7 +172,7 @@ async def test_create_schedule_success(
             name="new-schedule",
             repository_id=sample_repository.id,
             cron_expression="0 2 * * *",
-            source_path="/backup",
+            source_paths='["/backup"]',
         )
 
         assert result.success is True
@@ -202,7 +202,7 @@ async def test_create_schedule_repository_not_found(
             name="test-schedule",
             repository_id=999,
             cron_expression="0 2 * * *",
-            source_path="/data",
+            source_paths='["/data"]',
         )
 
         assert result.success is False
@@ -222,7 +222,7 @@ async def test_create_schedule_invalid_cron(
             name="test-schedule",
             repository_id=sample_repository.id,
             cron_expression="invalid cron",
-            source_path="/data",
+            source_paths='["/data"]',
         )
 
         assert result.success is False
@@ -245,7 +245,7 @@ async def test_create_schedule_scheduler_failure(
             name="test-schedule",
             repository_id=sample_repository.id,
             cron_expression="0 2 * * *",
-            source_path="/data",
+            source_paths='["/data"]',
         )
 
         assert result.success is False
@@ -273,7 +273,7 @@ async def test_update_schedule_success(
         schedule.name = "original-name"
         schedule.repository_id = sample_repository.id
         schedule.cron_expression = "0 2 * * *"
-        schedule.source_path = "/data"
+        schedule.source_paths = '["/data"]'
         schedule.enabled = True
         test_db.add(schedule)
         await test_db.commit()
@@ -315,7 +315,7 @@ async def test_toggle_schedule_enable(
         schedule.name = "test-schedule"
         schedule.repository_id = sample_repository.id
         schedule.cron_expression = "0 2 * * *"
-        schedule.source_path = "/data"
+        schedule.source_paths = '["/data"]'
         schedule.enabled = False
         test_db.add(schedule)
         await test_db.commit()
@@ -343,7 +343,7 @@ async def test_toggle_schedule_disable(
         schedule.name = "test-schedule"
         schedule.repository_id = sample_repository.id
         schedule.cron_expression = "0 2 * * *"
-        schedule.source_path = "/data"
+        schedule.source_paths = '["/data"]'
         schedule.enabled = True
         test_db.add(schedule)
         await test_db.commit()
@@ -382,7 +382,7 @@ async def test_toggle_schedule_scheduler_error(
         schedule.name = "test-schedule"
         schedule.repository_id = sample_repository.id
         schedule.cron_expression = "0 2 * * *"
-        schedule.source_path = "/data"
+        schedule.source_paths = '["/data"]'
         schedule.enabled = False
         test_db.add(schedule)
         await test_db.commit()
@@ -411,7 +411,7 @@ async def test_delete_schedule_success(
         schedule.name = "test-schedule"
         schedule.repository_id = sample_repository.id
         schedule.cron_expression = "0 2 * * *"
-        schedule.source_path = "/data"
+        schedule.source_paths = '["/data"]'
         test_db.add(schedule)
         await test_db.commit()
         await test_db.refresh(schedule)
@@ -456,7 +456,7 @@ async def test_delete_schedule_scheduler_error(
         schedule.name = "test-schedule"
         schedule.repository_id = sample_repository.id
         schedule.cron_expression = "0 2 * * *"
-        schedule.source_path = "/data"
+        schedule.source_paths = '["/data"]'
         test_db.add(schedule)
         await test_db.commit()
         await test_db.refresh(schedule)
@@ -486,7 +486,7 @@ async def test_schedule_lifecycle(
             name="lifecycle-test",
             repository_id=sample_repository.id,
             cron_expression="0 2 * * *",
-            source_path="/data",
+            source_paths='["/data"]',
         )
         assert result.success is True
         assert result.schedule is not None
diff --git a/tests/unit/schedules/test_schedule_validation_service.py b/tests/unit/schedules/test_schedule_validation_service.py
index 288f9251..e3ca721b 100644
--- a/tests/unit/schedules/test_schedule_validation_service.py
+++ b/tests/unit/schedules/test_schedule_validation_service.py
@@ -31,7 +31,7 @@ def test_validate_schedule_creation_data_valid_input(
             "name": "Daily Backup",
             "repository_id": "1",
             "cron_expression": "0 2 * * *",
-            "source_path": "/data",
+            "source_paths": ["/data"],
             "cloud_sync_config_id": "2",
             "prune_config_id": "3",
             "notification_config_id": "4",
@@ -47,7 +47,7 @@ def test_validate_schedule_creation_data_valid_input(
             "name": "Daily Backup",
             "repository_id": 1,
             "cron_expression": "0 2 * * *",
-            "source_path": "/data",
+            "source_paths": ["/data"],
             "cloud_sync_config_id": 2,
             "prune_config_id": 3,
             "notification_config_id": 4,
@@ -64,7 +64,7 @@ def test_validate_schedule_creation_data_minimal_valid_input(
             "name": "Test Schedule",
             "repository_id": "1",
             "cron_expression": "*/5 * * * *",
-            "source_path": "",
+            "source_paths": [],
             "cloud_sync_config_id": "",
             "prune_config_id": "",
             "notification_config_id": "",
@@ -80,7 +80,7 @@ def test_validate_schedule_creation_data_minimal_valid_input(
             "name": "Test Schedule",
             "repository_id": 1,
             "cron_expression": "*/5 * * * *",
-            "source_path": "",
+            "source_paths": [],
             "cloud_sync_config_id": None,
             "prune_config_id": None,
             "notification_config_id": None,
@@ -276,6 +276,114 @@ def test_validate_schedule_creation_data_optional_field_type_conversion(
             assert is_valid is True
             assert processed_data["cloud_sync_config_id"] == expected_output
 
+    def test_validate_source_paths_valid(
+        self, schedule_service: ScheduleService
+    ) -> None:
+        """Test that valid absolute source_paths are accepted and serialized."""
+        data = {
+            "name": "Test",
+            "repository_id": "1",
+            "cron_expression": "0 2 * * *",
+            "source_paths": ["/data", "/backup"],
+        }
+
+        is_valid, processed_data, error_msg = (
+            schedule_service.validate_schedule_creation_data(data)
+        )
+
+        assert is_valid is True
+        assert error_msg is None
+        assert processed_data["source_paths"] == ["/data", "/backup"]
+
+    def test_validate_source_paths_empty_after_filtering(
+        self, schedule_service: ScheduleService
+    ) -> None:
+        """All-empty source_paths are passed through as-is."""
+        data = {
+            "name": "Test",
+            "repository_id": "1",
+            "cron_expression": "0 2 * * *",
+            "source_paths": ["", "  "],
+        }
+
+        is_valid, processed_data, error_msg = (
+            schedule_service.validate_schedule_creation_data(data)
+        )
+
+        assert is_valid is True
+        assert processed_data["source_paths"] == ["", "  "]
+
+    def test_validate_source_paths_relative_passed_through(
+        self, schedule_service: ScheduleService
+    ) -> None:
+        """Relative paths are passed through as-is (validation happens elsewhere)."""
+        data = {
+            "name": "Test",
+            "repository_id": "1",
+            "cron_expression": "0 2 * * *",
+            "source_paths": ["relative/path"],
+        }
+
+        is_valid, processed_data, error_msg = (
+            schedule_service.validate_schedule_creation_data(data)
+        )
+
+        assert is_valid is True
+        assert processed_data["source_paths"] == ["relative/path"]
+
+    def test_validate_source_paths_mixed_passed_through(
+        self, schedule_service: ScheduleService
+    ) -> None:
+        """A mix of absolute and relative paths are passed through as-is."""
+        data = {
+            "name": "Test",
+            "repository_id": "1",
+            "cron_expression": "0 2 * * *",
+            "source_paths": ["/valid", "bad"],
+        }
+
+        is_valid, processed_data, error_msg = (
+            schedule_service.validate_schedule_creation_data(data)
+        )
+
+        assert is_valid is True
+        assert processed_data["source_paths"] == ["/valid", "bad"]
+
+    def test_validate_source_paths_strips_whitespace(
+        self, schedule_service: ScheduleService
+    ) -> None:
+        data = {
+            "name": "Test",
+            "repository_id": "1",
+            "cron_expression": "0 2 * * *",
+            "source_paths": ["  /data  "],
+        }
+
+        is_valid, processed_data, _ = (
+            schedule_service.validate_schedule_creation_data(data)
+        )
+
+        assert is_valid is True
+        assert processed_data["source_paths"] == ["  /data  "]
+
+    def test_validate_source_paths_string_fallback(
+        self, schedule_service: ScheduleService
+    ) -> None:
+        """When source_paths is absent, source_paths string is used as-is."""
+        data = {
+            "name": "Test",
+            "repository_id": "1",
+            "cron_expression": "0 2 * * *",
+            "source_paths": "/legacy",
+        }
+
+        is_valid, processed_data, _ = (
+            schedule_service.validate_schedule_creation_data(data)
+        )
+
+        assert is_valid is True
+        assert processed_data["source_paths"] == "/legacy"
+
     def test_validate_cron_expression_valid(
         self, schedule_service: ScheduleService
     ) -> None:
diff --git a/tests/unit/test_task_definition_builder.py b/tests/unit/test_task_definition_builder.py
index 1b31ce75..8fff318e 100644
--- a/tests/unit/test_task_definition_builder.py
+++ b/tests/unit/test_task_definition_builder.py
@@ -108,7 +108,7 @@ def test_build_backup_task_defaults(
             type=TaskTypeEnum.BACKUP,
             name="Backup test-repo",
             parameters={
-                "source_path": "/data",
+                "source_paths": [],
                 "compression": "zstd",
                 "dry_run": False,
                 "ignore_lock": False,
@@ -122,14 +122,14 @@ def test_build_backup_task_custom_params(
     ) -> None:
         """Test building backup task with custom parameters"""
         task = task_builder.build_backup_task(
-            "custom-repo", source_path="/custom/path", compression="lz4", dry_run=True
+            "custom-repo", source_paths=["/custom/path"], compression="lz4", dry_run=True
         )
 
         expected = TaskDefinition(
             type=TaskTypeEnum.BACKUP,
             name="Backup custom-repo",
             parameters={
-                "source_path": "/custom/path",
+                "source_paths": ["/custom/path"],
                 "compression": "lz4",
                 "dry_run": True,
                 "ignore_lock": False,
@@ -479,7 +479,7 @@ def mock_execute_side_effect(*args: Any, **kwargs: Any) -> MagicMock:
             mock_db,
             repository_name="test-repo",
             include_backup=True,
-            backup_params={"source_path": "/custom", "compression": "lz4"},
+            backup_params={"source_paths": ['/custom'], "compression": "lz4"},
             prune_config_id=1,
             check_config_id=1,
             include_cloud_sync=True,
@@ -500,7 +500,7 @@ def mock_execute_side_effect(*args: Any, **kwargs: Any) -> MagicMock:
 
         # Verify backup task uses custom params
         backup_task = next(task for task in tasks if task.type == TaskTypeEnum.BACKUP)
-        assert backup_task.parameters["source_path"] == "/custom"
+        assert backup_task.parameters["source_paths"] == ["/custom"]
         assert backup_task.parameters["compression"] == "lz4"
 
     async def test_build_task_list_minimal(
diff --git a/tests/utils/test_source_paths.py b/tests/utils/test_source_paths.py
new file mode 100644
index 00000000..94cc8421
--- /dev/null
+++ b/tests/utils/test_source_paths.py
@@ -0,0 +1,72 @@
+"""Tests for source_paths utility functions."""
+
+import json
+from borgitory.utils.source_paths import parse_source_paths, serialize_source_paths
+
+
+class TestParseSourcePaths:
+    def test_single_path_json_array(self):
+        assert parse_source_paths('["/data"]') == ["/data"]
+
+    def test_multiple_paths_json_array(self):
+        result = parse_source_paths('["/home/user/src", "/home/user/Documents"]')
+        assert result == ["/home/user/src", "/home/user/Documents"]
+
+    def test_empty_string(self):
+        assert parse_source_paths("") == []
+
+    def test_whitespace_only(self):
+        assert parse_source_paths("   ") == []
+
+    def test_empty_json_array(self):
+        assert parse_source_paths("[]") == []
+
+    def test_filters_empty_strings_in_array(self):
+        assert parse_source_paths('["/data", "", "  "]') == ["/data"]
+
+    def test_invalid_json_starting_with_bracket(self):
+        assert parse_source_paths("[not json") == []
+
+    def test_json_array_with_non_string_elements(self):
+        assert parse_source_paths('["/data", 123]') == ["/data"]
+
+    def test_three_paths(self):
+        paths = '["/appdata/app1", "/appdata/app2", "/appdata/app3"]'
+        result = parse_source_paths(paths)
+        assert result == ["/appdata/app1", "/appdata/app2", "/appdata/app3"]
+
+    def test_relative_paths_in_array_returned_as_is(self):
+        result = parse_source_paths('["/valid", "relative", "/also-valid"]')
+        assert result == ["/valid", "relative", "/also-valid"]
+
+    def test_non_json_input_returns_empty_list(self):
+        assert parse_source_paths("/data") == []
+        assert parse_source_paths("plain string") == []
+        assert parse_source_paths("relative/path") == []
+
+
+class TestSerializeSourcePaths:
+    def test_single_path(self):
+        result = serialize_source_paths(["/data"])
+        assert json.loads(result) == ["/data"]
+
+    def test_multiple_paths(self):
+        result = serialize_source_paths(["/src", "/Documents"])
+        assert json.loads(result) == ["/src", "/Documents"]
+
+    def test_empty_list(self):
+        assert serialize_source_paths([]) == "[]"
+
+    def test_filters_empty_strings(self):
+        result = serialize_source_paths(["/data", "", "  "])
+        assert json.loads(result) == ["/data"]
+
+    def test_strips_whitespace(self):
+        result = serialize_source_paths(["  /data  "])
+        assert json.loads(result) == ["/data"]
+
+    def test_roundtrip(self):
+        original = ["/home/user/src", "/home/user/Documents", "/backups"]
+        serialized = serialize_source_paths(original)
+        parsed = parse_source_paths(serialized)
+        assert parsed == original