fix: follow deployment boundary rules and fail fast (wxo) (#12539)

* feat(wxo): filter configs to key_value_creds, surface type and environment, harden mapper contract
Service layer (service.py):
- Filter list_configs to only return connections with security_scheme == "key_value_creds" in both tenant and deployment scopes
- Surface `environment` field alongside `type` in provider_data for config list items
- Normalize security_scheme via _normalize_optional_text to handle SDK enum/tuple quirks
- Extract _warn_if_expected_ids_missing helper to deduplicate staleness warnings
- Remove defensive isinstance checks on provider responses (trust the provider)
- Replace conflicting-binding error with last-write-wins (app_to_connection_id.update)
- Deduplicate connection IDs before calling get_drafts_by_ids
Mapper layer (mapper.py):
- Fail fast with HTTP 500 if config list item is missing a truthy `type`
- Conditionally include `environment` in shaped config list payload
Payloads (payloads.py):
- Add `environment: str | None` field to WatsonxApiConfigListItem with normalizing validator
Tests:
- Add test for deployment-scope key_value_creds filtering (mixed security schemes)
- Add test for tenant-scope key_value_creds filtering (oauth2 excluded)
- Add test for environment metadata passthrough in mapper and service
- Add tests for provider failure paths (tenant list failure, None response, tool fetch failure)
- Add tests for edge cases (latest-binding-wins, skip enrichment with no connections, malformed detailed connection)
- Update mapper tests to assert fail-fast on missing type
- Update existing fixtures to include security_scheme where required by new filtering

* refactor(deployments): move flow-version tool_name into provider_data
Move provider tool_name from a top-level flow-version response field into
provider_data, aligning API ownership boundaries for provider-originated
non-persisted fields.
- Remove top-level tool_name from DeploymentFlowVersionListItem
- Add tool_name to WatsonxApiDeploymentFlowVersionItemData with normalization
- Update WXO mapper to shape tool_name under provider_data
- Update frontend attachments type and consumer to read provider_data.tool_name
- Update backend tests for new response contract location
- Add explicit RULES.md requirement for non-persisted provider data placement

* make environment required

* refactor list configs method and fix broken tests

Author: HzaRashid

src/backend/base/langflow/api/v1/mappers/deployments/RULES.md

@@ -412,6 +412,7 @@ Use this checklist before merge:
 - [ ] Provider-account update logic lives in mapper, not in route conditionals
 - [ ] Provider-specific cross-field rules (e.g. tenant/URL coupling) are implemented as mapper overrides calling `super()`, not as base-class conditionals
 - [ ] Credential extraction uses `resolve_credential_fields`, not route-level assumptions about `provider_data` contents
+- [ ] Non-persisted provider-originated fields are placed in `provider_data` (never top-level)
 - [ ] DB-level consistency validators exist as defense-in-depth for cross-field invariants
 - [ ] Tests cover both base mapper defaults and provider overrides
 - [ ] Failure cases for missing/unexpected bindings are covered
@@ -556,3 +557,14 @@ When adding a new field to an execution or deployment response:
 1. Is Langflow the source of truth for this value? → top level.
 2. Does this value come from the provider and Langflow just relays it? → inside `provider_data`.
 3. Does the provider supply it but Langflow persists and indexes it (like `resource_key`)? → top level is acceptable.
+
+### 14.4 Hard placement rule for non-persisted provider data
+
+If data is not persisted in the Langflow DB and comes directly from the provider,
+it must go into `provider_data`.
+
+Rules:
+
+- Top-level response fields are reserved for values that Langflow persists and controls.
+- Provider-originated data that Langflow only relays must stay in `provider_data` without exception.
+- Examples: provider tool names, execution metadata/status/timestamps, connection types, environments.

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

@@ -10,7 +10,6 @@
 from lfx.services.adapters.deployment.schema import (
     BaseDeploymentData,
     BaseDeploymentDataUpdate,
-    ConfigListItem,
     ConfigListResult,
     DeploymentCreateResult,
     DeploymentListLlmsResult,
@@ -60,6 +59,7 @@
     WatsonxApiAddFlowItem,
     WatsonxApiAgentExecutionCreateResultData,
     WatsonxApiAgentExecutionStatusResultData,
+    WatsonxApiConfigListItem,
     WatsonxApiConfigListProviderData,
     WatsonxApiCreatedTool,
     WatsonxApiCreateUpsertToolItem,
@@ -173,6 +173,10 @@ class WatsonxOrchestrateDeploymentMapper(BaseDeploymentMapper):
             adapter_model=WatsonxApiConfigListProviderData,
             policy=PayloadSlotPolicy.VALIDATE_ONLY,
         ),
+        config_item_data=PayloadSlot(
+            adapter_model=WatsonxApiConfigListItem,
+            policy=PayloadSlotPolicy.VALIDATE_ONLY,
+        ),
         snapshot_list_result=PayloadSlot(
             adapter_model=WatsonxApiSnapshotListProviderData,
             policy=PayloadSlotPolicy.VALIDATE_ONLY,
@@ -1007,11 +1011,26 @@ def shape_config_list_result(
             msg = "Watsonx config_list_result payload slot is not configured."
             raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=msg)
 
-        items_all = [self._shape_config_list_item(item) for item in result.configs]
+        items_all: list[WatsonxApiConfigListItem] = []
+        for item in result.configs:
+            if not isinstance(item.provider_data, dict):
+                msg = "Invalid config item provider_data payload: expected non-null object."
+                raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=msg)
+            items_all.append(
+                self.shape_config_item_data(
+                    {
+                        **item.provider_data,
+                        "connection_id": item.id,
+                        "app_id": item.name,
+                    }
+                )
+            )
         total = len(items_all)
         offset = page_offset(page, size)
         provider_payload = {
-            "connections": items_all[offset : offset + size],
+            "connections": [
+                item.model_dump(mode="json", exclude_none=True) for item in items_all[offset : offset + size]
+            ],
             "page": page,
             "size": size,
             "total": total,
@@ -1093,8 +1112,10 @@ def shape_flow_version_list_result(
                 version_number=row.flow_version.version_number,
                 attached_at=row.attachment.created_at,
                 provider_snapshot_id=row.snapshot_id,
-                tool_name=snapshot_name_by_id.get(row.snapshot_id),
-                provider_data=self.shape_deployment_flow_version_item_data(snapshot_data_by_id.get(row.snapshot_id)),
+                provider_data=self.shape_deployment_flow_version_item_data(
+                    snapshot_data=snapshot_data_by_id.get(row.snapshot_id),
+                    tool_name=snapshot_name_by_id.get(row.snapshot_id),
+                ),
             )
             for row in normalized_rows
         ]
@@ -1159,12 +1180,13 @@ def _resolve_snapshot_name_by_id(
 
         Edge cases:
         - Provider unreachable / snapshot_result is None: returns ``{}``.
-          The ``tool_name`` field in the response will be ``None`` and the
-          frontend falls back to the Langflow flow name for display.
+          ``provider_data.tool_name`` will be absent/``None`` and the frontend
+          falls back to the Langflow flow name for display.
         - Tool renamed in wxO console: the new name is returned here since
           ``snapshot_result`` is fetched fresh on each request.
         - Tool deleted in wxO: missing from ``snapshot_result.snapshots``,
-          so no entry in the returned dict. ``tool_name`` will be ``None``.
+          so no entry in the returned dict. ``provider_data.tool_name`` will be
+          absent/``None``.
         """
         if not snapshot_result or not snapshot_result.snapshots:
             return {}
@@ -1178,17 +1200,21 @@ def _resolve_snapshot_name_by_id(
 
     def shape_deployment_flow_version_item_data(
         self,
+        *,
         snapshot_data: dict[str, Any] | None,
+        tool_name: str | None = None,
     ) -> dict[str, Any] | None:
-        if not snapshot_data:
-            return None
-        raw_connections = snapshot_data.get("connections")
-        if raw_connections is None or not isinstance(raw_connections, dict):
+        raw_connections = snapshot_data.get("connections") if snapshot_data else None
+        app_ids = list(raw_connections.keys()) if isinstance(raw_connections, dict) else []
+        if not app_ids and not tool_name:
             return None
         try:
             return self._validate_slot(
                 self.api_payloads.deployment_item_data,
-                {"app_ids": list(raw_connections.keys())},
+                {
+                    "app_ids": app_ids,
+                    "tool_name": tool_name,
+                },
             )
         except AdapterPayloadValidationError as exc:
             detail = exc.format_first_error()
@@ -1225,16 +1251,14 @@ def _shape_provider_deployment_list_entry(self, item: Any) -> dict[str, Any]:
                 detail=f"Invalid deployment list item provider_data payload: {detail}",
             ) from exc
 
-    def _shape_config_list_item(self, item: ConfigListItem) -> dict[str, Any]:
-        payload: dict[str, Any] = {
-            "connection_id": str(item.id).strip(),
-            "app_id": str(item.name).strip(),
-        }
-        item_provider_data = item.provider_data if isinstance(item.provider_data, dict) else {}
-        config_type = str(item_provider_data.get("type") or "").strip()
-        if config_type:
-            payload["type"] = config_type
-        return payload
+    def shape_config_item_data(self, provider_data: dict[str, Any]) -> WatsonxApiConfigListItem:
+        return self._parse_required_payload_slot(
+            slot=self.api_payloads.config_item_data,
+            slot_name="config_item_data",
+            raw=provider_data,
+            missing_payload_detail="Config item provider_data payload is missing.",
+            malformed_payload_detail="Invalid config item provider_data payload:",
+        )
 
     def _parse_required_payload_slot(
         self,

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

@@ -12,7 +12,6 @@
     BaseModel,
     Field,
     StringConstraints,
-    ValidationInfo,
     field_validator,
     model_validator,
 )
@@ -27,6 +26,16 @@
     ),
 ]
 
+# Keep API-boundary scalar normalization local to this module instead of
+# importing adapter-layer aliases, so mapper contracts can evolve independently.
+NormalizedStr = Annotated[
+    str,
+    StringConstraints(
+        strip_whitespace=True,
+        min_length=1,
+    ),
+]
+
 
 class WatsonxApiFlowArtifactProviderData(CreateFlowArtifactProviderData):
     """Watsonx create-time flow artifact provider_data contract."""
@@ -452,25 +461,10 @@ class WatsonxApiConfigListItem(BaseModel):
 
     model_config = {"extra": "forbid"}
 
-    connection_id: str = Field(min_length=1)
-    app_id: str = Field(min_length=1)
-    type: str | None = None
-
-    @field_validator("connection_id", "app_id", mode="before")
-    @classmethod
-    def normalize_required_strings(cls, value: Any, info: ValidationInfo) -> str:
-        normalized = str(value or "").strip()
-        if not normalized:
-            field_name = str(info.field_name)
-            msg = f"Config list item field '{field_name}' must be a non-empty string."
-            raise ValueError(msg)
-        return normalized
-
-    @field_validator("type", mode="before")
-    @classmethod
-    def normalize_optional_type(cls, value: Any) -> str | None:
-        normalized = str(value or "").strip()
-        return normalized or None
+    connection_id: NormalizedStr
+    app_id: NormalizedStr
+    type: NormalizedStr
+    environment: NormalizedStr
 
 
 class WatsonxApiConfigListProviderData(BaseModel):
@@ -520,6 +514,7 @@ class WatsonxApiDeploymentFlowVersionItemData(BaseModel):
     model_config = {"extra": "forbid"}
 
     app_ids: list[str] = Field(default_factory=list)
+    tool_name: str | None = None
 
     @field_validator("app_ids", mode="before")
     @classmethod
@@ -528,6 +523,22 @@ def normalize_app_ids(cls, value: Any) -> list[str]:
             return []
         return [str(app_id).strip() for app_id in value if str(app_id).strip()]
 
+    @field_validator("tool_name", mode="before")
+    @classmethod
+    def normalize_optional_tool_name(cls, value: Any) -> str | None:
+        normalized = str(value or "").strip()
+        return normalized or None
+
+
+class WatsonxApiRenameToolOperation(BaseModel):
+    """API-facing rename-tool operation payload."""
+
+    model_config = {"extra": "forbid"}
+
+    op: Literal["rename_tool"]
+    flow_version_id: str = Field(min_length=1)
+    tool_name: NormalizedStr = Field(min_length=1)
+
 
 class _WatsonxApiAgentExecutionResultBase(BaseModel):
     """Shared fields for API-facing agent execution result payloads.

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

@@ -331,15 +331,15 @@ class DeploymentFlowVersionListItem(BaseModel):
     distinguishes the following cases:
 
     * **Tool renamed in provider** — Same ``provider_snapshot_id``, different
-      ``tool_name``.  Langflow picks up the new name on the next fetch.
+      ``provider_data.tool_name``.  Langflow picks up the new name on the next fetch.
     * **Tool deleted in provider** — ``provider_snapshot_id`` no longer
-      resolves.  ``tool_name`` and ``provider_data`` will be ``None``.
+      resolves.  ``provider_data.tool_name`` may be missing/``None``.
     * **Tool deleted + new tool created with same name** — The new tool has
       a different ID.  Langflow's attachment still points to the old
       (missing) ID.  The new tool is invisible to Langflow until explicitly
       attached via an update operation.
 
-    Frontends should use ``tool_name`` for display and
+    Frontends should use ``provider_data.tool_name`` for display and
     ``provider_snapshot_id`` for identity / operations.
     """
 
@@ -358,15 +358,6 @@ class DeploymentFlowVersionListItem(BaseModel):
         default=None,
         description="Provider-owned snapshot/tool identifier linked by the attachment.",
     )
-    tool_name: str | None = Field(
-        default=None,
-        description=(
-            "Provider tool name for this snapshot. May differ from ``flow_name`` "
-            "if the user set a custom name at deploy time or renamed the tool in "
-            "the provider console. ``None`` when the provider is unreachable or "
-            "the tool has been deleted — frontends should fall back to ``flow_name``."
-        ),
-    )
     provider_data: dict[str, Any] | None = Field(
         default=None,
         description="Provider-owned opaque payload for this attached flow version.",