feat: add canonical workflow result models

This commit is contained in:
lda
2026-07-30 21:02:53 +07:00 Verified
parent 005a96c9ff
commit 93e690c4d0
18 changed files with 746 additions and 156 deletions
+14
View File
@@ -68,3 +68,17 @@
with no parity check or generated contract. Add an operation-inventory test
or a code-generation seam so server additions cannot silently remain absent
from TypeScript.
- A 2026-07-30 spike confirmed that `fastapi-jsonrpc` already exports a
complete OpenRPC document for all 70 registered methods. Request payloads
retain useful Pydantic schemas, so OpenRPC is a viable transport input.
- The first typed-result slice now gives `workflow.health` and all deployment
and run operations named transport-neutral result schemas: 11 of 70
methods. The remaining 59 success results still collapse to generic objects
because their Python API and JSON-RPC handlers return `dict[str, Any]`.
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.
Keep OpenRPC as an interchange format, but generate a small
transport-neutral contract manifest rather than adopting its client stack.
+3
View File
@@ -51,6 +51,9 @@ can continue to reach a loopback-only workflow RPC server.
- [`docs/wf_authoring_control_flow.md`](wf_authoring_control_flow.md): when to
use `branch`, `handle`, `match`, `when`, and `choose`.
- `wf_api.WorkflowApi`: process-local workflow application facade.
- `wf_api.models`: canonical transport-neutral request and result models shared
by process-local surfaces and remote transports. Legacy `wf_mcp` DTOs are
compatibility contracts, not a second source of truth.
- `wf_server.WorkflowServer`: durable workflow server composition object.
- `wf_transport_rpc_http.RpcWorkflowApiClient`: JSON-RPC client implementing
the workflow/admin surfaces over HTTP.
+30 -9
View File
@@ -15,10 +15,22 @@ from wf_artifacts import (
)
from wf_platform import CapabilitySource, hash_json_schema
from .models import (
DeleteDeploymentResult,
DeploymentSummary,
JsonProjector,
ListDeploymentsResult,
SaveDeploymentResult,
ValidateDeploymentResult,
WorkflowDeploymentPayload,
)
from .next_actions import NextActions
from .operation_context import WorkflowOperationContext
from .saved_subgraphs import resolve_saved_subgraph_tree, validate_saved_subgraph_tree
_PROJECT_DEPLOYMENT = JsonProjector(WorkflowDeploymentPayload)
_PROJECT_DEPLOYMENT_VALIDATION = JsonProjector(ValidateDeploymentResult)
class WorkflowDeploymentApi:
"""Saved deployment operations and dependency validation."""
@@ -31,7 +43,7 @@ class WorkflowDeploymentApi:
raise KeyError("workflow artifact store is not configured")
return self.context.artifact_store
async def list_deployments(self) -> dict[str, Any]:
async def list_deployments(self) -> ListDeploymentsResult:
if self.context.artifact_store is None:
return {"deployments": []}
return {
@@ -41,12 +53,18 @@ class WorkflowDeploymentApi:
]
}
async def inspect_deployment(self, *, deployment_id: str) -> dict[str, Any]:
return (
self._artifact_store().get_deployment(deployment_id).model_dump(mode="json")
async def inspect_deployment(
self, *, deployment_id: str
) -> WorkflowDeploymentPayload:
return _PROJECT_DEPLOYMENT(
self._artifact_store()
.get_deployment(deployment_id)
.model_dump(mode="json"),
)
async def save_deployment(self, deployment: dict[str, Any]) -> dict[str, Any]:
async def save_deployment(
self, deployment: dict[str, Any]
) -> SaveDeploymentResult:
workflow_deployment = WorkflowDeployment.model_validate(deployment)
self._artifact_store().save_deployment(workflow_deployment)
self.context.events.record_workflow_event(
@@ -65,7 +83,9 @@ class WorkflowDeploymentApi:
"saved": True,
}
async def delete_deployment(self, *, deployment_id: str) -> dict[str, Any]:
async def delete_deployment(
self, *, deployment_id: str
) -> DeleteDeploymentResult:
"""Delete one mutable deployment environment binding."""
self._artifact_store().delete_deployment(deployment_id)
self.context.events.record_workflow_event(
@@ -80,7 +100,7 @@ class WorkflowDeploymentApi:
*,
deployment_id: str,
live_check: bool = False,
) -> dict[str, Any]:
) -> ValidateDeploymentResult:
deployment, artifact, diagnostics, tree = self.deployment_validation(
deployment_id
)
@@ -91,7 +111,7 @@ class WorkflowDeploymentApi:
artifacts=[artifact, *tree.artifacts_by_ref.values()],
)
)
return {
result = {
"deployment_id": deployment.id,
"artifact_id": artifact.id,
"artifact_version": artifact.version,
@@ -104,6 +124,7 @@ class WorkflowDeploymentApi:
diagnostics=diagnostics,
).model_dump(mode="json"),
}
return _PROJECT_DEPLOYMENT_VALIDATION(result)
def deployment_validation(
self,
@@ -196,7 +217,7 @@ def _capability_name(qualified_name: str) -> str | None:
return parsed.name
def _deployment_summary(deployment: WorkflowDeployment) -> dict[str, Any]:
def _deployment_summary(deployment: WorkflowDeployment) -> DeploymentSummary:
"""Return compact deployment metadata for progressive list responses."""
return {
"id": deployment.id,
-43
View File
@@ -1,43 +0,0 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from pydantic import BaseModel, Field
from wf_core import Edge
from wf_core.models.steps import InputBinding, Step
@dataclass(frozen=True, slots=True)
class TraceRange:
"""Caller-bounded debug trace slice for durable deployment runs."""
start: int = 0
limit: int = 25
class RawWorkflowPlan(BaseModel):
"""Raw authoring plan using the same graph step and edge models as core."""
name: str
input_schema: dict[str, Any]
state_schema: dict[str, Any]
output_schema: dict[str, Any]
outcomes: list[str] = Field(
default_factory=lambda: ["ok"],
description=(
"Declared public workflow outcomes. Legacy plans without this field "
"default to ok."
),
)
output: list[InputBinding] = Field(
default_factory=list,
description=(
"Optional root workflow output bindings. Sources read graph paths "
"such as state.result and targets write the public output payload."
),
)
start: str
nodes: list[Step]
edges: list[Edge]
+67
View File
@@ -0,0 +1,67 @@
"""Transport-neutral workflow API models."""
from .common import (
ArtifactVersionPayload,
DependencyDiagnosticPayload,
GuidedResultPayload,
HealthResult,
JsonObject,
JsonProjector,
JsonSchema,
NextActionPatchExamplePayload,
NextActionsPayload,
RawWorkflowPlan,
TraceRange,
)
from .deployments import (
DeleteDeploymentResult,
DeploymentSummary,
ListDeploymentsResult,
SaveDeploymentResult,
SourceBindingPayload,
ValidateDeploymentResult,
WorkflowDeploymentPayload,
)
from .runs import (
InterruptPayload,
InterruptRoutePayload,
ListRunsResult,
ResumeReadiness,
RunResult,
RunStatus,
RunSummary,
RunTraceResult,
TraceEntryPayload,
WorkflowRefPayload,
)
__all__ = [
"ArtifactVersionPayload",
"DeleteDeploymentResult",
"DependencyDiagnosticPayload",
"DeploymentSummary",
"HealthResult",
"GuidedResultPayload",
"InterruptPayload",
"InterruptRoutePayload",
"JsonObject",
"JsonProjector",
"JsonSchema",
"ListDeploymentsResult",
"ListRunsResult",
"NextActionPatchExamplePayload",
"NextActionsPayload",
"RawWorkflowPlan",
"ResumeReadiness",
"RunResult",
"RunStatus",
"RunSummary",
"RunTraceResult",
"SaveDeploymentResult",
"SourceBindingPayload",
"TraceRange",
"TraceEntryPayload",
"ValidateDeploymentResult",
"WorkflowDeploymentPayload",
"WorkflowRefPayload",
]
+107
View File
@@ -0,0 +1,107 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Literal, TypedDict
from pydantic import BaseModel, Field, TypeAdapter
from wf_core import Edge
from wf_core.models.steps import InputBinding, Step
type JsonObject = dict[str, Any]
type JsonSchema = dict[str, object]
class JsonProjector[T]:
"""Validate a dictionary projection once against its canonical JSON shape."""
def __init__(self, schema: type[T]) -> None:
self._adapter = TypeAdapter(schema)
def __call__(self, value: object) -> T:
return self._adapter.validate_python(value)
class HealthResult(TypedDict):
"""Health response shared by self-describing workflow transports."""
status: Literal["ok"]
store_root: str
class ArtifactVersionPayload(TypedDict):
"""Identity shared by results that refer to one immutable artifact version."""
artifact_id: str
artifact_version: int
class DependencyDiagnosticPayload(TypedDict):
"""JSON projection of one deployment dependency diagnostic."""
severity: str
code: str
logical_ref: str
bound_source: str | None
message: str
repair_hint: str | None
class NextActionPatchExamplePayload(TypedDict):
"""Concrete follow-up operation suggested to an API caller."""
description: str
tool: str
request: JsonObject
class NextActionsPayload(TypedDict):
"""JSON projection of advisory workflow continuation guidance."""
can_continue: bool
can_save_now: bool | None
recommended_next_tool: str | None
reason: str
patch_examples: list[NextActionPatchExamplePayload]
warnings: list[str]
class GuidedResultPayload(TypedDict):
"""Diagnostics and continuation guidance shared by validated operations."""
diagnostics: list[DependencyDiagnosticPayload]
next_actions: NextActionsPayload
@dataclass(frozen=True, slots=True)
class TraceRange:
"""Caller-bounded debug trace slice for durable deployment runs."""
start: int = 0
limit: int = 25
class RawWorkflowPlan(BaseModel):
"""Raw authoring plan using the same graph step and edge models as core."""
name: str
input_schema: dict[str, Any]
state_schema: dict[str, Any]
output_schema: dict[str, Any]
outcomes: list[str] = Field(
default_factory=lambda: ["ok"],
description=(
"Declared public workflow outcomes. Legacy plans without this field "
"default to ok."
),
)
output: list[InputBinding] = Field(
default_factory=list,
description=(
"Optional root workflow output bindings. Sources read graph paths "
"such as state.result and targets write the public output payload."
),
)
start: str
nodes: list[Step]
edges: list[Edge]
+47
View File
@@ -0,0 +1,47 @@
from __future__ import annotations
from typing import Literal, TypedDict
from .common import ArtifactVersionPayload, GuidedResultPayload
class SourceBindingPayload(TypedDict):
"""JSON projection of a logical-to-concrete deployment source binding."""
logical_source: str
concrete_source: str
class WorkflowDeploymentPayload(ArtifactVersionPayload):
"""Serialized deployment accepted and returned by workflow API surfaces."""
id: str
bindings: list[SourceBindingPayload]
drift_policy: str
class DeploymentSummary(ArtifactVersionPayload):
"""Compact deployment row used by list operations."""
id: str
binding_count: int
drift_policy: str
class ListDeploymentsResult(TypedDict):
deployments: list[DeploymentSummary]
class SaveDeploymentResult(ArtifactVersionPayload):
deployment_id: str
saved: bool
class DeleteDeploymentResult(TypedDict):
deployment_id: str
deleted: bool
class ValidateDeploymentResult(ArtifactVersionPayload, GuidedResultPayload):
deployment_id: str
status: Literal["runnable", "unrunnable"]
+103
View File
@@ -0,0 +1,103 @@
from __future__ import annotations
from typing import NotRequired, TypedDict
from .common import (
ArtifactVersionPayload,
GuidedResultPayload,
JsonObject,
JsonSchema,
)
type RunStatus = str
type ResumeReadiness = str
class WorkflowRefPayload(TypedDict, total=False):
name: str
artifact_id: str
version: int
class InterruptRoutePayload(TypedDict):
frame_id: str
node_id: str
scope_id: str
lineage_id: str
parent_frame_id: str
workflow_ref: WorkflowRefPayload
class InterruptPayload(TypedDict):
"""Persisted typed interrupt contract exposed to API clients."""
id: str
frame_id: str
node_id: str
kind: str
payload: JsonObject
resumable: bool
route: InterruptRoutePayload | None
outcomes: list[str]
request_schema: JsonSchema
resume_schema: JsonSchema
typed: bool
class TraceEntryPayload(TypedDict):
frame_id: str
node_id: str
step_type: str
resolved_input: JsonObject
outcome: str
next_node_id: str
output: JsonObject
state_changes: JsonObject
class RunSummary(ArtifactVersionPayload):
run_id: str
deployment_id: str
status: RunStatus
resume_readiness: ResumeReadiness
diagnostic_count: int
created_at: str
updated_at: str
class ListRunsResult(TypedDict):
runs: list[RunSummary]
total: int
cursor: str | None
next_cursor: str | None
limit: int
class RunResultBase(ArtifactVersionPayload, GuidedResultPayload):
deployment_id: str
status: RunStatus
run_id: str | None
resume_readiness: ResumeReadiness | None
interrupt: InterruptPayload | None
outcome: str | None
error: str | None
output: JsonObject | None
trace_count: int
class RunResult(RunResultBase):
"""Run operation result with an optional caller-requested trace slice."""
trace: NotRequired[list[TraceEntryPayload]]
trace_start: NotRequired[int]
trace_limit: NotRequired[int]
trace_truncated: NotRequired[bool]
class RunTraceResult(RunResultBase):
"""Run result where a bounded trace slice is always present."""
trace: list[TraceEntryPayload]
trace_start: int
trace_limit: int
trace_truncated: bool
+32 -14
View File
@@ -15,6 +15,15 @@ from wf_core import RunState
from .artifact_plans import raw_plan_from_artifact
from .deployments import WorkflowDeploymentApi, _available_sources
from .models import (
InterruptPayload,
JsonProjector,
ListRunsResult,
RunResult,
RunSummary,
RunTraceResult,
TraceEntryPayload,
)
from .next_actions import NextActions
from .operation_context import WorkflowOperationContext
from .run_lifecycle import (
@@ -29,6 +38,10 @@ from .run_lifecycle import (
from .run_locks import AsyncKeyedLock
from .saved_subgraphs import saved_subgraph_tree_from_snapshots
_PROJECT_INTERRUPT = JsonProjector(InterruptPayload)
_PROJECT_RUN_RESULT = JsonProjector(RunResult)
_PROJECT_RUN_TRACE_RESULT = JsonProjector(RunTraceResult)
class TraceRangeLike(Protocol):
"""Small structural trace range accepted from MCP, CLI, or HTTP adapters."""
@@ -67,7 +80,7 @@ class WorkflowRunApi:
deployment_id: str,
workflow_input: dict[str, Any],
trace_range: TraceRangeLike | None = None,
) -> dict[str, Any]:
) -> RunResult:
trace_values = _trace_range_values(trace_range)
deployment, artifact, diagnostics, tree = (
self.deployments.deployment_validation(deployment_id)
@@ -118,7 +131,7 @@ class WorkflowRunApi:
resume_payload: dict[str, Any],
resume_outcome: str = "submitted",
trace_range: TraceRangeLike | None = None,
) -> dict[str, Any]:
) -> RunResult:
"""Resume one durable interrupted deployment run."""
# FileRunStore locks individual file writes only. The API layer owns the
# process-local read/execute/write critical section for one run id.
@@ -137,7 +150,7 @@ class WorkflowRunApi:
resume_payload: dict[str, Any],
resume_outcome: str,
trace_range: TraceRangeLike | None,
) -> dict[str, Any]:
) -> RunResult:
trace_values = _trace_range_values(trace_range)
record, stopped_run = restore_interrupted_run(self._run_store(), run_id)
environment = record.environment
@@ -201,7 +214,7 @@ class WorkflowRunApi:
status: str | None = None,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
) -> ListRunsResult:
"""Return compact persisted run summaries without trace or checkpoint state."""
if limit < 1 or limit > 100:
raise ValueError("limit must be between 1 and 100")
@@ -232,7 +245,7 @@ class WorkflowRunApi:
"limit": limit,
}
async def inspect_run(self, *, run_id: str) -> dict[str, Any]:
async def inspect_run(self, *, run_id: str) -> RunResult:
"""Return one durable stopped-run summary without debug trace entries."""
record, run = load_stored_run(self._run_store(), run_id)
environment = record.environment
@@ -255,12 +268,12 @@ class WorkflowRunApi:
*,
run_id: str,
trace_range: TraceRangeLike,
) -> dict[str, Any]:
) -> RunTraceResult:
"""Return only a caller-bounded debug trace slice from a stopped run."""
trace_values = _trace_range_values(trace_range)
record, run = load_stored_run(self._run_store(), run_id)
environment = record.environment
return _run_payload(
payload = _run_payload(
deployment=environment.deployment,
artifact=environment.root_artifact,
status=record.status.value,
@@ -270,6 +283,9 @@ class WorkflowRunApi:
trace_count=len(run.trace),
**_trace_slice_fields(run, trace_values),
)
# A concrete trace range makes _run_payload include the four trace
# fields required by the narrower trace-result contract.
return _PROJECT_RUN_TRACE_RESULT(payload)
def _trace_range_values(
@@ -300,7 +316,7 @@ def _cursor_offset(cursor: str | None) -> int:
return offset
def _run_summary(record: WorkflowRunRecord) -> dict[str, Any]:
def _run_summary(record: WorkflowRunRecord) -> RunSummary:
"""Return an operator-facing run row without heavy runtime state."""
environment = record.environment
return {
@@ -340,17 +356,17 @@ def _run_payload(
status: str,
run_id: str | None = None,
resume_readiness: str | None = None,
interrupt: dict[str, Any] | None = None,
interrupt: InterruptPayload | None = None,
outcome: str | None = None,
error: str | None = None,
diagnostics: list[DependencyDiagnostic] | None = None,
output: dict[str, Any] | None = None,
trace_count: int = 0,
trace: list[dict[str, Any]] | None = None,
trace: list[TraceEntryPayload] | None = None,
trace_start: int | None = None,
trace_limit: int | None = None,
trace_truncated: bool = False,
) -> dict[str, Any]:
) -> RunResult:
payload = {
"deployment_id": deployment.id,
"artifact_id": artifact.id,
@@ -380,10 +396,12 @@ def _run_payload(
payload["trace_limit"] = trace_limit
payload["trace"] = trace
payload["trace_truncated"] = trace_truncated
return payload
# This helper is the sole projection from runtime/Pydantic objects into the
# stable JSON dictionary described by RunResult.
return _PROJECT_RUN_RESULT(payload)
def _interrupt_payload(run: RunState) -> dict[str, Any] | None:
def _interrupt_payload(run: RunState) -> InterruptPayload | None:
"""Return a JSON-safe interrupt payload for the current run, if paused."""
if run.interrupt is None:
return None
@@ -395,4 +413,4 @@ def _interrupt_payload(run: RunState) -> dict[str, Any] | None:
workflow_ref = route["workflow_ref"]
if hasattr(workflow_ref, "model_dump"):
route["workflow_ref"] = workflow_ref.model_dump(mode="json")
return payload
return _PROJECT_INTERRUPT(payload)
+21 -11
View File
@@ -13,7 +13,17 @@ from .deployments import WorkflowDeploymentApi
from .draft_authoring import RouteSource, WorkflowDraftAuthoringApi
from .draft_updates import CapabilityStepUpdate
from .drafts import WorkflowDraftApi
from .models import RawWorkflowPlan
from .models import (
DeleteDeploymentResult,
ListDeploymentsResult,
ListRunsResult,
RawWorkflowPlan,
RunResult,
RunTraceResult,
SaveDeploymentResult,
ValidateDeploymentResult,
WorkflowDeploymentPayload,
)
from .operation_context import WorkflowOperationContext
from .runs import TraceRangeLike, WorkflowRunApi
@@ -726,27 +736,27 @@ class WorkflowApi:
# -- deployments --
async def list_deployments(self) -> dict[str, Any]:
async def list_deployments(self) -> ListDeploymentsResult:
return await self.deployments.list_deployments()
async def inspect_deployment(
self,
*,
deployment_id: str,
) -> dict[str, Any]:
) -> WorkflowDeploymentPayload:
return await self.deployments.inspect_deployment(deployment_id=deployment_id)
async def save_deployment(
self,
deployment: dict[str, Any],
) -> dict[str, Any]:
) -> SaveDeploymentResult:
return await self.deployments.save_deployment(deployment)
async def delete_deployment(
self,
*,
deployment_id: str,
) -> dict[str, Any]:
) -> DeleteDeploymentResult:
return await self.deployments.delete_deployment(deployment_id=deployment_id)
async def validate_deployment(
@@ -754,7 +764,7 @@ class WorkflowApi:
*,
deployment_id: str,
live_check: bool = False,
) -> dict[str, Any]:
) -> ValidateDeploymentResult:
return await self.deployments.validate_deployment(
deployment_id=deployment_id,
live_check=live_check,
@@ -768,7 +778,7 @@ class WorkflowApi:
status: str | None = None,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
) -> ListRunsResult:
return await self.runs.list_runs(
status=status,
cursor=cursor,
@@ -781,7 +791,7 @@ class WorkflowApi:
deployment_id: str,
workflow_input: dict[str, Any],
trace_range: TraceRangeLike | None = None,
) -> dict[str, Any]:
) -> RunResult:
return await self.runs.run_deployment(
deployment_id=deployment_id,
workflow_input=workflow_input,
@@ -795,7 +805,7 @@ class WorkflowApi:
resume_payload: dict[str, Any],
resume_outcome: str = "submitted",
trace_range: TraceRangeLike | None = None,
) -> dict[str, Any]:
) -> RunResult:
return await self.runs.resume_run(
run_id=run_id,
resume_payload=resume_payload,
@@ -807,7 +817,7 @@ class WorkflowApi:
self,
*,
run_id: str,
) -> dict[str, Any]:
) -> RunResult:
return await self.runs.inspect_run(run_id=run_id)
async def read_run_trace(
@@ -815,7 +825,7 @@ class WorkflowApi:
*,
run_id: str,
trace_range: TraceRangeLike,
) -> dict[str, Any]:
) -> RunTraceResult:
return await self.runs.read_run_trace(
run_id=run_id,
trace_range=trace_range,
+20 -10
View File
@@ -9,6 +9,16 @@ from wf_core.models.steps import InputBinding, OutputBinding
from .draft_authoring import RouteSource
from .draft_updates import CapabilityStepUpdate
from .models import (
DeleteDeploymentResult,
ListDeploymentsResult,
ListRunsResult,
RunResult,
RunTraceResult,
SaveDeploymentResult,
ValidateDeploymentResult,
WorkflowDeploymentPayload,
)
from .runs import TraceRangeLike
@@ -380,31 +390,31 @@ class WorkflowArtifactSurface(Protocol):
class WorkflowDeploymentSurface(Protocol):
"""Deployment methods exposed by workflow frontends."""
async def list_deployments(self) -> dict[str, Any]: ...
async def list_deployments(self) -> ListDeploymentsResult: ...
async def inspect_deployment(
self,
*,
deployment_id: str,
) -> dict[str, Any]: ...
) -> WorkflowDeploymentPayload: ...
async def save_deployment(
self,
deployment: dict[str, Any],
) -> dict[str, Any]: ...
) -> SaveDeploymentResult: ...
async def delete_deployment(
self,
*,
deployment_id: str,
) -> dict[str, Any]: ...
) -> DeleteDeploymentResult: ...
async def validate_deployment(
self,
*,
deployment_id: str,
live_check: bool = False,
) -> dict[str, Any]: ...
) -> ValidateDeploymentResult: ...
class WorkflowRunSurface(Protocol):
@@ -416,7 +426,7 @@ class WorkflowRunSurface(Protocol):
status: str | None = None,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]: ...
) -> ListRunsResult: ...
async def run_deployment(
self,
@@ -424,7 +434,7 @@ class WorkflowRunSurface(Protocol):
deployment_id: str,
workflow_input: dict[str, Any],
trace_range: TraceRangeLike | None = None,
) -> dict[str, Any]: ...
) -> RunResult: ...
async def resume_run(
self,
@@ -433,20 +443,20 @@ class WorkflowRunSurface(Protocol):
resume_payload: dict[str, Any],
resume_outcome: str = "submitted",
trace_range: TraceRangeLike | None = None,
) -> dict[str, Any]: ...
) -> RunResult: ...
async def inspect_run(
self,
*,
run_id: str,
) -> dict[str, Any]: ...
) -> RunResult: ...
async def read_run_trace(
self,
*,
run_id: str,
trace_range: TraceRangeLike,
) -> dict[str, Any]: ...
) -> RunTraceResult: ...
class WorkflowApiSurface(
+6 -3
View File
@@ -1,9 +1,12 @@
from __future__ import annotations
"""JSON-RPC application composition.
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 HealthResult
from wf_server import WorkflowServer
from .errors import WorkflowRpcError
@@ -39,7 +42,7 @@ def create_rpc_app(server: WorkflowServer, *, rpc_path: str = "/rpc") -> jsonrpc
return {"status": "ok"}
@entrypoint.method(name="workflow.health", errors=[WorkflowRpcError])
async def workflow_health() -> dict[str, Any]:
async def workflow_health() -> HealthResult:
return {
"status": "ok",
"store_root": str(server.config.store_root),
+43 -17
View File
@@ -1,6 +1,14 @@
from __future__ import annotations
from typing import Any
from typing import Any, cast
from wf_api.models import (
DeleteDeploymentResult,
ListDeploymentsResult,
SaveDeploymentResult,
ValidateDeploymentResult,
WorkflowDeploymentPayload,
)
from .base import RpcCaller
@@ -8,34 +16,52 @@ from .base import RpcCaller
class RpcDeploymentClientMixin:
"""JSON-RPC implementation of workflow deployment surface methods."""
async def list_deployments(self: RpcCaller) -> dict[str, Any]:
return await self._call("workflow.deployments.list", {})
async def list_deployments(self: RpcCaller) -> ListDeploymentsResult:
return cast(
ListDeploymentsResult,
await self._call("workflow.deployments.list", {}),
)
async def inspect_deployment(
self: RpcCaller, *, deployment_id: str
) -> dict[str, Any]:
return await self._call(
"workflow.deployments.inspect",
{"deployment_id": deployment_id},
) -> WorkflowDeploymentPayload:
return cast(
WorkflowDeploymentPayload,
await self._call(
"workflow.deployments.inspect",
{"deployment_id": deployment_id},
),
)
async def validate_deployment(
self: RpcCaller, *, deployment_id: str, live_check: bool = False
) -> dict[str, Any]:
return await self._call(
"workflow.deployments.validate",
{"deployment_id": deployment_id, "live_check": live_check},
) -> ValidateDeploymentResult:
return cast(
ValidateDeploymentResult,
await self._call(
"workflow.deployments.validate",
{"deployment_id": deployment_id, "live_check": live_check},
),
)
async def save_deployment(
self: RpcCaller, deployment: dict[str, Any]
) -> dict[str, Any]:
return await self._call("workflow.deployments.save", {"deployment": deployment})
) -> SaveDeploymentResult:
return cast(
SaveDeploymentResult,
await self._call(
"workflow.deployments.save",
{"deployment": deployment},
),
)
async def delete_deployment(
self: RpcCaller, *, deployment_id: str
) -> dict[str, Any]:
return await self._call(
"workflow.deployments.delete",
{"deployment_id": deployment_id},
) -> DeleteDeploymentResult:
return cast(
DeleteDeploymentResult,
await self._call(
"workflow.deployments.delete",
{"deployment_id": deployment_id},
),
)
+51 -35
View File
@@ -1,7 +1,8 @@
from __future__ import annotations
from typing import Any
from typing import Any, cast
from wf_api.models import ListRunsResult, RunResult, RunTraceResult
from wf_api.runs import TraceRangeLike
from .base import RpcCaller
@@ -16,14 +17,17 @@ class RpcRunClientMixin:
status: str | None = None,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
return await self._call(
"workflow.runs.list",
{
"status": status,
"cursor": cursor,
"limit": limit,
},
) -> ListRunsResult:
return cast(
ListRunsResult,
await self._call(
"workflow.runs.list",
{
"status": status,
"cursor": cursor,
"limit": limit,
},
),
)
async def run_deployment(
@@ -32,14 +36,17 @@ class RpcRunClientMixin:
deployment_id: str,
workflow_input: dict[str, Any],
trace_range: TraceRangeLike | None = None,
) -> dict[str, Any]:
return await self._call(
"workflow.runs.start",
{
"deployment_id": deployment_id,
"workflow_input": workflow_input,
"trace_range": _trace_range_payload(trace_range),
},
) -> RunResult:
return cast(
RunResult,
await self._call(
"workflow.runs.start",
{
"deployment_id": deployment_id,
"workflow_input": workflow_input,
"trace_range": _trace_range_payload(trace_range),
},
),
)
async def resume_run(
@@ -49,32 +56,41 @@ class RpcRunClientMixin:
resume_payload: dict[str, Any],
resume_outcome: str = "submitted",
trace_range: TraceRangeLike | None = None,
) -> dict[str, Any]:
return await self._call(
"workflow.runs.resume",
{
"run_id": run_id,
"resume_payload": resume_payload,
"resume_outcome": resume_outcome,
"trace_range": _trace_range_payload(trace_range),
},
) -> RunResult:
return cast(
RunResult,
await self._call(
"workflow.runs.resume",
{
"run_id": run_id,
"resume_payload": resume_payload,
"resume_outcome": resume_outcome,
"trace_range": _trace_range_payload(trace_range),
},
),
)
async def inspect_run(self: RpcCaller, *, run_id: str) -> dict[str, Any]:
return await self._call("workflow.runs.inspect", {"run_id": run_id})
async def inspect_run(self: RpcCaller, *, run_id: str) -> RunResult:
return cast(
RunResult,
await self._call("workflow.runs.inspect", {"run_id": run_id}),
)
async def read_run_trace(
self: RpcCaller,
*,
run_id: str,
trace_range: TraceRangeLike,
) -> dict[str, Any]:
return await self._call(
"workflow.runs.trace",
{
"run_id": run_id,
"trace_range": _trace_range_payload(trace_range),
},
) -> RunTraceResult:
return cast(
RunTraceResult,
await self._call(
"workflow.runs.trace",
{
"run_id": run_id,
"trace_range": _trace_range_payload(trace_range),
},
),
)
@@ -1,9 +1,18 @@
from __future__ import annotations
"""Deployment 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 (
DeleteDeploymentResult,
ListDeploymentsResult,
SaveDeploymentResult,
ValidateDeploymentResult,
WorkflowDeploymentPayload,
)
from wf_server import WorkflowServer
from ..errors import WorkflowRpcError, raise_workflow_rpc_error
@@ -26,7 +35,7 @@ def register_methods(
@entrypoint.method(name="workflow.deployments.save", errors=[WorkflowRpcError])
async def workflow_deployments_save(
params: SaveDeploymentParams = RpcParams(),
) -> dict[str, Any]:
) -> SaveDeploymentResult:
try:
return await server.api.save_deployment(params.deployment)
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
@@ -35,7 +44,7 @@ def register_methods(
@entrypoint.method(name="workflow.deployments.validate", errors=[WorkflowRpcError])
async def workflow_deployments_validate(
params: ValidateDeploymentParams = RpcParams(),
) -> dict[str, Any]:
) -> ValidateDeploymentResult:
try:
return await server.api.validate_deployment(
deployment_id=params.deployment_id,
@@ -47,7 +56,7 @@ def register_methods(
@entrypoint.method(name="workflow.deployments.list", errors=[WorkflowRpcError])
async def workflow_deployments_list(
params: ListDeploymentsParams = RpcParams(),
) -> dict[str, Any]:
) -> ListDeploymentsResult:
try:
return await server.api.list_deployments()
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
@@ -56,7 +65,7 @@ def register_methods(
@entrypoint.method(name="workflow.deployments.inspect", errors=[WorkflowRpcError])
async def workflow_deployments_inspect(
params: InspectDeploymentParams = RpcParams(),
) -> dict[str, Any]:
) -> WorkflowDeploymentPayload:
try:
return await server.api.inspect_deployment(
deployment_id=params.deployment_id,
@@ -67,7 +76,7 @@ def register_methods(
@entrypoint.method(name="workflow.deployments.delete", errors=[WorkflowRpcError])
async def workflow_deployments_delete(
params: DeleteDeploymentParams = RpcParams(),
) -> dict[str, Any]:
) -> DeleteDeploymentResult:
try:
return await server.api.delete_deployment(
deployment_id=params.deployment_id,
+10 -7
View File
@@ -1,9 +1,12 @@
from __future__ import annotations
"""Run 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 ListRunsResult, RunResult, RunTraceResult
from wf_server import WorkflowServer
from ..errors import WorkflowRpcError, raise_workflow_rpc_error
@@ -26,7 +29,7 @@ def register_methods(
@entrypoint.method(name="workflow.runs.list", errors=[WorkflowRpcError])
async def workflow_runs_list(
params: ListRunsParams = RpcParams(),
) -> dict[str, Any]:
) -> ListRunsResult:
try:
return await server.api.list_runs(
status=params.status,
@@ -39,7 +42,7 @@ def register_methods(
@entrypoint.method(name="workflow.runs.start", errors=[WorkflowRpcError])
async def workflow_runs_start(
params: StartRunParams = RpcParams(),
) -> dict[str, Any]:
) -> RunResult:
try:
return await server.api.run_deployment(
deployment_id=params.deployment_id,
@@ -56,7 +59,7 @@ def register_methods(
@entrypoint.method(name="workflow.runs.inspect", errors=[WorkflowRpcError])
async def workflow_runs_inspect(
params: InspectRunParams = RpcParams(),
) -> dict[str, Any]:
) -> RunResult:
try:
return await server.api.inspect_run(run_id=params.run_id)
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
@@ -65,7 +68,7 @@ def register_methods(
@entrypoint.method(name="workflow.runs.trace", errors=[WorkflowRpcError])
async def workflow_runs_trace(
params: ReadRunTraceParams = RpcParams(),
) -> dict[str, Any]:
) -> RunTraceResult:
try:
return await server.api.read_run_trace(
run_id=params.run_id,
@@ -77,7 +80,7 @@ def register_methods(
@entrypoint.method(name="workflow.runs.resume", errors=[WorkflowRpcError])
async def workflow_runs_resume(
params: ResumeRunParams = RpcParams(),
) -> dict[str, Any]:
) -> RunResult:
try:
return await server.api.resume_run(
run_id=params.run_id,
+22
View File
@@ -0,0 +1,22 @@
from __future__ import annotations
import pytest
from pydantic import ValidationError
from wf_api.models import HealthResult, JsonProjector
def test_json_projector_validates_transport_neutral_payloads() -> None:
project_health = JsonProjector(HealthResult)
assert project_health({"status": "ok", "store_root": "store"}) == {
"status": "ok",
"store_root": "store",
}
def test_json_projector_rejects_malformed_payloads() -> None:
project_health = JsonProjector(HealthResult)
with pytest.raises(ValidationError):
project_health({"status": "ok"})
@@ -0,0 +1,154 @@
from __future__ import annotations
from collections.abc import Collection
from pathlib import Path
from typing import Any
import pytest
from wf_server import build_local_static_workflow_server
from wf_transport_rpc_http import create_rpc_app
def _method_by_name(document: dict[str, Any], name: str) -> dict[str, Any]:
return next(method for method in document["methods"] if method["name"] == name)
def _assert_result_component(
document: dict[str, Any],
*,
method_name: str,
component_name: str,
properties: Collection[str],
) -> None:
method = _method_by_name(document, method_name)
assert method["result"]["schema"] == {
"$ref": f"#/components/schemas/{component_name}"
}
component = document["components"]["schemas"][component_name]
assert properties <= component["properties"].keys()
@pytest.fixture
def openrpc_document(tmp_path: Path) -> dict[str, Any]:
app = create_rpc_app(build_local_static_workflow_server(tmp_path / "store"))
return app.get_openrpc()
def test_openrpc_exposes_typed_health_result(
openrpc_document: dict[str, Any],
) -> None:
_assert_result_component(
openrpc_document,
method_name="workflow.health",
component_name="HealthResult",
properties={"status", "store_root"},
)
@pytest.mark.parametrize(
("method_name", "component_name", "properties"),
[
(
"workflow.deployments.list",
"ListDeploymentsResult",
{"deployments"},
),
(
"workflow.deployments.inspect",
"WorkflowDeploymentPayload",
{"id", "artifact_id", "artifact_version", "bindings", "drift_policy"},
),
(
"workflow.deployments.save",
"SaveDeploymentResult",
{"deployment_id", "artifact_id", "artifact_version", "saved"},
),
(
"workflow.deployments.delete",
"DeleteDeploymentResult",
{"deployment_id", "deleted"},
),
(
"workflow.deployments.validate",
"ValidateDeploymentResult",
{
"deployment_id",
"artifact_id",
"artifact_version",
"status",
"diagnostics",
"next_actions",
},
),
],
)
def test_openrpc_exposes_typed_deployment_results(
openrpc_document: dict[str, Any],
method_name: str,
component_name: str,
properties: set[str],
) -> None:
_assert_result_component(
openrpc_document,
method_name=method_name,
component_name=component_name,
properties=properties,
)
@pytest.mark.parametrize(
("method_name", "component_name", "properties"),
[
(
"workflow.runs.list",
"ListRunsResult",
{"runs", "total", "cursor", "next_cursor", "limit"},
),
(
"workflow.runs.start",
"RunResult",
{
"deployment_id",
"run_id",
"status",
"interrupt",
"output",
"next_actions",
},
),
(
"workflow.runs.inspect",
"RunResult",
{"run_id", "status", "resume_readiness", "trace_count"},
),
(
"workflow.runs.resume",
"RunResult",
{"run_id", "status", "resume_readiness", "trace_count"},
),
(
"workflow.runs.trace",
"RunTraceResult",
{
"run_id",
"trace",
"trace_start",
"trace_limit",
"trace_truncated",
},
),
],
)
def test_openrpc_exposes_typed_run_results(
openrpc_document: dict[str, Any],
method_name: str,
component_name: str,
properties: set[str],
) -> None:
_assert_result_component(
openrpc_document,
method_name=method_name,
component_name=component_name,
properties=properties,
)