feat(deployment): add typed update rollback contracts and wxo rollback payloads

Add a typed DeploymentUpdateRollback contract and update_rollback payload slot across deployment interfaces/services, and introduce typed watsonx rollback payload models (including stricter agent rollback/update field typing) with schema test updates.

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

@@ -39,6 +39,7 @@ from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import
     verify_langflow_owned,
 )
 from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import (
+    WatsonxAgentRollbackUpdatePayload,
     WatsonxAttachToolOperation,
     WatsonxBindOperation,
     WatsonxDeploymentUpdatePayload,
@@ -381,8 +382,12 @@ async def _apply_tool_renames(
     logger.debug("_apply_tool_renames: renamed %d tools: %s", len(tool_updates), tool_renames)
 
 
-def _build_agent_rollback_payload(*, agent: dict[str, Any], final_update_payload: dict[str, Any]) -> dict[str, Any]:
-    rollback_payload: dict[str, Any] = {}
+def _build_agent_rollback_payload(
+    *,
+    agent: dict[str, Any],
+    final_update_payload: dict[str, Any],
+) -> WatsonxAgentRollbackUpdatePayload:
+    rollback_payload: WatsonxAgentRollbackUpdatePayload = {}
     if "tools" in final_update_payload:
         rollback_payload["tools"] = extract_agent_tool_ids(agent)
     for update_field in ("name", "display_name", "description", "llm"):
@@ -395,7 +400,7 @@ async def _rollback_agent_update(
     *,
     clients: WxOClient,
     agent_id: str,
-    rollback_agent_payload: dict[str, Any],
+    rollback_agent_payload: WatsonxAgentRollbackUpdatePayload,
 ) -> None:
     if not rollback_agent_payload:
         return
@@ -457,7 +462,7 @@ async def apply_provider_update_plan_with_rollback(
     added_snapshot_ids: list[str] = []
     created_snapshot_bindings: list[WatsonxResultToolRefBinding] = []
     final_update_payload = dict(update_payload)
-    rollback_agent_payload: dict[str, Any] = {}
+    rollback_agent_payload: WatsonxAgentRollbackUpdatePayload = {}
     created_app_ids_journal: list[str] = []
 
     # Pre-seed resolved_connections with bindings already attached to the

src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/payloads.py

@@ -3,17 +3,38 @@
 from __future__ import annotations
 
 from collections import Counter
-from typing import Annotated, Any, Literal
+from typing import Annotated, Any, Literal, TypedDict
 
 from lfx.services.adapters.deployment.payloads import DeploymentPayloadSchemas
 from lfx.services.adapters.deployment.schema import BaseFlowArtifact, EnvVarKey, EnvVarValueSpec, NormalizedId
 from lfx.services.adapters.payload import AdapterPayload, PayloadSlot
-from pydantic import BaseModel, ConfigDict, Field, StringConstraints, field_validator, model_validator
+from pydantic import AfterValidator, BaseModel, ConfigDict, Field, StringConstraints, field_validator, model_validator
 
 RawToolName = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1)]
 NormalizedStr = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1)]
 
 
+def _validate_nonempty_str(value: str) -> str:
+    """Reject empty or whitespace-only strings without mutating the value."""
+    if not value or not value.strip():
+        msg = "Value must not be empty or whitespace-only."
+        raise ValueError(msg)
+    return value
+
+
+NonEmptyStr = Annotated[str, AfterValidator(_validate_nonempty_str)]
+
+
+class WatsonxAgentRollbackUpdatePayload(TypedDict, total=False):
+    """Partial wxO agent update payload used by update/rollback helpers."""
+
+    tools: list[NonEmptyStr]
+    name: NonEmptyStr
+    display_name: NonEmptyStr
+    description: NonEmptyStr
+    llm: NonEmptyStr
+
+
 class WatsonxFlowArtifactProviderData(BaseModel):
     """Provider metadata for watsonx flow artifacts."""
 
@@ -487,6 +508,29 @@ class WatsonxToolAppBinding(BaseModel):
         return [str(app_id).strip() for app_id in value if str(app_id).strip()]
 
 
