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

feat: type capability results

Browse Source
This commit is contained in:
lda
2026-07-31 14:35:06 +07:00 Verified
parent c96fee4e7d
commit 82f764f7a6
11 changed files with 423 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ISSUES.md
+5 -5
View File
@@ -73,11 +73,11 @@
retain useful Pydantic schemas, so OpenRPC is a viable transport input.
- Typed-result slices now give `workflow.health`, all artifact, deployment,
and run operations, every persisted draft-workspace operation, and the
capability-bootstrap alias named transport-neutral result schemas: 47 of
70 methods. The remaining 23 success results still collapse to generic
objects across capability discovery/call, stateless draft patch/validate,
source discovery, and source-registry/admin operations. Continue introducing
operation result DTOs before adopting generated TypeScript contracts.
capability discovery/call surface named transport-neutral result schemas:
50 of 70 methods. The remaining 20 success results still collapse to generic
objects across stateless draft patch/validate, source discovery, and
source-registry/admin operations. Continue introducing operation result DTOs
before adopting generated TypeScript contracts.
- The stock `@open-rpc/generator` TypeScript client is not suitable here. It
exhausted a 4 GB Node heap on the full contract and emitted invalid dotted
class members plus `any` results for a minimal `workflow.health` contract.
src/wf_api/capabilities.py
+46 -14
View File
@@ -22,9 +22,17 @@ from .draft_authoring import WorkflowDraftAuthoringApi
from .drafts import WorkflowDraftApi
from .listing import matches_query, paged_list_payload
from .models import (
CapabilityCallResult,
CapabilitySummary,
CreateDraftWorkspaceFromCapabilityResult,
InspectCapabilityResult,
JsonProjector,
ListCapabilitiesResult,
NextActionsPayload,
NodeSpecCapabilityDetail,
NodeSpecCapabilitySummary,
WrapperArtifactCapabilityDetail,
WrapperArtifactCapabilitySummary,
WrapperAuthoringHintsPayload,
)
from .next_actions import NextActions
@@ -37,6 +45,12 @@ from .wrapper_hints import (
)
_PROJECT_WRAPPER_HINTS = JsonProjector(WrapperAuthoringHintsPayload)
_PROJECT_CAPABILITY_LIST = JsonProjector(ListCapabilitiesResult)
_PROJECT_CAPABILITY_CALL = JsonProjector(CapabilityCallResult)
_PROJECT_NODE_SPEC_SUMMARY = JsonProjector(NodeSpecCapabilitySummary)
_PROJECT_WRAPPER_SUMMARY = JsonProjector(WrapperArtifactCapabilitySummary)
_PROJECT_NODE_SPEC_DETAIL = JsonProjector(NodeSpecCapabilityDetail)
_PROJECT_WRAPPER_DETAIL = JsonProjector(WrapperArtifactCapabilityDetail)
def _schema_field_names(schema: dict[str, Any]) -> list[str]:
@@ -82,9 +96,10 @@ class WorkflowCapabilityApi:
source_id: str | None = None,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
) -> ListCapabilitiesResult:
"""Return compact paged planner-visible workflow capability summaries."""
capabilities = [
capabilities: list[CapabilitySummary] = [
_PROJECT_NODE_SPEC_SUMMARY(
{
"name": detail.name,
"source_id": source.id,
@@ -97,6 +112,7 @@ class WorkflowCapabilityApi:
workflow_output_schema_for_authoring(detail.output_schema)
),
}
)
for source in sorted(
self.context.specs.capability_sources.values(),
key=lambda source: source.id,
@@ -114,14 +130,18 @@ class WorkflowCapabilityApi:
self._wrapper_capability_summaries(query=query, source_id=source_id)
)
capabilities.sort(key=lambda capability: capability["name"])
return paged_list_payload(
return _PROJECT_CAPABILITY_LIST(
paged_list_payload(
"capabilities",
capabilities,
cursor=cursor,
limit=limit,
)
)
async def inspect_capability(self, *, qualified_name: str) -> dict[str, Any]:
async def inspect_capability(
self, *, qualified_name: str
) -> InspectCapabilityResult:
"""Return one planner-visible workflow capability contract."""
for source in self.context.specs.capability_sources.values():
if not source.enabled or not source.visibility.planner:
@@ -129,13 +149,15 @@ class WorkflowCapabilityApi:
for detail in source.as_inventory().capabilities.node_spec_details:
if detail.name == qualified_name:
detail_payload = detail.model_dump(mode="json")
detail_payload["source_id"] = source.id
detail_payload["kind"] = "node_spec"
detail_payload["wrapper_hints"] = wrapper_hints_for_capability(
capability_name=detail.name,
input_schema=detail.input_schema,
output_schema=detail.output_schema,
outcomes=detail.outcomes,
).model_dump(mode="json")
return detail_payload
return _PROJECT_NODE_SPEC_DETAIL(detail_payload)
wrapper_detail = self._wrapper_capability_detail(qualified_name)
if wrapper_detail is not None:
return wrapper_detail
@@ -147,7 +169,7 @@ class WorkflowCapabilityApi:
qualified_name: str,
payload: dict[str, Any],
deployment_id: str | None = None,
) -> dict[str, Any]:
) -> CapabilityCallResult:
"""Execute one planner-visible workflow capability for authoring tests."""
wrapper_artifact = self._wrapper_artifact_for_capability_name(qualified_name)
if wrapper_artifact is not None:
@@ -166,7 +188,8 @@ class WorkflowCapabilityApi:
try:
result = await handler(payload, RuntimeContext(current_node_id=spec.name))
except Exception as exc:
return {
return _PROJECT_CAPABILITY_CALL(
{
"qualified_name": spec.name,
"source_id": source_id,
"kind": "node_spec",
@@ -190,7 +213,9 @@ class WorkflowCapabilityApi:
).model_dump(mode="json")
],
}
return {
)
return _PROJECT_CAPABILITY_CALL(
{
"qualified_name": spec.name,
"source_id": source_id,
"kind": "node_spec",
@@ -199,6 +224,7 @@ class WorkflowCapabilityApi:
"output": result["output"],
"diagnostics": [],
}
)
def _wrapper_artifact_for_capability_name(
self,
@@ -230,7 +256,7 @@ class WorkflowCapabilityApi:
*,
query: str | None,
source_id: str | None,
) -> list[dict[str, Any]]:
) -> list[WrapperArtifactCapabilitySummary]:
"""Project saved wrappers into workflow capability discovery rows.
Wrapper artifacts are not live source NodeSpecs, but authors need to
@@ -240,7 +266,7 @@ class WorkflowCapabilityApi:
"""
if source_id not in {None, "workflow"} or self.context.artifact_store is None:
return []
rows: list[dict[str, Any]] = []
rows: list[WrapperArtifactCapabilitySummary] = []
for artifact in self.context.artifact_store.list_artifacts():
if artifact.kind != "wrapper":
continue
@@ -252,6 +278,7 @@ class WorkflowCapabilityApi:
):
continue
rows.append(
_PROJECT_WRAPPER_SUMMARY(
{
"name": name,
"source_id": "workflow",
@@ -266,17 +293,19 @@ class WorkflowCapabilityApi:
"output_fields": _schema_field_names(artifact.output_schema),
}
)
)
return rows
def _wrapper_capability_detail(
self,
qualified_name: str,
) -> dict[str, Any] | None:
) -> WrapperArtifactCapabilityDetail | None:
"""Return a NodeSpec-like contract for one saved wrapper artifact."""
artifact = self._wrapper_artifact_for_capability_name(qualified_name)
if artifact is None:
return None
return {
return _PROJECT_WRAPPER_DETAIL(
{
"name": artifact_capability_id(artifact),
"source_id": "workflow",
"kind": "wrapper_artifact",
@@ -298,6 +327,7 @@ class WorkflowCapabilityApi:
outcomes=list(artifact.outcomes),
).model_dump(mode="json"),
}
)
async def _call_wrapper_artifact(
self,
@@ -305,7 +335,7 @@ class WorkflowCapabilityApi:
payload: dict[str, Any],
*,
deployment_id: str | None,
) -> dict[str, Any]:
) -> CapabilityCallResult:
"""Execute a saved wrapper artifact through the workflow runner."""
unsupported = direct_wrapper_interrupt_diagnostic(artifact)
if unsupported is not None:
@@ -334,7 +364,8 @@ class WorkflowCapabilityApi:
deployment=deployment,
artifact=artifact,
)
return {
return _PROJECT_CAPABILITY_CALL(
{
"qualified_name": artifact_capability_id(artifact),
"source_id": "workflow",
"kind": "wrapper_artifact",
@@ -343,6 +374,7 @@ class WorkflowCapabilityApi:
"output": run.output,
"diagnostics": [],
}
)
async def create_draft_workspace_from_capability(
self,
src/wf_api/models/__init__.py
+20
View File
@@ -11,6 +11,17 @@ from .artifacts import (
SaveArtifactResult,
WorkflowArtifactPayload,
)
from .capabilities import (
CapabilityCallResult,
CapabilitySummary,
CapabilitySummaryPayload,
InspectCapabilityResult,
ListCapabilitiesResult,
NodeSpecCapabilityDetail,
NodeSpecCapabilitySummary,
WrapperArtifactCapabilityDetail,
WrapperArtifactCapabilitySummary,
)
from .common import (
ArtifactVersionPayload,
DependencyDiagnosticPayload,
@@ -69,7 +80,10 @@ __all__ = [
"ArtifactCatalogEntryPayload",
"ArtifactKindPayload",
"CapabilityKindPayload",
"CapabilityCallResult",
"CapabilityRefPayload",
"CapabilitySummary",
"CapabilitySummaryPayload",
"CompileDraftWorkspaceResult",
"CompileDraftWorkspaceSuccess",
"CreateArtifactFromWorkspaceResult",
@@ -87,15 +101,19 @@ __all__ = [
"InterruptPayload",
"InterruptRoutePayload",
"InvalidDraftResult",
"InspectCapabilityResult",
"JsonObject",
"JsonProjector",
"JsonSchema",
"ListDeploymentsResult",
"ListDraftWorkspacesResult",
"ListArtifactsResult",
"ListCapabilitiesResult",
"ListRunsResult",
"NextActionPatchExamplePayload",
"NextActionsPayload",
"NodeSpecCapabilityDetail",
"NodeSpecCapabilitySummary",
"PageMetadataPayload",
"RawWorkflowPlan",
"ResumeReadiness",
@@ -116,6 +134,8 @@ __all__ = [
"WorkflowArtifactPayload",
"WorkflowRefPayload",
"WrapperAuthoringHintsPayload",
"WrapperArtifactCapabilityDetail",
"WrapperArtifactCapabilitySummary",
"WrapperMissingDecisionPayload",
"WrapperOutcomeCandidatePayload",
]
src/wf_api/models/capabilities.py
+90
View File
@@ -0,0 +1,90 @@
from __future__ import annotations
from typing import Literal, TypedDict
from .artifacts import RequiredCapabilityPayload
from .common import DependencyDiagnosticPayload, JsonObject, PageMetadataPayload
from .drafts import WrapperAuthoringHintsPayload
class CapabilitySummaryPayload(TypedDict):
"""Fields shared by compact planner-visible capability rows."""
name: str
source_id: str
description: str | None
outcomes: list[str]
is_async: bool
input_fields: list[str]
output_fields: list[str]
class NodeSpecCapabilitySummary(CapabilitySummaryPayload):
"""Compact discovery row for one executable source node spec."""
kind: Literal["node_spec"]
class WrapperArtifactCapabilitySummary(CapabilitySummaryPayload):
"""Compact discovery row for one saved wrapper artifact."""
kind: Literal["wrapper_artifact"]
artifact_id: str
version: int
title: str
type CapabilitySummary = NodeSpecCapabilitySummary | WrapperArtifactCapabilitySummary
class ListCapabilitiesResult(PageMetadataPayload):
"""Cursor-paged planner-visible capability discovery result."""
capabilities: list[NodeSpecCapabilitySummary | WrapperArtifactCapabilitySummary]
class CapabilityDetailPayload(TypedDict):
"""Fields shared by inspectable workflow capability contracts."""
name: str
source_id: str
description: str | None
outcomes: list[str]
is_async: bool
input_schema: JsonObject
output_schema: JsonObject
wrapper_hints: WrapperAuthoringHintsPayload
class NodeSpecCapabilityDetail(CapabilityDetailPayload):
"""Full executable contract for one source node spec."""
kind: Literal["node_spec"]
accepts_context: bool
class WrapperArtifactCapabilityDetail(CapabilityDetailPayload):
"""Full callable contract for one saved wrapper artifact."""
kind: Literal["wrapper_artifact"]
artifact_id: str
version: int
title: str
required_capabilities: dict[str, RequiredCapabilityPayload]
type InspectCapabilityResult = (
NodeSpecCapabilityDetail | WrapperArtifactCapabilityDetail
)
class CapabilityCallResult(TypedDict):
"""Outcome returned by a direct node-spec or wrapper capability call."""
qualified_name: str
source_id: str
kind: Literal["node_spec", "wrapper_artifact"]
deployment_id: str | None
outcome: str
output: JsonObject | None
diagnostics: list[DependencyDiagnosticPayload]
src/wf_api/service.py
+6 -3
View File
@@ -14,6 +14,7 @@ from .draft_authoring import RouteSource, WorkflowDraftAuthoringApi
from .draft_updates import CapabilityStepUpdate
from .drafts import WorkflowDraftApi
from .models import (
CapabilityCallResult,
CompileDraftWorkspaceResult,
CompileDraftWorkspaceSuccess,
CreateArtifactFromWorkspaceResult,
@@ -22,7 +23,9 @@ from .models import (
DeleteDeploymentResult,
DeleteDraftWorkspaceResult,
DraftWorkspaceResult,
InspectCapabilityResult,
ListArtifactsResult,
ListCapabilitiesResult,
ListDeploymentsResult,
ListDraftWorkspacesResult,
ListRunsResult,
@@ -66,7 +69,7 @@ class WorkflowApi:
source_id: str | None = None,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
) -> ListCapabilitiesResult:
return await self.capabilities.list_capabilities(
query=query,
source_id=source_id,
@@ -78,7 +81,7 @@ class WorkflowApi:
self,
*,
qualified_name: str,
) -> dict[str, Any]:
) -> InspectCapabilityResult:
return await self.capabilities.inspect_capability(qualified_name=qualified_name)
async def call_capability(
@@ -87,7 +90,7 @@ class WorkflowApi:
qualified_name: str,
payload: dict[str, Any],
deployment_id: str | None = None,
) -> dict[str, Any]:
) -> CapabilityCallResult:
return await self.capabilities.call_capability(
qualified_name=qualified_name,
payload=payload,
src/wf_api/surface.py
+6 -3
View File
@@ -10,6 +10,7 @@ from wf_core.models.steps import InputBinding, OutputBinding
from .draft_authoring import RouteSource
from .draft_updates import CapabilityStepUpdate
from .models import (
CapabilityCallResult,
CompileDraftWorkspaceResult,
CreateArtifactFromWorkspaceResult,
CreateDraftWorkspaceFromCapabilityResult,
@@ -17,7 +18,9 @@ from .models import (
DeleteDeploymentResult,
DeleteDraftWorkspaceResult,
DraftWorkspaceResult,
InspectCapabilityResult,
ListArtifactsResult,
ListCapabilitiesResult,
ListDeploymentsResult,
ListDraftWorkspacesResult,
ListRunsResult,
@@ -42,13 +45,13 @@ class WorkflowCapabilitySurface(Protocol):
source_id: str | None = None,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]: ...
) -> ListCapabilitiesResult: ...
async def inspect_capability(
self,
*,
qualified_name: str,
) -> dict[str, Any]: ...
) -> InspectCapabilityResult: ...
async def call_capability(
self,
@@ -56,7 +59,7 @@ class WorkflowCapabilitySurface(Protocol):
qualified_name: str,
payload: dict[str, Any],
deployment_id: str | None = None,
) -> dict[str, Any]: ...
) -> CapabilityCallResult: ...
class WorkflowDraftSurface(Protocol):
src/wf_transport_rpc_http/client/capabilities.py
+22 -7
View File
@@ -1,6 +1,12 @@
from __future__ import annotations
from typing import Any
from typing import Any, cast
from wf_api.models import (
CapabilityCallResult,
InspectCapabilityResult,
ListCapabilitiesResult,
)
from .base import RpcCaller
@@ -15,8 +21,10 @@ class RpcCapabilityClientMixin:
source_id: str | None = None,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
return await self._call(
) -> ListCapabilitiesResult:
return cast(
ListCapabilitiesResult,
await self._call(
"workflow.capabilities.list",
{
"query": query,
@@ -24,14 +32,18 @@ class RpcCapabilityClientMixin:
"cursor": cursor,
"limit": limit,
},
),
)
async def inspect_capability(
self: RpcCaller, *, qualified_name: str
) -> dict[str, Any]:
return await self._call(
) -> InspectCapabilityResult:
return cast(
InspectCapabilityResult,
await self._call(
"workflow.capabilities.inspect",
{"qualified_name": qualified_name},
),
)
async def call_capability(
@@ -40,12 +52,15 @@ class RpcCapabilityClientMixin:
qualified_name: str,
payload: dict[str, Any],
deployment_id: str | None = None,
) -> dict[str, Any]:
return await self._call(
) -> CapabilityCallResult:
return cast(
CapabilityCallResult,
await self._call(
"workflow.capabilities.call",
{
"qualified_name": qualified_name,
"payload": payload,
"deployment_id": deployment_id,
},
),
)
src/wf_transport_rpc_http/methods/capabilities.py
+12 -5
View File
@@ -1,9 +1,16 @@
from __future__ import annotations
"""Capability JSON-RPC method registration.
from typing import Any
Return annotations stay eagerly evaluated because fastapi-jsonrpc captures them
while registering nested handlers for response validation and OpenRPC output.
"""
import fastapi_jsonrpc as jsonrpc
from wf_api.models import (
CapabilityCallResult,
InspectCapabilityResult,
ListCapabilitiesResult,
)
from wf_server import WorkflowServer
from ..errors import WorkflowRpcError, raise_workflow_rpc_error
@@ -24,7 +31,7 @@ def register_methods(
@entrypoint.method(name="workflow.capabilities.list", errors=[WorkflowRpcError])
async def workflow_capabilities_list(
params: ListCapabilitiesParams = RpcParams(),
) -> dict[str, Any]:
) -> ListCapabilitiesResult:
try:
return await server.api.list_capabilities(
query=params.query,
@@ -38,7 +45,7 @@ def register_methods(
@entrypoint.method(name="workflow.capabilities.inspect", errors=[WorkflowRpcError])
async def workflow_capabilities_inspect(
params: InspectCapabilityParams = RpcParams(),
) -> dict[str, Any]:
) -> InspectCapabilityResult:
try:
return await server.api.inspect_capability(
qualified_name=params.qualified_name,
@@ -49,7 +56,7 @@ def register_methods(
@entrypoint.method(name="workflow.capabilities.call", errors=[WorkflowRpcError])
async def workflow_capabilities_call(
params: CallCapabilityParams = RpcParams(),
) -> dict[str, Any]:
) -> CapabilityCallResult:
try:
return await server.api.call_capability(
qualified_name=params.qualified_name,
tests/wf_api/test_capability_api.py
+2
View File
@@ -84,6 +84,8 @@ async def test_inspect_capability_returns_detail_with_wrapper_hints(
detail = await api.inspect_capability(qualified_name="demo.personal.echo_tool")
assert detail["name"] == "demo.personal.echo_tool"
assert detail["source_id"] == "demo.personal"
assert detail["kind"] == "node_spec"
assert "wrapper_hints" in detail
hints = detail["wrapper_hints"]
assert hints["capability_name"] == "demo.personal.echo_tool"
tests/wf_transport_rpc_http/test_app.py
+44
View File
@@ -163,6 +163,50 @@ async def test_rpc_health_and_capability_methods(tmp_path) -> None:
assert called["result"]["output"] == {"value": "hello direct rpc"}
async def test_rpc_capability_methods_preserve_saved_wrapper_fields(tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
await server.api.create_artifact_from_plan(
artifact_id="rpc_wrapper",
version=1,
title="RPC Wrapper",
kind="wrapper",
plan=_constant_plan(),
outcomes=["ok"],
)
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
listed = await _rpc(
client,
"workflow.capabilities.list",
{"source_id": "workflow", "limit": 10},
)
inspected = await _rpc(
client,
"workflow.capabilities.inspect",
{"qualified_name": "workflow.rpc_wrapper.v1"},
)
called = await _rpc(
client,
"workflow.capabilities.call",
{
"qualified_name": "workflow.rpc_wrapper.v1",
"payload": {},
},
)
wrapper = listed["result"]["capabilities"][0]
assert wrapper["kind"] == "wrapper_artifact"
assert wrapper["artifact_id"] == "rpc_wrapper"
assert wrapper["version"] == 1
assert wrapper["title"] == "RPC Wrapper"
assert inspected["result"]["kind"] == "wrapper_artifact"
assert inspected["result"]["artifact_id"] == "rpc_wrapper"
assert "required_capabilities" in inspected["result"]
assert called["result"]["kind"] == "wrapper_artifact"
assert called["result"]["output"] == {"result": "hello over rpc"}
async def test_rpc_unknown_method_returns_json_rpc_error(tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server)
tests/wf_transport_rpc_http/test_openrpc_contract.py
+71
View File
@@ -46,6 +46,77 @@ def test_openrpc_exposes_typed_health_result(
)
def test_openrpc_exposes_typed_capability_list_result(
openrpc_document: dict[str, Any],
) -> None:
_assert_result_component(
openrpc_document,
method_name="workflow.capabilities.list",
component_name="ListCapabilitiesResult",
properties={"capabilities", "next_cursor", "total"},
)
capabilities = openrpc_document["components"]["schemas"]["ListCapabilitiesResult"][
"properties"
]["capabilities"]
assert capabilities["items"] == {
"anyOf": [
{"$ref": "#/components/schemas/NodeSpecCapabilitySummary"},
{"$ref": "#/components/schemas/WrapperArtifactCapabilitySummary"},
]
}
def test_openrpc_exposes_typed_capability_inspect_result(
openrpc_document: dict[str, Any],
) -> None:
method = _method_by_name(openrpc_document, "workflow.capabilities.inspect")
assert method["result"]["schema"] == {
"$ref": "#/components/schemas/InspectCapabilityResult"
}
schemas = openrpc_document["components"]["schemas"]
assert schemas["InspectCapabilityResult"] == {
"anyOf": [
{"$ref": "#/components/schemas/NodeSpecCapabilityDetail"},
{"$ref": "#/components/schemas/WrapperArtifactCapabilityDetail"},
]
}
assert schemas["NodeSpecCapabilityDetail"]["properties"]["wrapper_hints"] == {
"$ref": "#/components/schemas/WrapperAuthoringHintsPayload"
}
assert schemas["NodeSpecCapabilityDetail"]["properties"]["kind"]["const"] == (
"node_spec"
)
assert (
schemas["WrapperArtifactCapabilityDetail"]["properties"]["kind"]["const"]
== "wrapper_artifact"
)
assert schemas["WrapperArtifactCapabilityDetail"]["properties"][
"required_capabilities"
]["additionalProperties"] == {
"$ref": "#/components/schemas/RequiredCapabilityPayload"
}
def test_openrpc_exposes_typed_capability_call_result(
openrpc_document: dict[str, Any],
) -> None:
_assert_result_component(
openrpc_document,
method_name="workflow.capabilities.call",
component_name="CapabilityCallResult",
properties={
"qualified_name",
"source_id",
"kind",
"deployment_id",
"outcome",
"output",
"diagnostics",
},
)
@pytest.mark.parametrize(
("method_name", "component_name", "properties"),
[