From 370e1bd57cbce6c5febf95f032f5b6aa9b05d00e Mon Sep 17 00:00:00 2001 From: lda Date: Mon, 31 Aug 2026 01:22:58 +0700 Subject: [PATCH] feat: add editable remote workflow lifecycle --- .../lda_report_workflow/build_workflow.py | 16 +- src/wf_client/__init__.py | 16 ++ src/wf_client/app.py | 28 ++- src/wf_client/authoring.py | 237 ++++++++++++++++++ src/wf_client/codec.py | 23 +- src/wf_client/workflows.py | 116 +++++++++ tests/wf_client/test_authoring.py | 120 +++++++++ 7 files changed, 535 insertions(+), 21 deletions(-) create mode 100644 src/wf_client/authoring.py create mode 100644 src/wf_client/workflows.py create mode 100644 tests/wf_client/test_authoring.py diff --git a/examples/lda_report_workflow/build_workflow.py b/examples/lda_report_workflow/build_workflow.py index 5585600e..d5a97569 100644 --- a/examples/lda_report_workflow/build_workflow.py +++ b/examples/lda_report_workflow/build_workflow.py @@ -19,12 +19,7 @@ WORKFLOW_OUTPUT = [ def build_workflow() -> Workflow: - """Build the demo workflow with the public authoring API. - - `WorkflowBuilder` does not yet expose a workflow-output setter, so this - module adds the final output projection in `_with_workflow_output()` after - compiling the graph. Keep that seam small and validated. - """ + """Build the demo workflow with the public authoring API.""" builder = WorkflowBuilder( name="lda_report_case_study", input_schema={ @@ -185,13 +180,8 @@ def build_workflow() -> Workflow: builder.connect(create_issues, "ok", finalise) builder.connect(finalise, "ok", end_completed) builder.connect(revision_requested, "ok", end_cancelled) - return _with_workflow_output(builder.compile()) - - -def _with_workflow_output(workflow: Workflow) -> Workflow: - payload = workflow.model_dump(mode="json", by_alias=True) - payload["output"] = WORKFLOW_OUTPUT - return Workflow.model_validate(payload) + builder.set_output(WORKFLOW_OUTPUT) + return builder.compile() def workflow_plan_payload() -> dict[str, Any]: diff --git a/src/wf_client/__init__.py b/src/wf_client/__init__.py index 7972503e..d599f25d 100644 --- a/src/wf_client/__init__.py +++ b/src/wf_client/__init__.py @@ -3,6 +3,7 @@ from wf_platform import CapabilityRef, Page from .app import App +from .authoring import EditableWorkflow from .capabilities import CapabilityResult, CapabilitySummary, RemoteCapability from .codec import ( DecodedRunResult, @@ -15,6 +16,7 @@ from .codec import ( decode_deployment, decode_run_result, decode_trace_result, + decode_validate_artifact_plan, decode_workflow_artifact, ) from .errors import ( @@ -31,17 +33,26 @@ from .errors import ( WorkflowClientError, ) from .protocols import WorkflowClientPort +from .workflows import ( + ArtifactRef, + Diagnostic, + WorkflowArtifact, + WorkflowDiagnostic, + WorkflowValidation, +) __all__ = [ "ArtifactNotFound", "ArtifactVersionConflict", "App", + "ArtifactRef", "CapabilityNotFound", "CapabilityRef", "CapabilityResult", "CapabilitySummary", "DecodedRunResult", "DecodedTracePage", + "Diagnostic", "DeploymentNotRunnable", "DeploymentRequired", "InvalidResponse", @@ -53,6 +64,10 @@ __all__ = [ "ValidationFailed", "WorkflowClientError", "WorkflowClientPort", + "EditableWorkflow", + "WorkflowArtifact", + "WorkflowDiagnostic", + "WorkflowValidation", "decode_capabilities_page", "decode_capability_call", "decode_capability_diagnostics", @@ -61,5 +76,6 @@ __all__ = [ "decode_deployment", "decode_run_result", "decode_trace_result", + "decode_validate_artifact_plan", "decode_workflow_artifact", ] diff --git a/src/wf_client/app.py b/src/wf_client/app.py index 83bd1ffa..94703d78 100644 --- a/src/wf_client/app.py +++ b/src/wf_client/app.py @@ -6,17 +6,19 @@ from collections.abc import Sequence from dataclasses import dataclass, field from typing import Any -from wf_authoring import WorkflowBuilder from wf_platform import CapabilityRef, Page, SourceRef from wf_transport_rpc_http import RpcWorkflowApiClient +from .authoring import EditableWorkflow from .capabilities import CapabilitySummary, RemoteCapability from .codec import ( decode_capabilities_page, decode_capability_inspect, + decode_workflow_artifact, ) from .errors import InvalidResponse from .protocols import WorkflowClientPort +from .workflows import WorkflowArtifact def _capability_ref(qualified_name: str, source_id: str) -> CapabilityRef: @@ -145,12 +147,32 @@ class App: state_schema: Any, output_schema: Any, outcomes: Sequence[str] = ("ok",), - ) -> WorkflowBuilder: + ) -> EditableWorkflow: """Construct a local builder; remote operations remain opt-in/async.""" - return WorkflowBuilder( + return EditableWorkflow( + _port=self._port, name=name, input_schema=input_schema, state_schema=state_schema, output_schema=output_schema, outcomes=outcomes, ) + + async def workflow(self, artifact_id: str, *, version: int) -> WorkflowArtifact: + """Inspect and reconstruct one exact immutable workflow artifact version.""" + artifact, workflow = decode_workflow_artifact( + await self._port.inspect_artifact( + artifact_id=artifact_id, + version=version, + ) + ) + return WorkflowArtifact(self._port, artifact, workflow) + + async def edit_workflow( + self, + artifact_id: str, + *, + version: int, + ) -> EditableWorkflow: + """Inspect an exact artifact version and seed an editable builder.""" + return (await self.workflow(artifact_id, version=version)).edit() diff --git a/src/wf_client/authoring.py b/src/wf_client/authoring.py new file mode 100644 index 00000000..1d7726c9 --- /dev/null +++ b/src/wf_client/authoring.py @@ -0,0 +1,237 @@ +"""Editable workflow authoring over the transport-independent builder.""" + +from __future__ import annotations + +from collections.abc import Sequence +from copy import deepcopy +from dataclasses import dataclass, field +from typing import Any, overload + +from wf_authoring import WorkflowBuilder +from wf_authoring.builder.mapping import OutputBindingArg, StepInputBindingArg +from wf_authoring.nodes import NodeSpec +from wf_core import ( + NodeDef, + NodeUse, + SchemaRef, + SubgraphNode, + ValidationReport, + Workflow, +) + +from .capabilities import RemoteCapability +from .codec import decode_validate_artifact_plan, decode_workflow_artifact +from .protocols import WorkflowClientPort +from .workflows import ( + ArtifactRef, + WorkflowArtifact, + WorkflowDiagnostic, + WorkflowValidation, +) + + +@dataclass(slots=True) +class EditableWorkflow(WorkflowBuilder): + """A mutable ``WorkflowBuilder`` carrying the client port used to save it.""" + + _port: WorkflowClientPort = field(repr=False, kw_only=True) + based_on: ArtifactRef | None = field(default=None, kw_only=True) + artifact_title: str | None = field(default=None, kw_only=True) + artifact_description: str | None = field(default=None, kw_only=True) + _source_plan: dict[str, Any] | None = field(default=None, repr=False, kw_only=True) + _source_workflow: Workflow | None = field( + default=None, repr=False, kw_only=True + ) + + @classmethod + def from_artifact(cls, artifact: WorkflowArtifact) -> EditableWorkflow: + """Copy every canonical graph field from an immutable artifact snapshot.""" + builder = WorkflowBuilder.from_workflow(artifact.workflow) + _seed_remote_node_defs(builder, artifact) + return cls( + _port=artifact._port, + based_on=artifact.ref, + artifact_title=artifact.title, + artifact_description=artifact.description, + name=builder.name, + input_schema=builder.input_schema, + state_schema=builder.state_schema, + output_schema=builder.output_schema, + outcomes=builder.outcomes, + start=builder.start, + reducers=builder.reducers, + node_specs=dict(builder.node_specs), + nodes=builder.nodes, + edges=builder.edges, + workflow_output=builder.workflow_output, + seeded_node_defs=builder.seeded_node_defs, + prepared_subgraphs=builder.prepared_subgraphs, + _source_plan=deepcopy(artifact.artifact.plan), + _source_workflow=builder._build_workflow(start=builder.start or ""), + ) + + @overload + def use( + self, + spec: NodeSpec[Any, Any], + *, + id: str | None = None, + input: Sequence[StepInputBindingArg] | None = None, + output: Sequence[OutputBindingArg] | None = None, + desc: str | None = None, + ) -> NodeUse: ... + + @overload + def use( + self, + spec: RemoteCapability, + *, + id: str | None = None, + input: Sequence[StepInputBindingArg] | None = None, + output: Sequence[OutputBindingArg] | None = None, + desc: str | None = None, + ) -> NodeUse: ... + + def use( + self, + spec: NodeSpec[Any, Any] | RemoteCapability, + **kwargs: Any, + ) -> NodeUse: + if isinstance(spec, RemoteCapability): + return self.use_contract(spec.node_def(), **kwargs) + return super().use(spec, **kwargs) + + def subgraph( + self, + workflow: WorkflowArtifact, + *, + id: str | None = None, + input: Sequence[StepInputBindingArg] | None = None, + output: Sequence[OutputBindingArg] | None = None, + desc: str | None = None, + ) -> SubgraphNode: + """Add a native subgraph pinned to an immutable artifact version.""" + return super().subgraph( + workflow=workflow.workflow, + workflow_ref={ + "artifact_id": workflow.ref.artifact_id, + "version": workflow.ref.version, + }, + id=id, + input=input, + output=output, + desc=desc, + ) + + def validate_local(self) -> ValidationReport: + """Validate graph structure locally without touching the transport.""" + return self.validate_structure() + + def _plan(self) -> tuple[Workflow, dict[str, Any]]: + workflow = self.compile() + if ( + self._source_plan is not None + and self._source_workflow is not None + and workflow.model_dump(mode="json", by_alias=True) + == self._source_workflow.model_dump(mode="json", by_alias=True) + ): + # Pydantic canonical models add omitted defaults during a round trip + # (for example ``required=[]``). Keep an untouched artifact's raw + # plan byte-for-byte structural equivalent until it is edited. + return workflow, deepcopy(self._source_plan) + plan = workflow.model_dump(mode="json", by_alias=True) + # Node definitions are local execution metadata, not part of the raw + # persisted artifact plan; the server inventories these from node refs. + plan.pop("node_defs", None) + return workflow, plan + + async def validate(self) -> WorkflowValidation: + local = self.validate_local() + if not local.ok: + return WorkflowValidation(local, "not_run", ()) + + _workflow, plan = self._plan() + wire = decode_validate_artifact_plan( + await self._port.validate_artifact_plan( + plan=plan, + outcomes=tuple(self.outcomes), + ) + ) + diagnostics = tuple( + WorkflowDiagnostic( + severity=item["severity"], + code=item["code"], + path=item["path"], + message=item["message"], + repair_hint=item["repair_hint"], + ) + for item in wire["diagnostics"] + ) + return WorkflowValidation(local, wire["status"], diagnostics) + + async def save( + self, + *, + artifact_id: str | None = None, + version: int, + title: str | None = None, + description: str | None = None, + ) -> WorkflowArtifact: + validation = await self.validate() + validation.raise_for_errors() + _workflow, plan = self._plan() + saved_id = artifact_id or (self.based_on.artifact_id if self.based_on else self.name) + saved_title = title if title is not None else self.artifact_title or self.name + saved_description = ( + description if description is not None else self.artifact_description + ) + await self._port.create_artifact_from_plan( + artifact_id=saved_id, + version=version, + title=saved_title, + plan=plan, + outcomes=tuple(self.outcomes), + description=saved_description, + ) + # The acknowledgement is only an identity signal. Inspecting the exact + # requested version ensures server normalization is retained losslessly. + inspected = await self._port.inspect_artifact( + artifact_id=saved_id, + version=version, + ) + artifact, workflow = decode_workflow_artifact(inspected) + return WorkflowArtifact(self._port, artifact, workflow) + + +def _seed_remote_node_defs( + builder: WorkflowBuilder, + artifact: WorkflowArtifact, +) -> None: + """Restore remote node contracts retained as artifact dependency snapshots.""" + node_name_by_step_id = { + node.id: node.node + for node in artifact.workflow.nodes + if isinstance(node, NodeUse) + } + outcomes_by_node: dict[str, list[str]] = {} + for edge in artifact.workflow.edges: + node_name = node_name_by_step_id.get(edge.from_) + if node_name is not None: + outcomes_by_node.setdefault(node_name, []).append(edge.outcome) + for requirement in artifact.required_capabilities: + if requirement.kind != "node_spec": + continue + input_schema = requirement.input_schema_snapshot + output_schema = requirement.output_schema_snapshot + if not isinstance(input_schema, dict) or not isinstance(output_schema, dict): + continue + name = str(requirement.capability_ref()) + builder.seeded_node_defs.setdefault( + name, + NodeDef( + name=name, + input_schema=SchemaRef.model_validate(input_schema), + output_schema=SchemaRef.model_validate(output_schema), + outcomes=outcomes_by_node.get(name, ["ok"]), + ), + ) diff --git a/src/wf_client/codec.py b/src/wf_client/codec.py index 32095024..a4d00f3b 100644 --- a/src/wf_client/codec.py +++ b/src/wf_client/codec.py @@ -16,6 +16,7 @@ from wf_api.models import ( RawWorkflowPlan, RunResult, RunTraceResult, + ValidateArtifactPlanResult, WorkflowArtifactPayload, WorkflowDeploymentPayload, ) @@ -109,6 +110,15 @@ def decode_capabilities_page(payload: object) -> ListCapabilitiesResult: ) +def decode_validate_artifact_plan(payload: object) -> ValidateArtifactPlanResult: + """Validate a non-persisting artifact-plan response at the client boundary.""" + return _validate( + payload, + ValidateArtifactPlanResult, + "workflow.artifacts.validate_plan", + ) + + _ModelT = TypeVar("_ModelT", bound=BaseModel) @@ -129,11 +139,14 @@ def decode_workflow_artifact( 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, - ) + workflow_payload = raw_plan.model_dump(mode="python", by_alias=True) + # Raw artifact plans normally omit node definitions because the server + # inventories remote contracts. Preserve them when a caller supplies them + # so an inspect/edit/save round trip remains lossless. + raw_node_defs = wire["plan"].get("node_defs") + if isinstance(raw_node_defs, list): + workflow_payload["node_defs"] = raw_node_defs + workflow = _model_validate(Workflow, workflow_payload, operation) return artifact, workflow diff --git a/src/wf_client/workflows.py b/src/wf_client/workflows.py new file mode 100644 index 00000000..927567b8 --- /dev/null +++ b/src/wf_client/workflows.py @@ -0,0 +1,116 @@ +"""Immutable workflow artifacts and validation snapshots.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Literal + +from wf_artifacts.models import ( + RequiredCapability, +) +from wf_artifacts.models import ( + WorkflowArtifact as ArtifactDomainModel, +) +from wf_core import ValidationReport, Workflow + +from .errors import ValidationFailed + +if TYPE_CHECKING: + from .authoring import EditableWorkflow + from .protocols import WorkflowClientPort + + +@dataclass(frozen=True, slots=True) +class ArtifactRef: + """Stable identity of one immutable workflow artifact version.""" + + artifact_id: str + version: int + + +@dataclass(frozen=True, slots=True) +class WorkflowDiagnostic: + """One server-side workflow-plan diagnostic returned by validation.""" + + severity: str + code: str + path: str + message: str + repair_hint: str | None = None + + +# Keep the short name used by the public design available without requiring a +# second diagnostic implementation. +Diagnostic = WorkflowDiagnostic + + +@dataclass(frozen=True, slots=True) +class WorkflowValidation: + """Combined deterministic local and server-side plan validation result.""" + + local: ValidationReport + remote_status: Literal["valid", "invalid", "not_run"] + remote_diagnostics: tuple[WorkflowDiagnostic, ...] + + @property + def ok(self) -> bool: + return self.local.ok and self.remote_status == "valid" + + def raise_for_errors(self) -> None: + """Raise a useful error for either local or remote validation failures.""" + self.local.raise_for_errors() + if self.remote_status != "invalid": + return + rendered = "\n".join( + f"- [{diagnostic.code}] {diagnostic.path}: {diagnostic.message}" + for diagnostic in self.remote_diagnostics + ) + raise ValidationFailed( + "Workflow server validation failed" + + (f":\n{rendered}" if rendered else ".") + ) + + +@dataclass(frozen=True, slots=True) +class WorkflowArtifact: + """Immutable client snapshot retaining the validated artifact and workflow.""" + + _port: WorkflowClientPort = field(repr=False, compare=False) + artifact: ArtifactDomainModel + workflow: Workflow + + @property + def ref(self) -> ArtifactRef: + return ArtifactRef(self.artifact.id, self.artifact.version) + + @property + def title(self) -> str: + return self.artifact.title + + @property + def description(self) -> str | None: + return self.artifact.description + + @property + def required_capabilities(self) -> tuple[RequiredCapability, ...]: + return tuple( + capability + if isinstance(capability, RequiredCapability) + else RequiredCapability.model_validate(capability) + for capability in self.artifact.required_capabilities + ) + + @property + def workflow_dependencies(self) -> dict[str, int]: + return dict(self.artifact.workflow_dependencies) + + def inspect(self) -> Workflow: + """Return a deep copy so inspecting an artifact cannot mutate its snapshot.""" + return self.workflow.model_copy(deep=True) + + def edit(self) -> EditableWorkflow: + """Seed an editable builder from this exact immutable artifact version.""" + # Import lazily to avoid the authoring/workflows module cycle. + from .authoring import EditableWorkflow + + return EditableWorkflow.from_artifact(self) diff --git a/tests/wf_client/test_authoring.py b/tests/wf_client/test_authoring.py new file mode 100644 index 00000000..9e63ec47 --- /dev/null +++ b/tests/wf_client/test_authoring.py @@ -0,0 +1,120 @@ +from __future__ import annotations + +from typing import Any, cast + +import pytest + +from wf_authoring import WorkflowBuilder +from wf_client import App, ArtifactRef, EditableWorkflow +from wf_client.protocols import WorkflowClientPort + + +class FakePort: + def __init__(self) -> None: + self.calls: list[tuple[str, dict[str, Any]]] = [] + self.validate_artifact_plan_result: dict[str, Any] = { + "status": "valid", + "diagnostics": [], + "required_capabilities": [], + "workflow_dependencies": {}, + } + self.inspect_artifact_result: dict[str, Any] | None = None + + async def validate_artifact_plan(self, **params: Any) -> object: + self.calls.append(("validate_artifact_plan", params)) + return self.validate_artifact_plan_result + + async def create_artifact_from_plan(self, **params: Any) -> object: + self.calls.append(("create_artifact_from_plan", params)) + return {"artifact_id": params["artifact_id"], "version": params["version"], "saved": True} + + async def inspect_artifact(self, **params: Any) -> object: + self.calls.append(("inspect_artifact", params)) + assert self.inspect_artifact_result is not None + return self.inspect_artifact_result + + +def valid_plan(version: int = 1) -> dict[str, Any]: + return { + "id": "report", + "version": version, + "title": "Report", + "kind": "workflow", + "description": None, + "input_schema": {"type": "object", "properties": {}}, + "output_schema": {"type": "object", "properties": {"value": {"type": "string"}}}, + "outcomes": ["ok"], + "plan": { + "name": "report", + "input_schema": {"type": "object", "properties": {}}, + "state_schema": {"type": "object", "properties": {"value": {"type": "string"}}}, + "output_schema": {"type": "object", "properties": {"value": {"type": "string"}}}, + "outcomes": ["ok"], + "output": [{"path": "state.value", "target": "value"}], + "start": "done", + "nodes": [{"id": "done", "type": "end", "outcome": "ok"}], + "edges": [], + }, + "required_capabilities": [], + "workflow_dependencies": {}, + "created_from_catalog_version": None, + } + + +@pytest.mark.asyncio +async def test_validate_stops_before_remote_call_when_local_graph_is_invalid() -> None: + port = FakePort() + graph = App._from_port(cast(WorkflowClientPort, port)).new_workflow( + "invalid", + input_schema={"type": "object", "properties": {}}, + state_schema={"type": "object", "properties": {}}, + output_schema={"type": "object", "properties": {}}, + ) + + result = await graph.validate() + + assert result.local.ok is False + assert result.remote_status == "not_run" + assert port.calls == [] + + +@pytest.mark.asyncio +async def test_validate_runs_local_and_server_validation() -> None: + port = FakePort() + graph = App._from_port(cast(WorkflowClientPort, port)).new_workflow( + "report", + input_schema={"type": "object", "properties": {}}, + state_schema={"type": "object", "properties": {"value": {"type": "string"}}}, + output_schema={"type": "object", "properties": {"value": {"type": "string"}}}, + ) + done = graph.end("ok", id="done") + graph.set_entry_point(done) + + result = await graph.validate() + + assert result.ok is True + assert result.local.ok is True + assert result.remote_status == "valid" + assert port.calls[-1][0] == "validate_artifact_plan" + + +@pytest.mark.asyncio +async def test_edit_and_save_inspects_exact_saved_version() -> None: + port = FakePort() + port.inspect_artifact_result = valid_plan(version=1) + app = App._from_port(cast(WorkflowClientPort, port)) + + graph = await app.edit_workflow("report", version=1) + assert isinstance(graph, WorkflowBuilder) + assert isinstance(graph, EditableWorkflow) + assert all(hasattr(graph, name) for name in ("when", "choose", "match", "foreach", "interrupt", "end", "connect", "set_entry_point")) + + port.inspect_artifact_result = valid_plan(version=2) + saved = await graph.save(version=2) + + create = next(params for operation, params in port.calls if operation == "create_artifact_from_plan") + assert create["plan"] == valid_plan(version=1)["plan"] + inspect = [params for operation, params in port.calls if operation == "inspect_artifact"][-1] + assert inspect == {"artifact_id": "report", "version": 2} + assert saved.ref == ArtifactRef("report", 2) + assert str(saved.workflow.output[0].target) == "value"