+class WatsonxDeploymentUpdateRollback(BaseModel):
+    """Pre-update rollback snapshot captured during a forward deployment update."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    rollback_agent_payload: WatsonxAgentRollbackUpdatePayload = Field(
+        default_factory=dict,
+        description="Pre-update agent restore payload (tools, name, display_name, description, llm).",
+    )
+    original_tools: dict[NonEmptyStr, dict] = Field(
+        default_factory=dict,
+        description="Writable pre-update payloads for mutated existing tools, keyed by tool id.",
+    )
+    created_tool_ids: list[NonEmptyStr] = Field(
+        default_factory=list,
+        description="Tool ids created during the forward update.",
+    )
+    created_app_ids: list[NonEmptyStr] = Field(
+        default_factory=list,
+        description="App/config ids created during the forward update.",
+    )
+
+
 class WatsonxDeploymentUpdateResultData(BaseModel):
     """Normalized provider result payload for deployment update.
 
@@ -519,6 +563,10 @@ class WatsonxDeploymentUpdateResultData(BaseModel):
     # Full operation correlation set (created + existing refs).
     referenced_snapshot_bindings: list[WatsonxResultToolRefBinding] = Field(default_factory=list)
     tool_app_bindings: list[WatsonxToolAppBinding] | None = None
+    rollback_data: WatsonxDeploymentUpdateRollback | None = Field(
+        default=None,
+        description="Pre-update rollback snapshot for DB-commit-failure compensation (adapter-only).",
+    )
 
     @field_validator("created_app_ids", mode="before")
     @classmethod
