diff --git a/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/update.py b/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/update.py index a839eb1821..8c963cd68f 100644 --- a/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/update.py +++ b/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 diff --git a/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/payloads.py b/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/payloads.py index 85be21d8e5..61c5d943ad 100644 --- a/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/payloads.py +++ b/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), diff --git a/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/service.py b/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/service.py index 2d5cc383a6..f364cb03c4 100644 --- a/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/service.py +++ b/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, *, diff --git a/src/backend/tests/unit/services/deployment/test_watsonx_orchestrate_update_schema.py b/src/backend/tests/unit/services/deployment/test_watsonx_orchestrate_update_schema.py index 4867cd7f3f..571d811095 100644 --- a/src/backend/tests/unit/services/deployment/test_watsonx_orchestrate_update_schema.py +++ b/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 diff --git a/src/lfx/src/lfx/services/adapters/deployment/base.py b/src/lfx/src/lfx/services/adapters/deployment/base.py index 32e5d2a42d..d4e8ad6c00 100644 --- a/src/lfx/src/lfx/services/adapters/deployment/base.py +++ b/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, diff --git a/src/lfx/src/lfx/services/adapters/deployment/payloads.py b/src/lfx/src/lfx/services/adapters/deployment/payloads.py index b59ee3e0aa..30438ff758 100644 --- a/src/lfx/src/lfx/services/adapters/deployment/payloads.py +++ b/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 diff --git a/src/lfx/src/lfx/services/adapters/deployment/schema.py b/src/lfx/src/lfx/services/adapters/deployment/schema.py index c9e07f8f02..c9a81df996 100644 --- a/src/lfx/src/lfx/services/adapters/deployment/schema.py +++ b/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.""" diff --git a/src/lfx/src/lfx/services/adapters/deployment/service.py b/src/lfx/src/lfx/services/adapters/deployment/service.py index 6c951cdbf1..0a22f8d86e 100644 --- a/src/lfx/src/lfx/services/adapters/deployment/service.py +++ b/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, *, diff --git a/src/lfx/src/lfx/services/interfaces.py b/src/lfx/src/lfx/services/interfaces.py index 5cbf8ebaf8..2a51b21168 100644 --- a/src/lfx/src/lfx/services/interfaces.py +++ b/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 diff --git a/src/lfx/tests/unit/services/adapter_test_helpers.py b/src/lfx/tests/unit/services/adapter_test_helpers.py index f1d8aee8c7..9fe678f7dc 100644 --- a/src/lfx/tests/unit/services/adapter_test_helpers.py +++ b/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): ... diff --git a/src/lfx/tests/unit/services/deployment/test_deployment_exceptions.py b/src/lfx/tests/unit/services/deployment/test_deployment_exceptions.py index 23cfcb41a3..3bd920797f 100644 --- a/src/lfx/tests/unit/services/deployment/test_deployment_exceptions.py +++ b/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}), diff --git a/src/lfx/tests/unit/services/deployment/test_deployment_schema.py b/src/lfx/tests/unit/services/deployment/test_deployment_schema.py index fa9725d709..195f438b44 100644 --- a/src/lfx/tests/unit/services/deployment/test_deployment_schema.py +++ b/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}