From 00b51d2d960f6dda2e4e441bba1f9fc9fc69de88 Mon Sep 17 00:00:00 2001 From: lda Date: Mon, 31 Aug 2026 00:45:40 +0700 Subject: [PATCH] feat: add validated workflow client boundary --- src/wf_client/__init__.py | 47 +++++ src/wf_client/codec.py | 191 +++++++++++++++++++++ src/wf_client/errors.py | 59 +++++++ src/wf_client/protocols.py | 152 ++++++++++++++++ src/wf_transport_rpc_http/client/base.py | 37 +++- tests/wf_client/__init__.py | 1 + tests/wf_client/conftest.py | 73 ++++++++ tests/wf_client/test_codec.py | 175 +++++++++++++++++++ tests/wf_transport_rpc_http/test_client.py | 31 ++++ 9 files changed, 762 insertions(+), 4 deletions(-) create mode 100644 src/wf_client/__init__.py create mode 100644 src/wf_client/codec.py create mode 100644 src/wf_client/errors.py create mode 100644 src/wf_client/protocols.py create mode 100644 tests/wf_client/__init__.py create mode 100644 tests/wf_client/conftest.py create mode 100644 tests/wf_client/test_codec.py diff --git a/src/wf_client/__init__.py b/src/wf_client/__init__.py new file mode 100644 index 00000000..69337a28 --- /dev/null +++ b/src/wf_client/__init__.py @@ -0,0 +1,47 @@ +"""Transport-independent workflow client primitives.""" + +from .codec import ( + DecodedRunResult, + DecodedTracePage, + decode_dependency_diagnostics, + decode_deployment, + decode_run_result, + decode_trace_result, + decode_workflow_artifact, +) +from .errors import ( + ArtifactNotFound, + ArtifactVersionConflict, + CapabilityNotFound, + DeploymentNotRunnable, + DeploymentRequired, + InvalidResponse, + ProtocolError, + RevisionConflict, + TransportError, + ValidationFailed, + WorkflowClientError, +) +from .protocols import WorkflowClientPort + +__all__ = [ + "ArtifactNotFound", + "ArtifactVersionConflict", + "CapabilityNotFound", + "DecodedRunResult", + "DecodedTracePage", + "DeploymentNotRunnable", + "DeploymentRequired", + "InvalidResponse", + "ProtocolError", + "RevisionConflict", + "TransportError", + "ValidationFailed", + "WorkflowClientError", + "WorkflowClientPort", + "decode_dependency_diagnostics", + "decode_deployment", + "decode_run_result", + "decode_trace_result", + "decode_workflow_artifact", +] diff --git a/src/wf_client/codec.py b/src/wf_client/codec.py new file mode 100644 index 00000000..651d4047 --- /dev/null +++ b/src/wf_client/codec.py @@ -0,0 +1,191 @@ +"""Validated conversions from workflow API wire payloads to domain values.""" + +from __future__ import annotations + +from dataclasses import dataclass +from dataclasses import fields as dataclass_fields +from typing import Any, TypeVar + +from pydantic import BaseModel, TypeAdapter, ValidationError + +from wf_api.models import ( + DependencyDiagnosticPayload, + RawWorkflowPlan, + RunResult, + RunTraceResult, + WorkflowArtifactPayload, + WorkflowDeploymentPayload, +) +from wf_artifacts.models import ( + DependencyDiagnostic, + WorkflowArtifact, + WorkflowDeployment, +) +from wf_core.models.workflow import Workflow + +from .errors import InvalidResponse + + +@dataclass(frozen=True, slots=True) +class _DecodedRunFields: + """Domain-shaped run fields shared by normal and trace responses.""" + + artifact_id: str + artifact_version: int + deployment_id: str + status: str + run_id: str | None + resume_readiness: str | None + interrupt: dict[str, Any] | None + outcome: str | None + error: str | None + output: dict[str, Any] | None + trace_count: int + diagnostics: tuple[DependencyDiagnostic, ...] + next_actions: dict[str, Any] + trace: tuple[dict[str, Any], ...] | None = None + trace_start: int | None = None + trace_limit: int | None = None + trace_truncated: bool | None = None + + +@dataclass(frozen=True, slots=True) +class DecodedRunResult(_DecodedRunFields): + """Validated run result without exposing a wire ``TypedDict``.""" + + +@dataclass(frozen=True, slots=True) +class DecodedTracePage(_DecodedRunFields): + """Validated bounded trace result without exposing wire dictionaries.""" + + +_PayloadT = TypeVar("_PayloadT") + + +def _validate(payload: object, schema: type[_PayloadT], operation: str) -> _PayloadT: + try: + return TypeAdapter(schema).validate_python(payload) + except (TypeError, ValueError, ValidationError) as exc: + raise InvalidResponse(operation=operation, details=str(exc)) from exc + + +_ModelT = TypeVar("_ModelT", bound=BaseModel) + + +def _model_validate(model: type[_ModelT], payload: object, operation: str) -> _ModelT: + try: + # Pydantic models are the canonical domain validation boundary. This + # helper keeps all malformed-response errors tied to their operation. + return model.model_validate(payload) + except (TypeError, ValueError, ValidationError) as exc: + raise InvalidResponse(operation=operation, details=str(exc)) from exc + + +def decode_workflow_artifact( + payload: object, +) -> tuple[WorkflowArtifact, Workflow]: + """Validate an artifact envelope and reconstruct its executable workflow.""" + operation = "workflow.artifacts.inspect" + wire = _validate(payload, WorkflowArtifactPayload, operation) + artifact = _model_validate(WorkflowArtifact, wire, operation) + raw_plan = _model_validate(RawWorkflowPlan, wire["plan"], operation) + workflow = _model_validate( + Workflow, + raw_plan.model_dump(mode="python", by_alias=True), + operation, + ) + return artifact, workflow + + +def decode_deployment(payload: object) -> WorkflowDeployment: + """Validate and reconstruct one immutable deployment model.""" + operation = "workflow.deployments.inspect" + wire = _validate(payload, WorkflowDeploymentPayload, operation) + return _model_validate(WorkflowDeployment, wire, operation) + + +def decode_dependency_diagnostics( + payload: object, +) -> tuple[DependencyDiagnostic, ...]: + """Decode deployment diagnostics into domain models.""" + return _decode_dependency_diagnostics(payload, "workflow.deployments.validate") + + +def _decode_dependency_diagnostics( + payload: object, + operation: str, +) -> tuple[DependencyDiagnostic, ...]: + wire = _validate(payload, list[DependencyDiagnosticPayload], operation) + return tuple( + _model_validate(DependencyDiagnostic, diagnostic, operation) + for diagnostic in wire + ) + + +def _decode_run_fields( + wire: RunResult | RunTraceResult, + *, + trace_required: bool, + operation: str, +) -> _DecodedRunFields: + diagnostics = _decode_dependency_diagnostics(wire["diagnostics"], operation) + raw_trace = wire.get("trace") + trace = None + if raw_trace is not None: + # Copy validated TypedDict values into ordinary dictionaries at the + # boundary so callers never receive transport DTO instances/types. + trace = tuple(dict(entry) for entry in raw_trace) + if trace_required and trace is None: + # RunTraceResult validation makes this unreachable; retain a defensive + # guard should its contract evolve. + raise InvalidResponse( + operation=operation, + details="trace result did not include a trace page", + ) + return _DecodedRunFields( + artifact_id=wire["artifact_id"], + artifact_version=wire["artifact_version"], + deployment_id=wire["deployment_id"], + status=wire["status"], + run_id=wire["run_id"], + resume_readiness=wire["resume_readiness"], + interrupt=dict(wire["interrupt"]) if wire["interrupt"] is not None else None, + outcome=wire["outcome"], + error=wire["error"], + output=dict(wire["output"]) if wire["output"] is not None else None, + trace_count=wire["trace_count"], + diagnostics=diagnostics, + next_actions=dict(wire["next_actions"]), + trace=trace, + trace_start=wire.get("trace_start"), + trace_limit=wire.get("trace_limit"), + trace_truncated=wire.get("trace_truncated"), + ) + + +def decode_run_result(payload: object) -> DecodedRunResult: + """Validate and decode a start/inspect/resume run response.""" + operation = "workflow.runs.inspect" + wire = _validate(payload, RunResult, operation) + fields = _decode_run_fields( + wire, + trace_required=False, + operation=operation, + ) + return DecodedRunResult( + *(getattr(fields, field.name) for field in dataclass_fields(_DecodedRunFields)) + ) + + +def decode_trace_result(payload: object) -> DecodedTracePage: + """Validate and decode a bounded run trace response.""" + operation = "workflow.runs.trace" + wire = _validate(payload, RunTraceResult, operation) + fields = _decode_run_fields( + wire, + trace_required=True, + operation=operation, + ) + return DecodedTracePage( + *(getattr(fields, field.name) for field in dataclass_fields(_DecodedRunFields)) + ) diff --git a/src/wf_client/errors.py b/src/wf_client/errors.py new file mode 100644 index 00000000..7069bebd --- /dev/null +++ b/src/wf_client/errors.py @@ -0,0 +1,59 @@ +"""Stable exceptions raised by the transport-independent workflow client.""" + +from __future__ import annotations + +from dataclasses import dataclass + + +class WorkflowClientError(Exception): + """Base class for errors that can be handled by workflow callers.""" + + +class TransportError(WorkflowClientError): + """The client could not communicate with the workflow service.""" + + +class ProtocolError(WorkflowClientError): + """The service returned a response that violates its protocol contract.""" + + +@dataclass(slots=True) +class InvalidResponse(ProtocolError): + """A response failed validation for one named workflow operation.""" + + operation: str + details: str + + def __post_init__(self) -> None: + object.__setattr__(self, "args", (str(self),)) + + def __str__(self) -> str: + return f"invalid response from {self.operation}: {self.details}" + + +class CapabilityNotFound(WorkflowClientError): + """The requested remote capability does not exist.""" + + +class ArtifactNotFound(WorkflowClientError): + """The requested immutable artifact version does not exist.""" + + +class ArtifactVersionConflict(WorkflowClientError): + """An artifact version conflicts with an existing saved version.""" + + +class DeploymentRequired(WorkflowClientError): + """An operation requires a deployment selection.""" + + +class DeploymentNotRunnable(WorkflowClientError): + """A selected deployment failed its readiness checks.""" + + +class ValidationFailed(WorkflowClientError): + """A workflow or dependency validation operation reported errors.""" + + +class RevisionConflict(WorkflowClientError): + """An editable workflow revision is stale.""" diff --git a/src/wf_client/protocols.py b/src/wf_client/protocols.py new file mode 100644 index 00000000..b294e333 --- /dev/null +++ b/src/wf_client/protocols.py @@ -0,0 +1,152 @@ +"""The narrow transport port consumed by rich workflow client objects.""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any, Literal, Protocol + +from wf_api.models import ( + CapabilityCallResult, + InspectCapabilityResult, + ListArtifactsResult, + ListCapabilitiesResult, + ListDeploymentsResult, + ListRunsResult, + RunResult, + RunTraceResult, + SaveArtifactResult, + SaveDeploymentResult, + ValidateArtifactPlanResult, + ValidateDeploymentResult, + WorkflowArtifactPayload, + WorkflowDeploymentPayload, +) +from wf_api.runs import TraceRangeLike + + +class WorkflowClientPort(Protocol): + """Minimum operation port required by transport-independent rich objects. + + This deliberately does not inherit ``WorkflowApiSurface``: that protocol + includes draft, source-admin, registry, and other operations that rich + workflow objects do not need. + """ + + async def list_capabilities( + self, + *, + query: str | None = None, + source_id: str | None = None, + cursor: str | None = None, + limit: int = 50, + ) -> ListCapabilitiesResult: ... + + async def inspect_capability( + self, + *, + qualified_name: str, + ) -> InspectCapabilityResult: ... + + async def call_capability( + self, + *, + qualified_name: str, + payload: dict[str, Any], + deployment_id: str | None = None, + ) -> CapabilityCallResult: ... + + async def list_artifacts( + self, + *, + query: str | None = None, + kind: Literal["workflow", "wrapper"] | None = None, + cursor: str | None = None, + limit: int = 50, + ) -> ListArtifactsResult: ... + + async def inspect_artifact( + self, + *, + artifact_id: str, + version: int, + ) -> WorkflowArtifactPayload: ... + + async def save_artifact(self, artifact: dict[str, Any]) -> SaveArtifactResult: ... + + async def create_artifact_from_plan( + self, + *, + artifact_id: str, + version: int, + title: str, + plan: dict[str, Any], + outcomes: Sequence[str], + kind: Literal["workflow", "wrapper"] = "workflow", + description: str | None = None, + required_capabilities: dict[str, dict[str, Any]] | None = None, + source_bindings: dict[str, str] | None = None, + created_from_catalog_version: str | None = None, + ) -> SaveArtifactResult: ... + + async def validate_artifact_plan( + self, + *, + plan: dict[str, Any], + outcomes: Sequence[str], + required_capabilities: dict[str, dict[str, Any]] | None = None, + source_bindings: dict[str, str] | None = None, + ) -> ValidateArtifactPlanResult: ... + + async def list_deployments(self) -> ListDeploymentsResult: ... + + async def inspect_deployment( + self, + *, + deployment_id: str, + ) -> WorkflowDeploymentPayload: ... + + async def save_deployment( + self, + deployment: dict[str, Any], + ) -> SaveDeploymentResult: ... + + async def validate_deployment( + self, + *, + deployment_id: str, + live_check: bool = False, + ) -> ValidateDeploymentResult: ... + + async def list_runs( + self, + *, + status: str | None = None, + cursor: str | None = None, + limit: int = 50, + ) -> ListRunsResult: ... + + async def run_deployment( + self, + *, + deployment_id: str, + workflow_input: dict[str, Any], + trace_range: TraceRangeLike | None = None, + ) -> RunResult: ... + + async def inspect_run(self, *, run_id: str) -> RunResult: ... + + async def resume_run( + self, + *, + run_id: str, + resume_payload: dict[str, Any], + resume_outcome: str = "submitted", + trace_range: TraceRangeLike | None = None, + ) -> RunResult: ... + + async def read_run_trace( + self, + *, + run_id: str, + trace_range: TraceRangeLike, + ) -> RunTraceResult: ... diff --git a/src/wf_transport_rpc_http/client/base.py b/src/wf_transport_rpc_http/client/base.py index 9d094c0b..6dd4b20c 100644 --- a/src/wf_transport_rpc_http/client/base.py +++ b/src/wf_transport_rpc_http/client/base.py @@ -7,6 +7,31 @@ from uuid import uuid4 import httpx +@dataclass(frozen=True, slots=True) +class RpcProtocolError(RuntimeError): + """Structured JSON-RPC application error returned by a remote endpoint. + + The exception intentionally remains a ``RuntimeError`` for compatibility + with the existing CLI's remote-operation handling, while retaining the + server's machine-readable code and data for richer clients. + """ + + code: int | str | None + message: str + data: object = None + + def __post_init__(self) -> None: + # ``Exception`` stores positional arguments separately from dataclass + # fields; populate them so generic exception tooling sees the useful + # rendered message as well. + object.__setattr__(self, "args", (str(self),)) + + def __str__(self) -> str: + if isinstance(self.data, dict) and isinstance(self.data.get("message"), str): + return f"{self.message}: {self.data['message']}" + return self.message + + class RpcCaller(Protocol): """Transport primitive required by domain RPC client mixins.""" @@ -41,11 +66,15 @@ class RpcClientTransport: payload = response.json() if "error" in payload: error = payload["error"] + if not isinstance(error, dict): + raise RpcProtocolError(None, "JSON-RPC error", error) + code = error.get("code") + if not isinstance(code, (int, str)): + code = None message = error.get("message", "JSON-RPC error") - data = error.get("data") - if isinstance(data, dict) and data.get("message"): - message = f"{message}: {data['message']}" - raise RuntimeError(message) + if not isinstance(message, str): + message = "JSON-RPC error" + raise RpcProtocolError(code, message, error.get("data")) result = payload.get("result") if not isinstance(result, dict): raise RuntimeError("JSON-RPC response result must be an object") diff --git a/tests/wf_client/__init__.py b/tests/wf_client/__init__.py new file mode 100644 index 00000000..f2442955 --- /dev/null +++ b/tests/wf_client/__init__.py @@ -0,0 +1 @@ +"""Tests for the transport-independent workflow client boundary.""" diff --git a/tests/wf_client/conftest.py b/tests/wf_client/conftest.py new file mode 100644 index 00000000..26492796 --- /dev/null +++ b/tests/wf_client/conftest.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +from typing import Any + +import pytest + + +class FakeWorkflowClient: + """Small configurable adapter used to test rich clients without HTTP.""" + + def __init__(self, **responses: object) -> None: + self.responses = responses + self.calls: list[tuple[str, dict[str, Any]]] = [] + + def response(self, method: str) -> object: + return self.responses[method] + + def _response(self, method: str, params: dict[str, Any]) -> object: + self.calls.append((method, params)) + return self.response(method) + + async def list_capabilities(self, **params: Any) -> object: + return self._response("workflow.capabilities.list", params) + + async def inspect_capability(self, **params: Any) -> object: + return self._response("workflow.capabilities.inspect", params) + + async def call_capability(self, **params: Any) -> object: + return self._response("workflow.capabilities.call", params) + + async def inspect_artifact(self, **params: Any) -> object: + return self._response("workflow.artifacts.inspect", params) + + async def save_artifact(self, artifact: dict[str, Any]) -> object: + return self._response("workflow.artifacts.save", {"artifact": artifact}) + + async def validate_artifact_plan(self, **params: Any) -> object: + return self._response("workflow.artifacts.validate_plan", params) + + async def create_artifact_from_plan(self, **params: Any) -> object: + return self._response("workflow.artifacts.create_from_plan", params) + + async def list_deployments(self) -> object: + return self._response("workflow.deployments.list", {}) + + async def inspect_deployment(self, **params: Any) -> object: + return self._response("workflow.deployments.inspect", params) + + async def save_deployment(self, deployment: dict[str, Any]) -> object: + return self._response("workflow.deployments.save", {"deployment": deployment}) + + async def validate_deployment(self, **params: Any) -> object: + return self._response("workflow.deployments.validate", params) + + async def run_deployment(self, **params: Any) -> object: + return self._response("workflow.runs.start", params) + + async def inspect_run(self, **params: Any) -> object: + return self._response("workflow.runs.inspect", params) + + async def resume_run(self, **params: Any) -> object: + return self._response("workflow.runs.resume", params) + + async def read_run_trace(self, **params: Any) -> object: + return self._response("workflow.runs.trace", params) + + async def _call(self, method: str, params: dict[str, Any]) -> object: + return self._response(method, params) + + +@pytest.fixture +def fake_client() -> FakeWorkflowClient: + return FakeWorkflowClient() diff --git a/tests/wf_client/test_codec.py b/tests/wf_client/test_codec.py new file mode 100644 index 00000000..3c4e8c05 --- /dev/null +++ b/tests/wf_client/test_codec.py @@ -0,0 +1,175 @@ +from __future__ import annotations + +from typing import Any + +import pytest + +from wf_client.codec import ( + decode_dependency_diagnostics, + decode_deployment, + decode_run_result, + decode_trace_result, + decode_workflow_artifact, +) +from wf_client.errors import InvalidResponse + + +def _constant_plan_payload() -> dict[str, Any]: + return { + "name": "report", + "input_schema": {"type": "object", "properties": {}}, + "state_schema": { + "type": "object", + "properties": {"result": {"type": "string", "reducer": "wf.std.replace"}}, + }, + "output_schema": { + "type": "object", + "properties": {"result": {"type": "string"}}, + "required": ["result"], + }, + "outcomes": ["ok"], + "start": "constant", + "nodes": [ + { + "id": "constant", + "type": "node", + "node": "wf.std.constant", + "input": [{"value": "hello", "target": "local.value"}], + "output": [{"source": "local.value", "target": "state.result"}], + } + ], + "edges": [{"from": "constant", "outcome": "ok", "to": "__end__"}], + "output": [{"path": "state.result", "target": "result"}], + } + + +def _workflow_artifact_payload(*, plan: dict[str, Any]) -> dict[str, Any]: + return { + "id": "report", + "version": 1, + "title": "Report", + "kind": "workflow", + "description": None, + "input_schema": {"type": "object", "properties": {}}, + "output_schema": { + "type": "object", + "properties": {"result": {"type": "string"}}, + }, + "outcomes": ["ok"], + "plan": plan, + "required_capabilities": [], + "workflow_dependencies": {}, + "created_from_catalog_version": None, + } + + +def test_decode_workflow_artifact_validates_plan() -> None: + artifact, workflow = decode_workflow_artifact( + _workflow_artifact_payload(plan=_constant_plan_payload()) + ) + + assert artifact.id == "report" + assert workflow.name == "report" + assert workflow.start == "constant" + + +def test_decode_workflow_artifact_rejects_invalid_nested_plan() -> None: + with pytest.raises(InvalidResponse, match="workflow.artifacts.inspect"): + decode_workflow_artifact(_workflow_artifact_payload(plan={"name": "broken"})) + + +def test_decode_deployment_returns_domain_model() -> None: + deployment = decode_deployment( + { + "id": "production", + "artifact_id": "report", + "artifact_version": 1, + "bindings": [], + "drift_policy": "block", + } + ) + + assert deployment.id == "production" + assert deployment.artifact_id == "report" + + +def test_decode_dependency_diagnostics_returns_domain_models() -> None: + diagnostics = decode_dependency_diagnostics( + [ + { + "severity": "error", + "code": "missing_source", + "logical_ref": "demo", + "bound_source": None, + "message": "missing", + "repair_hint": "bind it", + } + ] + ) + + assert diagnostics[0].code == "missing_source" + assert diagnostics[0].severity.value == "error" + + +def _run_payload(*, trace: list[dict[str, Any]] | None = None) -> dict[str, Any]: + payload: dict[str, Any] = { + "artifact_id": "report", + "artifact_version": 1, + "deployment_id": "production", + "status": "completed", + "run_id": "run-1", + "resume_readiness": None, + "interrupt": None, + "outcome": "ok", + "error": None, + "output": {"result": "hello"}, + "trace_count": 0 if trace is None else len(trace), + "diagnostics": [], + "next_actions": { + "can_continue": False, + "can_save_now": None, + "recommended_next_tool": None, + "reason": "done", + "patch_examples": [], + "warnings": [], + }, + } + if trace is not None: + payload.update( + trace=trace, + trace_start=0, + trace_limit=25, + trace_truncated=False, + ) + return payload + + +def test_decode_run_result_returns_typed_domain_boundary() -> None: + result = decode_run_result(_run_payload()) + + assert result.run_id == "run-1" + assert result.output == {"result": "hello"} + assert result.diagnostics == () + + +def test_decode_trace_result_decodes_bounded_trace() -> None: + result = decode_trace_result( + _run_payload( + trace=[ + { + "frame_id": "root", + "node_id": "constant", + "step_type": "node", + "resolved_input": {}, + "outcome": "ok", + "next_node_id": "__end__", + "output": {}, + "state_changes": {}, + } + ] + ) + ) + + assert result.trace_start == 0 + assert result.trace is not None + assert result.trace[0]["node_id"] == "constant" diff --git a/tests/wf_transport_rpc_http/test_client.py b/tests/wf_transport_rpc_http/test_client.py index acfbdbf8..7df5626b 100644 --- a/tests/wf_transport_rpc_http/test_client.py +++ b/tests/wf_transport_rpc_http/test_client.py @@ -26,10 +26,41 @@ from wf_core.models.steps import ( from wf_core.paths import GraphSourcePath, LocalPath, StatePath from wf_server import build_local_static_workflow_server from wf_transport_rpc_http import RpcWorkflowApiClient, create_rpc_app +from wf_transport_rpc_http.client.base import RpcProtocolError from wf_transport_rpc_http.client.drafts import RpcDraftClientMixin from wf_transport_rpc_http.client.sources import RpcSourceAdminClientMixin +async def test_rpc_client_preserves_structured_jsonrpc_error() -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + json={ + "jsonrpc": "2.0", + "id": "request", + "error": { + "code": "missing_source", + "message": "workflow operation failed", + "data": {"message": "source is not configured", "hint": "bind it"}, + }, + }, + ) + + http_client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + async with http_client: + client = RpcWorkflowApiClient(url="http://test/rpc", http_client=http_client) + with pytest.raises(RpcProtocolError) as raised: + await client.list_capabilities() + + assert raised.value.code == "missing_source" + assert raised.value.message == "workflow operation failed" + assert raised.value.data == { + "message": "source is not configured", + "hint": "bind it", + } + assert str(raised.value) == ("workflow operation failed: source is not configured") + + def _constant_plan() -> RawWorkflowPlan: return RawWorkflowPlan.model_validate( {