diff --git a/docs/superpowers/plans/2026-08-30-python-workflow-client.md b/docs/superpowers/plans/2026-08-30-python-workflow-client.md new file mode 100644 index 00000000..87f5b561 --- /dev/null +++ b/docs/superpowers/plans/2026-08-30-python-workflow-client.md @@ -0,0 +1,1348 @@ +# Python Workflow Client Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Deliver a public async-native `wf_client.App` whose returned Python objects can inspect and call capabilities, author and edit workflows through `WorkflowBuilder`, validate locally and remotely, save immutable artifacts, deploy them, and operate durable runs. + +**Architecture:** `wf_client` is a deep public module over a narrow internal `WorkflowClientPort`; the existing `RpcWorkflowApiClient` is its production HTTP adapter. `EditableWorkflow` subclasses the transport-free `WorkflowBuilder`, while codecs reconstruct wire payloads into existing `wf_core` and `wf_artifacts` domain models at the adapter seam. + +**Tech Stack:** Python 3.14, Pydantic 2, jsonschema 4.26, httpx, FastAPI JSON-RPC, pytest, pytest-asyncio, ruff, basedpyright. + +**Spec:** `docs/superpowers/specs/2026-08-30-python-workflow-client-design.md` + +## Global Constraints + +- Python baseline is 3.14 (`requires-python = ">=3.14"`). +- `App` is the only public connection entry point. +- Network operations are async; do not add an event-loop-owning synchronous facade. +- Local graph mutations remain synchronous. +- JSON serialization and result validation stay at the transport/codec seam. +- Reuse `wf_core`, `wf_artifacts`, and `wf_authoring` models; do not create parallel workflow semantics. +- Workflow artifacts are immutable; editing seeds a new mutable graph. +- Capability use and native subgraphs remain distinct operations. +- The client never guesses between multiple source accounts or deployments. +- Draft models and draft operations must not be imported or exposed by `wf_client`. +- Add docstrings or comments at non-obvious reconstruction, routing-replacement, and validation seams. +- Use focused pytest commands first; run repository-wide verification only after the focused suite passes. + +## File Structure + +Create: + +```text +src/wf_client/ +├── __init__.py # stable public exports +├── app.py # App construction and top-level lookup +├── authoring.py # EditableWorkflow subclass +├── capabilities.py # RemoteCapability and CapabilityResult +├── codec.py # validated wire/domain reconstruction +├── deployments.py # Deployment lifecycle object +├── errors.py # structured client exception hierarchy +├── protocols.py # narrow WorkflowClientPort +├── runs.py # Run and TracePage lifecycle objects +└── workflows.py # WorkflowArtifact and save/edit behavior + +tests/wf_client/ +├── __init__.py +├── conftest.py +├── test_app.py +├── test_authoring.py +├── test_capabilities.py +├── test_codec.py +├── test_deployments.py +├── test_http_integration.py +├── test_repr.py +└── test_runs.py +``` + +Modify focused existing modules instead of duplicating them: + +- `src/wf_authoring/builder/core.py`: lossless workflow output, reconstruction, and focused graph editing. +- `src/wf_authoring/builder/mapping.py`: schema-contract auto-binding shared by local and remote capability authoring. +- `src/wf_api/artifacts.py`, `src/wf_api/service.py`, and `src/wf_api/surface.py`: non-persisting plan validation. +- `src/wf_api/models/artifacts.py` and `src/wf_api/models/__init__.py`: validation result contract. +- `src/wf_transport_rpc_http/models.py`, `methods/artifacts.py`, and `client/artifacts.py`: JSON-RPC validation operation. +- `src/wf_transport_rpc_http/client/base.py`: preserve structured JSON-RPC errors. +- `contracts/workflow-api.manifest.json` and generated TypeScript contract files: checked operation inventory. +- `docs/project_map.md`, `docs/source_architecture.md`, `docs/wf_api_architecture.md`, and `docs/current_roadmap.md`: public architecture and completed-slice status. + +--- + +### Task 1: Make `WorkflowBuilder` lossless and safely editable + +**Files:** + +- Modify: `src/wf_authoring/builder/core.py:205` +- Modify: `src/wf_authoring/builder/mapping.py:138` +- Modify: `src/wf_authoring/builder/__init__.py` +- Test: `tests/authoring/test_builder.py` +- Test: `tests/authoring/test_subgraph.py` + +**Interfaces:** + +- Consumes: canonical `Workflow`, `NodeDef`, `Step`, `Edge`, `InputBinding`, and existing `WorkflowBuilder` operations. +- Produces: `WorkflowBuilder.from_workflow(workflow)`, `set_output(bindings)`, `set_route(source, outcome, target)`, `remove_route(source, outcome)`, `remove_step(step)`, `use_contract(...)`, `validate_structure()`, and lossless `compile()`. + +- [ ] **Step 1: Write failing workflow-output and reconstruction tests** + +Add tests that construct a workflow with a path final-output binding, seed a builder from it, mutate the returned builder, and prove the original remains unchanged: + +```python +def test_builder_round_trip_preserves_complete_workflow() -> None: + original = Workflow( + name="round_trip", + input_schema={"type": "object", "properties": {"topic": {"type": "string"}}}, + state_schema={ + "type": "object", + "properties": {"result": {"type": "string"}}, + }, + output_schema={ + "type": "object", + "properties": {"result": {"type": "string"}}, + }, + output=[{"path": "state.result", "target": "result"}], + node_defs=[ + { + "name": "app.default.search", + "input_schema": {"type": "object", "properties": {}}, + "output_schema": { + "type": "object", + "properties": {"result": {"type": "string"}}, + }, + "outcomes": ["ok"], + } + ], + outcomes=["ok"], + start="search", + nodes=[ + { + "id": "search", + "type": "node", + "node": "app.default.search", + "input": [], + "output": [ + {"source": "result", "target": "state.result"}, + ], + }, + {"id": "end_ok", "type": "end", "outcome": "ok"}, + ], + edges=[{"from": "search", "outcome": "ok", "to": "end_ok"}], + ) + + builder = WorkflowBuilder.from_workflow(original) + rebuilt = builder.compile() + + assert rebuilt.model_dump(mode="json", by_alias=True) == original.model_dump( + mode="json", by_alias=True + ) + builder.set_output([{"value": "changed", "target": "result"}]) + original_output = original.output[0] + assert isinstance(original_output, InputPathBinding) + assert str(original_output.path) == "state.result" +``` + +Add focused tests proving `set_route()` replaces rather than duplicates, `remove_route()` removes only the requested source/outcome pair, and `remove_step()` rejects referenced steps: + +```python +def test_set_route_replaces_unique_source_outcome_edge() -> None: + builder = _editable_three_step_builder() + + builder.set_route("first", "ok", "third") + + matching = [ + edge + for edge in builder.edges + if edge.from_ == "first" and edge.outcome == "ok" + ] + assert [(edge.from_, edge.outcome, edge.to) for edge in matching] == [ + ("first", "ok", "third") + ] + + +def test_remove_step_rejects_referenced_step() -> None: + builder = _editable_three_step_builder() + + with pytest.raises(ValueError, match="still referenced by route"): + builder.remove_step("second") +``` + +- [ ] **Step 2: Run the focused tests and confirm the missing behavior** + +Run: + +```powershell +uv run pytest tests/authoring/test_builder.py tests/authoring/test_subgraph.py -q +``` + +Expected: the new tests fail because reconstruction, workflow-output storage, and focused edit methods do not exist. + +- [ ] **Step 3: Add schema-contract mapping helpers** + +Generalize the auto-mapping functions so remote schema contracts do not need fake Pydantic classes: + +```python +def auto_input_map_from_schema( + capability_input_schema: SchemaRef, + *, + input_schema: SchemaRef, + state_schema: StateSchema, +) -> dict[str, str]: + return { + _auto_source_path( + field, + input_schema=input_schema, + state_schema=state_schema, + ): field + for field in capability_input_schema.properties + } + + +def auto_output_map_from_schema( + capability_output_schema: SchemaRef, + *, + state_schema: StateSchema, +) -> dict[str, str]: + state_fields = state_schema.field_map() + return { + field: f"state.{field}" + for field in capability_output_schema.properties + if field in state_fields + } +``` + +Keep `auto_input_map(spec, ...)` and `auto_output_map(spec, ...)` as shallow compatibility callers of these schema-based helpers. + +- [ ] **Step 4: Implement lossless builder state and focused editing** + +Add builder-owned workflow output and seeded node definitions: + +```python +workflow_output: list[InputBinding] = field(default_factory=list) +seeded_node_defs: dict[str, NodeDef] = field(default_factory=dict, repr=False) +``` + +Implement the public reconstruction and mutation signatures: + +```python +@classmethod +def from_workflow(cls, workflow: Workflow) -> WorkflowBuilder: + return cls( + name=workflow.name, + input_schema=workflow.input_schema.model_copy(deep=True), + state_schema=workflow.state_schema.model_copy(deep=True), + output_schema=workflow.output_schema.model_copy(deep=True), + outcomes=tuple(workflow.outcomes), + start=workflow.start, + nodes=[node.model_copy(deep=True) for node in workflow.nodes], + edges=[edge.model_copy(deep=True) for edge in workflow.edges], + workflow_output=[binding.model_copy(deep=True) for binding in workflow.output], + seeded_node_defs={ + node_def.name: node_def.model_copy(deep=True) + for node_def in workflow.node_defs + }, + ) + + +def set_output(self, bindings: Sequence[StepInputBindingArg]) -> None: + self.workflow_output = normalize_step_input_bindings(bindings) + + +def set_route(self, source: StepRef, outcome: str, target: StepRef) -> None: + source_id = step_id(source) + target_id = step_id(target) + self.edges = [ + edge + for edge in self.edges + if not (edge.from_ == source_id and edge.outcome == outcome) + ] + self.edges.append( + Edge.model_validate({"from": source_id, "outcome": outcome, "to": target_id}) + ) +``` + +Implement `remove_route()` and `remove_step()` with explicit missing/referrer errors. Implement `use_contract()` to register a `NodeDef`, derive auto-bindings from its schemas, and lower to `use_ref()` without inventing a local handler: + +```python +def use_contract( + self, + node_def: NodeDef, + *, + id: str | None = None, + input: Sequence[StepInputBindingArg] | None = None, + output: Sequence[OutputBindingArg] | None = None, + desc: str | None = None, +) -> NodeUse: ... +``` + +Update `compile()` to merge copied seeded definitions with definitions derived from local `NodeSpec`s, reject incompatible duplicate names, and set `output=self.workflow_output`. + +Extract one private `_build_workflow(start: str) -> Workflow` constructor used +by both public operations. `compile()` retains its explicit missing-start error; +`validate_structure()` passes `self.start or ""` so an unset entry point becomes +the canonical `UNKNOWN_START` validation issue instead of an exception: + +```python +def validate_structure(self) -> ValidationReport: + return self._build_workflow(start=self.start or "").validate_structure() + + +def compile(self) -> Workflow: + if self.start is None: + raise WorkflowExecutionError( + "workflow builder requires an explicit start; " + "call set_entry_point(...) or pass start=..." + ) + return self._build_workflow(start=self.start) +``` + +- [ ] **Step 5: Run the focused authoring tests** + +Run: + +```powershell +uv run pytest tests/authoring/test_builder.py tests/authoring/test_subgraph.py -q +uv run basedpyright --level error src/wf_authoring tests/authoring/test_builder.py tests/authoring/test_subgraph.py +``` + +Expected: all focused tests pass and basedpyright reports zero errors. + +- [ ] **Step 6: Commit the builder foundation** + +```powershell +git add src/wf_authoring/builder tests/authoring/test_builder.py tests/authoring/test_subgraph.py +git commit -m "feat: make workflow builder edits lossless" +``` + +--- + +### Task 2: Add non-persisting server-side plan validation + +**Files:** + +- Modify: `src/wf_api/models/artifacts.py` +- Modify: `src/wf_api/models/__init__.py` +- Modify: `src/wf_api/artifacts.py` +- Modify: `src/wf_api/service.py` +- Modify: `src/wf_api/surface.py` +- Modify: `src/wf_transport_rpc_http/models.py` +- Modify: `src/wf_transport_rpc_http/methods/artifacts.py` +- Modify: `src/wf_transport_rpc_http/client/artifacts.py` +- Modify: `tests/wf_api/test_artifact_api.py` +- Modify: `tests/wf_transport_rpc_http/test_app.py` +- Modify: `tests/wf_transport_rpc_http/test_client.py` +- Modify: `tests/wf_transport_rpc_http/test_openrpc_contract.py` +- Modify: `contracts/workflow-api.manifest.json` +- Modify: generated contract files under `web/packages/rpc/src/generated/` + +**Interfaces:** + +- Consumes: `create_workflow_artifact_from_plan`, `RawWorkflowPlan`, `observed_node_specs(context)`, existing artifact creation arguments. +- Produces: `WorkflowArtifactApi.validate_artifact_plan(...) -> ValidateArtifactPlanResult`, surface/client method `validate_artifact_plan`, and JSON-RPC operation `workflow.artifacts.validate_plan`. + +- [ ] **Step 1: Write failing API tests for valid and invalid plans** + +Define the response contract: + +```python +class ArtifactPlanDiagnosticPayload(TypedDict): + severity: Literal["error", "warning"] + code: str + path: str + message: str + repair_hint: str | None + + +class ValidateArtifactPlanResult(TypedDict): + status: Literal["valid", "invalid"] + diagnostics: list[ArtifactPlanDiagnosticPayload] + required_capabilities: list[RequiredCapabilityPayload] + workflow_dependencies: dict[str, int] +``` + +Test that validation derives dependencies without writing the artifact store: + +```python +async def test_validate_artifact_plan_does_not_persist(tmp_path) -> None: + server = build_local_static_workflow_server(tmp_path / "store") + plan = _constant_plan().model_dump(mode="json", by_alias=True) + + result = await server.api.validate_artifact_plan( + plan=plan, + outcomes=["ok"], + source_bindings={}, + ) + + assert result["status"] == "valid" + assert result["diagnostics"] == [] + assert await server.api.list_artifacts(query="constant") == { + "nodes": [], + "next_cursor": None, + "total": 0, + } +``` + +Add an invalid-plan test asserting a stable diagnostic code and path rather than a raised traceback. + +- [ ] **Step 2: Run the focused API test and confirm it fails** + +```powershell +uv run pytest tests/wf_api/test_artifact_api.py -q +``` + +Expected: failure because `validate_artifact_plan()` and its result model do not exist. + +- [ ] **Step 3: Extract one shared preparation seam** + +In `wf_api.artifacts`, add a private pure helper used by both validation and save: + +```python +def _prepare_artifact_from_plan( + context: WorkflowOperationContext, + *, + artifact_id: str, + version: int, + title: str, + kind: ArtifactKind, + description: str | None, + plan: RawWorkflowPlan | dict[str, Any], + outcomes: Sequence[str], + required_capabilities: dict[str, dict[str, Any]] | None, + source_bindings: dict[str, str] | None, + created_from_catalog_version: str | None, +) -> WorkflowArtifact: + typed_plan = ( + plan if isinstance(plan, RawWorkflowPlan) else RawWorkflowPlan.model_validate(plan) + ) + return build_workflow_artifact_from_plan( + artifact_id=artifact_id, + version=version, + title=title, + kind=kind, + description=description, + plan=typed_plan.model_dump(mode="json", by_alias=True), + outcomes=tuple(outcomes), + required_capabilities={ + name: RequiredCapability.model_validate(capability) + for name, capability in (required_capabilities or {}).items() + }, + source_bindings=source_bindings, + observed_node_specs=observed_node_specs(context), + created_from_catalog_version=created_from_catalog_version, + ) +``` + +Make `create_artifact_from_plan()` call this helper before its existing store write and event. Implement validation with reserved in-memory metadata, catch only expected model/workflow validation exceptions, and project them to diagnostics. Do not catch programming errors. + +- [ ] **Step 4: Expose the operation through the stable surface and JSON-RPC adapter** + +Use the exact signature at every layer: + +```python +async def validate_artifact_plan( + self, + *, + plan: dict[str, Any], + outcomes: Sequence[str], + required_capabilities: dict[str, dict[str, Any]] | None = None, + source_bindings: dict[str, str] | None = None, +) -> ValidateArtifactPlanResult: ... +``` + +Add `ValidateArtifactPlanParams` in the transport models and register `workflow.artifacts.validate_plan` beside artifact create/inspect operations. Add one ASGI RPC test and one `RpcWorkflowApiClient` test proving no artifact is persisted. + +- [ ] **Step 5: Regenerate and verify checked contracts** + +Run: + +```powershell +uv run python -m wf_contract_manifest write +pnpm --dir web --filter @lda/workflow-rpc contract:write +uv run pytest tests/wf_api/test_artifact_api.py tests/wf_transport_rpc_http/test_app.py tests/wf_transport_rpc_http/test_client.py tests/wf_transport_rpc_http/test_openrpc_contract.py tests/wf_contract_manifest/test_generate.py tests/wf_contract_manifest/test_committed_manifest.py -q +pnpm --dir web --filter @lda/workflow-rpc test +``` + +Expected: the new operation has a named success schema, manifest drift checks pass, and TypeScript generated-contract tests pass. Do not add it to the browser-authorized Effect RPC cohort; this task expands the transport inventory, not browser policy. + +- [ ] **Step 6: Commit server validation** + +```powershell +git add src/wf_api src/wf_transport_rpc_http tests/wf_api/test_artifact_api.py tests/wf_transport_rpc_http contracts/workflow-api.manifest.json web/packages/rpc/src/generated web/packages/rpc/scripts +git commit -m "feat: validate workflow plans without saving" +``` + +--- + +### Task 3: Build the narrow client port, codecs, and structured errors + +**Files:** + +- Create: `src/wf_client/protocols.py` +- Create: `src/wf_client/codec.py` +- Create: `src/wf_client/errors.py` +- Create: `tests/wf_client/__init__.py` +- Create: `tests/wf_client/conftest.py` +- Create: `tests/wf_client/test_codec.py` +- Modify: `src/wf_transport_rpc_http/client/base.py` +- Modify: `tests/wf_transport_rpc_http/test_client.py` + +**Interfaces:** + +- Consumes: typed result contracts from Task 2, existing `RpcWorkflowApiClient` methods, `WorkflowArtifact`, `WorkflowDeployment`, `Workflow`, `DependencyDiagnostic`, and run payload contracts. +- Produces: `WorkflowClientPort`, `RpcProtocolError`, `WorkflowClientError` hierarchy, and named codec functions used by all later rich objects. + +- [ ] **Step 1: Write failing codec and error-preservation tests** + +Create a fake narrow adapter in `tests/wf_client/conftest.py` whose methods return configurable payloads and record calls. Test artifact reconstruction: + +```python +def test_decode_workflow_artifact_validates_plan() -> None: + payload = workflow_artifact_payload(plan=_constant_plan_payload()) + + artifact, workflow = decode_workflow_artifact(payload) + + assert artifact.id == "report" + assert workflow.name == "report" + assert workflow.start == "constant" +``` + +Test malformed nested plan rejection: + +```python +def test_decode_workflow_artifact_rejects_invalid_nested_plan() -> None: + payload = workflow_artifact_payload(plan={"name": "broken"}) + + with pytest.raises(InvalidResponse, match="workflow.artifacts.inspect"): + decode_workflow_artifact(payload) +``` + +Add an RPC transport test proving JSON-RPC `error.code`, `error.message`, and `error.data` survive in a structured exception. + +- [ ] **Step 2: Run focused tests and confirm failures** + +```powershell +uv run pytest tests/wf_client/test_codec.py tests/wf_transport_rpc_http/test_client.py -q +``` + +Expected: failures because `wf_client` and `RpcProtocolError` do not exist. + +- [ ] **Step 3: Define the narrow port** + +Create a `Protocol` containing only methods rich client objects require: + +```python +class WorkflowClientPort(Protocol): + async def list_capabilities( + self, + *, + query: str | None = None, + source_id: str | None = None, + cursor: str | None = None, + limit: int = 50, + ) -> ListCapabilitiesResult: ... + + async def inspect_capability( + self, + *, + qualified_name: str, + ) -> InspectCapabilityResult: ... + + async def call_capability( + self, + *, + qualified_name: str, + payload: dict[str, Any], + deployment_id: str | None = None, + ) -> CapabilityCallResult: ... + + async def validate_artifact_plan( + self, + *, + plan: dict[str, Any], + outcomes: Sequence[str], + required_capabilities: dict[str, dict[str, Any]] | None = None, + source_bindings: dict[str, str] | None = None, + ) -> ValidateArtifactPlanResult: ... +``` + +Include exact existing artifact inspect/create, deployment list/inspect/save/validate, and run start/inspect/resume/trace signatures. Do not inherit the 70-operation `WorkflowApiSurface`. + +- [ ] **Step 4: Preserve structured JSON-RPC failures** + +Add this transport exception in `wf_transport_rpc_http.client.base`: + +```python +@dataclass(frozen=True, slots=True) +class RpcProtocolError(RuntimeError): + code: int | str | None + message: str + data: object = None + + def __str__(self) -> str: + if isinstance(self.data, dict) and isinstance(self.data.get("message"), str): + return f"{self.message}: {self.data['message']}" + return self.message +``` + +Define `InvalidResponse` concretely so every codec reports the failed +operation: + +```python +@dataclass(frozen=True, slots=True) +class InvalidResponse(WorkflowClientError): + operation: str + details: str + + def __str__(self) -> str: + return f"invalid response from {self.operation}: {self.details}" +``` + +Raise it from `_call()` instead of flattening the envelope into `RuntimeError`. It remains a `RuntimeError` subclass so existing CLI error handling remains compatible. + +- [ ] **Step 5: Implement codec functions with existing models** + +Create named functions instead of a generic reflection layer: + +```python +def decode_workflow_artifact( + payload: object, +) -> tuple[WorkflowArtifact, Workflow]: ... + + +def decode_deployment(payload: object) -> WorkflowDeployment: ... + + +def decode_dependency_diagnostics( + payload: object, +) -> tuple[DependencyDiagnostic, ...]: ... + + +def decode_run_result(payload: object) -> DecodedRunResult: ... + + +def decode_trace_result(payload: object) -> DecodedTracePage: ... +``` + +Use `TypeAdapter` for wire `TypedDict` validation, then `WorkflowArtifact.model_validate()`, `RawWorkflowPlan.model_validate()`, and `Workflow.model_validate()`. Wrap validation failures as `InvalidResponse(operation=..., details=...)`; do not weaken schemas to accept malformed server data. + +- [ ] **Step 6: Run focused tests and type checks** + +```powershell +uv run pytest tests/wf_client/test_codec.py tests/wf_transport_rpc_http/test_client.py -q +uv run basedpyright --level error src/wf_client src/wf_transport_rpc_http/client tests/wf_client/test_codec.py +``` + +Expected: codec and structured-error tests pass with zero type errors. + +- [ ] **Step 7: Commit the client foundation** + +```powershell +git add src/wf_client/protocols.py src/wf_client/codec.py src/wf_client/errors.py src/wf_transport_rpc_http/client/base.py tests/wf_client tests/wf_transport_rpc_http/test_client.py +git commit -m "feat: add validated workflow client boundary" +``` + +--- + +### Task 4: Add `App` and callable remote capabilities + +**Files:** + +- Create: `src/wf_client/app.py` +- Create: `src/wf_client/capabilities.py` +- Create: `src/wf_client/__init__.py` +- Create: `tests/wf_client/test_app.py` +- Create: `tests/wf_client/test_capabilities.py` + +**Interfaces:** + +- Consumes: `WorkflowClientPort`, `RpcWorkflowApiClient`, capability result codecs, JSON Schema Draft 2020-12 validation, and `WorkflowBuilder.use_contract()` from Task 1. +- Produces: `App.from_http_jsonrpc()`, `App.capability()`, `App.capabilities()`, `Page[CapabilitySummary]`, `RemoteCapability`, `CapabilityResult`, and `App.new_workflow()` construction hook. + +- [ ] **Step 1: Write failing connection and capability tests** + +Test construction performs no network operation: + +```python +def test_from_http_jsonrpc_is_lazy(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 == [] +``` + +Test callable behavior and graph contract: + +```python +async def test_remote_capability_is_callable_and_graph_usable(fake_port) -> None: + fake_port.inspect_capability_result = search_capability_payload() + fake_port.call_capability_result = { + "qualified_name": "app.default.search", + "source_id": "app.default", + "kind": "node_spec", + "deployment_id": None, + "outcome": "ok", + "output": {"results": ["one"]}, + "diagnostics": [], + } + app = App._from_port(fake_port) + + search = await app.capability("app.default.search") + result = await search(query="workflow") + + assert search.ref == CapabilityRef( + source="app.default", + capability_key="search", + ) + assert result.outcome == "ok" + assert result.output == {"results": ["one"]} +``` + +Add tests rejecting positional-plus-keyword payloads, invalid local inputs before I/O, invalid outputs after I/O, and reserved client options only through `call()`. + +Add a paged discovery test proving list rows become attribute-bearing immutable +objects rather than leaked dictionaries: + +```python +async def test_capability_discovery_returns_rich_page(fake_port) -> None: + fake_port.list_capabilities_result = capability_page_payload() + + page = await App._from_port(fake_port).capabilities(query="search", limit=10) + + assert page.total == 1 + assert page.next_cursor is None + assert page.items[0].qualified_name == "app.default.search" + assert page.items[0].outcomes == ("ok",) +``` + +- [ ] **Step 2: Run tests and confirm missing public objects** + +```powershell +uv run pytest tests/wf_client/test_app.py tests/wf_client/test_capabilities.py -q +``` + +Expected: import failures for `App` and `RemoteCapability`. + +- [ ] **Step 3: Implement `RemoteCapability`** + +Use a frozen dataclass retaining its port and inspected contract: + +```python +@dataclass(frozen=True, slots=True) +class RemoteCapability: + _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 + + 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) + + def node_def(self) -> NodeDef: + return NodeDef( + name=self.qualified_name, + input_schema=self.input_schema, + output_schema=self.output_schema, + outcomes=list(self.outcomes), + ) +``` + +Use `jsonschema.Draft202012Validator` for input/output validation. Reject an inspected schema that is not itself a valid Draft 2020-12 schema when constructing the object. + +- [ ] **Step 4: Implement `App` and public exports** + +Provide a private/test construction seam and the public transport factory: + +```python +@dataclass(frozen=True, slots=True) +class App: + _port: WorkflowClientPort = field(repr=False) + endpoint: str + + @classmethod + def from_http_jsonrpc( + cls, + url: str, + *, + timeout_seconds: float = 30.0, + ) -> App: + return cls( + _port=RpcWorkflowApiClient( + url=url, + timeout_seconds=timeout_seconds, + ), + endpoint=url, + ) + + @classmethod + def _from_port(cls, port: WorkflowClientPort) -> App: + return cls(_port=port, endpoint="in-process") +``` + +`capability()` validates the inspect payload and reconstructs `CapabilityRef` structurally from `source_id` plus the source-relative capability key. Do not split a dotted qualified name heuristically; add/reuse the platform capability-ref parser. + +Use an exact prefix check so capability keys may themselves contain dots: + +```python +def _capability_ref(qualified_name: str, source_id: str) -> CapabilityRef: + 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}" + ), + ) + return CapabilityRef( + source=source_id, + capability_key=qualified_name.removeprefix(prefix), + ) +``` + +Define `Page[T]` and `CapabilitySummary` as frozen dataclasses. `Page.items` and +summary outcomes are tuples so decoded discovery results cannot be mutated +behind the client's validation seam. + +- [ ] **Step 5: Run focused tests and type checks** + +```powershell +uv run pytest tests/wf_client/test_app.py tests/wf_client/test_capabilities.py -q +uv run basedpyright --level error src/wf_client tests/wf_client/test_app.py tests/wf_client/test_capabilities.py +``` + +Expected: all tests pass and public imports resolve from `wf_client`. + +- [ ] **Step 6: Commit capability ergonomics** + +```powershell +git add src/wf_client tests/wf_client/test_app.py tests/wf_client/test_capabilities.py +git commit -m "feat: add callable remote workflow capabilities" +``` + +--- + +### Task 5: Add editable workflows and immutable artifact round trips + +**Files:** + +- Create: `src/wf_client/authoring.py` +- Create: `src/wf_client/workflows.py` +- Create: `tests/wf_client/test_authoring.py` +- Modify: `src/wf_client/app.py` +- Modify: `src/wf_client/__init__.py` +- Modify: `examples/lda_report_workflow/build_workflow.py:24` + +**Interfaces:** + +- Consumes: `WorkflowBuilder.from_workflow()`, `use_contract()`, Task 2 validation operation, Task 3 codecs/port, and Task 4 `RemoteCapability`. +- Produces: `EditableWorkflow(WorkflowBuilder)`, `WorkflowValidation`, `WorkflowArtifact`, `App.new_workflow()`, `App.workflow()`, `App.edit_workflow()`, and lossless `save()`. + +- [ ] **Step 1: Write failing local/remote validation tests** + +Prove local failures avoid I/O and valid plans run both sides: + +```python +async def test_validate_stops_before_remote_call_when_local_graph_is_invalid( + fake_port, +) -> None: + graph = App._from_port(fake_port).new_workflow( + "invalid", + input_schema={"type": "object", "properties": {}}, + state_schema={"type": "object", "properties": {}}, + output_schema={"type": "object", "properties": {}}, + ) + + result = await graph.validate() + + assert result.local.ok is False + assert result.remote_status == "not_run" + assert fake_port.calls == [] + + +async def test_validate_runs_local_and_server_validation(fake_port) -> None: + graph = valid_editable_workflow(App._from_port(fake_port)) + fake_port.validate_artifact_plan_result = { + "status": "valid", + "diagnostics": [], + "required_capabilities": [], + "workflow_dependencies": {}, + } + + result = await graph.validate() + + assert result.ok is True + assert result.local.ok is True + assert result.remote_status == "valid" + assert fake_port.calls[-1].operation == "validate_artifact_plan" +``` + +- [ ] **Step 2: Write failing save/edit round-trip tests** + +Use an artifact payload containing composite inputs, final-output bindings, explicit end nodes, and a native subgraph. Assert: + +```python +async def test_edit_and_save_preserve_complete_artifact_plan(fake_port) -> None: + fake_port.inspect_artifact_result = complex_artifact_payload() + app = App._from_port(fake_port) + + graph = await app.edit_workflow("report", version=1) + fake_port.inspect_artifact_result = complex_artifact_payload(version=2) + saved = await graph.save(version=2) + + create_call = next( + call + for call in fake_port.calls + if call.operation == "create_artifact_from_plan" + ) + assert create_call.operation == "create_artifact_from_plan" + assert create_call.params["plan"] == complex_artifact_payload(version=1)["plan"] + assert saved.ref == ArtifactRef("report", 2) +``` + +Add an assertion that `EditableWorkflow` is a `WorkflowBuilder` and directly exposes `when`, `choose`, `match`, `foreach`, `interrupt`, `end`, `connect`, and `set_entry_point`. + +- [ ] **Step 3: Run tests and confirm failures** + +```powershell +uv run pytest tests/wf_client/test_authoring.py -q +``` + +Expected: failures because editable workflow and artifact objects do not exist. + +- [ ] **Step 4: Implement `EditableWorkflow` as a real subclass** + +Use explicit dataclass fields compatible with `WorkflowBuilder` initialization; do not proxy through `__getattr__`: + +```python +@dataclass(slots=True) +class EditableWorkflow(WorkflowBuilder): + _port: WorkflowClientPort = field(repr=False, kw_only=True) + based_on: ArtifactRef | None = field(default=None, kw_only=True) + artifact_title: str | None = field(default=None, kw_only=True) + artifact_description: str | None = field(default=None, kw_only=True) + + def use( + self, + spec: NodeSpec[Any, Any] | RemoteCapability, + **kwargs: Any, + ) -> NodeUse: + if isinstance(spec, RemoteCapability): + return self.use_contract(spec.node_def(), **kwargs) + return super().use(spec, **kwargs) +``` + +Preserve precise overloads in the actual implementation; the broad body signature is not the public typing contract. Implement `validate_local()`, async `validate()`, and `save()` exactly as specified. `save()` must call validation first and refuse invalid results before `create_artifact_from_plan()`. `validate_local()` delegates to inherited `validate_structure()` so incomplete graphs produce ordinary `ValidationReport` issues without network I/O. + +- [ ] **Step 5: Implement immutable `WorkflowArtifact` and App lookups** + +Use a frozen dataclass retaining both canonical models: + +```python +@dataclass(frozen=True, slots=True) +class WorkflowArtifact: + _port: WorkflowClientPort = field(repr=False, compare=False) + artifact: ArtifactDomainModel + workflow: Workflow + + @property + def ref(self) -> ArtifactRef: + return ArtifactRef(self.artifact.id, self.artifact.version) + + def inspect(self) -> Workflow: + return self.workflow.model_copy(deep=True) + + def edit(self) -> EditableWorkflow: + return EditableWorkflow.from_artifact(self) +``` + +`App.workflow()` calls inspect and codec reconstruction. `App.edit_workflow()` is `return (await self.workflow(...)).edit()`. `App.new_workflow()` creates the subclass with no `based_on` ref. After `save()` validates and calls `create_artifact_from_plan()`, it inspects the newly saved exact version and returns that reconstructed immutable object; it never fabricates an artifact from the save acknowledgement. + +- [ ] **Step 6: Remove the documented builder output workaround** + +Update `examples/lda_report_workflow/build_workflow.py` to call `builder.set_output(...)` instead of compiling then applying `model_copy(update={"output": ...})`. Keep the example behavior identical and delete the stale limitation comment. + +- [ ] **Step 7: Run authoring and artifact verification** + +```powershell +uv run pytest tests/wf_client/test_authoring.py tests/authoring/test_builder.py tests/authoring/test_subgraph.py tests/examples/test_lda_report_workflow_example.py -q +uv run basedpyright --level error src/wf_client src/wf_authoring tests/wf_client/test_authoring.py +``` + +Expected: lossless round trips, subclass behavior, and the existing report example pass. + +- [ ] **Step 8: Commit workflow authoring** + +```powershell +git add src/wf_client src/wf_authoring examples/lda_report_workflow/build_workflow.py tests/wf_client/test_authoring.py tests/authoring tests/examples/test_lda_report_workflow_example.py +git commit -m "feat: add editable remote workflow lifecycle" +``` + +--- + +### Task 6: Add deployment and durable run objects + +**Files:** + +- Create: `src/wf_client/deployments.py` +- Create: `src/wf_client/runs.py` +- Create: `tests/wf_client/test_deployments.py` +- Create: `tests/wf_client/test_runs.py` +- Modify: `src/wf_client/app.py` +- Modify: `src/wf_client/workflows.py` +- Modify: `src/wf_client/__init__.py` + +**Interfaces:** + +- Consumes: Task 3 codecs/port, existing deployment and run operations, `TraceRange`, and Task 5 `WorkflowArtifact`. +- Produces: `Deployment`, `DeploymentValidation`, `Run`, `TracePage`, `WorkflowArtifact.deploy()`, strict `WorkflowArtifact.run()`, `App.deployment()`, and `App.run()`. + +- [ ] **Step 1: Write failing explicit deployment tests** + +Test save, validate, and run: + +```python +async def test_artifact_deploys_with_explicit_bindings(fake_port, workflow_artifact) -> None: + fake_port.validate_deployment_result = runnable_deployment_result() + + deployment = await workflow_artifact.deploy( + "report.production", + bindings={"app.default": "company.production"}, + ) + + assert deployment.deployment_id == "report.production" + assert deployment.bindings == {"app.default": "company.production"} + assert deployment.runnable is True + assert [call.operation for call in fake_port.calls[-3:]] == [ + "save_deployment", + "inspect_deployment", + "validate_deployment", + ] +``` + +- [ ] **Step 2: Write failing convenience-run policy tests** + +Cover the three allowed paths and ambiguity rejection: + +```python +async def test_artifact_run_rejects_ambiguous_deployments( + fake_port, + workflow_artifact, +) -> None: + fake_port.list_deployments_result = { + "deployments": [ + deployment_summary("report.dev", "report", 1), + deployment_summary("report.prod", "report", 1), + ] + } + + with pytest.raises(DeploymentRequired) as captured: + await workflow_artifact.run({"topic": "workflow"}) + + assert captured.value.candidate_deployment_ids == ( + "report.dev", + "report.prod", + ) + assert not any(call.operation == "run_deployment" for call in fake_port.calls) +``` + +Also prove an artifact with no matching deployment creates deterministic id +`report.v1.default` with the caller-supplied bindings or an empty binding map, +then asks the server to validate it. It runs only when the server returns +`runnable`; otherwise it raises `DeploymentRequired` containing the server +diagnostics and leaves the named deployment available for deliberate repair. +A supplied `deployment_id` bypasses discovery. + +- [ ] **Step 3: Write failing run lifecycle tests** + +```python +async def test_interrupted_run_resumes_and_reads_bounded_trace(fake_port) -> None: + run = Run.from_payload(fake_port, interrupted_run_payload()) + fake_port.resume_run_result = completed_run_payload() + fake_port.run_trace_result = trace_payload(start=0, limit=25) + + completed = await run.resume({"approved": True}) + trace = await completed.trace(limit=25) + + assert completed.status == "completed" + assert completed.output == {"result": "done"} + assert trace.start == 0 + assert trace.limit == 25 + assert len(trace.frames) == 1 +``` + +Add tests for `refresh()`, non-resumable runs, negative trace starts, and trace limits outside the existing server bound. + +- [ ] **Step 4: Run focused tests and confirm missing lifecycle objects** + +```powershell +uv run pytest tests/wf_client/test_deployments.py tests/wf_client/test_runs.py -q +``` + +Expected: import or attribute failures for deployment/run objects. + +- [ ] **Step 5: Implement deployment objects and strict artifact convenience** + +Create immutable snapshots: + +```python +@dataclass(frozen=True, slots=True) +class Deployment: + _port: WorkflowClientPort = field(repr=False, compare=False) + model: WorkflowDeployment + diagnostics: tuple[DependencyDiagnostic, ...] = () + runnable: bool | None = None + + @property + def deployment_id(self) -> str: + return self.model.id + + async def validate(self) -> DeploymentValidation: ... + async def run(self, workflow_input: Mapping[str, Any]) -> Run: ... +``` + +Implement the exact convenience policy: + +1. If `deployment_id` is supplied, inspect and run only that deployment. +2. Otherwise filter deployment summaries to the exact artifact id and version. +3. Use the sole match. +4. Reject multiple matches with sorted candidates. +5. With no match, save `.v.default` using explicitly + supplied bindings or `{}` and immediately request server validation. +6. Run the default only when server validation says `runnable`; otherwise raise + `DeploymentRequired` with its diagnostics. Do not infer source type from + names or artifact payloads. +7. Never overwrite an existing deterministic deployment bound to another + artifact version. + +- [ ] **Step 6: Implement immutable run snapshots** + +```python +@dataclass(frozen=True, slots=True) +class Run: + _port: WorkflowClientPort = field(repr=False, compare=False) + run_id: str + deployment_id: str + status: str + outcome: str | None + output: dict[str, Any] | None + interrupt: InterruptRequest | None + diagnostics: tuple[DependencyDiagnostic, ...] + trace_count: int + + async def refresh(self) -> Run: ... + + async def resume( + self, + response: Mapping[str, Any], + *, + outcome: str = "submitted", + ) -> Run: ... + + async def trace(self, *, start: int = 0, limit: int = 25) -> TracePage: ... +``` + +Methods return new snapshots and never mutate `self`. Validate trace bounds before I/O and preserve structured diagnostics. + +Reuse `wf_core.InterruptRequest` and `wf_core.TraceEntry` for reconstructed +interrupts and trace frames. `TracePage.frames` is `tuple[TraceEntry, ...]`; +do not expose `InterruptPayload`, `TraceEntryPayload`, or other wire +`TypedDict`s from the public objects. + +`Deployment.run()` constructs `Run` only when `run_deployment()` returns a +non-null `run_id`. An unrunnable or rejected result raises +`DeploymentNotRunnable` with reconstructed diagnostics, outcome, and server +error text; a missing id must not become a fake Python run. + +- [ ] **Step 7: Run lifecycle verification** + +```powershell +uv run pytest tests/wf_client/test_deployments.py tests/wf_client/test_runs.py tests/wf_api/test_deployment_api.py tests/wf_api/test_run_api.py -q +uv run basedpyright --level error src/wf_client tests/wf_client/test_deployments.py tests/wf_client/test_runs.py +``` + +- [ ] **Step 8: Commit deployment and run ergonomics** + +```powershell +git add src/wf_client tests/wf_client/test_deployments.py tests/wf_client/test_runs.py +git commit -m "feat: add Python deployment and run objects" +``` + +--- + +### Task 7: Prove HTTP usage, add bounded IPython representations, and document the package + +**Files:** + +- Create: `tests/wf_client/test_http_integration.py` +- Create: `tests/wf_client/test_repr.py` +- Modify: `src/wf_client/capabilities.py` +- Modify: `src/wf_client/workflows.py` +- Modify: `src/wf_client/deployments.py` +- Modify: `src/wf_client/runs.py` +- Modify: `docs/project_map.md` +- Modify: `docs/source_architecture.md` +- Modify: `docs/wf_api_architecture.md` +- Modify: `docs/current_roadmap.md` +- Modify: `CONTEXT.md` only if implementation changes a term from the approved spec + +**Interfaces:** + +- Consumes: all public objects from Tasks 4-6 and the real local JSON-RPC ASGI application. +- Produces: verified `App.from_http_jsonrpc()` lifecycle, safe `repr()`/`_repr_html_()`, and current user/architecture documentation. + +- [ ] **Step 1: Write a failing end-to-end HTTP test** + +Use `httpx.ASGITransport` with the real RPC app and give its configured +`RpcWorkflowApiClient` to the private in-process construction seam. The public +HTTP factory remains transport-detail free: + +```python +async def test_http_app_calls_authors_saves_deploys_and_runs(tmp_path) -> None: + server = build_local_static_workflow_server(tmp_path / "store") + rpc_app = create_rpc_app(server) + transport = httpx.ASGITransport(app=rpc_app) + + async with httpx.AsyncClient( + transport=transport, + base_url="http://test", + ) as http_client: + app = App._from_port( + RpcWorkflowApiClient( + url="http://test/rpc", + http_client=http_client, + ) + ) + constant = await app.capability("wf.std.constant") + graph = empty_string_workflow(app) + step = graph.use( + constant, + id="constant", + input=[input_value("value", "hello")], + output=[output_to("value", state_path("value"))], + ) + end = graph.end("ok", id="end_ok") + graph.set_entry_point(step) + graph.connect(step, "ok", end) + graph.set_output([input_from(state_path("value"), "value")]) + + validation = await graph.validate() + artifact = await graph.save(version=1, title="HTTP client proof") + run = await artifact.run({}) + + assert validation.ok is True + assert artifact.ref == ArtifactRef("http_client_proof", 1) + assert run.status == "completed" + assert run.output == {"value": "hello"} +``` + +- [ ] **Step 2: Write representation safety tests** + +For every rich object, clear the fake adapter call log, render both representations, and assert no calls occurred: + +```python +def test_capability_html_repr_is_bounded_and_does_not_call_port( + fake_port, + remote_capability, +) -> None: + fake_port.calls.clear() + + rendered = remote_capability._repr_html_() + + assert "app.default.search" in rendered + assert "input schema" in rendered.lower() + assert fake_port.calls == [] +``` + +Add tests that large schemas/diagnostics/outputs are summarized by count and bounded previews, and secret-like fields are redacted with the repository's existing redaction helper rather than a second ad-hoc implementation. + +- [ ] **Step 3: Run the new tests and confirm missing representation/integration behavior** + +```powershell +uv run pytest tests/wf_client/test_http_integration.py tests/wf_client/test_repr.py -q +``` + +Expected: failures until the HTTP-client injection seam and rich representations are implemented. + +- [ ] **Step 4: Implement bounded, inert representations** + +Each object gets concise `__repr__()` plus `_repr_html_()` that uses only loaded fields. Centralize HTML escaping and bounded rendering in a private helper inside `wf_client`; do not add a public templating interface. Representations must never call `await`, access the port, or show an unbounded trace/output. + +- [ ] **Step 5: Update live documentation** + +Add `wf_client` to the package maps and document this lifecycle exactly: + +```python +app = App.from_http_jsonrpc("http://localhost:8765/rpc") +capability = await app.capability("wf.std.constant") +graph = app.new_workflow( + "example", + input_schema=InputModel, + state_schema=StateModel, + output_schema=OutputModel, +) +step = graph.use(capability) +graph.set_entry_point(step) +validation = await graph.validate() +validation.raise_for_errors() +artifact = await graph.save(version=1) +run = await artifact.run({}) +``` + +State explicitly that drafts are not part of `wf_client` but remain registered in the server until the separate draft opt-out plan is executed. Update `docs/current_roadmap.md` to mark the Python-client slice complete only after every verification step below passes. + +- [ ] **Step 6: Run focused and cross-layer verification** + +```powershell +uv run pytest tests/wf_client tests/authoring/test_builder.py tests/authoring/test_subgraph.py tests/wf_api/test_artifact_api.py tests/wf_transport_rpc_http/test_client.py tests/wf_transport_rpc_http/test_app.py tests/wf_transport_rpc_http/test_openrpc_contract.py tests/wf_contract_manifest/test_generate.py tests/wf_contract_manifest/test_committed_manifest.py -q +uv run ruff check src/wf_client src/wf_authoring src/wf_api src/wf_transport_rpc_http tests/wf_client tests/authoring/test_builder.py tests/wf_api/test_artifact_api.py +uv run ruff format --check src/wf_client src/wf_authoring src/wf_api src/wf_transport_rpc_http tests/wf_client tests/authoring/test_builder.py tests/wf_api/test_artifact_api.py +uv run basedpyright --level error src/wf_client src/wf_authoring src/wf_api src/wf_transport_rpc_http +uv run python -m wf_contract_manifest check +pnpm --dir web --filter @lda/workflow-rpc contract:check +pnpm --dir web --filter @lda/workflow-rpc test +git diff --check +``` + +Expected: every command passes with no manifest drift, formatting changes, type errors, or whitespace errors. + +- [ ] **Step 7: Run the full Python regression suite** + +```powershell +uv run pytest -q +``` + +Expected: the full suite passes. If environment-backed tests require local configuration, run `uv run --env-file .env pytest -q` and report exact external failures separately from deterministic test failures. + +- [ ] **Step 8: Commit the completed package and docs** + +```powershell +git add src/wf_client tests/wf_client docs/project_map.md docs/source_architecture.md docs/wf_api_architecture.md docs/current_roadmap.md CONTEXT.md +git commit -m "feat: deliver Python workflow client" +``` + +## Follow-Up Plan Boundary + +After this plan is complete, write and execute a separate focused plan for +draft opt-out. That plan must change all three current initialization seams: + +1. `WorkflowApi` unconditionally constructs `WorkflowDraftApi` and + `WorkflowDraftAuthoringApi`. +2. durable context validation currently requires `draft_workspace_store`. +3. `create_rpc_app()` unconditionally registers draft JSON-RPC methods. + +The follow-up target is `drafts=False` by default with an explicit opt-in, +while artifact, deployment, and run durability remain available without a +draft store. Do not mix that server-composition change into `wf_client` tasks.