diff --git a/docs/control-plane/ui-api-guide.md b/docs/control-plane/ui-api-guide.md
index 93e40c1b..67b16faf 100644
--- a/docs/control-plane/ui-api-guide.md
+++ b/docs/control-plane/ui-api-guide.md
@@ -161,7 +161,7 @@ accepts repeated `schema_type` (top-level path segment, e.g. `meta`).
 | PUT | `/sources/{name}` | Update source |
 | PATCH | `/sources/{name}` | Set lifecycle state (`state: active\|dormant\|disabled`, or compat `enabled`); no new version; `409 match_conflict` if activating duplicates another non-disabled source's match |
 | DELETE | `/sources/{name}` | Delete source |
-| POST | `/sources/bulk` | Bulk action (`enable` / `disable` / `dormant` / `delete`) on named sources |
+| POST | `/sources/bulk` | Bulk action (`enable` / `disable` / `dormant` / `delete` / `deploy`) on named sources |
 | POST | `/sources/seed` | Seed defaults |
 
 The landing source is engine-owned: it reads as `resource_type: core`, and POST,
diff --git a/docs/data-plane/source-registry.md b/docs/data-plane/source-registry.md
index 9ee0e18e..a3d4a39e 100644
--- a/docs/data-plane/source-registry.md
+++ b/docs/data-plane/source-registry.md
@@ -125,7 +125,7 @@ POST   /api/v1/sources/{name}/deploy    # Apply plan DDL to ClickHouse, set depl
 PUT    /api/v1/sources/{name}           # Update source (bumps major version when a deployed pin changes)
 PATCH  /api/v1/sources/{name}           # Set lifecycle state (state or enabled) - no new version
 DELETE /api/v1/sources/{name}           # Delete the definition
