wire ts in the builder

This commit is contained in:
lda
2026-05-25 02:38:04 +07:00 Verified
parent d3df4754ce
commit 16d5204fbc
4 changed files with 77 additions and 5 deletions
+3
View File
@@ -256,6 +256,9 @@ Today there are two different authoring paths:
`Workflow`: child input schema, output schema, and workflow outcomes are
copied into the boundary. Runtime execution still raises until native
subgraph scopes are implemented.
- `WorkflowBuilder.subgraph(...)` is the builder-facing version of
`subgraph_ref`: it appends the native boundary step and returns it as a
`StepRef` for `connect()` / `set_entry_point()`.
In the future, a compiled workflow or subgraph can also be exposed as a reusable
`NodeSpec`, likely by treating workflow input and output schemas as the node's
@@ -306,7 +306,7 @@ composition.
Current helper:
```python
child = subgraph_ref(
child = parent.subgraph(
id="run_child",
workflow=child_builder.compile(),
input=[input_from(state_path("request"), "request")],
@@ -314,10 +314,12 @@ child = subgraph_ref(
)
```
This copies the compiled child workflow contract into a core `SubgraphNode` but
does not make the child executable yet. `workflow` is still a string reference
inside the core model; higher layers need a structural workflow reference before
saved/deployed workflow dependencies become stable.
This copies the compiled child workflow contract into a core `SubgraphNode`,
appends it to the builder, and returns the step for normal routing. It does not
make the child executable yet. `workflow` is still a string reference inside the
core model; higher layers need a structural workflow reference before
saved/deployed workflow dependencies become stable. The lower-level
`subgraph_ref(...)` helper exists for code that wants only the core step object.
Possible API:
+28
View File
@@ -17,6 +17,7 @@ from wf_core import (
NodeUse,
SchemaRef,
StateSchema,
SubgraphNode,
Workflow,
RunState,
execute_workflow,
@@ -39,6 +40,7 @@ from ..nodes.callables import SyncRegistryHandler
from ..nodes.registry import build_registry
from ..reducers import ReducerCatalog
from ..schemas import SchemaLike, StateSchemaLike, schema_ref_from, state_schema_from
from ..subgraph import subgraph_ref
from ..nodes import NodeSpec
from .ids import next_step_id, slug_id
from .mapping import (
@@ -355,6 +357,32 @@ class WorkflowBuilder:
self.nodes.append(node)
return node
def subgraph(
self,
*,
workflow: Workflow,
id: str | None = None,
input: Sequence[InputBindingArg] | None = None,
output: Sequence[OutputBindingArg] | None = None,
workflow_ref: str | None = None,
desc: str | None = None,
) -> SubgraphNode:
"""Add a native subgraph boundary using a compiled child workflow contract.
This only authors the graph boundary. Runtime execution still raises
until wf_core grows child workflow scope/frame execution.
"""
node = subgraph_ref(
id=id or self._next_step_id(slug_id(workflow_ref or workflow.name)),
workflow=workflow,
input=normalize_input_bindings(input),
output=normalize_output_bindings(output),
workflow_ref=workflow_ref,
desc=desc,
)
self.nodes.append(node)
return node
def _next_step_id(self, base: str) -> str:
"""Return a stable unused step id based on the requested base name."""
return next_step_id(base, self.nodes)
+39
View File
@@ -200,3 +200,42 @@ def test_subgraph_ref_contract_validates_in_parent_workflow() -> None:
)
assert parent.validate_structure().errors == []
def test_workflow_builder_subgraph_adds_native_subgraph_node() -> None:
class ParentInput(BaseModel):
folder_id: str
should_email: bool
class ParentState(BaseModel):
summary: str
class ParentOutput(BaseModel):
summary: str
child = build_demo_workflow()
parent = WorkflowBuilder(
name="native_parent",
input_schema=ParentInput,
state_schema=ParentState,
output_schema=ParentOutput,
)
step = parent.subgraph(
workflow=child,
id="run_child",
input=[
input_from(input_path("folder_id"), "folder_id"),
input_from(input_path("should_email"), "should_email"),
],
output=[output_to("summary", state_path("summary"))],
)
parent.set_entry_point(step)
parent.connect(step, "ok", END)
workflow = parent.compile()
assert workflow.nodes[0].type == "subgraph"
assert workflow.nodes[0].id == "run_child"
assert workflow.nodes[0].outcomes == child.outcomes
assert workflow.validate_structure().errors == []