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

feat: type stateless draft results

Browse Source
This commit is contained in:
lda
2026-08-01 08:27:47 +07:00 Verified
parent 3891a62138
commit fbcd3a61a5
12 changed files with 338 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ISSUES.md
+4 -4
View File
@@ -74,10 +74,10 @@
- Typed-result slices now give `workflow.health`, all artifact, deployment,
and run operations, every persisted draft-workspace operation, and both the
capability and source-discovery surfaces named transport-neutral result
schemas: 53 of 70 methods. The remaining 17 success results still collapse
to generic objects across stateless draft patch/validate and the
source-registry/admin operations. Continue introducing operation result DTOs
before adopting generated TypeScript contracts.
schemas: 55 of 70 methods. The remaining 15 success results still collapse
to generic objects across the 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/drafts.py
+33 -7
View File
@@ -60,6 +60,11 @@ from .models import (
InvalidDraftResult,
JsonProjector,
ListDraftWorkspacesResult,
PatchDraftResult,
PatchedDraftInvalidResult,
PatchedDraftValidResult,
ValidateDraftResult,
ValidDraftResult,
)
from .operation_context import WorkflowOperationContext
from .schema_projection import project_property_to_schema_path, schema_path_exists
@@ -69,6 +74,23 @@ _PROJECT_DRAFT_WORKSPACE_LIST = JsonProjector(ListDraftWorkspacesResult)
_PROJECT_DRAFT_WORKSPACE_DELETE = JsonProjector(DeleteDraftWorkspaceResult)
_PROJECT_DRAFT_COMPILE = JsonProjector(CompileDraftWorkspaceSuccess)
_PROJECT_INVALID_DRAFT = JsonProjector(InvalidDraftResult)
_PROJECT_VALID_DRAFT = JsonProjector(ValidDraftResult)
_PROJECT_PATCHED_DRAFT_VALID = JsonProjector(PatchedDraftValidResult)
_PROJECT_PATCHED_DRAFT_INVALID = JsonProjector(PatchedDraftInvalidResult)
def _project_validate_draft(payload: dict[str, Any]) -> ValidateDraftResult:
"""Project one validation result through the matching status variant."""
if payload.get("status") == "valid":
return _PROJECT_VALID_DRAFT(payload)
return _PROJECT_INVALID_DRAFT(payload)
def _project_patch_draft(payload: dict[str, Any]) -> PatchDraftResult:
"""Project one patch result while preserving an optional invalid draft."""
if payload.get("status") == "valid":
return _PROJECT_PATCHED_DRAFT_VALID(payload)
return _PROJECT_PATCHED_DRAFT_INVALID(payload)
def _empty_object_schema() -> dict[str, Any]:
@@ -174,12 +196,14 @@ class WorkflowDraftApi:
node_defs.append(spec.to_node_def())
return node_defs
async def validate_draft(self, *, draft: dict[str, Any]) -> dict[str, Any]:
return validate_workflow_draft(
async def validate_draft(self, *, draft: dict[str, Any]) -> ValidateDraftResult:
return _project_validate_draft(
validate_workflow_draft(
draft,
outcome_lookup=self._outcomes_for_capability,
node_defs=self._node_defs_for_draft(draft),
)
)
async def compile_draft(
self, *, draft: dict[str, Any]
@@ -203,12 +227,14 @@ class WorkflowDraftApi:
*,
draft: dict[str, Any],
patch: list[dict[str, Any]],
) -> dict[str, Any]:
return patch_workflow_draft(
) -> PatchDraftResult:
return _project_patch_draft(
patch_workflow_draft(
draft,
patch,
node_defs_for_draft=self._node_defs_for_draft,
)
)
async def list_draft_workspaces(self) -> ListDraftWorkspacesResult:
"""Return compact summaries for stored draft workspaces."""
@@ -887,14 +913,14 @@ def _path_text(value: Any, *, expected_root: str | None = None) -> str:
def _with_workspace_repair_hints(
payload: dict[str, Any],
payload: Mapping[str, Any],
*,
workspace_id: str,
revision: int,
) -> dict[str, Any]:
diagnostics = payload.get("diagnostics")
if not isinstance(diagnostics, list):
return payload
return dict(payload)
enriched = []
changed = False
for diagnostic in diagnostics:
@@ -912,7 +938,7 @@ def _with_workspace_repair_hints(
changed = True
enriched.append(repaired)
if not changed:
return payload
return dict(payload)
return {**payload, "diagnostics": enriched}
src/wf_api/models/__init__.py
+10
View File
@@ -56,8 +56,13 @@ from .drafts import (
DraftWorkspaceSummary,
InvalidDraftResult,
ListDraftWorkspacesResult,
PatchDraftResult,
PatchedDraftInvalidResult,
PatchedDraftValidResult,
SavedDraftArtifactResult,
UnsavedDraftArtifactResult,
ValidateDraftResult,
ValidDraftResult,
WrapperAuthoringHintsPayload,
WrapperMissingDecisionPayload,
WrapperOutcomeCandidatePayload,
@@ -136,6 +141,9 @@ __all__ = [
"NodeSpecCapabilitySummary",
"NodeSpecInventoryPayload",
"PageMetadataPayload",
"PatchDraftResult",
"PatchedDraftInvalidResult",
"PatchedDraftValidResult",
"RawWorkflowPlan",
"ReducerInventoryPayload",
"ResumeReadiness",
@@ -163,7 +171,9 @@ __all__ = [
"TraceRange",
"TraceEntryPayload",
"UnsavedDraftArtifactResult",
"ValidateDraftResult",
"ValidateDeploymentResult",
"ValidDraftResult",
"WorkflowDeploymentPayload",
"WorkflowArtifactPayload",
"WorkflowRefPayload",
src/wf_api/models/drafts.py
+28
View File
@@ -61,6 +61,34 @@ class InvalidDraftResult(TypedDict):
diagnostics: list[DraftDiagnosticPayload]
class ValidDraftResult(TypedDict):
"""Successful validation of one stateless draft document."""
status: Literal["valid"]
diagnostics: list[DraftDiagnosticPayload]
compiled_plan: JsonObject
type ValidateDraftResult = ValidDraftResult | InvalidDraftResult
class PatchedDraftValidResult(ValidDraftResult):
"""Valid draft document produced after applying a JSON Patch."""
draft: JsonObject
class PatchedDraftInvalidResult(InvalidDraftResult):
"""Invalid patch result, optionally retaining the applied draft document."""
# Malformed JSON Patch cannot produce a document. Applied patches retain
# their invalid document so a later edit can repair it.
draft: NotRequired[JsonObject]
type PatchDraftResult = PatchedDraftValidResult | PatchedDraftInvalidResult
class CompileDraftWorkspaceSuccess(TypedDict):
"""Compiled raw plan and dependencies for one valid draft workspace."""
src/wf_api/service.py
+4 -2
View File
@@ -29,6 +29,7 @@ from .models import (
ListDeploymentsResult,
ListDraftWorkspacesResult,
ListRunsResult,
PatchDraftResult,
RawWorkflowPlan,
RunResult,
RunTraceResult,
@@ -36,6 +37,7 @@ from .models import (
SavedDraftArtifactResult,
SaveDeploymentResult,
ValidateDeploymentResult,
ValidateDraftResult,
WorkflowArtifactPayload,
WorkflowDeploymentPayload,
)
@@ -254,7 +256,7 @@ class WorkflowApi:
self,
*,
draft: dict[str, Any],
) -> dict[str, Any]:
) -> ValidateDraftResult:
return await self.drafts.validate_draft(draft=draft)
async def compile_draft(
@@ -269,7 +271,7 @@ class WorkflowApi:
*,
draft: dict[str, Any],
patch: list[dict[str, Any]],
) -> dict[str, Any]:
) -> PatchDraftResult:
return await self.drafts.patch_draft(draft=draft, patch=patch)
# -- draft workspaces --
src/wf_api/surface.py
+15
View File
@@ -26,12 +26,14 @@ from .models import (
ListDraftWorkspacesResult,
ListRunsResult,
ListSourcesResult,
PatchDraftResult,
RunResult,
RunTraceResult,
SaveArtifactResult,
SaveDeploymentResult,
SourceDiagnosisResult,
ValidateDeploymentResult,
ValidateDraftResult,
WorkflowArtifactPayload,
WorkflowDeploymentPayload,
)
@@ -72,6 +74,19 @@ class WorkflowDraftSurface(Protocol):
not every same-process authoring helper on ``WorkflowApi``.
"""
async def validate_draft(
self,
*,
draft: dict[str, Any],
) -> ValidateDraftResult: ...
async def patch_draft(
self,
*,
draft: dict[str, Any],
patch: list[dict[str, Any]],
) -> PatchDraftResult: ...
async def list_draft_workspaces(self) -> ListDraftWorkspacesResult: ...
async def get_draft_workspace(
src/wf_transport_rpc_http/client/drafts.py
+26
View File
@@ -11,6 +11,8 @@ from wf_api.models import (
DeleteDraftWorkspaceResult,
DraftWorkspaceResult,
ListDraftWorkspacesResult,
PatchDraftResult,
ValidateDraftResult,
)
from wf_api.surface import RouteSource
from wf_artifacts.drafts.models import DraftStep
@@ -31,6 +33,30 @@ async def _call_draft_workspace(
class RpcDraftClientMixin:
"""JSON-RPC implementation of workflow draft workspace surface methods."""
async def validate_draft(
self: RpcCaller,
*,
draft: dict[str, Any],
) -> ValidateDraftResult:
return cast(
ValidateDraftResult,
await self._call("workflow.drafts.validate", {"draft": draft}),
)
async def patch_draft(
self: RpcCaller,
*,
draft: dict[str, Any],
patch: list[dict[str, Any]],
) -> PatchDraftResult:
return cast(
PatchDraftResult,
await self._call(
"workflow.drafts.patch",
{"draft": draft, "patch": patch},
),
)
async def list_draft_workspaces(self: RpcCaller) -> ListDraftWorkspacesResult:
return cast(
ListDraftWorkspacesResult,
src/wf_transport_rpc_http/methods/drafts.py
+4 -4
View File
@@ -4,8 +4,6 @@ Return annotations stay eagerly evaluated because fastapi-jsonrpc captures them
while registering nested handlers for response validation and OpenRPC output.
"""
from typing import Any
import fastapi_jsonrpc as jsonrpc
from wf_api.models import (
@@ -15,6 +13,8 @@ from wf_api.models import (
DeleteDraftWorkspaceResult,
DraftWorkspaceResult,
ListDraftWorkspacesResult,
PatchDraftResult,
ValidateDraftResult,
)
from wf_api.surface import RouteSource
from wf_server import WorkflowServer
@@ -77,7 +77,7 @@ def register_methods(
@entrypoint.method(name="workflow.drafts.patch", errors=[WorkflowRpcError])
async def workflow_drafts_patch(
params: PatchDraftParams = RpcParams(),
) -> dict[str, Any]:
) -> PatchDraftResult:
try:
return await server.api.patch_draft(draft=params.draft, patch=params.patch)
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
@@ -86,7 +86,7 @@ def register_methods(
@entrypoint.method(name="workflow.drafts.validate", errors=[WorkflowRpcError])
async def workflow_drafts_validate(
params: ValidateDraftParams = RpcParams(),
) -> dict[str, Any]:
) -> ValidateDraftResult:
try:
return await server.api.validate_draft(draft=params.draft)
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
tests/wf_api/test_drafts_service.py
+30
View File
@@ -740,6 +740,36 @@ async def test_patch_draft_applies_json_patch(tmp_path: Path) -> None:
}
@pytest.mark.asyncio
async def test_patch_draft_projects_valid_applied_document(tmp_path: Path) -> None:
artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_patch_valid")
api, _service, _authoring = _draft_api(artifact_store, register_echo=True)
result = await api.patch_draft(
draft=_echo_draft(),
patch=[{"op": "replace", "path": "/name", "value": "renamed_echo"}],
)
assert result["status"] == "valid"
assert result["draft"]["name"] == "renamed_echo"
assert result["compiled_plan"]["name"] == "renamed_echo"
@pytest.mark.asyncio
async def test_patch_draft_malformed_patch_has_no_draft(tmp_path: Path) -> None:
artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_patch_malformed")
api, _service, _authoring = _draft_api(artifact_store, register_echo=True)
result = await api.patch_draft(
draft=_echo_draft(),
patch=[{"op": "remove", "path": "/missing"}],
)
assert result["status"] == "invalid"
assert result["diagnostics"][0]["code"] == "patch_invalid"
assert "draft" not in result
@pytest.mark.asyncio
async def test_create_draft_workspace_creates_workspace(tmp_path: Path) -> None:
artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_create_workspace")
tests/wf_transport_rpc_http/test_app.py
+104 -40
View File
@@ -31,6 +31,48 @@ async def _rpc(
return response.json()
def _rpc_constant_draft() -> dict[str, Any]:
"""Return the canonical keyed draft shared by stateless RPC tests."""
return {
"name": "rpc_constant",
"input_schema": {"type": "object", "properties": {}},
"state_schema": {
"type": "object",
"properties": {"result": {"type": "string", "reducer": "wf.std.replace"}},
},
"output_schema": {
"type": "object",
"properties": {"result": {"type": "string"}},
"required": ["result"],
},
"start": "constant",
"steps": {
"constant": {
"use": "wf.std.constant",
"input": [
{
"value": "hello over rpc",
"target": {"root": "local", "parts": ["value"]},
}
],
"output": [
{
"source": {"root": "local", "parts": ["value"]},
"target": {"root": "state", "parts": ["result"]},
}
],
}
},
"routes": {"constant": {"ok": "__end__"}},
"output": [
{
"path": {"root": "state", "parts": ["result"]},
"target": {"root": "local", "parts": ["result"]},
}
],
}
def test_update_capability_step_params_preserve_nested_field_presence() -> None:
params = UpdateCapabilityStepParams.model_validate(
{
@@ -193,6 +235,67 @@ async def test_rpc_source_discovery_preserves_inventory_contracts(tmp_path) -> N
assert diagnosed["result"]["status"] == "unknown"
async def test_rpc_stateless_draft_methods_preserve_result_variants(tmp_path) -> None:
app = create_rpc_app(build_local_static_workflow_server(tmp_path / "store"))
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
validated = await _rpc(
client,
"workflow.drafts.validate",
{"draft": {}},
)
patched_valid = await _rpc(
client,
"workflow.drafts.patch",
{
"draft": _rpc_constant_draft(),
"patch": [
{
"op": "replace",
"path": "/name",
"value": "renamed_rpc_constant",
}
],
},
)
patched_invalid = await _rpc(
client,
"workflow.drafts.patch",
{
"draft": _rpc_constant_draft(),
"patch": [
{
"op": "replace",
"path": "/routes/constant/ok",
"value": "missing_step",
}
],
},
)
patched_malformed = await _rpc(
client,
"workflow.drafts.patch",
{
"draft": {},
"patch": [{"op": "remove", "path": "/missing"}],
},
)
assert validated["result"]["status"] == "invalid"
assert validated["result"]["diagnostics"]
assert patched_valid["result"]["status"] == "valid"
assert patched_valid["result"]["draft"]["name"] == "renamed_rpc_constant"
assert patched_valid["result"]["compiled_plan"]["name"] == "renamed_rpc_constant"
assert patched_invalid["result"]["status"] == "invalid"
assert patched_invalid["result"]["diagnostics"]
assert patched_invalid["result"]["draft"]["routes"]["constant"]["ok"] == (
"missing_step"
)
assert patched_malformed["result"]["status"] == "invalid"
assert patched_malformed["result"]["diagnostics"][0]["code"] == "patch_invalid"
assert "draft" not in patched_malformed["result"]
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(
@@ -285,46 +388,7 @@ async def test_rpc_draft_artifact_deployment_lifecycle(tmp_path) -> None:
},
)
draft = {
"name": "rpc_constant",
"input_schema": {"type": "object", "properties": {}},
"state_schema": {
"type": "object",
"properties": {
"result": {"type": "string", "reducer": "wf.std.replace"}
},
},
"output_schema": {
"type": "object",
"properties": {"result": {"type": "string"}},
"required": ["result"],
},
"start": "constant",
"steps": {
"constant": {
"use": "wf.std.constant",
"input": [
{
"value": "hello over rpc",
"target": {"root": "local", "parts": ["value"]},
}
],
"output": [
{
"source": {"root": "local", "parts": ["value"]},
"target": {"root": "state", "parts": ["result"]},
}
],
}
},
"routes": {"constant": {"ok": "__end__"}},
"output": [
{
"path": {"root": "state", "parts": ["result"]},
"target": {"root": "local", "parts": ["result"]},
}
],
}
draft = _rpc_constant_draft()
validate_draft = await _rpc(
client,
tests/wf_transport_rpc_http/test_client.py
+27
View File
@@ -399,6 +399,33 @@ async def test_rpc_client_sends_exact_draft_lifecycle_payloads() -> None:
]
async def test_rpc_client_sends_exact_stateless_draft_payloads() -> 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 {"status": "invalid", "diagnostics": []}
client = Client()
draft = {"name": "report"}
patch = [{"op": "replace", "path": "/name", "value": "renamed"}]
await client.validate_draft(draft=draft)
await client.patch_draft(draft=draft, patch=patch)
assert calls == [
{
"method": "workflow.drafts.validate",
"params": {"draft": draft},
},
{
"method": "workflow.drafts.patch",
"params": {"draft": draft, "patch": patch},
},
]
async def test_rpc_client_sends_exact_replace_document_payload() -> None:
calls: list[dict[str, Any]] = []
tests/wf_transport_rpc_http/test_openrpc_contract.py
+47
View File
@@ -350,6 +350,53 @@ def test_openrpc_exposes_typed_compile_draft_workspace_result(
assert schemas["InvalidDraftResult"]["properties"]["status"]["const"] == "invalid"
def test_openrpc_exposes_typed_stateless_draft_validation_result(
openrpc_document: dict[str, Any],
) -> None:
method = _method_by_name(openrpc_document, "workflow.drafts.validate")
schemas = openrpc_document["components"]["schemas"]
assert method["result"]["schema"] == {
"$ref": "#/components/schemas/ValidateDraftResult"
}
assert schemas["ValidateDraftResult"] == {
"anyOf": [
{"$ref": "#/components/schemas/ValidDraftResult"},
{"$ref": "#/components/schemas/InvalidDraftResult"},
]
}
assert schemas["ValidDraftResult"]["required"] == [
"status",
"diagnostics",
"compiled_plan",
]
def test_openrpc_exposes_typed_stateless_draft_patch_result(
openrpc_document: dict[str, Any],
) -> None:
method = _method_by_name(openrpc_document, "workflow.drafts.patch")
schemas = openrpc_document["components"]["schemas"]
assert method["result"]["schema"] == {
"$ref": "#/components/schemas/PatchDraftResult"
}
assert schemas["PatchDraftResult"] == {
"anyOf": [
{"$ref": "#/components/schemas/PatchedDraftValidResult"},
{"$ref": "#/components/schemas/PatchedDraftInvalidResult"},
]
}
assert "draft" in schemas["PatchedDraftValidResult"]["required"]
assert "draft" not in schemas["PatchedDraftInvalidResult"]["required"]
assert schemas["PatchedDraftValidResult"]["properties"]["draft"] == {
"$ref": "#/components/schemas/JsonObject"
}
assert schemas["PatchedDraftInvalidResult"]["properties"]["draft"] == {
"$ref": "#/components/schemas/JsonObject"
}
def test_openrpc_exposes_typed_capability_bootstrap_result(
openrpc_document: dict[str, Any],
) -> None: