feat: e2e tests for deployments api (#12767)

* update e2e tests

* fix(watsonx): harden existing-resource create flow and rollback journaling
- move the direct Watsonx adapter E2E runner to `scripts/e2e_deployment_tests/watsonx_orchestrate/adapter.py` (rename-only) so deployment E2E assets are consolidated under one folder
- add `scripts/e2e_deployment_tests/watsonx_orchestrate/api.py` with a full `/api/v1/deployments` matrix covering create/update happy paths, validation rejections, attachment patching, rollback/error paths, concurrency races, large payload tiers, and failpoint scenarios with owned-resource cleanup
- clarify existing-resource create behavior in `src/backend/base/langflow/api/v1/deployments.py`: DB-only onboarding keeps `created_*` fields empty unless provider mutation operations are requested
- add `util_create_result_from_existing_resource` in `src/backend/base/langflow/api/v1/mappers/deployments/watsonx_orchestrate/mapper.py` to normalize non-mutating onboard responses into a create-style result with empty `app_ids` and `tools_with_refs`
- extend `src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/config.py` to accept `created_app_ids_journal` and append app ids immediately after successful provider connection creation for rollback safety
- update shared connection orchestration in `src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/shared.py` to normalize provider app ids via `RawConnectionCreatePlan.__post_init__`, propagate `created_app_ids_journal`, dedupe rollback ids, and wrap validation-stage failures as `ConnectionCreateBatchError` with rollback metadata
- wire journaling into create/update rollback flows in `src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/create.py` and `src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/update.py` so partial provider-side connection creation is always captured for cleanup
- add mapper coverage in `src/backend/tests/unit/api/v1/test_deployment_mapper_watsonx.py` for existing-resource create-result normalization
- expand `src/backend/tests/unit/services/deployment/test_watsonx_orchestrate.py` with coverage for provider app-id normalization, validation-failure rollback metadata, and create/update rollback when failures occur after provider connection creation; update mocks/monkeypatch targets to match the shared connection entrypoint signature

* simplify create result logic (improves error friendliness

(cherry picked from commit 9f42c9e707)

src/backend/base/langflow/api/v1/deployments.py

@@ -485,10 +485,14 @@ async def create_deployment(
                 db=session,
             )
     else:
+        # Existing-resource create starts as DB-only onboarding: no provider
+        # mutation is performed and created_* response fields stay empty.
         provider_create_result = deployment_mapper.util_create_result_from_existing_resource(
             existing_resource_key=str(existing_resource_key),
         )
         if should_mutate_existing_resource:
+            # When create payload includes add_flows/upsert_tools, run provider
+            # update and normalize the update result into create-style created_*.
             adapter_payload = await deployment_mapper.resolve_deployment_update_for_existing_create(
                 user_id=current_user.id,
                 project_id=project_id,

src/backend/base/langflow/api/v1/mappers/deployments/watsonx_orchestrate/mapper.py

@@ -504,6 +504,28 @@ class WatsonxOrchestrateDeploymentMapper(BaseDeploymentMapper):
             provider_result=create_provider_result,
         )
 
+    def util_create_result_from_existing_resource(
+        self,
+        *,
+        existing_resource_key: str,
+    ) -> DeploymentCreateResult:
+        """Build a create-style result payload for DB-only onboarding.
+
+        This path is used when create request includes ``existing_agent_id``
+        without create-time mutation operations. ``created_*`` fields represent
+        what this request created, so they are intentionally empty here.
+        """
+        create_provider_result = self._parse_required_payload_slot(
+            slot=WXO_ADAPTER_PAYLOAD_SCHEMAS.deployment_create_result,
+            slot_name="deployment_create_result",
+            raw={"app_ids": [], "tools_with_refs": []},
+            operation="building the create response for the existing resource",
+        )
+        return DeploymentCreateResult(
+            id=existing_resource_key,
+            provider_result=create_provider_result.model_dump(mode="json"),
+        )
+
     async def _resolve_provider_payload_from_create_api(
         self,
         *,

src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/config.py

@@ -61,13 +61,21 @@ async def create_config(
     config: DeploymentConfig,
     user_id: IdLike,
     db: AsyncSession,
+    created_app_ids_journal: list[str] | None = None,
 ) -> str:
-    """Create/update a wxO draft key-value connection config plus runtime credentials."""
+    """Create/update a wxO draft key-value connection config plus runtime credentials.
+
+    When ``created_app_ids_journal`` is provided, ``app_id`` is appended
+    immediately after provider connection creation succeeds so rollback can
+    clean up partially completed creates.
+    """
     app_id = validate_wxo_name(config.name)
     env_var_keys = list((config.environment_variables or {}).keys())
     logger.debug("create_config: app_id='%s', env_var_keys=%s", app_id, env_var_keys)
 
     await asyncio.to_thread(clients.connections.create, payload={"app_id": app_id})
+    if created_app_ids_journal is not None:
+        created_app_ids_journal.append(app_id)
 
     wxo_config = ConnectionConfiguration(
         app_id=app_id,

src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/create.py

@@ -28,7 +28,6 @@ from langflow.services.adapters.deployment.watsonx_orchestrate.core.shared impor
     OrderedUniqueStrs,
     RawConnectionCreatePlan,
     RawToolCreatePlan,
-    create_connection_with_conflict_mapping,
     create_raw_tools_with_bindings,
     log_batch_errors,
     resolve_connections_for_operations,
@@ -199,11 +198,15 @@ async def apply_provider_create_plan_with_rollback(
     # - operation_to_provider_app_id: operation app_id → provider app_id
     #     (identity mapping for both existing and raw-created connections).
     # - resolved_connections: provider_app_id → connection_id map for bind calls.
+    # - created_app_ids_journal: app_ids recorded immediately after successful
+    #     provider connection creation; used to ensure rollback sees partial
+    #     successes even if create later fails before returning.
     created_snapshot_bindings: list[WatsonxToolRefBinding] = []
     created_tool_app_bindings: list[WatsonxToolAppBinding] = []
     agent_create_response = None
     operation_to_provider_app_id: dict[str, str] = {}
     resolved_connections: dict[str, str] = {}
+    created_app_ids_journal: list[str] = []
 
     try:
         try:
@@ -215,7 +218,7 @@ async def apply_provider_create_plan_with_rollback(
                 raw_connections_to_create=plan.raw_connections_to_create,
                 error_prefix=ErrorPrefix.CREATE.value,
                 validate_connection_fn=validate_connection,
-                create_connection_fn=create_connection_with_conflict_mapping,
+                created_app_ids_journal=created_app_ids_journal,
             )
             operation_to_provider_app_id = connection_result.operation_to_provider_app_id
             resolved_connections = connection_result.resolved_connections

src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/shared.py

@@ -22,7 +22,11 @@ from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import
     create_and_upload_wxo_flow_tools_with_bindings,
 )
 from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import WatsonxResultToolRefBinding
-from langflow.services.adapters.deployment.watsonx_orchestrate.utils import extract_error_detail
+from langflow.services.adapters.deployment.watsonx_orchestrate.utils import (
+    dedupe_list,
+    extract_error_detail,
+    validate_wxo_name,
+)
 
 if TYPE_CHECKING:
     from collections.abc import Awaitable, Callable, Iterator
@@ -72,6 +76,12 @@ class RawConnectionCreatePlan:
     provider_app_id: str
     payload: WatsonxConnectionRawPayload
 
+    def __post_init__(self) -> None:
+        # operations[*].app_ids use operation_app_id as the caller-visible key.
+        # provider_app_id is used for provider calls and must follow wxO rules.
+        # Normalizing here keeps create/validate/rollback on one canonical id.
+        self.provider_app_id = validate_wxo_name(self.provider_app_id)
+
 
 @dataclass(slots=True)
 class RawToolCreatePlan:
@@ -116,6 +126,7 @@ async def create_connection_with_conflict_mapping(
     user_id: IdLike,
     db: AsyncSession,
     error_prefix: str,
+    created_app_ids_journal: list[str] | None = None,
 ) -> str:
     from lfx.services.adapters.deployment.schema import DeploymentConfig
 
@@ -133,6 +144,7 @@ async def create_connection_with_conflict_mapping(
             config=config_payload,
             user_id=user_id,
             db=db,
+            created_app_ids_journal=created_app_ids_journal,
         )
     except (ClientAPIException, HTTPException) as exc:
         if isinstance(exc, ClientAPIException):
@@ -163,7 +175,7 @@ async def resolve_connections_for_operations(
     raw_connections_to_create: list[RawConnectionCreatePlan],
     error_prefix: str,
     validate_connection_fn: Callable[..., Awaitable[object]] = validate_connection,
-    create_connection_fn: Callable[..., Awaitable[str]] = create_connection_with_conflict_mapping,
+    created_app_ids_journal: list[str] | None = None,
 ) -> ConnectionResolutionResult:
     logger.debug(
         "resolve_connections_for_operations: existing_app_ids=%s, raw_to_create=%d",
@@ -187,15 +199,17 @@ async def resolve_connections_for_operations(
             created_app_ids=[],
         )
 
+    journal = created_app_ids_journal if created_app_ids_journal is not None else []
     created_connections_results = await asyncio.gather(
         *(
-            create_connection_fn(
+            create_connection_with_conflict_mapping(
                 clients=clients,
                 app_id=create_plan.provider_app_id,
                 payload=create_plan.payload,
                 user_id=user_id,
                 db=db,
                 error_prefix=error_prefix,
+                created_app_ids_journal=journal,
             )
             for create_plan in raw_connections_to_create
         ),
@@ -203,7 +217,7 @@ async def resolve_connections_for_operations(
     )
 
     create_connection_errors: list[Exception] = []
-    created_app_ids_journal: list[str] = []
+    created_app_ids: list[str] = []
     for result in created_connections_results:
         if isinstance(result, BaseException):
             if isinstance(result, Exception):
@@ -213,26 +227,34 @@ async def resolve_connections_for_operations(
                     RuntimeError(f"Connection create failed with non-standard exception: {type(result).__name__}")
                 )
             continue
-        created_app_ids_journal.append(result)
-    created_app_ids = list(dict.fromkeys(created_app_ids_journal))
+        created_app_ids.append(result)
     if create_connection_errors:
+        rollback_app_ids = dedupe_list([*journal, *created_app_ids])
         logger.debug(
             "resolve_connections_for_operations: %d errors, created_app_ids=%s",
             len(create_connection_errors),
-            created_app_ids,
+            rollback_app_ids,
         )
-        raise ConnectionCreateBatchError(created_app_ids=created_app_ids, errors=create_connection_errors)
+        raise ConnectionCreateBatchError(created_app_ids=rollback_app_ids, errors=create_connection_errors)
 
-    validated_created_connections: list[object] = await asyncio.gather(
-        *(
-            retry_create(
-                validate_connection_fn,
-                clients.connections,
-                app_id=create_plan.provider_app_id,
+    try:
+        validated_created_connections: list[object] = await asyncio.gather(
+            *(
+                retry_create(
+                    validate_connection_fn,
+                    clients.connections,
+                    app_id=create_plan.provider_app_id,
+                )
+                for create_plan in raw_connections_to_create
             )
-            for create_plan in raw_connections_to_create
         )
-    )
+    except Exception as exc:
+        rollback_app_ids = dedupe_list([*journal, *created_app_ids])
+        logger.debug(
+            "resolve_connections_for_operations: validation error, created_app_ids=%s",
+            rollback_app_ids,
+        )
+        raise ConnectionCreateBatchError(created_app_ids=rollback_app_ids, errors=[exc]) from exc
     for create_plan, connection in zip(raw_connections_to_create, validated_created_connections, strict=True):
         operation_to_provider_app_id[create_plan.operation_app_id] = create_plan.provider_app_id
         resolved_connections[create_plan.provider_app_id] = connection.connection_id  # type: ignore[attr-defined]

src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/update.py

@@ -25,7 +25,6 @@ from langflow.services.adapters.deployment.watsonx_orchestrate.core.shared impor
     OrderedUniqueStrs,
     RawConnectionCreatePlan,
     RawToolCreatePlan,
-    create_connection_with_conflict_mapping,
     create_raw_tools_with_bindings,
     log_batch_errors,
     resolve_connections_for_operations,
@@ -449,6 +448,9 @@ async def apply_provider_update_plan_with_rollback(
     # - referenced_snapshot_bindings: full operation correlation set.
     # - final_update_payload: outbound agent patch payload (spec + tools).
     # - rollback_agent_payload: best-effort restore payload for agent rollback.
+    # - created_app_ids_journal: app_ids recorded immediately after successful
+    #     provider connection creation; used to ensure rollback sees partial
+    #     successes even if create later fails before returning.
     resolved_connections: dict[str, str] = {}
     operation_to_provider_app_id: dict[str, str] = {app_id: app_id for app_id in plan.existing_app_ids}
     created_snapshot_ids: list[str] = []
@@ -456,6 +458,7 @@ async def apply_provider_update_plan_with_rollback(
     created_snapshot_bindings: list[WatsonxResultToolRefBinding] = []
     final_update_payload = dict(update_payload)
     rollback_agent_payload: dict[str, Any] = {}
+    created_app_ids_journal: list[str] = []
 
     # Pre-seed resolved_connections with bindings already attached to the
     # agent's existing tools.  This lets new tools reuse the same connections
@@ -499,7 +502,7 @@ async def apply_provider_update_plan_with_rollback(
                 raw_connections_to_create=plan.raw_connections_to_create,
                 error_prefix=ErrorPrefix.UPDATE.value,
                 validate_connection_fn=validate_connection,
-                create_connection_fn=create_connection_with_conflict_mapping,
+                created_app_ids_journal=created_app_ids_journal,
             )
             operation_to_provider_app_id.update(connection_result.operation_to_provider_app_id)
             resolved_connections.update(connection_result.resolved_connections)

src/backend/tests/unit/api/v1/test_deployment_mapper_watsonx.py

@@ -896,6 +896,16 @@ def test_watsonx_mapper_create_result_from_existing_update_normalizes_slot_paylo
     }
 
 
+def test_watsonx_mapper_create_result_from_existing_resource_includes_empty_payload() -> None:
+    mapper = WatsonxOrchestrateDeploymentMapper()
+    create_result = mapper.util_create_result_from_existing_resource(existing_resource_key="existing-agent-1")
+
+    assert create_result.id == "existing-agent-1"
+    assert isinstance(create_result.provider_result, dict)
+    assert create_result.provider_result.get("app_ids") == []
+    assert create_result.provider_result.get("tools_with_refs") == []
+
+
 def test_watsonx_mapper_resolve_verify_credentials_for_update_returns_none_without_provider_data() -> None:
     from langflow.services.database.models.deployment_provider_account.model import DeploymentProviderAccount
 

src/backend/tests/unit/services/deployment/test_watsonx_orchestrate.py

@@ -1621,6 +1621,125 @@ async def test_apply_provider_create_plan_rolls_back_mutated_existing_tools_with
     assert rollback_payload["binding"]["langflow"]["connections"] == {"old": "conn-old"}
 
 
+@pytest.mark.anyio
+async def test_resolve_connections_for_operations_normalizes_provider_app_id_and_uses_created_id_for_validation(
+    monkeypatch,
+):
+    raw_payload = payloads_module.WatsonxConnectionRawPayload.model_validate(
+        {"app_id": "cfg-s-003", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}}
+    )
+    raw_connections_to_create = [
+        shared_core_module.RawConnectionCreatePlan(
+            operation_app_id="cfg-s-003",
+            provider_app_id="cfg-s-003",
+            payload=raw_payload,
+        )
+    ]
+    fake_clients = SimpleNamespace(connections=FakeConnectionsClient())
+    validated_app_ids: list[str] = []
+
+    async def mock_create_connection_with_conflict_mapping(
+        *,
+        clients,
+        app_id,
+        payload,
+        user_id,
+        db,
+        error_prefix,
+        created_app_ids_journal=None,
+    ):
+        _ = clients, app_id, payload, user_id, db, error_prefix
+        created_id = "cfg_s_003"
+        if created_app_ids_journal is not None:
+            created_app_ids_journal.append(created_id)
+        return created_id
+
+    async def mock_validate_connection(connections_client, *, app_id):  # noqa: ARG001
+        validated_app_ids.append(app_id)
+        return SimpleNamespace(connection_id=f"conn-{app_id}")
+
+    monkeypatch.setattr(
+        shared_core_module,
+        "create_connection_with_conflict_mapping",
+        mock_create_connection_with_conflict_mapping,
+    )
+
+    result = await shared_core_module.resolve_connections_for_operations(
+        clients=fake_clients,
+        user_id="user-1",
+        db=object(),
+        existing_app_ids=[],
+        raw_connections_to_create=raw_connections_to_create,
+        error_prefix="CREATE",
+        validate_connection_fn=mock_validate_connection,
+    )
+
+    assert raw_connections_to_create[0].provider_app_id == "cfg_s_003"
+    assert validated_app_ids == ["cfg_s_003"]
+    assert result.operation_to_provider_app_id == {"cfg-s-003": "cfg_s_003"}
+    assert result.resolved_connections == {"cfg_s_003": "conn-cfg_s_003"}
+    assert result.created_app_ids == ["cfg_s_003"]
+
+
+@pytest.mark.anyio
+async def test_resolve_connections_for_operations_wraps_validation_failure_with_rollback_metadata(monkeypatch):
+    raw_payload = payloads_module.WatsonxConnectionRawPayload.model_validate(
+        {"app_id": "cfg-s-003", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}}
+    )
+    raw_connections_to_create = [
+        shared_core_module.RawConnectionCreatePlan(
+            operation_app_id="cfg-s-003",
+            provider_app_id="cfg-s-003",
+            payload=raw_payload,
+        )
+    ]
+    fake_clients = SimpleNamespace(connections=FakeConnectionsClient())
+    created_app_ids_journal: list[str] = []
+
+    async def mock_create_connection_with_conflict_mapping(
+        *,
+        clients,
+        app_id,
+        payload,
+        user_id,
+        db,
+        error_prefix,
+        created_app_ids_journal=None,
+    ):
+        _ = clients, app_id, payload, user_id, db, error_prefix
+        created_id = "cfg_s_003"
+        if created_app_ids_journal is not None:
+            created_app_ids_journal.append(created_id)
+        return created_id
+
+    async def mock_validate_connection(connections_client, *, app_id):  # noqa: ARG001
+        msg = f"Connection '{app_id}' not found. Ensure the connection exists with a draft configuration."
+        raise InvalidContentError(message=msg)
+
+    monkeypatch.setattr(
+        shared_core_module,
+        "create_connection_with_conflict_mapping",
+        mock_create_connection_with_conflict_mapping,
+    )
+
+    with pytest.raises(shared_core_module.ConnectionCreateBatchError) as exc_info:
+        await shared_core_module.resolve_connections_for_operations(
+            clients=fake_clients,
+            user_id="user-1",
+            db=object(),
+            existing_app_ids=[],
+            raw_connections_to_create=raw_connections_to_create,
+            error_prefix="CREATE",
+            validate_connection_fn=mock_validate_connection,
+            created_app_ids_journal=created_app_ids_journal,
+        )
+
+    assert created_app_ids_journal == ["cfg_s_003"]
+    assert exc_info.value.created_app_ids == ["cfg_s_003"]
+    assert len(exc_info.value.errors) == 1
+    assert isinstance(exc_info.value.errors[0], InvalidContentError)
+
+
 @pytest.mark.anyio
 async def test_apply_provider_create_plan_rolls_back_successfully_created_raw_connections_on_partial_batch_failure(
     monkeypatch,
@@ -1651,7 +1770,17 @@ async def test_apply_provider_create_plan_rolls_back_successfully_created_raw_co
     fake_clients = SimpleNamespace(connections=FakeConnectionsClient())
     captured: dict[str, Any] = {}
 
-    async def mock_create_connection_with_conflict_mapping(*, clients, app_id, payload, user_id, db, error_prefix):  # noqa: ARG001
+    async def mock_create_connection_with_conflict_mapping(
+        *,
+        clients,
+        app_id,
+        payload,
+        user_id,
+        db,
+        error_prefix,
+        created_app_ids_journal=None,
+    ):
+        _ = clients, payload, user_id, db, error_prefix, created_app_ids_journal
         if app_id.endswith("cfg-a"):
             return app_id
         msg = "boom-create-connection"
@@ -1661,7 +1790,7 @@ async def test_apply_provider_create_plan_rolls_back_successfully_created_raw_co
         captured["rollback_app_ids"] = list(app_ids or [])
 
     monkeypatch.setattr(
-        create_core_module,
+        shared_core_module,
         "create_connection_with_conflict_mapping",
         mock_create_connection_with_conflict_mapping,
     )
@@ -1714,7 +1843,17 @@ async def test_apply_provider_create_plan_rolls_back_all_journaled_raw_connectio
     fake_clients = SimpleNamespace(connections=FakeConnectionsClient())
     captured: dict[str, Any] = {}
 
-    async def mock_create_connection_with_conflict_mapping(*, clients, app_id, payload, user_id, db, error_prefix):  # noqa: ARG001
+    async def mock_create_connection_with_conflict_mapping(
+        *,
+        clients,
+        app_id,
+        payload,
+        user_id,
+        db,
+        error_prefix,
+        created_app_ids_journal=None,
+    ):
+        _ = clients, payload, user_id, db, error_prefix, created_app_ids_journal
         if app_id.endswith("cfg-c"):
             msg = "boom-create-connection"
             raise RuntimeError(msg)
@@ -1724,7 +1863,7 @@ async def test_apply_provider_create_plan_rolls_back_all_journaled_raw_connectio
         captured["rollback_app_ids"] = list(app_ids or [])
 
     monkeypatch.setattr(
-        create_core_module,
+        shared_core_module,
         "create_connection_with_conflict_mapping",
         mock_create_connection_with_conflict_mapping,
     )
@@ -1746,6 +1885,77 @@ async def test_apply_provider_create_plan_rolls_back_all_journaled_raw_connectio
     assert captured["rollback_app_ids"] == ["cfg-a", "cfg-b"]
 
 
+@pytest.mark.anyio
+async def test_apply_provider_create_plan_rolls_back_journaled_app_ids_when_create_fails_after_provider_create(
+    monkeypatch,
+):
+    provider_create = payloads_module.WatsonxDeploymentCreatePayload.model_validate(
+        {
+            "tools": {},
+            "connections": {
+                "raw_payloads": [
+                    {"app_id": "cfg-a", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}},
+                ]
+            },
+            "llm": TEST_WXO_LLM,
+            "operations": [
+                {
+                    "op": "bind",
+                    "tool": {"tool_id_with_ref": _tool_ref("tool-existing-1")},
+                    "app_ids": ["cfg-a"],
+                },
+            ],
+        }
+    )
+    plan = create_core_module.build_provider_create_plan(
+        deployment_name="my deployment",
+        provider_create=provider_create,
+    )
+    fake_clients = SimpleNamespace(connections=FakeConnectionsClient())
+    captured: dict[str, Any] = {}
+
+    async def mock_create_connection_with_conflict_mapping(
+        *,
+        clients,
+        app_id,
+        payload,
+        user_id,
+        db,
+        error_prefix,
+        created_app_ids_journal=None,
+    ):
+        _ = clients, payload, user_id, db, error_prefix
+        if created_app_ids_journal is not None:
+            created_app_ids_journal.append(app_id)
+        msg = "boom-after-provider-create"
+        raise RuntimeError(msg)
+
+    async def mock_rollback_created_resources(*, clients, agent_id, tool_ids, app_ids=None):  # noqa: ARG001
+        captured["rollback_app_ids"] = list(app_ids or [])
+
+    monkeypatch.setattr(
+        shared_core_module,
+        "create_connection_with_conflict_mapping",
+        mock_create_connection_with_conflict_mapping,
+    )
+    monkeypatch.setattr(create_core_module, "rollback_created_resources", mock_rollback_created_resources)
+
+    with pytest.raises(RuntimeError, match="boom-after-provider-create"):
+        await create_core_module.apply_provider_create_plan_with_rollback(
+            clients=fake_clients,
+            user_id="user-1",
+            db=object(),
+            deployment_spec=BaseDeploymentData(
+                name="my deployment",
+                description="desc",
+                type=DeploymentType.AGENT,
+            ),
+            plan=plan,
+        )
+
+    assert captured["rollback_app_ids"] == ["cfg-a"]
+
+
 @pytest.mark.anyio
 async def test_apply_provider_update_plan_rolls_back_successfully_created_raw_connections_on_partial_batch_failure(
     monkeypatch,
@@ -1780,7 +1990,17 @@ async def test_apply_provider_update_plan_rolls_back_successfully_created_raw_co
     )
     captured: dict[str, Any] = {}
 
-    async def mock_create_connection_with_conflict_mapping(*, clients, app_id, payload, user_id, db, error_prefix):  # noqa: ARG001
+    async def mock_create_connection_with_conflict_mapping(
+        *,
+        clients,
+        app_id,
+        payload,
+        user_id,
+        db,
+        error_prefix,
+        created_app_ids_journal=None,
+    ):
+        _ = clients, payload, user_id, db, error_prefix, created_app_ids_journal
         if app_id.endswith("cfg-a"):
             return app_id
         msg = "boom-update-connection"
@@ -1793,7 +2013,7 @@ async def test_apply_provider_update_plan_rolls_back_successfully_created_raw_co
         captured["rolled_back_app_ids"] = list(created_app_ids)
 
     monkeypatch.setattr(
-        update_core_module,
+        shared_core_module,
         "create_connection_with_conflict_mapping",
         mock_create_connection_with_conflict_mapping,
     )
@@ -1849,7 +2069,17 @@ async def test_apply_provider_update_plan_rolls_back_all_journaled_raw_connectio
     )
     captured: dict[str, Any] = {}
 
-    async def mock_create_connection_with_conflict_mapping(*, clients, app_id, payload, user_id, db, error_prefix):  # noqa: ARG001
+    async def mock_create_connection_with_conflict_mapping(
+        *,
+        clients,
+        app_id,
+        payload,
+        user_id,
+        db,
+        error_prefix,
+        created_app_ids_journal=None,
+    ):
+        _ = clients, payload, user_id, db, error_prefix, created_app_ids_journal
         if app_id.endswith("cfg-c"):
             msg = "boom-update-connection"
             raise RuntimeError(msg)
@@ -1862,7 +2092,7 @@ async def test_apply_provider_update_plan_rolls_back_all_journaled_raw_connectio
         captured["rolled_back_app_ids"] = list(created_app_ids)
 
     monkeypatch.setattr(
-        update_core_module,
+        shared_core_module,
         "create_connection_with_conflict_mapping",
         mock_create_connection_with_conflict_mapping,
     )
@@ -1883,6 +2113,83 @@ async def test_apply_provider_update_plan_rolls_back_all_journaled_raw_connectio
     assert captured["rolled_back_app_ids"] == ["cfg-a", "cfg-b"]
 
 
+@pytest.mark.anyio
+async def test_apply_provider_update_plan_rolls_back_journaled_app_ids_when_create_fails_after_provider_create(
+    monkeypatch,
+):
+    provider_update = payloads_module.WatsonxDeploymentUpdatePayload.model_validate(
+        {
+            "tools": {},
+            "connections": {
+                "raw_payloads": [
+                    {"app_id": "cfg-a", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}},
+                ]
+            },
+            "llm": TEST_WXO_LLM,
+            "operations": [
+                {
+                    "op": "bind",
+                    "tool": {"tool_id_with_ref": _tool_ref("tool-existing-1")},
+                    "app_ids": ["cfg-a"],
+                },
+            ],
+        }
+    )
+    plan = update_core_module.build_provider_update_plan(
+        agent={"id": "dep-1", "tools": ["tool-existing-1"]},
+        provider_update=provider_update,
+    )
+    fake_clients = SimpleNamespace(
+        agent=FakeAgentClient({"id": "dep-1", "tools": ["tool-existing-1"]}),
+        tool=FakeToolClient([]),
+        connections=FakeConnectionsClient(),
+    )
+    captured: dict[str, Any] = {}
+
+    async def mock_create_connection_with_conflict_mapping(
+        *,
+        clients,
+        app_id,
+        payload,
+        user_id,
+        db,
+        error_prefix,
+        created_app_ids_journal=None,
+    ):
+        _ = clients, payload, user_id, db, error_prefix
+        if created_app_ids_journal is not None:
+            created_app_ids_journal.append(app_id)
+        msg = "boom-after-provider-create"
+        raise RuntimeError(msg)
+
+    async def mock_rollback_update_resources(*, clients, created_tool_ids, created_app_id, original_tools):  # noqa: ARG001
+        _ = (created_tool_ids, created_app_id, original_tools)
+
+    async def mock_rollback_created_app_ids(*, clients, created_app_ids):  # noqa: ARG001
+        captured["rolled_back_app_ids"] = list(created_app_ids)
+
+    monkeypatch.setattr(
+        shared_core_module,
+        "create_connection_with_conflict_mapping",
+        mock_create_connection_with_conflict_mapping,
+    )
+    monkeypatch.setattr(update_core_module, "rollback_update_resources", mock_rollback_update_resources)
+    monkeypatch.setattr(update_core_module, "rollback_created_app_ids", mock_rollback_created_app_ids)
+
+    with pytest.raises(RuntimeError, match="boom-after-provider-create"):
+        await update_core_module.apply_provider_update_plan_with_rollback(
+            clients=fake_clients,
+            user_id="user-1",
+            db=object(),
+            agent_id="dep-1",
+            agent={"id": "dep-1", "tools": ["tool-existing-1"]},
+            update_payload={},
+            plan=plan,
+        )
+
+    assert captured["rolled_back_app_ids"] == ["cfg-a"]
+
+
 @pytest.mark.anyio
 async def test_create_provider_data_prefixes_tool_and_deployment_names_but_not_connection_app_ids(monkeypatch):
     service = WatsonxOrchestrateDeploymentService(DummySettingsService())
@@ -2351,7 +2658,19 @@ async def test_create_provider_data_rolls_back_partially_created_raw_tools(monke
     async def mock_get_provider_clients(*, user_id, db):  # noqa: ARG001
         return fake_clients
 
-    async def mock_create_connection_with_conflict_mapping(*, clients, app_id, payload, user_id, db, error_prefix):  # noqa: ARG001
+    async def mock_create_connection_with_conflict_mapping(
+        *,
+        clients,
+        app_id,
+        payload,
+        user_id,
+        db,
+        error_prefix,  # noqa: ARG001
+        created_app_ids_journal=None,
+    ):
+        _ = clients, payload, user_id, db
+        if created_app_ids_journal is not None:
+            created_app_ids_journal.append(app_id)
         fake_connections._connections_by_app_id[app_id] = f"conn-{app_id}"
         return app_id
 
@@ -2368,7 +2687,7 @@ async def test_create_provider_data_rolls_back_partially_created_raw_tools(monke
 
     monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients)
     monkeypatch.setattr(
-        create_core_module,
+        shared_core_module,
         "create_connection_with_conflict_mapping",
         mock_create_connection_with_conflict_mapping,
     )