-POST   /api/v1/sources/bulk             # Bulk action: enable | disable | dormant | delete
+POST   /api/v1/sources/bulk             # Bulk action: enable | disable | dormant | delete | deploy
 POST   /api/v1/sources/seed             # Seed built-in sources (non-destructive)
 ```
 
diff --git a/openapi-spec/openapi.json b/openapi-spec/openapi.json
index 6856b8f6..46ae163e 100644
--- a/openapi-spec/openapi.json
+++ b/openapi-spec/openapi.json
@@ -5793,7 +5793,7 @@
           "Sources"
         ],
         "summary": "Bulk Action",
-        "description": "Perform a bulk action (enable, disable, dormant, delete) on multiple sources.",
+        "description": "Perform a bulk action on multiple sources.\n\n``enable`` / ``disable`` / ``dormant`` / ``delete`` apply immediately.\n``deploy`` submits one ``source:deploy`` background task per accepted source\n(poll ``GET /tasks/{task_id}``); refused names land in ``failed``.",
         "operationId": "bulk_action_api_v1_sources_bulk_post",
         "requestBody": {
           "content": {
@@ -24039,6 +24039,33 @@
         ],
         "title": "AppSummary"
       },
+      "ApplyDefaultsPin": {
+        "properties": {
+          "source": {
+            "type": "string",
+            "title": "Source",
+            "description": "Source name (_source label)"
+          },
+          "version": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Version",
+            "description": "Version id that was pinned or already matched; null when unset"
+          }
+        },
+        "type": "object",
+        "required": [
+          "source"
+        ],
+        "title": "ApplyDefaultsPin",
+        "description": "One source whose current version was considered for the table defaults."
+      },
       "ApplyDefaultsRequest": {
         "properties": {
           "sources": {
@@ -24062,17 +24089,19 @@
         "properties": {
           "updated": {
             "items": {
-              "type": "string"
+              "$ref": "#/components/schemas/ApplyDefaultsPin"
             },
             "type": "array",
-            "title": "Updated"
+            "title": "Updated",
+            "description": "Sources whose current version was written with the defaults"
           },
           "unchanged": {
             "items": {
-              "type": "string"
+              "$ref": "#/components/schemas/ApplyDefaultsPin"
             },
             "type": "array",
-            "title": "Unchanged"
+            "title": "Unchanged",
+            "description": "Sources whose current version already matched the defaults"
           },
           "live": {
             "items": {
@@ -24454,20 +24483,54 @@
         ],
         "title": "Body_elastic_converter_api_v1_schemas_elastic_converter_post"
       },
+      "BulkActionItem": {
+        "properties": {
+          "source": {
+            "type": "string",
+            "title": "Source",
+            "description": "Source name (_source label)"
+          },
+          "version": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Version",
+            "description": "Deploy only: version id to deploy. Omitted uses the working current. Ignored for enable/disable/dormant/delete."
+          }
+        },
+        "type": "object",
+        "required": [
+          "source"
+        ],
+        "title": "BulkActionItem",
+        "description": "One source in a bulk request."
+      },
       "BulkActionRequest": {
         "properties": {
           "action": {
             "type": "string",
             "title": "Action",
-            "description": "Action: enable, disable, dormant, delete"
+            "description": "Action: enable, disable, dormant, delete, deploy"
           },
           "sources": {
             "items": {
-              "type": "string"
+              "$ref": "#/components/schemas/BulkActionItem"
             },
             "type": "array",
+            "minItems": 1,
             "title": "Sources",
-            "description": "Source names to act on"
+            "description": "Sources to act on. Plain name strings are accepted."
+          },
+          "dry_run": {
+            "type": "boolean",
+            "title": "Dry Run",
+            "description": "Deploy only: plan/validate DDL without applying it. Ignored otherwise.",
+            "default": false
           }
         },
         "type": "object",
@@ -24501,6 +24564,14 @@
             "type": "array",
             "title": "Failed",
             "description": "One entry per source left untouched: its name, the same code the single-source route answers with, and the message"
+          },
+          "deployments": {
+            "items": {
+              "$ref": "#/components/schemas/BulkDeployTask"
+            },
+            "type": "array",
+            "title": "Deployments",
+            "description": "Deploy only: one ``source:deploy`` task per accepted source. Poll via GET /tasks/{task_id}."
           }
         },
         "type": "object",
@@ -24510,6 +24581,44 @@
         "title": "BulkActionResponse",
         "description": "Result of a bulk operation."
       },
+      "BulkDeployTask": {
+        "properties": {
+          "source": {
+            "type": "string",
+            "title": "Source"
+          },
+          "version": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Version",
+            "description": "Version requested; null means the working current at task start"
+          },
+          "task_id": {
+            "type": "string",
+            "title": "Task Id",
+            "description": "Task id for polling via GET /tasks/{task_id}"
+          },
+          "kind": {
+            "type": "string",
+            "title": "Kind",
+            "description": "Task kind",
+            "default": "source:deploy"
+          }
+        },
+        "type": "object",
+        "required": [
+          "source",
+          "task_id"
+        ],
+        "title": "BulkDeployTask",
+        "description": "One accepted deploy submitted as a background task."
+      },
       "BundleSchema-Input": {
         "properties": {
           "pins": {
diff --git a/src/dfe_engine/api/v1/sources.py b/src/dfe_engine/api/v1/sources.py
index 6757dac1..5e7af987 100644
--- a/src/dfe_engine/api/v1/sources.py
+++ b/src/dfe_engine/api/v1/sources.py
@@ -16,7 +16,7 @@
 PUT    /api/v1/sources/{name}           -> Update source
 PATCH  /api/v1/sources/{name}           -> Enable or disable source
 DELETE /api/v1/sources/{name}           -> Delete source
-POST   /api/v1/sources/bulk             -> Bulk enable/disable/delete
+POST   /api/v1/sources/bulk             -> Bulk enable/disable/dormant/delete/deploy
 POST   /api/v1/sources/seed             -> Seed built-in defaults
 POST   /api/v1/sources/reconcile-apps   -> Bring the derived app state into step with the sources
 
@@ -36,8 +36,8 @@
 from datetime import UTC, datetime
 from typing import Any, Literal, NamedTuple, NoReturn
 
-from fastapi import APIRouter, Depends, HTTPException, Query, Request
-from pydantic import BaseModel, ConfigDict, Field, model_validator
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 from scalo.concurrency import run_blocking
 from scalo.logger import logger
 
@@ -448,11 +448,52 @@ def target_state(self) -> SourceState:
         return "active" if self.enabled else "disabled"
 
 
+class BulkActionItem(BaseModel):
+    """One source in a bulk request."""
+
+    source: str = Field(description="Source name (_source label)")
+    version: str | None = Field(
+        default=None,
+        description=(
+            "Deploy only: version id to deploy. Omitted uses the working current. "
+            "Ignored for enable/disable/dormant/delete."
+        ),
+    )
+
+
 class BulkActionRequest(BaseModel):
     """Bulk operation on multiple sources."""
 
-    action: str = Field(description="Action: enable, disable, dormant, delete")
-    sources: list[str] = Field(description="Source names to act on")
+    action: str = Field(
+        description="Action: enable, disable, dormant, delete, deploy",
+    )
+    sources: list[BulkActionItem] = Field(
+        min_length=1,
+        description="Sources to act on. Plain name strings are accepted.",
+    )
+    dry_run: bool = Field(
+        default=False,
+        description="Deploy only: plan/validate DDL without applying it. Ignored otherwise.",
+    )
+
+    @field_validator("sources", mode="before")
+    @classmethod
+    def _coerce_source_names(cls, value: Any) -> Any:
+        if not isinstance(value, list):
+            return value
+        return [{"source": item} if isinstance(item, str) else item for item in value]
+
+
+class BulkDeployTask(BaseModel):
+    """One accepted deploy submitted as a background task."""
+
+    source: str
+    version: str | None = Field(
+        default=None,
+        description="Version requested; null means the working current at task start",
+    )
+    task_id: str = Field(description="Task id for polling via GET /tasks/{task_id}")
+    kind: str = Field(default="source:deploy", description="Task kind")
 
 
 class BulkActionResponse(BaseModel):
@@ -467,6 +508,13 @@ class BulkActionResponse(BaseModel):
             "single-source route answers with, and the message"
         ),
     )
+    deployments: list[BulkDeployTask] = Field(
+        default_factory=list,
+        description=(
+            "Deploy only: one ``source:deploy`` task per accepted source. "
+            "Poll via GET /tasks/{task_id}."
+        ),
+    )
 
 
 class SeedResponse(BaseModel):
@@ -1378,33 +1426,20 @@ def _remove_source_topics(source: Any, settings: Any) -> tuple[list[str], list[s
     return outcome.removed, [name for name, _ in outcome.failed]
 
 
-@router.post(
-    "/{name}/deploy",
-    response_model=SchemaDeployResult,
-    dependencies=[Depends(require_action(scopes_dict["source_write"]))],
-)
-async def deploy_source_schema(
+_SOURCE_DEPLOY_TASK_KIND = "source:deploy"
+
+
+async def _execute_source_deploy(
+    *,
     name: str,
-    user: CurrentUser,
-    registry: SourceReg,
-    settings: Settings,
+    user: Any,
+    registry: Any,
+    settings: Any,
     request: Request,
-    version: str | None = Query(
-        None, description="Source version id (defaults to the source's current version)"
-    ),
-    dry_run: bool = Query(
-        False, description="Plan only: generate + validate the DDL without applying it"
-    ),
+    version: str | None,
+    dry_run: bool,
 ) -> SchemaDeployResult:
-    """Plan (``dry_run=true``) or deploy a source version's schema to ClickHouse.
-
-    Runs the v2 YAML -> DDL pipeline. In plan mode the CREATE TABLE (+ any standard
-    views) and validation errors are returned for review WITHOUT touching
-    ClickHouse. In deploy mode the DDL is applied: an absent table is created, and
-    an existing one gains the columns it lacks and moves to the version's TTL, so a
-    re-deploy changes only what differs. The engine and the type of an existing
-    column are left as created. A schema that failed validation is never deployed.
-    """
+    """Plan or deploy one source version. Shared by the single and bulk deploy paths."""
     ddl_settings = await run_blocking(functools.partial(ttl_settings, request, settings))
     applied = await run_blocking(
         functools.partial(_apply_source_schema, name, registry, ddl_settings, version, dry_run)
@@ -1483,6 +1518,84 @@ async def deploy_source_schema(
     return deploy_result
 
 
+def _http_exception_message(exc: HTTPException) -> str:
+    """A task-facing message from an HTTPException detail."""
+    detail = exc.detail
+    if isinstance(detail, dict):
+        message = detail.get("message")
+        if isinstance(message, str) and message:
+            return message
+    return str(detail)
+
+
+async def _run_source_deploy_task(
+    name: str,
+    version: str | None,
+    dry_run: bool,
+    user: Any,
+    registry: Any,
+    settings: Any,
+    request: Request,
+    *,
+    task,
+) -> dict[str, Any]:
+    """Background worker for one source deploy (bulk ``action=deploy``)."""
+    label = f"{name}" + (f"@{version}" if version else "")
+    task.set_progress(5, f"{'Planning' if dry_run else 'Deploying'} {label}")
+    try:
+        result = await _execute_source_deploy(
+            name=name,
+            user=user,
+            registry=registry,
+            settings=settings,
+            request=request,
+            version=version,
+            dry_run=dry_run,
+        )
+    except HTTPException as exc:
+        raise RuntimeError(_http_exception_message(exc)) from exc
+    task.set_progress(100, f"{'Planned' if dry_run else 'Deployed'} {label}")
+    return result.model_dump(mode="json")
+
+
+@router.post(
+    "/{name}/deploy",
+    response_model=SchemaDeployResult,
+    dependencies=[Depends(require_action(scopes_dict["source_write"]))],
+)
+async def deploy_source_schema(
+    name: str,
+    user: CurrentUser,
+    registry: SourceReg,
+    settings: Settings,
+    request: Request,
+    version: str | None = Query(
+        None, description="Source version id (defaults to the source's current version)"
+    ),
+    dry_run: bool = Query(
+        False, description="Plan only: generate + validate the DDL without applying it"
+    ),
+) -> SchemaDeployResult:
+    """Plan (``dry_run=true``) or deploy a source version's schema to ClickHouse.
+
+    Runs the v2 YAML -> DDL pipeline. In plan mode the CREATE TABLE (+ any standard
+    views) and validation errors are returned for review WITHOUT touching
+    ClickHouse. In deploy mode the DDL is applied: an absent table is created, and
+    an existing one gains the columns it lacks and moves to the version's TTL, so a
+    re-deploy changes only what differs. The engine and the type of an existing
+    column are left as created. A schema that failed validation is never deployed.
+    """
+    return await _execute_source_deploy(
+        name=name,
+        user=user,
+        registry=registry,
+        settings=settings,
+        request=request,
+        version=version,
+        dry_run=dry_run,
+    )
+
+
 class _AppliedSchema(NamedTuple):
     """A source version whose DDL has reached ClickHouse, and what it took."""
 
@@ -2019,20 +2132,31 @@ async def bulk_action(
     registry: SourceReg,
     settings: Settings,
     request: Request,
+    response: Response,
 ):
-    """Perform a bulk action (enable, disable, dormant, delete) on multiple sources."""
+    """Perform a bulk action on multiple sources.
+
+    ``enable`` / ``disable`` / ``dormant`` / ``delete`` apply immediately.
+    ``deploy`` submits one ``source:deploy`` background task per accepted source
+    (poll ``GET /tasks/{task_id}``); refused names land in ``failed``.
+    """
     action_to_state = {"enable": "active", "disable": "disabled", "dormant": "dormant"}
-    if body.action not in (*action_to_state, "delete"):
+    if body.action not in (*action_to_state, "delete", "deploy"):
         raise HTTPException(
             status_code=422,
             detail={
                 "code": "validation_error",
                 "message": (
-                    f"Unknown action {body.action!r}. Must be: enable, disable, dormant, delete"
+                    f"Unknown action {body.action!r}. Must be: enable, disable, "
+                    "dormant, delete, deploy"
                 ),
             },
         )
 
+    if body.action == "deploy":
+        response.status_code = 202
+        return _submit_bulk_deploys(body, user, registry, settings, request)
+
     succeeded, failed, deleted = await run_blocking(
         functools.partial(_apply_bulk, body, action_to_state, user, registry)
     )
@@ -2051,6 +2175,89 @@ async def bulk_action(
     return result
 
 
+def _submit_bulk_deploys(
+    body: BulkActionRequest,
+    user: Any,
+    registry: Any,
+    settings: Any,
+    request: Request,
+) -> BulkActionResponse:
+    """Pre-check and submit one ``source:deploy`` task per accepted source."""
+    manager = request.app.state.task_manager
+    deployments: list[BulkDeployTask] = []
+    failed: list[dict[str, str]] = []
+
+    for item in body.sources:
+        if not registry.source_exists(item.source):
+            failed.append(
+                {
+                    "source": item.source,
+                    "code": "not_found",
+                    "error": f"Source '{item.source}' not found",
+                }
+            )
+            continue
+        if item.version is not None:
+            try:
+                source = registry.get_source(item.source)
+            except SourceNotFoundError:
+                failed.append(
+                    {
+                        "source": item.source,
+                        "code": "not_found",
+                        "error": f"Source '{item.source}' not found",
+                    }
+                )
+                continue
+            if item.version not in source.versions:
+                failed.append(
+                    {
+                        "source": item.source,
+                        "code": "not_found",
+                        "error": (f"Version '{item.version}' not found for source '{item.source}'"),
+                    }
+                )
+                continue
+
+        info = manager.submit(
+            _SOURCE_DEPLOY_TASK_KIND,
+            _run_source_deploy_task,
+            item.source,
+            item.version,
+            body.dry_run,
+            user,
+            registry,
+            settings,
+            request,
+        )
+        deployments.append(
+            BulkDeployTask(
+                source=item.source,
+                version=item.version,
+                task_id=info.id,
+                kind=info.kind,
+            )
+        )
+
+    if deployments:
+        audit_resource_change(
+            user.user_id,
+            "schema",
+            ",".join(entry.source for entry in deployments),
+            "bulk_deploy_submitted",
+            details={
+                "task_ids": [entry.task_id for entry in deployments],
+                "dry_run": body.dry_run,
+            },
+        )
+    return BulkActionResponse(
+        action="deploy",
+        succeeded=[entry.source for entry in deployments],
+        failed=failed,
+        deployments=deployments,
+    )
+
+
 def _apply_bulk(
     body: BulkActionRequest, action_to_state: dict[str, str], user: Any, registry: Any
 ) -> tuple[list[str], list[dict[str, str]], list[Source]]:
@@ -2058,7 +2265,8 @@ def _apply_bulk(
     succeeded: list[str] = []
     failed: list[dict[str, str]] = []
     deleted: list[Source] = []
-    for name in body.sources:
+    for item in body.sources:
+        name = item.source
         try:
             if body.action == "delete":
                 # Read before the delete, for the topics it owns. A name with no
diff --git a/src/dfe_engine/api/v1/system.py b/src/dfe_engine/api/v1/system.py
index 9447c9f7..885e384f 100644
--- a/src/dfe_engine/api/v1/system.py
+++ b/src/dfe_engine/api/v1/system.py
@@ -923,11 +923,25 @@ def _live_body(outcome: SourceLiveOutcome) -> SourceLiveApply:
     )
 
 
+class ApplyDefaultsPin(BaseModel):
+    """One source whose current version was considered for the table defaults."""
+
+    source: str = Field(description="Source name (_source label)")
+    version: str | None = Field(
+        default=None,
+        description="Version id that was pinned or already matched; null when unset",
+    )
+
+
 class ApplyDefaultsResponse(BaseModel):
     """Which named sources were pinned, which already had these defaults, and their tables."""
 
-    updated: list[str]
-    unchanged: list[str]
+    updated: list[ApplyDefaultsPin] = Field(
+        description="Sources whose current version was written with the defaults"
+    )
+    unchanged: list[ApplyDefaultsPin] = Field(
+        description="Sources whose current version already matched the defaults"
+    )
     live: list[SourceLiveApply] = Field(
         description="One entry per named source, in request order: what reached its table."
     )
@@ -990,11 +1004,12 @@ def apply_defaults(
             ) from None
         loaded.append(source)
 
-    updated: list[str] = []
-    unchanged: list[str] = []
+    updated: list[ApplyDefaultsPin] = []
+    unchanged: list[ApplyDefaultsPin] = []
     saved: list[Source] = []
     author = git_author(user)
     for source in loaded:
+        pin = ApplyDefaultsPin(source=source.source, version=source.current)
         pinned = pin_table_defaults(
             source,
             header_type=table.header_type,
@@ -1003,7 +1018,7 @@ def apply_defaults(
             engine=table.engine,
         )
         if pinned is None:
-            unchanged.append(source.source)
+            unchanged.append(pin)
             saved.append(source)
             continue
         stored = registry.save_source(
@@ -1018,13 +1033,14 @@ def apply_defaults(
             source.source,
             "updated",
             {
+                "version": source.current,
                 "ttl_days": ttl.effective,
                 "common_header_type": table.header_type,
                 "common_header_version": table.header_version,
                 "engine": table.engine,
             },
         )
-        updated.append(source.source)
+        updated.append(pin)
 
     def connect() -> Any:
         return get_clickhouse_client(settings)
diff --git a/tests/integration/test_system_defaults_apply_ch.py b/tests/integration/test_system_defaults_apply_ch.py
index 6fd29269..4f727ae5 100644
--- a/tests/integration/test_system_defaults_apply_ch.py
+++ b/tests/integration/test_system_defaults_apply_ch.py
@@ -201,7 +201,7 @@ def test_apply_moves_a_deployed_tables_ttl_without_a_restart(ch_client, stack):
     body = _apply(stack, "syslog-ttl")
 
     assert _live_ttl_days(ch_client, db, "syslog-ttl") == 91
-    assert body["updated"] == ["syslog-ttl"]
+    assert body["updated"] == [{"source": "syslog-ttl", "version": "1.0.0"}]
     assert body["live"] == [
         {
             "source": "syslog-ttl",
@@ -265,7 +265,7 @@ def test_an_engine_change_needs_a_rebuild_and_stays_drift(ch_client, stack):
         "live": "MergeTree",
     }
     assert drift["ttl_days"]["live"] == 90
-    assert body["updated"] == ["syslog-engine"]
+    assert body["updated"] == [{"source": "syslog-engine", "version": "1.0.0"}]
     live = body["live"][0]
     assert live["status"] == "unchanged"
     assert live["not_applied"] == [{"field": "engine", "reason": REBUILD}]
@@ -340,7 +340,10 @@ def test_one_table_clickhouse_refuses_does_not_hide_the_others(ch_client, stack)
         logger.remove(handler_id)
 
     assert _live_ttl_days(ch_client, db, "syslog-good") == 91
-    assert body["updated"] == ["syslog-bad", "syslog-good"]
+    assert body["updated"] == [
+        {"source": "syslog-bad", "version": "1.0.0"},
+        {"source": "syslog-good", "version": "1.0.0"},
+    ]
     bad, good = body["live"]
     assert bad["source"] == "syslog-bad"
     assert bad["status"] == "failed"
diff --git a/tests/support/loopback.py b/tests/support/loopback.py
index 20370dc0..5f553619 100644
--- a/tests/support/loopback.py
+++ b/tests/support/loopback.py
@@ -33,11 +33,23 @@ def wake_accept(listener: socket.socket) -> None:
 
     Closing the socket alone leaves that thread parked, and a later child exit
     resumes it on the reused fd. Shutting the socket down first makes the
-    ``accept()`` raise. An ``OSError`` here (not connected, already closed) is
-    ignored: there is nothing left to wake.
+    ``accept()`` raise on platforms that allow ``shutdown`` on a listening
+    socket. On macOS that call fails with ``ENOTCONN`` and does not wake
+    ``accept``, so a short connect to the port is used instead: ``accept``
+    returns, and the caller closes or finishes the request as usual.
     """
-    with contextlib.suppress(OSError):
+    try:
         listener.shutdown(socket.SHUT_RDWR)
+        return
+    except OSError:
+        pass
+    try:
+        host, port = listener.getsockname()[:2]
+    except OSError:
+        return
+    with contextlib.suppress(OSError):
+        with socket.create_connection((host, port), timeout=1.0):
+            pass
 
 
 def _end_after_first_bytes(conn: socket.socket, how: EndOfConnection) -> None:
diff --git a/tests/unit/test_api/test_no_backend_error_text.py b/tests/unit/test_api/test_no_backend_error_text.py
index 26029fb7..ec582628 100644
--- a/tests/unit/test_api/test_no_backend_error_text.py
+++ b/tests/unit/test_api/test_no_backend_error_text.py
@@ -405,7 +405,13 @@ def delete_source(self, name: str, **_kwargs) -> None:
                     raise SourceNotFoundError("Source 'gone' not found")
                 raise RuntimeError(SENTINEL)
 
-        body = SimpleNamespace(action="delete", sources=["acme", "gone"])
+        body = SimpleNamespace(
+            action="delete",
+            sources=[
+                sources_module.BulkActionItem(source="acme"),
+                sources_module.BulkActionItem(source="gone"),
+            ],
+        )
         user = AuthContext(user_id="admin")
 
         _, failed, _ = sources_module._apply_bulk(body, {}, user, _Registry())
diff --git a/tests/unit/test_api/test_sources.py b/tests/unit/test_api/test_sources.py
index e14bd70d..86c9b736 100644
--- a/tests/unit/test_api/test_sources.py
+++ b/tests/unit/test_api/test_sources.py
@@ -1198,6 +1198,86 @@ def test_bulk_invalid_action(self, client: TestClient, admin_headers: dict):
         assert resp.status_code == 422
 
 
+class TestBulkDeploy:
+    """POST /api/v1/sources/bulk with action=deploy"""
+
+    def test_submits_one_task_per_source(
+        self, client: TestClient, admin_headers: dict, app, monkeypatch
+    ):
+        from dfe_engine.api.v1 import sources as sources_api
+        from dfe_engine.source.deployment import SchemaDeployResult
+
+        for name in ("deploy-a", "deploy-b"):
+            client.post(
+                "/api/v1/sources",
+                json={
+                    "source": name,
+                    "match": {"field": "tags.collector.type", "value": name},
+                    "schema": {"engine": "MergeTree"},
+                },
+                headers=admin_headers,
+            )
+
+        async def _fake_deploy(*, name, version, dry_run, **_kwargs):
+            return SchemaDeployResult(
+                source_name=name,
+                version=version or "1.0.0",
+                dry_run=dry_run,
+                applied=not dry_run,
+                create_table="CREATE TABLE t",
+                views={},
+                validation_errors=[],
+            )
+
+        monkeypatch.setattr(sources_api, "_execute_source_deploy", _fake_deploy)
+
+        resp = client.post(
+            "/api/v1/sources/bulk",
+            json={
+                "action": "deploy",
+                "sources": ["deploy-a", {"source": "deploy-b", "version": "1.0.0"}],
+            },
+            headers=admin_headers,
+        )
+        assert resp.status_code == 202, resp.text
+        body = resp.json()
+        assert body["action"] == "deploy"
+        assert set(body["succeeded"]) == {"deploy-a", "deploy-b"}
+        assert body["failed"] == []
+        assert {row["source"] for row in body["deployments"]} == {"deploy-a", "deploy-b"}
+        assert all(row["kind"] == "source:deploy" for row in body["deployments"])
+        assert len({row["task_id"] for row in body["deployments"]}) == 2
+
+        for row in body["deployments"]:
+            task = app.state.task_manager.get(row["task_id"])
+            assert task is not None
+            assert task.kind == "source:deploy"
+
+    def test_refuses_missing_source_and_version(
+        self, client: TestClient, admin_headers: dict, sample_source: dict
+    ):
+        client.post("/api/v1/sources", json=sample_source, headers=admin_headers)
+        resp = client.post(
+            "/api/v1/sources/bulk",
+            json={
+                "action": "deploy",
+                "sources": [
+                    "missing-source",
+                    {"source": sample_source["source"], "version": "9.9.9"},
+                ],
+            },
+            headers=admin_headers,
+        )
+        assert resp.status_code == 202
+        body = resp.json()
+        assert body["deployments"] == []
+        assert body["succeeded"] == []
+        by_source = {row["source"]: row for row in body["failed"]}
+        assert by_source["missing-source"]["code"] == "not_found"
+        assert by_source[sample_source["source"]]["code"] == "not_found"
+        assert "9.9.9" in by_source[sample_source["source"]]["error"]
+
+
 class TestSeedSources:
     """POST /api/v1/sources/seed"""
 
diff --git a/tests/unit/test_api/test_system_defaults.py b/tests/unit/test_api/test_system_defaults.py
index 9dbcedfd..edea7ab7 100644
--- a/tests/unit/test_api/test_system_defaults.py
+++ b/tests/unit/test_api/test_system_defaults.py
@@ -256,7 +256,7 @@ def test_only_the_named_sources_are_pinned(self, admin_headers, app, client, tmp
 
         assert resp.status_code == 200, resp.text
         body = resp.json()
-        assert body["updated"] == ["alpha"]
+        assert body["updated"] == [{"source": "alpha", "version": "1.0.0"}]
         assert body["unchanged"] == []
         assert [entry["source"] for entry in body["live"]] == ["alpha"]
         alpha = _version(client, admin_headers, "alpha")
@@ -281,7 +281,7 @@ def test_a_source_already_on_the_defaults_is_unchanged(
 
         assert second.status_code == 200, second.text
         assert second.json()["updated"] == []
-        assert second.json()["unchanged"] == ["alpha"]
+        assert second.json()["unchanged"] == [{"source": "alpha", "version": "1.0.0"}]
         assert [entry["source"] for entry in second.json()["live"]] == ["alpha"]
 
     def test_a_source_never_deployed_is_pinned_without_reaching_clickhouse(
@@ -293,7 +293,7 @@ def test_a_source_never_deployed_is_pinned_without_reaching_clickhouse(
         resp = client.post(APPLY, headers=admin_headers, json={"sources": ["alpha"]})
 
         assert resp.status_code == 200, resp.text
-        assert resp.json()["updated"] == ["alpha"]
+        assert resp.json()["updated"] == [{"source": "alpha", "version": "1.0.0"}]
         assert resp.json()["live"] == [
             {
                 "source": "alpha",