diff --git a/docs/project_map.md b/docs/project_map.md index 00d26321..ff2bc72d 100644 --- a/docs/project_map.md +++ b/docs/project_map.md @@ -134,20 +134,30 @@ arguments may be JSON Schema dictionaries or the application's schema model values: ```python +from pydantic import BaseModel + from wf_client import App from wf_authoring import input_from, input_value, output_to, state_path + +class Input(BaseModel): + request_id: str + + +class State(BaseModel): + value: str | None = None + + +class Output(BaseModel): + value: str + app = App.from_http_jsonrpc("http://localhost:8765/rpc") capability = await app.capability("wf.std.constant") graph = app.new_workflow( "example", - input_schema={"type": "object", "properties": {}}, - state_schema={"type": "object", "properties": {"value": {"type": "string"}}}, - output_schema={ - "type": "object", - "properties": {"value": {"type": "string"}}, - "required": ["value"], - }, + input_schema=Input, + state_schema=State, + output_schema=Output, ) step = graph.use( capability, @@ -162,12 +172,16 @@ graph.set_output([input_from(state_path("value"), "value")]) validation = await graph.validate() validation.raise_for_errors() artifact = await graph.save(version=1) -run = await artifact.run({}) +run = await artifact.run({"request_id": "request-1"}) ``` The graph is a local, mutable builder. `validate()` checks its structure locally and then asks the server to validate the serialized plan. `save()` persists an immutable artifact version; it does not deploy or execute the graph. +Pydantic models are normalized into canonical schemas. During authoring, +`set_contract()` can atomically replace selected input, state, output, or +outcome contracts; it preserves existing bindings so validation can expose any +paths made invalid by the replacement. `artifact.run()` selects or creates a deployment, validates its source bindings, and starts a durable run. The returned run is a loaded snapshot; call `refresh()`, `resume()`, or bounded `trace(start=..., limit=...)` when more diff --git a/docs/wf_api_architecture.md b/docs/wf_api_architecture.md index 0c18acc1..c6f21a7c 100644 --- a/docs/wf_api_architecture.md +++ b/docs/wf_api_architecture.md @@ -114,20 +114,30 @@ Hypothetically, an application that wants to turn a discovered capability into a durable run would use the following complete flow: ```python +from pydantic import BaseModel + from wf_client import App from wf_authoring import input_from, input_value, output_to, state_path + +class Input(BaseModel): + request_id: str + + +class State(BaseModel): + value: str | None = None + + +class Output(BaseModel): + value: str + app = App.from_http_jsonrpc("http://localhost:8765/rpc") capability = await app.capability("wf.std.constant") graph = app.new_workflow( "example", - input_schema={"type": "object", "properties": {}}, - state_schema={"type": "object", "properties": {"value": {"type": "string"}}}, - output_schema={ - "type": "object", - "properties": {"value": {"type": "string"}}, - "required": ["value"], - }, + input_schema=Input, + state_schema=State, + output_schema=Output, ) step = graph.use( capability, @@ -142,9 +152,16 @@ graph.set_output([input_from(state_path("value"), "value")]) validation = await graph.validate() validation.raise_for_errors() artifact = await graph.save(version=1) -run = await artifact.run({}) +run = await artifact.run({"request_id": "request-1"}) ``` +Pydantic models, typed mappings, and raw JSON Schema are accepted, but Python +applications should normally keep their contracts as Python types. If a +contract changes while the graph is being authored, call +`graph.set_contract(state_schema=..., output_schema=..., outcomes=...)`. +Supplied fields replace their whole contract atomically; omitted fields and +existing graph bindings remain, so validate after the replacement. + The graph is an in-process builder. Validation is local structural checking plus a server plan check. Saving creates an immutable, versioned artifact; it does not execute anything. A deployment is the server's runnable configuration diff --git a/skills/wf-python/SKILL.md b/skills/wf-python/SKILL.md new file mode 100644 index 00000000..8d2b47a0 --- /dev/null +++ b/skills/wf-python/SKILL.md @@ -0,0 +1,211 @@ +--- +name: wf-python +description: Use when writing, reviewing, or debugging Python code that uses wf_client App, RemoteCapability, EditableWorkflow, WorkflowArtifact, Deployment, Run, or the Python workflow lifecycle. Prefer this skill for Python workflow authoring, typed contracts, artifact editing, deployment selection, interrupts, and traces; use wf-cli instead for shell-first work. +--- + +# wf Python Client + +Use `wf_client` as the public Python object API over the workflow service. Keep +JSON-RPC payloads, codecs, stores, and draft workspaces below this boundary. + +## Start With The Right Object + +```python +from wf_client import App + +app = App.from_http_jsonrpc("http://localhost:8765/rpc") +capability = await app.capability("app.default.thing") +``` + +Choose the object that matches the operation: + +- `RemoteCapability`: inspect or directly probe one capability. +- `EditableWorkflow`: author or revise a graph locally. +- `WorkflowArtifact`: inspect one immutable saved version. +- `Deployment`: bind one artifact version to concrete sources and validate it. +- `Run`: inspect, refresh, resume, or trace one durable execution. + +Do not collapse artifact saving, deployment configuration, and execution into +one invented "publish" operation. + +## Prefer Python Contract Types + +Declare workflow contracts with Pydantic models or supported Python types. The +builder converts them to canonical schemas. + +```python +from pydantic import BaseModel + +class Input(BaseModel): + topic: str + +class State(BaseModel): + result: str | None = None + +class Output(BaseModel): + result: str + +graph = app.new_workflow( + "report", + input_schema=Input, + state_schema=State, + output_schema=Output, +) +``` + +Use raw schema dictionaries only when the caller already owns a JSON Schema or +needs a construct the Python type system cannot express. + +### Replace A Contract During Authoring + +Use `set_contract()` instead of assigning normalized builder fields directly: + +```python +graph.set_contract( + state_schema=ExpandedState, + output_schema=FinalOutput, + outcomes=("completed", "rejected"), +) +``` + +Supplied fields replace the whole corresponding contract. Omitted fields remain +unchanged. Replacement is atomic: normalization completes before the builder is +mutated. Existing bindings remain, so run local validation after replacement to +find paths invalidated by the new contract. + +## Know What The Builder Infers + +`graph.use(remote_capability)` registers the capability's node contract. When +bindings are omitted, it can auto-bind matching capability fields against the +workflow's already-declared input and state fields. + +It does not invent the workflow's public contract or topology. Keep these +explicit: + +- Input, state, and output contract declarations. +- Workflow outcomes when they differ from the default `("ok",)`. +- Entry point. +- Routes and terminal nodes. +- Final workflow-output projection. +- Deployment bindings and selection when the environment is ambiguous. + +Prefer canonical `input=[...]` and `output=[...]` binding lists when the +dataflow matters. Auto-binding is useful for exact field-name matches, not a +substitute for deciding semantics. + +## Author, Validate, Save + +```python +from wf_authoring import input_from, input_value, output_to, state_path + +step = graph.use( + capability, + id="constant", + input=[input_value("value", "hello")], + output=[output_to("value", state_path("result"))], +) +end = graph.end("ok", id="end_ok") +graph.set_entry_point(step) +graph.connect(step, "ok", end) +graph.set_output([input_from(state_path("result"), "result")]) + +local = graph.validate_local() +local.raise_for_errors() + +validation = await graph.validate() +validation.raise_for_errors() + +artifact = await graph.save(version=1, title="Report") +``` + +`validate_local()` performs no remote I/O. `validate()` adds server plan and +dependency validation. `save()` validates, saves, then inspects the exact saved +version before returning its immutable snapshot. + +## Edit An Existing Artifact + +```python +artifact = await app.workflow("report", version=3) +graph = artifact.edit() +# Equivalent: graph = await app.edit_workflow("report", version=3) +``` + +Use ordinary builder methods on `graph`. Do not reconstruct an existing graph +from scratch: the seeded editor preserves schemas, bindings, routes, outcomes, +subgraphs, and retained dependency contracts. Save edits as a new immutable +version unless the caller explicitly requests otherwise. + +Never guess step IDs or state paths. Inspect `artifact.inspect()` or the +editable graph before choosing mutation targets. + +## Deploy And Run + +```python +deployment = await artifact.deploy( + "report.production", + bindings={"logical.documents": "production.documents"}, +) +readiness = await deployment.validate() +if not readiness.runnable: + for diagnostic in readiness.diagnostics: + print(diagnostic.code, diagnostic.message) + raise RuntimeError("deployment is not runnable") + +run = await deployment.run({"topic": "workflow systems"}) +``` + +`artifact.run(input)` is convenient only when deployment selection is +unambiguous. Handle `DeploymentRequired` rather than guessing an account or +source binding. + +Run snapshots are immutable: + +```python +run = await run.refresh() +if run.status == "interrupted" and run.interrupt is not None: + run = await run.resume({"approved": True}) + +trace = await run.trace(start=0, limit=25) +for frame in trace.frames: + print(frame) +``` + +Trace reads must stay bounded (`limit` is 1 through 100). + +## Handle Public Errors + +Catch `WorkflowClientError` or its public subclasses. Do not import transport +exceptions or decode wire payloads yourself. + +```python +from wf_client import ( + DeploymentRequired, + ProtocolError, + TransportError, + WorkflowClientError, +) + +try: + run = await artifact.run(payload) +except DeploymentRequired as error: + print(error.candidate_deployment_ids, error.unresolved_logical_sources) +except TransportError as error: + print("workflow service unavailable", error) +except ProtocolError as error: + print(error.code, error.message, error.data) +except WorkflowClientError as error: + print(type(error).__name__, error) +``` + +## Boundaries + +- `wf_client` intentionally has no draft workspace API. Use server/admin or + console surfaces only when the task is genuinely about drafts. +- Rich representations are inert debugging aids, not a secrecy boundary. + Schema-level sensitivity metadata is the appropriate future source of truth; + do not rely on repr redaction to protect credentials. +- Use the public objects before inspecting `wf_api`, RPC clients, codecs, or + stores. Drop below the client boundary only when implementing the client. + +Read [references/python-lifecycle.md](references/python-lifecycle.md) when a +complete typed lifecycle or an editing/debugging recipe is needed. diff --git a/skills/wf-python/evals/evals.json b/skills/wf-python/evals/evals.json new file mode 100644 index 00000000..0a35dd72 --- /dev/null +++ b/skills/wf-python/evals/evals.json @@ -0,0 +1,41 @@ +{ + "skill_name": "wf-python", + "evals": [ + { + "id": 1, + "prompt": "Write Python code using the new wf_client API to connect to http://localhost:8765/rpc, discover wf.std.constant, build a typed workflow with Pydantic input/state/output models, then halfway through authoring replace the state/output contract with revised Pydantic models, validate, save artifact version 1, and run it. Explain what WorkflowBuilder infers and what must stay explicit.", + "expected_output": "A public Python lifecycle using Pydantic contracts and set_contract(), with explicit topology/output and correct validation/save/run behavior.", + "files": [], + "assertions": [ + "Uses Pydantic models for workflow contracts instead of hand-written JSON Schema", + "Uses graph.set_contract() for mid-authoring replacement", + "Keeps entry point, routes, terminal, and final workflow output explicit", + "Uses only public wf_client and wf_authoring APIs" + ] + }, + { + "id": 2, + "prompt": "I have immutable workflow artifact report.v3 on the remote service. Show me Python code using wf_client to load it, edit it with the normal graph builder methods, add a discovered app.default.summarize capability, validate locally and remotely, save version 4, deploy it with a source binding, and start a durable run. Preserve everything I did not edit.", + "expected_output": "A lossless artifact.edit()/edit_workflow flow that inspects rather than guesses graph identifiers and keeps artifact, deployment, and run separate.", + "files": [], + "assertions": [ + "Loads exact artifact version 3 and edits the seeded builder rather than rebuilding", + "Warns that step IDs and state paths must be inspected rather than guessed", + "Validates locally and remotely before saving version 4", + "Creates and validates a deployment before starting a durable run" + ] + }, + { + "id": 3, + "prompt": "Using only the public Python workflow client, show how to diagnose why artifact invoice.v2 cannot run, select or create a deployment with bindings, validate it, start a run, handle an interrupt, refresh it, and read only 25 trace frames. Include public error handling. Do not use draft workspaces or raw JSON-RPC.", + "expected_output": "Public object-based diagnosis with DeploymentRequired, explicit bindings, validation, immutable refresh/resume, and a bounded TracePage.", + "files": [], + "assertions": [ + "Uses no draft workspace, raw JSON-RPC, codec, or store API", + "Handles DeploymentRequired without guessing bindings", + "Uses run.resume(), run.refresh(), and run.trace(start=0, limit=25)", + "Uses trace.frames and public WorkflowClientError subclasses accurately" + ] + } + ] +} diff --git a/skills/wf-python/references/python-lifecycle.md b/skills/wf-python/references/python-lifecycle.md new file mode 100644 index 00000000..3f05bfd6 --- /dev/null +++ b/skills/wf-python/references/python-lifecycle.md @@ -0,0 +1,130 @@ +# Python Workflow Lifecycle + +This reference contains complete patterns for the public `wf_client` API. + +## Typed Authoring With Contract Replacement + +```python +from pydantic import BaseModel + +from wf_authoring import input_from, input_value, output_to, state_path +from wf_client import App + + +class InitialInput(BaseModel): + request_id: str + + +class InitialState(BaseModel): + value: str | None = None + + +class InitialOutput(BaseModel): + value: str + + +class ExpandedState(BaseModel): + value: str | None = None + source: str | None = None + + +class FinalOutput(BaseModel): + value: str + + +app = App.from_http_jsonrpc("http://localhost:8765/rpc") +constant = await app.capability("wf.std.constant") + +graph = app.new_workflow( + "typed_example", + input_schema=InitialInput, + state_schema=InitialState, + output_schema=InitialOutput, +) +step = graph.use( + constant, + id="constant", + input=[input_value("value", "hello")], + output=[output_to("value", state_path("value"))], +) + +graph.set_contract(state_schema=ExpandedState, output_schema=FinalOutput) + +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")]) + +graph.validate_local().raise_for_errors() +validation = await graph.validate() +validation.raise_for_errors() +artifact = await graph.save(version=1, title="Typed example") +run = await artifact.run({"request_id": "request-1"}) +``` + +## Lossless Editing + +```python +artifact = await app.workflow("report", version=3) +workflow = artifact.inspect() +print([node.id for node in workflow.nodes]) + +summarize = await app.capability("app.default.summarize") +graph = artifact.edit() + +summary = graph.use( + summarize, + id="summarize", + input=[input_from(state_path("draft"), "text")], + output=[output_to("summary", state_path("summary"))], +) +graph.set_route("draft_report", "ok", summary) +graph.connect(summary, "ok", "end_ok") + +graph.validate_local().raise_for_errors() +(await graph.validate()).raise_for_errors() +report_v4 = await graph.save(version=4) +``` + +The step IDs and paths above are examples. Inspect the loaded artifact and use +its real contract; do not assume those names exist. + +## Deployment Diagnosis And Durable Runs + +```python +from wf_client import DeploymentRequired, WorkflowClientError + +artifact = await app.workflow("invoice", version=2) + +try: + run = await artifact.run({"invoice_id": "INV-1001"}) +except DeploymentRequired as error: + print("candidates", error.candidate_deployment_ids) + print("unresolved", error.unresolved_logical_sources) + for diagnostic in error.diagnostics: + print(diagnostic.code, diagnostic.message) + +deployment = await artifact.deploy( + "invoice.production", + bindings={"billing": "production.billing"}, +) +readiness = await deployment.validate() +if not readiness.runnable: + for diagnostic in readiness.diagnostics: + print(diagnostic.code, diagnostic.message) + raise RuntimeError("invoice.production is not runnable") + +run = await deployment.run({"invoice_id": "INV-1001"}) + +if run.status == "interrupted" and run.interrupt is not None: + run = await run.resume({"approved": True}) + +run = await run.refresh() +trace = await run.trace(start=0, limit=25) +for frame in trace.frames: + print(frame) +``` + +Catch the specific public errors useful to the application and retain a final +`WorkflowClientError` fallback. Unknown server errors remain inspectable +`ProtocolError` values with `code`, `message`, and `data`. diff --git a/src/wf_authoring/builder/core.py b/src/wf_authoring/builder/core.py index 8b4a0a8f..d1198210 100644 --- a/src/wf_authoring/builder/core.py +++ b/src/wf_authoring/builder/core.py @@ -483,6 +483,44 @@ class WorkflowBuilder: list[InputBinding], normalize_step_input_bindings(bindings) ) + def set_contract( + self, + *, + input_schema: SchemaLike | None = None, + state_schema: StateSchemaLike | None = None, + output_schema: SchemaLike | None = None, + outcomes: Sequence[str] | None = None, + ) -> None: + """Replace selected workflow contract fields atomically. + + Omitted fields retain their current normalized values. Supplied Python + models, typed mappings, and raw schema dictionaries pass through the + same normalization used by the constructor. Existing graph bindings + are left intact so structural validation can report anything the new + contract invalidates. + """ + next_input = ( + self.input_schema + if input_schema is None + else schema_ref_from(input_schema) + ) + next_state = ( + self.state_schema + if state_schema is None + else state_schema_from(state_schema) + ) + next_output = ( + self.output_schema + if output_schema is None + else schema_ref_from(output_schema) + ) + next_outcomes = self.outcomes if outcomes is None else tuple(outcomes) + + self.input_schema = next_input + self.state_schema = next_state + self.output_schema = next_output + self.outcomes = next_outcomes + def set_route(self, source: StepRef, outcome: str, target: StepRef) -> None: """Replace the unique route for one source/outcome pair.""" source_id = step_id(source) diff --git a/tests/authoring/test_schemas.py b/tests/authoring/test_schemas.py index ea717ae2..86613519 100644 --- a/tests/authoring/test_schemas.py +++ b/tests/authoring/test_schemas.py @@ -1,7 +1,9 @@ from __future__ import annotations +from copy import deepcopy from typing import Annotated +import pytest from pydantic import BaseModel, Field from tests.authoring.helpers import ( @@ -25,6 +27,18 @@ class DotAliasState(BaseModel): ] +class RevisedInput(BaseModel): + query: str + + +class RevisedState(BaseModel): + result: str | None = None + + +class RevisedOutput(BaseModel): + result: str + + def test_builder_accepts_basemodel_classes_for_workflow_schemas() -> None: builder = WorkflowBuilder( name="model_schema_demo", @@ -45,6 +59,50 @@ def test_builder_accepts_basemodel_classes_for_workflow_schemas() -> None: assert fields["tags"].type == "array" +def test_set_contract_replaces_selected_fields_from_python_models() -> None: + builder = WorkflowBuilder( + name="editable_contract", + input_schema=WorkflowInput, + state_schema=WorkflowState, + output_schema=WorkflowOutput, + outcomes=("ok",), + start="start", + ) + original_input = deepcopy(builder.input_schema) + + builder.set_contract( + state_schema=RevisedState, + output_schema=RevisedOutput, + outcomes=("completed", "rejected"), + ) + workflow = builder.compile() + + assert workflow.input_schema == original_input + assert set(workflow.state_schema.field_map()) == {"result"} + assert workflow.output_schema.properties["result"]["type"] == "string" + assert workflow.outcomes == ["completed", "rejected"] + + +def test_set_contract_replacement_is_atomic_when_normalization_fails() -> None: + builder = WorkflowBuilder( + name="atomic_contract", + input_schema=WorkflowInput, + state_schema=WorkflowState, + output_schema=WorkflowOutput, + outcomes=("ok",), + start="start", + ) + original = builder.compile() + + with pytest.raises(ValueError): + builder.set_contract( + input_schema=RevisedInput, + output_schema={"type": "definitely-not-a-json-schema-type"}, + ) + + assert builder.compile() == original + + def test_builder_accepts_typeddict_for_json_schema_refs() -> None: builder = WorkflowBuilder( name="typed_dict_schema_demo",