From d6f5a2e31d2c77ea55e83c52f80232a3c5fd5c96 Mon Sep 17 00:00:00 2001 From: lda Date: Mon, 31 Aug 2026 01:01:40 +0700 Subject: [PATCH] feat: add callable remote workflow capabilities --- src/wf_client/__init__.py | 18 +++ src/wf_client/app.py | 148 ++++++++++++++++++++++ src/wf_client/capabilities.py | 177 +++++++++++++++++++++++++++ src/wf_client/codec.py | 42 ++++++- tests/wf_client/test_app.py | 121 ++++++++++++++++++ tests/wf_client/test_capabilities.py | 132 ++++++++++++++++++++ 6 files changed, 637 insertions(+), 1 deletion(-) create mode 100644 src/wf_client/app.py create mode 100644 src/wf_client/capabilities.py create mode 100644 tests/wf_client/test_app.py create mode 100644 tests/wf_client/test_capabilities.py diff --git a/src/wf_client/__init__.py b/src/wf_client/__init__.py index 69337a28..7972503e 100644 --- a/src/wf_client/__init__.py +++ b/src/wf_client/__init__.py @@ -1,8 +1,16 @@ """Transport-independent workflow client primitives.""" +from wf_platform import CapabilityRef, Page + +from .app import App +from .capabilities import CapabilityResult, CapabilitySummary, RemoteCapability from .codec import ( DecodedRunResult, DecodedTracePage, + decode_capabilities_page, + decode_capability_call, + decode_capability_diagnostics, + decode_capability_inspect, decode_dependency_diagnostics, decode_deployment, decode_run_result, @@ -27,18 +35,28 @@ from .protocols import WorkflowClientPort __all__ = [ "ArtifactNotFound", "ArtifactVersionConflict", + "App", "CapabilityNotFound", + "CapabilityRef", + "CapabilityResult", + "CapabilitySummary", "DecodedRunResult", "DecodedTracePage", "DeploymentNotRunnable", "DeploymentRequired", "InvalidResponse", + "Page", "ProtocolError", + "RemoteCapability", "RevisionConflict", "TransportError", "ValidationFailed", "WorkflowClientError", "WorkflowClientPort", + "decode_capabilities_page", + "decode_capability_call", + "decode_capability_diagnostics", + "decode_capability_inspect", "decode_dependency_diagnostics", "decode_deployment", "decode_run_result", diff --git a/src/wf_client/app.py b/src/wf_client/app.py new file mode 100644 index 00000000..e7413f80 --- /dev/null +++ b/src/wf_client/app.py @@ -0,0 +1,148 @@ +"""The public entry point for transport-independent workflow clients.""" + +from __future__ import annotations + +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 .capabilities import CapabilitySummary, RemoteCapability +from .codec import ( + decode_capabilities_page, + decode_capability_inspect, +) +from .errors import InvalidResponse +from .protocols import WorkflowClientPort + + +def _capability_ref(qualified_name: str, source_id: str) -> CapabilityRef: + """Build a ref by removing the exact source prefix, preserving dotted keys.""" + prefix = f"{source_id}." + if not qualified_name.startswith(prefix): + raise InvalidResponse( + operation="workflow.capabilities.inspect", + details=( + f"qualified name {qualified_name!r} does not belong to " + f"source {source_id!r}" + ), + ) + try: + return CapabilityRef( + source=SourceRef.parse(source_id), + name=qualified_name.removeprefix(prefix), + ) + except ValueError as exc: + raise InvalidResponse( + operation="workflow.capabilities.inspect", + details=f"invalid capability reference: {exc}", + ) from exc + + +def _summary_from_wire(row: object) -> CapabilitySummary: + """Project one wire discovery row into an attribute-bearing value object.""" + data = dict(row) if isinstance(row, dict) else {} + return CapabilitySummary( + qualified_name=data["name"], + source_id=data["source_id"], + kind=data["kind"], + description=data["description"], + outcomes=tuple(data["outcomes"]), + is_async=data["is_async"], + input_fields=tuple(data["input_fields"]), + output_fields=tuple(data["output_fields"]), + artifact_id=data.get("artifact_id"), + version=data.get("version"), + title=data.get("title"), + ) + + +@dataclass(frozen=True, slots=True) +class App: + """Connected workflow service facade; all remote methods are asynchronous.""" + + _port: WorkflowClientPort = field(repr=False) + endpoint: str + + @classmethod + def from_http_jsonrpc( + cls, + url: str, + *, + timeout_seconds: float = 30.0, + ) -> App: + """Configure a lazy HTTP JSON-RPC connection without performing I/O.""" + return cls( + _port=RpcWorkflowApiClient( + url=url, + timeout_seconds=timeout_seconds, + ), + endpoint=url, + ) + + @classmethod + def _from_port(cls, port: WorkflowClientPort) -> App: + """Construct an app over an in-memory/test port.""" + return cls(_port=port, endpoint="in-process") + + async def capability(self, name: str) -> RemoteCapability: + """Inspect and reconstruct one callable remote capability.""" + wire = decode_capability_inspect( + await self._port.inspect_capability(qualified_name=name) + ) + qualified_name = wire["name"] + source_id = wire["source_id"] + return RemoteCapability( + _port=self._port, + ref=_capability_ref(qualified_name, source_id), + qualified_name=qualified_name, + description=wire["description"], + input_schema=dict(wire["input_schema"]), + output_schema=dict(wire["output_schema"]), + outcomes=tuple(wire["outcomes"]), + is_async=wire["is_async"], + ) + + async def capabilities( + self, + *, + query: str | None = None, + source_id: str | None = None, + cursor: str | None = None, + limit: int = 50, + ) -> Page[CapabilitySummary]: + """List planner-visible capabilities as immutable rich rows.""" + wire = decode_capabilities_page( + await self._port.list_capabilities( + query=query, + source_id=source_id, + cursor=cursor, + limit=limit, + ) + ) + return Page( + items=tuple(_summary_from_wire(row) for row in wire["capabilities"]), + next_cursor=wire["next_cursor"], + total=wire["total"], + ) + + def new_workflow( + self, + name: str, + *, + input_schema: Any, + state_schema: Any, + output_schema: Any, + outcomes: Sequence[str] = ("ok",), + ) -> WorkflowBuilder: + """Construct a local builder; remote operations remain opt-in/async.""" + return WorkflowBuilder( + name=name, + input_schema=input_schema, + state_schema=state_schema, + output_schema=output_schema, + outcomes=outcomes, + ) diff --git a/src/wf_client/capabilities.py b/src/wf_client/capabilities.py new file mode 100644 index 00000000..08f7cef8 --- /dev/null +++ b/src/wf_client/capabilities.py @@ -0,0 +1,177 @@ +"""Rich, transport-independent objects for remote workflow capabilities.""" + +from __future__ import annotations + +from collections.abc import Mapping +from copy import deepcopy +from dataclasses import dataclass, field +from typing import Any + +from jsonschema import Draft202012Validator, SchemaError, ValidationError + +from wf_artifacts.models import DependencyDiagnostic +from wf_core.models.schemas import NodeDef, SchemaRef +from wf_platform import CapabilityRef + +from .codec import decode_capability_call, decode_capability_diagnostics +from .errors import InvalidResponse +from .protocols import WorkflowClientPort + + +@dataclass(frozen=True, slots=True) +class CapabilitySummary: + """Compact immutable discovery row for a planner-visible capability.""" + + qualified_name: str + source_id: str + kind: str + description: str | None + outcomes: tuple[str, ...] + is_async: bool + input_fields: tuple[str, ...] + output_fields: tuple[str, ...] + artifact_id: str | None = None + version: int | None = None + title: str | None = None + + @property + def name(self) -> str: + """Compatibility alias for the wire row's ``name`` field.""" + return self.qualified_name + + +@dataclass(frozen=True, slots=True) +class CapabilityResult: + """Validated result of invoking one remote capability.""" + + outcome: str + output: dict[str, Any] | None + diagnostics: tuple[DependencyDiagnostic, ...] + + +def _check_schema(schema: object, *, operation: str) -> dict[str, Any]: + if not isinstance(schema, Mapping): + raise InvalidResponse( + operation=operation, + details="capability schema must be a JSON object", + ) + schema_copy = deepcopy(dict(schema)) + try: + Draft202012Validator.check_schema(schema_copy) + except SchemaError as exc: + raise InvalidResponse( + operation=operation, + details=f"invalid JSON Schema: {exc.message}", + ) from exc + return schema_copy + + +@dataclass(frozen=True, slots=True) +class RemoteCapability: + """Inspected remote capability that validates calls against its contract.""" + + _port: WorkflowClientPort = field(repr=False, compare=False) + ref: CapabilityRef + qualified_name: str + description: str | None + input_schema: dict[str, Any] + output_schema: dict[str, Any] + outcomes: tuple[str, ...] + is_async: bool + + def __post_init__(self) -> None: + # Freeze the public container shape at construction. The nested JSON + # values remain ordinary JSON objects because callers expect to inspect + # and pass schemas directly to existing pydantic/core APIs. + if not self.outcomes: + raise InvalidResponse( + operation="workflow.capabilities.inspect", + details="capability contract must declare at least one outcome", + ) + object.__setattr__( + self, + "input_schema", + _check_schema( + self.input_schema, + operation="workflow.capabilities.inspect", + ), + ) + object.__setattr__( + self, + "output_schema", + _check_schema( + self.output_schema, + operation="workflow.capabilities.inspect", + ), + ) + object.__setattr__(self, "outcomes", tuple(self.outcomes)) + + async def __call__( + self, + payload: Mapping[str, Any] | None = None, + /, + **fields: Any, + ) -> CapabilityResult: + if payload is not None and fields: + raise TypeError("pass a payload mapping or keyword fields, not both") + return await self.call(dict(payload) if payload is not None else fields) + + async def call( + self, + payload: Mapping[str, Any], + *, + deployment_id: str | None = None, + ) -> CapabilityResult: + """Validate input locally, invoke remotely, and validate its result.""" + input_payload = dict(payload) + input_validator = Draft202012Validator(self.input_schema) + # jsonschema.ValidationError intentionally remains the local input + # error: no transport operation has happened when it is raised. + input_validator.validate(input_payload) + + wire = decode_capability_call( + await self._port.call_capability( + qualified_name=self.qualified_name, + payload=input_payload, + deployment_id=deployment_id, + ) + ) + if wire["qualified_name"] != self.qualified_name: + raise InvalidResponse( + operation="workflow.capabilities.call", + details=( + f"result qualified name {wire['qualified_name']!r} does not " + f"match requested {self.qualified_name!r}" + ), + ) + if wire["outcome"] not in self.outcomes: + raise InvalidResponse( + operation="workflow.capabilities.call", + details=f"unknown capability outcome {wire['outcome']!r}", + ) + + output = wire["output"] + if output is not None: + try: + Draft202012Validator(self.output_schema).validate(output) + except ValidationError as exc: + # Schema validation errors are expected server-contract + # failures; do not leak jsonschema internals as public output. + raise InvalidResponse( + operation="workflow.capabilities.call", + details=f"output does not match capability schema: {exc}", + ) from exc + return CapabilityResult( + outcome=wire["outcome"], + output=deepcopy(output) if output is not None else None, + diagnostics=decode_capability_diagnostics(wire["diagnostics"]), + ) + + def node_def(self) -> NodeDef: + """Return the schema contract consumed by ``WorkflowBuilder.use_contract``.""" + return NodeDef( + name=self.qualified_name, + input_schema=SchemaRef.model_validate(deepcopy(self.input_schema)), + output_schema=SchemaRef.model_validate(deepcopy(self.output_schema)), + outcomes=list(self.outcomes), + ) diff --git a/src/wf_client/codec.py b/src/wf_client/codec.py index 651d4047..32095024 100644 --- a/src/wf_client/codec.py +++ b/src/wf_client/codec.py @@ -9,7 +9,10 @@ from typing import Any, TypeVar from pydantic import BaseModel, TypeAdapter, ValidationError from wf_api.models import ( + CapabilityCallResult, DependencyDiagnosticPayload, + InspectCapabilityResult, + ListCapabilitiesResult, RawWorkflowPlan, RunResult, RunTraceResult, @@ -62,13 +65,50 @@ class DecodedTracePage(_DecodedRunFields): _PayloadT = TypeVar("_PayloadT") -def _validate(payload: object, schema: type[_PayloadT], operation: str) -> _PayloadT: +def _validate(payload: object, schema: object, 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 +def decode_capability_inspect(payload: object) -> InspectCapabilityResult: + """Validate one capability contract returned by discovery inspection.""" + return _validate( + payload, + InspectCapabilityResult, + "workflow.capabilities.inspect", + ) + + +def decode_capability_call(payload: object) -> CapabilityCallResult: + """Validate one direct capability-call result at the transport boundary.""" + return _validate( + payload, + CapabilityCallResult, + "workflow.capabilities.call", + ) + + +def decode_capability_diagnostics( + payload: object, +) -> tuple[DependencyDiagnostic, ...]: + """Decode diagnostics attached to a capability invocation.""" + return _decode_dependency_diagnostics( + payload, + "workflow.capabilities.call", + ) + + +def decode_capabilities_page(payload: object) -> ListCapabilitiesResult: + """Validate one cursor-paged capability discovery response.""" + return _validate( + payload, + ListCapabilitiesResult, + "workflow.capabilities.list", + ) + + _ModelT = TypeVar("_ModelT", bound=BaseModel) diff --git a/tests/wf_client/test_app.py b/tests/wf_client/test_app.py new file mode 100644 index 00000000..e78ef8a1 --- /dev/null +++ b/tests/wf_client/test_app.py @@ -0,0 +1,121 @@ +from __future__ import annotations + +from typing import Any, cast + +import httpx +import pytest + +from wf_client import App, CapabilitySummary, Page +from wf_client.protocols import WorkflowClientPort +from wf_platform import CapabilityRef + + +def _inspect_payload() -> dict[str, Any]: + return { + "name": "app.default.search", + "source_id": "app.default", + "kind": "node_spec", + "description": "Search things", + "outcomes": ["ok"], + "is_async": False, + "input_schema": {"type": "object", "properties": {}}, + "output_schema": {"type": "object", "properties": {}}, + "wrapper_hints": { + "capability_name": "app.default.search", + "confidence": "high", + "declared_outcomes": ["ok"], + "suggested_wrapper_outcomes": ["ok"], + "outcome_policy": "preserve_declared", + "input_schema": {"type": "object", "properties": {}}, + "state_schema": {"type": "object", "properties": {}}, + "output_schema": {"type": "object", "properties": {}}, + "input_map": {}, + "output_map": {}, + "outcome_candidates": [], + "missing_decisions": [], + "notes": [], + }, + "accepts_context": False, + } + + +class _Port: + def __init__(self, *, capability_name: str = "app.default.search") -> None: + self.calls: list[tuple[str, dict[str, Any]]] = [] + self.capability_name = capability_name + + async def inspect_capability(self, **params: Any) -> object: + self.calls.append(("inspect", params)) + payload = _inspect_payload() + payload["name"] = self.capability_name + payload["wrapper_hints"]["capability_name"] = self.capability_name + return payload + + async def list_capabilities(self, **params: Any) -> object: + self.calls.append(("list", params)) + return { + "next_cursor": None, + "total": 1, + "capabilities": [ + { + "name": "app.default.search", + "source_id": "app.default", + "kind": "node_spec", + "description": "Search things", + "outcomes": ["ok"], + "is_async": False, + "input_fields": [], + "output_fields": [], + } + ], + } + + +def _app(*, capability_name: str = "app.default.search") -> App: + return App._from_port( + cast(WorkflowClientPort, _Port(capability_name=capability_name)) + ) + + +def test_from_http_jsonrpc_is_lazy(monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[str] = [] + monkeypatch.setattr( + httpx.AsyncClient, + "post", + lambda *args, **kwargs: calls.append("post"), + ) + + app = App.from_http_jsonrpc("http://localhost:8765/rpc") + + assert app.endpoint == "http://localhost:8765/rpc" + assert calls == [] + + +@pytest.mark.asyncio +async def test_capability_discovery_returns_rich_page() -> None: + page = await _app().capabilities(query="search", limit=10) + + assert isinstance(page, Page) + assert page.total == 1 + assert page.next_cursor is None + assert isinstance(page.items[0], CapabilitySummary) + assert page.items[0].qualified_name == "app.default.search" + assert page.items[0].outcomes == ("ok",) + + +@pytest.mark.asyncio +async def test_capability_reconstructs_structural_reference() -> None: + capability = await _app().capability("app.default.search") + + assert capability.ref == CapabilityRef.parse("app.default.search") + assert capability.ref.source.parts == ("app", "default") + + +@pytest.mark.asyncio +async def test_capability_reference_keeps_dotted_local_key() -> None: + capability = await _app(capability_name="app.default.search.v2").capability( + "app.default.search.v2" + ) + + assert capability.ref.source.parts == ("app", "default") + assert capability.ref.name == "search.v2" diff --git a/tests/wf_client/test_capabilities.py b/tests/wf_client/test_capabilities.py new file mode 100644 index 00000000..2f471c47 --- /dev/null +++ b/tests/wf_client/test_capabilities.py @@ -0,0 +1,132 @@ +from __future__ import annotations + +from typing import Any, cast + +import pytest + +from wf_client import CapabilityResult, RemoteCapability +from wf_client.errors import InvalidResponse +from wf_client.protocols import WorkflowClientPort +from wf_platform import CapabilityRef, SourceRef + + +def _inspect_payload() -> dict[str, Any]: + return { + "name": "app.default.search", + "source_id": "app.default", + "kind": "node_spec", + "description": "Search things", + "outcomes": ["ok", "error"], + "is_async": True, + "input_schema": { + "type": "object", + "properties": {"query": {"type": "string"}}, + "required": ["query"], + }, + "output_schema": { + "type": "object", + "properties": {"results": {"type": "array"}}, + "required": ["results"], + }, + "wrapper_hints": {}, + "accepts_context": False, + } + + +class _Port: + def __init__(self) -> None: + self.calls: list[dict[str, Any]] = [] + + async def call_capability(self, **params: Any) -> object: + self.calls.append(params) + return { + "qualified_name": "app.default.search", + "source_id": "app.default", + "kind": "node_spec", + "deployment_id": None, + "outcome": "ok", + "output": {"results": ["one"]}, + "diagnostics": [], + } + + +def _port() -> WorkflowClientPort: + return cast(WorkflowClientPort, _Port()) + + +def test_remote_capability_preserves_dotted_local_key() -> None: + capability = RemoteCapability( + _port=_port(), + ref=CapabilityRef(source=SourceRef.parse("app.default"), name="search.v2"), + qualified_name="app.default.search.v2", + description=None, + input_schema={"type": "object"}, + output_schema={"type": "object"}, + outcomes=("ok",), + is_async=False, + ) + + assert capability.ref.name == "search.v2" + assert capability.node_def().name == "app.default.search.v2" + + +@pytest.mark.asyncio +async def test_remote_capability_is_callable_and_validates_result() -> None: + port = _Port() + capability = RemoteCapability( + _port=cast(WorkflowClientPort, port), + ref=CapabilityRef.parse("app.default.search"), + qualified_name="app.default.search", + description=None, + input_schema={ + "type": "object", + "properties": {"query": {"type": "string"}}, + "required": ["query"], + }, + output_schema={"type": "object", "required": ["results"]}, + outcomes=("ok",), + is_async=False, + ) + + result = await capability(query="workflow") + + assert isinstance(result, CapabilityResult) + assert result.output == {"results": ["one"]} + assert port.calls == [ + { + "qualified_name": "app.default.search", + "payload": {"query": "workflow"}, + "deployment_id": None, + } + ] + + +@pytest.mark.asyncio +async def test_remote_capability_rejects_mixed_payload_forms() -> None: + capability = RemoteCapability( + _port=_port(), + ref=CapabilityRef.parse("app.default.search"), + qualified_name="app.default.search", + description=None, + input_schema={"type": "object"}, + output_schema={"type": "object"}, + outcomes=("ok",), + is_async=False, + ) + + with pytest.raises(TypeError, match="not both"): + await capability({"query": "workflow"}, query="again") + + +def test_remote_capability_rejects_invalid_inspected_schema() -> None: + with pytest.raises(InvalidResponse, match="invalid JSON Schema"): + RemoteCapability( + _port=_port(), + ref=CapabilityRef.parse("app.default.search"), + qualified_name="app.default.search", + description=None, + input_schema={"type": "not-a-json-schema-type"}, + output_schema={"type": "object"}, + outcomes=("ok",), + is_async=False, + )