@@ -671,6 +719,7 @@ PAYLOAD_SCHEMAS = DeploymentPayloadSchemas(
     deployment_create_result=PayloadSlot(WatsonxDeploymentCreateResultData),
     deployment_update=PayloadSlot(WatsonxDeploymentUpdatePayload),
     deployment_update_result=PayloadSlot(WatsonxDeploymentUpdateResultData),
+    update_rollback=PayloadSlot(WatsonxDeploymentUpdateRollback),
     execution_create_result=PayloadSlot(WatsonxAgentExecutionResultData),
     execution_status_result=PayloadSlot(WatsonxAgentExecutionResultData),
     deployment_llm_list_result=PayloadSlot(WatsonxDeploymentLlmListResultData),

src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/service.py

@@ -46,6 +46,7 @@ from lfx.services.adapters.deployment.schema import (
     DeploymentType,
     DeploymentUpdate,
     DeploymentUpdateResult,
+    DeploymentUpdateRollback,
     ExecutionCreate,
     ExecutionCreateResult,
     ExecutionStatusResult,
@@ -282,6 +283,19 @@ class WatsonxOrchestrateDeploymentService(BaseDeploymentService):
             app_ids=result_data.app_ids,
         )
 
+    async def rollback_update_result(
+        self,
+        *,
+        user_id: IdLike,
+        deployment_id: IdLike,
+        payload: DeploymentUpdateRollback,
+        db: AsyncSession,
+    ) -> None:
+        """Best-effort rollback of provider-side update mutations from a pre-update journal."""
+        _ = (user_id, deployment_id, payload, db)
+        msg = f"{ErrorPrefix.UPDATE.value} Update rollback executor is not implemented."
+        raise DeploymentError(message=msg, error_code="deployment_error")
+
     async def list_types(
         self,
         *,

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

@@ -21,6 +21,7 @@ from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import (
     WatsonxDeploymentCreateResultData,
     WatsonxDeploymentUpdatePayload,
     WatsonxDeploymentUpdateResultData,
+    WatsonxDeploymentUpdateRollback,
     WatsonxFlowArtifactProviderData,
 )
 from langflow.services.adapters.deployment.watsonx_orchestrate.service import WatsonxOrchestrateDeploymentService
@@ -60,12 +61,42 @@ def test_payload_schema_slot_registered_for_deployment_update() -> None:
     assert slot.deployment_update.adapter_model is WatsonxDeploymentUpdatePayload
     assert slot.deployment_update_result is not None
     assert slot.deployment_update_result.adapter_model is WatsonxDeploymentUpdateResultData
+    assert slot.update_rollback is not None
+    assert slot.update_rollback.adapter_model is WatsonxDeploymentUpdateRollback
     assert slot.execution_create_result is not None
     assert slot.execution_create_result.adapter_model is WatsonxAgentExecutionResultData
     assert slot.execution_status_result is not None
     assert slot.execution_status_result.adapter_model is WatsonxAgentExecutionResultData
 
 
+def test_update_rollback_journal_round_trips_through_slot() -> None:
+    slot = WatsonxOrchestrateDeploymentService.payload_schemas
+    assert slot is not None
+    assert slot.update_rollback is not None
+
+    journal = {
+        "rollback_agent_payload": {"tools": ["tool-1"], "name": "agent-a"},
+        "original_tools": {"tool-2": {"name": "tool-2"}},
+        "created_tool_ids": ["tool-new"],
+        "created_app_ids": ["app-new"],
+    }
+    parsed = slot.update_rollback.parse(journal)
+    assert isinstance(parsed, WatsonxDeploymentUpdateRollback)
+    assert slot.update_rollback.dump(parsed) == journal
+
+
+def test_update_result_data_accepts_optional_rollback_journal() -> None:
+    result = WatsonxDeploymentUpdateResultData(
+        created_app_ids=["app-1"],
+        rollback_data=WatsonxDeploymentUpdateRollback(
+            rollback_agent_payload={"tools": ["tool-1"]},
+            created_tool_ids=["tool-new"],
+        ),
+    )
+    assert result.rollback_data is not None
+    assert result.rollback_data.created_tool_ids == ["tool-new"]
+
+
 def test_create_schema_accepts_raw_tool_pool_and_shared_connection_refs() -> None:
     slot = WatsonxOrchestrateDeploymentService.payload_schemas
     assert slot is not None

src/lfx/src/lfx/services/adapters/deployment/base.py

@@ -27,6 +27,7 @@ if TYPE_CHECKING:
         DeploymentType,
         DeploymentUpdate,
         DeploymentUpdateResult,
+        DeploymentUpdateRollback,
         ExecutionCreate,
         ExecutionCreateResult,
         ExecutionStatusResult,
@@ -114,6 +115,17 @@ class BaseDeploymentService(Service, ABC):
     ) -> DeploymentUpdateResult:
         """Update deployment inputs and apply changes in the provider."""
 
+    @abstractmethod
+    async def rollback_update_result(
+        self,
+        *,
+        user_id: IdLike,
+        deployment_id: IdLike,
+        payload: DeploymentUpdateRollback,
+        db: AsyncSession,
+    ) -> None:
+        """Rollback provider-side update mutations using a pre-update journal."""
+
     @abstractmethod
     async def redeploy(
         self,

src/lfx/src/lfx/services/adapters/deployment/payloads.py

@@ -24,6 +24,9 @@ T_DeploymentConfigModel = TypeVar("T_DeploymentConfigModel", bound=BaseModel, de
 T_DeploymentUpdate = TypeVar("T_DeploymentUpdate", default=AdapterPayload)
 T_DeploymentUpdateModel = TypeVar("T_DeploymentUpdateModel", bound=BaseModel, default=BaseModel)
 
+T_DeploymentUpdateRollback = TypeVar("T_DeploymentUpdateRollback", default=AdapterPayload)
+T_DeploymentUpdateRollbackModel = TypeVar("T_DeploymentUpdateRollbackModel", bound=BaseModel, default=BaseModel)
+
 T_DeploymentCreate = TypeVar("T_DeploymentCreate", default=AdapterPayload)
 T_DeploymentCreateModel = TypeVar("T_DeploymentCreateModel", bound=BaseModel, default=BaseModel)
 
@@ -119,6 +122,7 @@ class DeploymentPayloadFields(ProviderPayloadSchemas):
     deployment_create: PayloadSlot[T_DeploymentCreateModel] | None = None
     deployment_config: PayloadSlot[T_DeploymentConfigModel] | None = None
     deployment_update: PayloadSlot[T_DeploymentUpdateModel] | None = None
+    update_rollback: PayloadSlot[T_DeploymentUpdateRollbackModel] | None = None
     execution_input: PayloadSlot[T_ExecutionInputModel] | None = None
     deployment_list_params: PayloadSlot[T_DeploymentListParamsModel] | None = None
     config_list_params: PayloadSlot[T_ConfigListParamsModel] | None = None

src/lfx/src/lfx/services/adapters/deployment/schema.py

@@ -22,6 +22,7 @@ from lfx.services.adapters.deployment.payloads import (
     T_DeploymentStatusData,
     T_DeploymentUpdate,
     T_DeploymentUpdateResult,
+    T_DeploymentUpdateRollback,
     T_ExecutionCreateResult,
     T_ExecutionInput,
     T_ExecutionResult,
@@ -587,6 +588,14 @@ class DeploymentUpdateResult(DeploymentOperationResult[T_DeploymentUpdateResult]
     """Model representing a result for a deployment update operation."""
 
 
+class DeploymentUpdateRollback(BaseModel, Generic[T_DeploymentUpdateRollback]):
+    """Compensating rollback payload for failed deployment update DB commits."""
+
+    provider_data: T_DeploymentUpdateRollback = Field(
+        description="Provider-specific rollback journal captured before the forward update.",
+    )
+
+
 class RedeployResult(DeploymentOperationResult):
     """Model representing a deployment redeployment operation result."""
 

src/lfx/src/lfx/services/adapters/deployment/service.py

@@ -28,6 +28,7 @@ if TYPE_CHECKING:
         DeploymentType,
         DeploymentUpdate,
         DeploymentUpdateResult,
+        DeploymentUpdateRollback,
         ExecutionCreate,
         ExecutionCreateResult,
         ExecutionStatusResult,
@@ -114,6 +115,17 @@ class DeploymentService(BaseDeploymentService):
         """Update deployment inputs and apply changes in the provider."""
         raise DeploymentNotConfiguredError(method="update")
 
+    async def rollback_update_result(
+        self,
+        *,
+        user_id: IdLike,
+        deployment_id: IdLike,
+        payload: DeploymentUpdateRollback,
+        db: AsyncSession,
+    ) -> None:
+        """Rollback provider-side update mutations using a pre-update journal."""
+        raise DeploymentNotConfiguredError(method="rollback_update_result")
+
     async def redeploy(
         self,
         *,

src/lfx/src/lfx/services/interfaces.py

@@ -27,6 +27,7 @@ if TYPE_CHECKING:
         DeploymentType,
         DeploymentUpdate,
         DeploymentUpdateResult,
+        DeploymentUpdateRollback,
         ExecutionCreate,
         ExecutionCreateResult,
         ExecutionStatusResult,
@@ -314,8 +315,18 @@ class DeploymentServiceProtocol(Protocol):
         db: AsyncSession,
     ) -> DeploymentUpdateResult:
         """Update deployment inputs and apply changes in the provider."""
-        # TODO: Add a rollback-update interface contract for adapters so callers
-        # can compensate provider-side updates when downstream local sync fails.
+        ...
+
+    @abstractmethod
+    async def rollback_update_result(
+        self,
+        *,
+        user_id: IdLike,
+        deployment_id: IdLike,
+        payload: DeploymentUpdateRollback,
+        db: AsyncSession,
+    ) -> None:
+        """Rollback provider-side update mutations using a pre-update journal."""
         ...
 
     @abstractmethod

src/lfx/tests/unit/services/adapter_test_helpers.py

@@ -18,6 +18,7 @@ class DeploymentAdapterStub(BaseDeploymentService):
     async def list(self, **kw): ...
     async def get(self, **kw): ...
     async def update(self, **kw): ...
+    async def rollback_update_result(self, **kw): ...
     async def redeploy(self, **kw): ...
     async def duplicate(self, **kw): ...
     async def delete(self, **kw): ...

src/lfx/tests/unit/services/deployment/test_deployment_exceptions.py

@@ -146,6 +146,10 @@ def test_deployment_service_is_base_deployment_service() -> None:
         ("list_snapshots", {"user_id": "u1", "db": None}),
         ("get", {"user_id": "u1", "deployment_id": "d1", "db": None}),
         ("update", {"user_id": "u1", "deployment_id": "d1", "payload": None, "db": None}),
+        (
+            "rollback_update_result",
+            {"user_id": "u1", "deployment_id": "d1", "payload": None, "db": None},
+        ),
         ("redeploy", {"user_id": "u1", "deployment_id": "d1", "db": None}),
         ("duplicate", {"user_id": "u1", "deployment_id": "d1", "db": None}),
         ("delete", {"user_id": "u1", "deployment_id": "d1", "db": None}),

src/lfx/tests/unit/services/deployment/test_deployment_schema.py

@@ -20,6 +20,7 @@ from lfx.services.adapters.deployment.schema import (
     DeploymentType,
     DeploymentUpdate,
     DeploymentUpdateResult,
+    DeploymentUpdateRollback,
     EnvVarSource,
     EnvVarValueSpec,
     ExecutionCreate,
@@ -460,6 +461,16 @@ def test_deployment_update_result_uses_base_operation_shape() -> None:
     assert result.provider_result is None
 
 
+def test_deployment_update_rollback_works_with_provider_data() -> None:
+    rollback = DeploymentUpdateRollback(provider_data={"created_tool_ids": ["tool-1"]})
+    assert rollback.provider_data == {"created_tool_ids": ["tool-1"]}
+
+
+def test_deployment_update_rollback_rejects_missing_provider_data() -> None:
+    with pytest.raises(ValidationError, match="provider_data"):
+        DeploymentUpdateRollback()
+
+
 def test_operation_results_share_provider_result_contract() -> None:
     provider_result = {"accepted": True}