lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: expose output bindings over rpc

Browse Source
This commit is contained in:
lda
2026-07-23 03:34:37 +07:00 Verified
parent 418719e3c6
commit 56dddd6a31
6 changed files with 158 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wf_transport_rpc_http/__init__.py
+2
View File
@@ -45,6 +45,7 @@ from .models import (
SetDraftStartParams,
SetStepInputBindingsParams,
SetStepInputMapParams,
SetStepOutputBindingsParams,
SetStepOutputMapParams,
SetWorkflowOutputMapParams,
StartRunParams,
@@ -96,6 +97,7 @@ __all__ = [
"SetDraftStartParams",
"SetStepInputBindingsParams",
"SetStepInputMapParams",
"SetStepOutputBindingsParams",
"SetStepOutputMapParams",
"SetWorkflowOutputMapParams",
"StartRunParams",
src/wf_transport_rpc_http/client/drafts.py
+19 -1
View File
@@ -5,7 +5,7 @@ from typing import Any, Literal
from wf_api.surface import RouteSource
from wf_artifacts.drafts.models import DraftStep
from wf_core.models.steps import InputBinding
from wf_core.models.steps import InputBinding, OutputBinding
from .base import RpcCaller
@@ -201,6 +201,24 @@ class RpcDraftClientMixin:
},
)
async def set_step_output_bindings(
self: RpcCaller,
*,
workspace_id: str,
revision: int,
step_id: str,
bindings: Sequence[OutputBinding],
) -> dict[str, Any]:
return await self._call(
"workflow.draft_workspaces.set_step_output_bindings",
{
"workspace_id": workspace_id,
"revision": revision,
"step_id": step_id,
"bindings": [binding.model_dump(mode="json") for binding in bindings],
},
)
async def set_step_output_map(
self: RpcCaller,
*,
src/wf_transport_rpc_http/methods/drafts.py
+18
View File
@@ -33,6 +33,7 @@ from ..models import (
SetDraftStartParams,
SetStepInputBindingsParams,
SetStepInputMapParams,
SetStepOutputBindingsParams,
SetStepOutputMapParams,
SetWorkflowOutputMapParams,
ValidateDraftParams,
@@ -244,6 +245,23 @@ def register_methods(
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
raise_workflow_rpc_error(exc)
@entrypoint.method(
name="workflow.draft_workspaces.set_step_output_bindings",
errors=[WorkflowRpcError],
)
async def workflow_draft_workspaces_set_step_output_bindings(
params: SetStepOutputBindingsParams = RpcParams(),
) -> dict[str, Any]:
try:
return await server.api.set_step_output_bindings(
workspace_id=params.workspace_id,
revision=params.revision,
step_id=params.step_id,
bindings=params.bindings,
)
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
raise_workflow_rpc_error(exc)
@entrypoint.method(
name="workflow.draft_workspaces.set_step_output_map",
errors=[WorkflowRpcError],
src/wf_transport_rpc_http/models.py
+8 -1
View File
@@ -6,7 +6,7 @@ from pydantic import BaseModel, ConfigDict, Field, model_validator
from wf_api.models import TraceRange
from wf_artifacts.drafts.models import DraftStep
from wf_core.models.steps import InputBinding
from wf_core.models.steps import InputBinding, OutputBinding
class RpcParamsModel(BaseModel):
@@ -216,6 +216,13 @@ class SetStepInputBindingsParams(RpcParamsModel):
bindings: list[InputBinding]
class SetStepOutputBindingsParams(RpcParamsModel):
workspace_id: str = Field(min_length=1)
revision: int = Field(ge=1)
step_id: str = Field(min_length=1)
bindings: list[OutputBinding]
class SetStepOutputMapParams(RpcParamsModel):
workspace_id: str = Field(min_length=1)
revision: int = Field(ge=1)
tests/wf_transport_rpc_http/test_app.py
+76
View File
@@ -930,6 +930,82 @@ async def test_rpc_set_step_input_bindings_preserves_canonical_order(tmp_path) -
]
async def test_rpc_set_step_output_bindings_preserves_order_and_source_fan_out(
tmp_path,
) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
await _rpc(
client,
"workflow.draft_workspaces.create_from_capability",
{
"workspace_id": "draft-rpc-output-bindings",
"capability_name": "wf.std.constant",
"name": "output_bindings",
},
)
response = await _rpc(
client,
"workflow.draft_workspaces.set_step_output_bindings",
{
"workspace_id": "draft-rpc-output-bindings",
"revision": 1,
"step_id": "call",
"bindings": [
{"source": "value", "target": "state.report.title"},
{"source": "value", "target": "state.audit.title"},
],
},
)
inspected = await _rpc(
client,
"workflow.draft_workspaces.get",
{"workspace_id": "draft-rpc-output-bindings", "include_draft": True},
)
assert response["result"]["revision"] == 2
assert inspected["result"]["draft"]["steps"]["call"]["output"] == [
{"source": "value", "target": "state.report.title"},
{"source": "value", "target": "state.audit.title"},
]
@pytest.mark.parametrize(
"binding",
[
{"target": "state.report.title"},
{"source": "value", "target": "state"},
{"source": "value", "target": "state.report.title", "extra": True},
{
"source": {"root": "state", "parts": ["value"]},
"target": "state.report.title",
},
],
)
async def test_rpc_set_step_output_bindings_rejects_malformed_binding(
tmp_path,
binding: dict[str, Any],
) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
rejected = await _rpc(
client,
"workflow.draft_workspaces.set_step_output_bindings",
{
"workspace_id": "missing_ws",
"revision": 1,
"step_id": "call",
"bindings": [binding],
},
)
assert rejected["error"]["code"] == -32602
@pytest.mark.parametrize(
"binding",
[
tests/wf_transport_rpc_http/test_client.py
+35 -2
View File
@@ -15,8 +15,8 @@ from wf_artifacts.drafts.models import (
DraftStep,
)
from wf_core import END
from wf_core.models.steps import InputPathBinding, InputValueBinding
from wf_core.paths import GraphSourcePath, LocalPath
from wf_core.models.steps import InputPathBinding, InputValueBinding, OutputBinding
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.drafts import RpcDraftClientMixin
@@ -731,6 +731,39 @@ async def test_rpc_client_serializes_canonical_step_input_bindings() -> None:
)
async def test_rpc_client_serializes_step_output_bindings() -> None:
calls: list[dict[str, Any]] = []
class Client(RpcDraftClientMixin):
async def _call(self, method: str, params: dict[str, object]):
calls.append({"method": method, "params": params})
return {"revision": 3}
client = Client()
await client.set_step_output_bindings(
workspace_id="client_ws",
revision=2,
step_id="analyze",
bindings=[
OutputBinding(
source=LocalPath.parse("report.title"),
target=StatePath.parse("state.report.title"),
),
OutputBinding(
source=LocalPath.parse("report.title"),
target=StatePath.parse("state.audit.title"),
),
],
)
assert calls[-1]["method"] == "workflow.draft_workspaces.set_step_output_bindings"
assert calls[-1]["params"]["bindings"] == [
{"source": "report.title", "target": "state.report.title"},
{"source": "report.title", "target": "state.audit.title"},
]
async def test_rpc_client_draft_remove_methods(tmp_path) -> None:
calls: list[dict[str, Any]] = []