lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
00b51d2d960f6dda2e4e441bba1f9fc9fc69de88
lda-wf/docs/superpowers/plans/2026-08-30-python-workflow-client.md
T

50 KiB
Raw Blame History

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:

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:

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:

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:

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:

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:

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:

@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:

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 NodeSpecs, 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:

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:

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
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:

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:

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
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:

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:

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:

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
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:

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:

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
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:

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:

@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:

@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:

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
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
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:

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:

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:

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
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:

@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:

@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:

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
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
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:

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:

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
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__:

@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:

@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
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
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:

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:

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
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
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:

@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 <artifact_id>.v<version>.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
@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 TypedDicts 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
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
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:

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:

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
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:

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
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
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
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.

Reference in New Issue View Git Blame Copy Permalink