From 327a7b24b042af1c662a16450e0f5ad213d99a4f Mon Sep 17 00:00:00 2001 From: lda Date: Thu, 3 Sep 2026 14:49:44 +0700 Subject: [PATCH] test: prove saved artifact subgraphs via Python client --- docs/wf_api_architecture.md | 17 +++- .../wf-python/references/python-lifecycle.md | 41 +++++++++ tests/wf_client/test_http_integration.py | 91 ++++++++++++++++++- 3 files changed, 147 insertions(+), 2 deletions(-) diff --git a/docs/wf_api_architecture.md b/docs/wf_api_architecture.md index fbc88fa1..48590ca9 100644 --- a/docs/wf_api_architecture.md +++ b/docs/wf_api_architecture.md @@ -117,7 +117,7 @@ a durable run would use the following complete flow: from pydantic import BaseModel from wf_client import App -from wf_authoring import input_from, input_value, output_to, state_path +from wf_authoring import input_from, input_path, input_value, output_to, state_path class Input(BaseModel): @@ -181,6 +181,21 @@ paged immutable summary rows; `app.deployments()` returns an immutable tuple. Call `app.workflow(id, version=...)`, `app.deployment(id)`, or `app.run(id)` to reconstruct the selected rich object. +A saved workflow artifact can be used directly as a native subgraph: + +```python +child = await app.workflow("child", version=2) +child_step = parent.subgraph( + child, + input=[input_from(input_path("prompt"), "prompt")], + output=[output_to("value", state_path("result"))], +) +``` + +The boundary snapshots the child's public input/output contract for local +validation. The saved parent retains the exact child artifact ID and version; +deployment validation and execution resolve that separate saved dependency. + ## WorkflowApiSurface And Domain Services `WorkflowApiSurface` is the public application contract shared by local and diff --git a/skills/wf-python/references/python-lifecycle.md b/skills/wf-python/references/python-lifecycle.md index 3f05bfd6..313729bb 100644 --- a/skills/wf-python/references/python-lifecycle.md +++ b/skills/wf-python/references/python-lifecycle.md @@ -62,6 +62,47 @@ artifact = await graph.save(version=1, title="Typed example") run = await artifact.run({"request_id": "request-1"}) ``` +## Saved Workflow As A Native Subgraph + +Load the exact child artifact before authoring the parent. Its public contract +is copied into the parent boundary for local validation; its artifact ID and +version remain the runtime dependency. + +```python +from wf_authoring import input_from, input_path, output_to, state_path + +child = await app.workflow("child", version=2) +parent = app.new_workflow( + "parent", + input_schema=ParentInput, + state_schema=ParentState, + output_schema=ParentOutput, +) +run_child = parent.subgraph( + child, + id="run_child", + input=[input_from(input_path("prompt"), "prompt")], + output=[output_to("value", state_path("result"))], +) +parent.set_entry_point(run_child) +parent.connect(run_child, "ok", parent.end("ok", id="parent_done")) +parent.set_output([input_from(state_path("result"), "result")]) + +parent.validate_local().raise_for_errors() +(await parent.validate()).raise_for_errors() +parent_v1 = await parent.save(version=1) + +deployment = await parent_v1.deploy("parent.production") +readiness = await deployment.validate() +if not readiness.runnable: + raise RuntimeError(readiness.diagnostics) +run = await deployment.run({"prompt": "hello"}) +``` + +Choose binding paths from `child.inspect()` and the parent models. Saving the +parent does not duplicate the child plan: the saved parent retains the exact +`child.v2` dependency, which deployment validation and execution resolve. + ## Lossless Editing ```python diff --git a/tests/wf_client/test_http_integration.py b/tests/wf_client/test_http_integration.py index 1ab7c035..52e668c8 100644 --- a/tests/wf_client/test_http_integration.py +++ b/tests/wf_client/test_http_integration.py @@ -1,9 +1,12 @@ from __future__ import annotations +from pathlib import Path + import httpx import pytest +from pydantic import BaseModel -from wf_authoring import input_from, input_value, output_to, state_path +from wf_authoring import input_from, input_path, input_value, output_to, state_path from wf_client import App, ArtifactRef from wf_server import build_local_static_workflow_server from wf_transport_rpc_http import RpcWorkflowApiClient, create_rpc_app @@ -67,3 +70,89 @@ async def test_http_app_calls_authors_saves_deploys_and_runs(tmp_path) -> None: ] assert deployments[0].artifact_id == "http_client_proof" assert runs.items[0].run_id == run.run_id + + +class _ChildInput(BaseModel): + prompt: str + + +class _ChildState(BaseModel): + value: str | None = None + + +class _ChildOutput(BaseModel): + value: str + + +class _ParentInput(BaseModel): + prompt: str + + +class _ParentState(BaseModel): + result: str | None = None + + +class _ParentOutput(BaseModel): + result: str + + +@pytest.mark.asyncio +async def test_http_app_runs_saved_workflow_artifact_as_native_subgraph( + tmp_path: Path, +) -> None: + """Catch public-client subgraphs losing exact saved-child resolution.""" + server = build_local_static_workflow_server(tmp_path / "store") + rpc_app = create_rpc_app(server) + transport = httpx.ASGITransport(app=rpc_app) + + async with httpx.AsyncClient( + transport=transport, + base_url="http://test", + ) as http_client: + app = App._from_port( + RpcWorkflowApiClient(url="http://test/rpc", http_client=http_client) + ) + constant = await app.capability("wf.std.constant") + + child = app.new_workflow( + "saved_child", + input_schema=_ChildInput, + state_schema=_ChildState, + output_schema=_ChildOutput, + ) + child_step = child.use( + constant, + id="copy_prompt", + input=[input_from(input_path("prompt"), "value")], + output=[output_to("value", state_path("value"))], + ) + child.set_entry_point(child_step) + child.connect(child_step, "ok", child.end("ok", id="child_done")) + child.set_output([input_from(state_path("value"), "value")]) + child_artifact = await child.save(version=1, title="Saved child") + + parent = app.new_workflow( + "saved_parent", + input_schema=_ParentInput, + state_schema=_ParentState, + output_schema=_ParentOutput, + ) + child_boundary = parent.subgraph( + child_artifact, + id="run_child", + input=[input_from(input_path("prompt"), "prompt")], + output=[output_to("value", state_path("result"))], + ) + parent.set_entry_point(child_boundary) + parent.connect(child_boundary, "ok", parent.end("ok", id="parent_done")) + parent.set_output([input_from(state_path("result"), "result")]) + parent_artifact = await parent.save(version=1, title="Saved parent") + + deployment = await parent_artifact.deploy("saved_parent.production") + readiness = await deployment.validate() + run = await deployment.run({"prompt": "hello from parent"}) + + assert readiness.runnable is True + assert parent_artifact.workflow_dependencies == {"saved_child": 1} + assert run.status == "completed" + assert run.output == {"result": "hello from parent"}