lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

wf-core reorg 2 split the Models

Browse Source
This commit is contained in:
lda
2026-05-08 21:39:49 +07:00 Verified
parent 4c860659e4
commit d37756246b
23 changed files with 321 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/wf_core_architecture.md
+7 -9
View File
@@ -11,7 +11,8 @@ and user-facing control belong in `wf_mcp`.
| Package / module | Responsibility |
| --- | --- |
| `wf_core.model` | Pydantic workflow schema: node definitions, node uses, control-flow nodes, edges, schemas, and node results. |
| `wf_core.models` | Pydantic workflow schema package: schemas, condition expressions, executable steps, workflow graph, and node results. |
| `wf_core.model` | Compatibility facade for older imports of core model objects. |
| `wf_core.run_state` | Serializable execution state: run status, frames, trace entries, interrupt requests, and runtime context. |
| `wf_core.runtime` | Public execution interface: execute, resume, and step in sync or async mode. |
| `wf_core.runtime.ops` | Executor-only operations used behind `wf_core.runtime`: node execution, state writes, frame movement, foreach, interrupts, indexes, and schema checks. |
@@ -20,9 +21,9 @@ and user-facing control belong in `wf_mcp`.
| `wf_core.paths` | Graph path parsing, reading, existence checks, and nested state writes. |
| `wf_core.tokens` | Importable graph boundary tokens: `START` and `END`. |
Root modules such as `wf_core.node_exec`, `wf_core.state_ops`, and
`wf_core.validate` are compatibility shims. New internal imports should prefer
the concern package directly.
Root modules such as `wf_core.model`, `wf_core.node_exec`, `wf_core.state_ops`,
and `wf_core.validate` are compatibility shims. New internal imports should
prefer the concern package directly.
## Runtime Flow
@@ -54,7 +55,8 @@ raising at the first failure.
## Dependency Rules
- `wf_core` must not import `wf_authoring` or `wf_mcp`.
- `wf_core.model` and `wf_core.run_state` should stay mostly data-only.
- `wf_core.models` and `wf_core.run_state` should stay mostly data-only.
- `wf_core.model` should stay a thin import facade.
- `wf_core.runtime` may import `runtime.ops`, but callers should not need to.
- `wf_core.runtime.ops` may use model, run state, paths, conditions, and errors.
- `wf_core.validation` may inspect model and path rules, but should not execute
@@ -63,9 +65,6 @@ raising at the first failure.
## What This Cleanup Does Not Solve Yet
- `wf_core.model` is still a single dense schema file. It is coherent today, but
if control-flow schemas grow, split it into `model/schemas.py`,
`model/steps.py`, and `model/workflow.py`.
- `demo_workflow.py` is still large because it is a fixture/demo, not core
runtime. If it becomes a permanent example suite, move it out of `wf_core`.
- Foreach is still serial-only. Parallel foreach needs an explicit scheduling
@@ -76,4 +75,3 @@ raising at the first failure.
- Runtime errors are still ordinary exceptions plus failed run status. A richer
error payload can be added later, but should be designed as part of trace/run
state rather than scattered exceptions.
src/wf_core/flow_ops.py
-1
View File
@@ -1,4 +1,3 @@
"""Compatibility shim for runtime flow operations."""
from wf_core.runtime.ops.flow import * # noqa: F403
src/wf_core/foreach_ops.py
-1
View File
@@ -1,4 +1,3 @@
"""Compatibility shim for foreach runtime operations."""
from wf_core.runtime.ops.foreach import * # noqa: F403
src/wf_core/frame_ops.py
-1
View File
@@ -1,4 +1,3 @@
"""Compatibility shim for frame runtime operations."""
from wf_core.runtime.ops.frames import * # noqa: F403
src/wf_core/interrupt_ops.py
-1
View File
@@ -1,4 +1,3 @@
"""Compatibility shim for interrupt runtime operations."""
from wf_core.runtime.ops.interrupts import * # noqa: F403
src/wf_core/model.py
+46 -153
View File
@@ -1,161 +1,54 @@
from __future__ import annotations
"""Compatibility facade for core workflow model objects."""
from typing import TYPE_CHECKING, Annotated, Any, Literal, cast
from wf_core.models import (
BinaryCondition,
Condition,
ConditionNode,
Edge,
ExistsCondition,
ForeachNode,
InterruptNode,
JoinNode,
LiteralOperand,
NodeDef,
NodeResult,
NodeUse,
NotCondition,
Operand,
PathOperand,
SchemaRef,
StateField,
StateSchema,
Step,
VariadicCondition,
Workflow,
)
from pydantic import BaseModel, ConfigDict, Field
if TYPE_CHECKING:
from wf_core.validation.issues import ValidationReport
class SchemaRef(BaseModel):
model_config = ConfigDict(extra="allow")
title: str | None = None
type: str | None = None
properties: dict[str, Any] = Field(default_factory=dict)
required: list[str] = Field(default_factory=list)
class StateField(BaseModel):
type: str
merge_strategy: Literal["replace", "append", "merge_object"] = "replace"
trace: bool = True
default: Any = None
class StateSchema(BaseModel):
model_config = ConfigDict(extra="allow")
fields: dict[str, StateField] = Field(default_factory=dict)
class NodeDef(BaseModel):
name: str
input_schema: SchemaRef
output_schema: SchemaRef
outcomes: list[str] = Field(min_length=1)
retry: int | None = Field(default=None, ge=0)
timeout_seconds: int | None = Field(default=None, gt=0)
class NodeUse(BaseModel):
id: str
type: Literal["node"]
node: str
desc: str | None = None
in_map: dict[str, str] = Field(default_factory=dict)
out_map: dict[str, str] = Field(default_factory=dict)
retry: int | None = Field(default=None, ge=0)
timeout_seconds: int | None = Field(default=None, gt=0)
class PathOperand(BaseModel):
path: str
class LiteralOperand(BaseModel):
value: Any
Operand = Annotated[PathOperand | LiteralOperand, Field(discriminator=None)]
class ExistsCondition(BaseModel):
op: Literal["exists"]
path: str
class NotCondition(BaseModel):
op: Literal["not"]
arg: "Condition"
class VariadicCondition(BaseModel):
op: Literal["and", "or"]
args: list["Condition"] = Field(min_length=1)
class BinaryCondition(BaseModel):
op: Literal["eq", "ne", "gt", "lt"]
left: PathOperand | LiteralOperand
right: PathOperand | LiteralOperand
Condition = Annotated[
ExistsCondition | NotCondition | VariadicCondition | BinaryCondition,
Field(discriminator="op"),
__all__ = [
"BinaryCondition",
"Condition",
"ConditionNode",
"Edge",
"ExistsCondition",
"ForeachNode",
"InterruptNode",
"JoinNode",
"LiteralOperand",
"NodeDef",
"NodeResult",
"NodeUse",
"NotCondition",
"Operand",
"PathOperand",
"SchemaRef",
"StateField",
"StateSchema",
"Step",
"VariadicCondition",
"Workflow",
]
class ConditionNode(BaseModel):
id: str
type: Literal["condition"]
check: Condition
class ForeachNode(BaseModel):
model_config = ConfigDict(populate_by_name=True)
id: str
type: Literal["foreach"]
over: str
as_: str = Field(alias="as")
mode: Literal["serial", "parallel"] = "serial"
on_item_error: Literal["fail", "collect", "skip"] = "fail"
class JoinNode(BaseModel):
id: str
type: Literal["join"]
class InterruptNode(BaseModel):
id: str
type: Literal["interrupt"]
kind: str
request_map: dict[str, str] = Field(default_factory=dict)
out_map: dict[str, str] = Field(default_factory=dict)
outcomes: list[str] = Field(default_factory=lambda: ["submitted"])
Step = Annotated[
NodeUse | ConditionNode | ForeachNode | JoinNode | InterruptNode,
Field(discriminator="type"),
]
class Edge(BaseModel):
from_: str = Field(alias="from")
outcome: str
to: str
class Workflow(BaseModel):
name: str
input_schema: SchemaRef
state_schema: StateSchema
output_schema: SchemaRef
node_defs: list[NodeDef] = Field(default_factory=list)
start: str
nodes: list[Step]
edges: list[Edge]
def validate_structure(self) -> "ValidationReport":
from importlib import import_module
validation = import_module("wf_core.validation.core")
return cast("ValidationReport", validation.validate_workflow(self))
class NodeResult(BaseModel):
model_config = ConfigDict(extra="allow")
outcome: str
output: dict[str, Any] = Field(default_factory=dict)
meta: dict[str, Any] = Field(default_factory=dict)
if __name__ == "__main__":
import json
src/wf_core/models/__init__.py
+45
View File
@@ -0,0 +1,45 @@
from wf_core.models.conditions import (
BinaryCondition,
Condition,
ExistsCondition,
LiteralOperand,
NotCondition,
Operand,
PathOperand,
VariadicCondition,
)
from wf_core.models.results import NodeResult
from wf_core.models.schemas import NodeDef, SchemaRef, StateField, StateSchema
from wf_core.models.steps import (
ConditionNode,
ForeachNode,
InterruptNode,
JoinNode,
NodeUse,
Step,
)
from wf_core.models.workflow import Edge, Workflow
__all__ = [
"BinaryCondition",
"Condition",
"ConditionNode",
"Edge",
"ExistsCondition",
"ForeachNode",
"InterruptNode",
"JoinNode",
"LiteralOperand",
"NodeDef",
"NodeResult",
"NodeUse",
"NotCondition",
"Operand",
"PathOperand",
"SchemaRef",
"StateField",
"StateSchema",
"Step",
"VariadicCondition",
"Workflow",
]
src/wf_core/models/conditions.py
+57
View File
@@ -0,0 +1,57 @@
from __future__ import annotations
from typing import Annotated, Any, Literal
from pydantic import BaseModel, Field
class PathOperand(BaseModel):
"""Condition operand resolved from a workflow graph path."""
path: str
class LiteralOperand(BaseModel):
"""Condition operand that carries a literal value."""
value: Any
Operand = Annotated[PathOperand | LiteralOperand, Field(discriminator=None)]
"""Condition operand accepted by binary condition expressions."""
class ExistsCondition(BaseModel):
"""Condition that is true when a graph path resolves to a present value."""
op: Literal["exists"]
path: str
class NotCondition(BaseModel):
"""Condition that negates another condition expression."""
op: Literal["not"]
arg: "Condition"
class VariadicCondition(BaseModel):
"""Condition that combines one or more child conditions."""
op: Literal["and", "or"]
args: list["Condition"] = Field(min_length=1)
class BinaryCondition(BaseModel):
"""Condition that compares two operands."""
op: Literal["eq", "ne", "gt", "lt"]
left: PathOperand | LiteralOperand
right: PathOperand | LiteralOperand
Condition = Annotated[
ExistsCondition | NotCondition | VariadicCondition | BinaryCondition,
Field(discriminator="op"),
]
"""Discriminated union of all supported condition expressions."""
src/wf_core/models/results.py
+15
View File
@@ -0,0 +1,15 @@
from __future__ import annotations
from typing import Any
from pydantic import BaseModel, ConfigDict, Field
class NodeResult(BaseModel):
"""Normalized result returned by a node handler after execution."""
model_config = ConfigDict(extra="allow")
outcome: str
output: dict[str, Any] = Field(default_factory=dict)
meta: dict[str, Any] = Field(default_factory=dict)
src/wf_core/models/schemas.py
+44
View File
@@ -0,0 +1,44 @@
from __future__ import annotations
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field
class SchemaRef(BaseModel):
"""JSON-schema-like shape used at workflow boundaries."""
model_config = ConfigDict(extra="allow")
title: str | None = None
type: str | None = None
properties: dict[str, Any] = Field(default_factory=dict)
required: list[str] = Field(default_factory=list)
class StateField(BaseModel):
"""Declared root state field plus its runtime merge behavior."""
type: str
merge_strategy: Literal["replace", "append", "merge_object"] = "replace"
trace: bool = True
default: Any = None
class StateSchema(BaseModel):
"""Workflow state schema keyed by declared root field name."""
model_config = ConfigDict(extra="allow")
fields: dict[str, StateField] = Field(default_factory=dict)
class NodeDef(BaseModel):
"""Reusable node contract referenced by one or more node uses."""
name: str
input_schema: SchemaRef
output_schema: SchemaRef
outcomes: list[str] = Field(min_length=1)
retry: int | None = Field(default=None, ge=0)
timeout_seconds: int | None = Field(default=None, gt=0)
src/wf_core/models/steps.py
+66
View File
@@ -0,0 +1,66 @@
from __future__ import annotations
from typing import Annotated, Literal
from pydantic import BaseModel, ConfigDict, Field
from wf_core.models.conditions import Condition
class NodeUse(BaseModel):
"""Concrete use of a reusable node definition inside a workflow graph."""
id: str
type: Literal["node"]
node: str
desc: str | None = None
in_map: dict[str, str] = Field(default_factory=dict)
out_map: dict[str, str] = Field(default_factory=dict)
retry: int | None = Field(default=None, ge=0)
timeout_seconds: int | None = Field(default=None, gt=0)
class ConditionNode(BaseModel):
"""Control-flow step that routes through `true` or `false` outcomes."""
id: str
type: Literal["condition"]
check: Condition
class ForeachNode(BaseModel):
"""Control-flow step that iterates over an input or state list."""
model_config = ConfigDict(populate_by_name=True)
id: str
type: Literal["foreach"]
over: str
as_: str = Field(alias="as")
mode: Literal["serial", "parallel"] = "serial"
on_item_error: Literal["fail", "collect", "skip"] = "fail"
class JoinNode(BaseModel):
"""Control-flow step that marks a branch or frame as joined."""
id: str
type: Literal["join"]
class InterruptNode(BaseModel):
"""Control-flow step that pauses a run and waits for resume input."""
id: str
type: Literal["interrupt"]
kind: str
request_map: dict[str, str] = Field(default_factory=dict)
out_map: dict[str, str] = Field(default_factory=dict)
outcomes: list[str] = Field(default_factory=lambda: ["submitted"])
Step = Annotated[
NodeUse | ConditionNode | ForeachNode | JoinNode | InterruptNode,
Field(discriminator="type"),
]
"""Discriminated union of all executable workflow graph steps."""
src/wf_core/models/workflow.py
+38
View File
@@ -0,0 +1,38 @@
from __future__ import annotations
from importlib import import_module
from typing import TYPE_CHECKING, cast
from pydantic import BaseModel, Field
from wf_core.models.schemas import NodeDef, SchemaRef, StateSchema
from wf_core.models.steps import Step
if TYPE_CHECKING:
from wf_core.validation.issues import ValidationReport
class Edge(BaseModel):
"""Outcome-specific transition from one workflow step to another."""
from_: str = Field(alias="from")
outcome: str
to: str
class Workflow(BaseModel):
"""Serializable workflow graph consumed by the core runtime."""
name: str
input_schema: SchemaRef
state_schema: StateSchema
output_schema: SchemaRef
node_defs: list[NodeDef] = Field(default_factory=list)
start: str
nodes: list[Step]
edges: list[Edge]
def validate_structure(self) -> "ValidationReport":
"""Return all structural validation issues for this workflow."""
validation = import_module("wf_core.validation.core")
return cast("ValidationReport", validation.validate_workflow(self))
src/wf_core/node_exec.py
-1
View File
@@ -1,4 +1,3 @@
"""Compatibility shim for node execution operations."""
from wf_core.runtime.ops.nodes import * # noqa: F403
src/wf_core/run_factory.py
-1
View File
@@ -1,4 +1,3 @@
"""Compatibility shim for run-state construction."""
from wf_core.runtime.ops.runs import * # noqa: F403
src/wf_core/runtime/ops/__init__.py
-1
View File
@@ -4,4 +4,3 @@ Root modules such as `wf_core.node_exec` remain as compatibility shims. New
runtime internals should import from this package so the execution seam stays
easy to navigate.
"""
src/wf_core/schema_tools.py
-1
View File
@@ -1,4 +1,3 @@
"""Compatibility shim for runtime schema validation helpers."""
from wf_core.runtime.ops.schemas import * # noqa: F403
src/wf_core/state_ops.py
-1
View File
@@ -1,4 +1,3 @@
"""Compatibility shim for state mutation operations."""
from wf_core.runtime.ops.state import * # noqa: F403
src/wf_core/step_handlers.py
-1
View File
@@ -1,4 +1,3 @@
"""Compatibility shim for non-node step handlers."""
from wf_core.runtime.ops.handlers import * # noqa: F403
src/wf_core/validation/__init__.py
-1
View File
@@ -11,4 +11,3 @@ __all__ = [
"ValidationReport",
"validate_workflow",
]
src/wf_core/validation/core.py
-1
View File
@@ -166,4 +166,3 @@ def _validate_reachable_outcomes(
f"nodes[{node_id}]",
f"reachable node is missing edges for outcomes {sorted(missing)!r}",
)
src/wf_core/validation/issues.py
-1
View File
@@ -50,4 +50,3 @@ class ValidationReport:
f"- [{issue.code}] {issue.path}: {issue.message}" for issue in self.errors
)
raise ValueError(f"Workflow validation failed:\n{rendered}")
src/wf_core/validation/outcomes.py
+3 -2
View File
@@ -19,7 +19,9 @@ def declared_outcomes_for_step(step: Step, node_defs: dict[str, NodeDef]) -> set
return set()
def reachable_node_ids(start: str, edges: list[Edge], nodes_by_id: dict[str, Step]) -> set[str]:
def reachable_node_ids(
start: str, edges: list[Edge], nodes_by_id: dict[str, Step]
) -> set[str]:
if start not in nodes_by_id:
return set()
@@ -38,4 +40,3 @@ def reachable_node_ids(start: str, edges: list[Edge], nodes_by_id: dict[str, Ste
seen.add(node_id)
stack.extend(adjacency.get(node_id, []))
return seen
src/wf_core/validation/steps.py
-1
View File
@@ -220,4 +220,3 @@ def validate_operand(
path,
f"invalid operand path {operand.path!r}",
)