init
This commit is contained in:
@@ -0,0 +1,33 @@
|
||||
from .model import (
|
||||
ConditionNode,
|
||||
Edge,
|
||||
ForeachNode,
|
||||
JoinNode,
|
||||
NodeDef,
|
||||
NodeResult,
|
||||
NodeUse,
|
||||
StateField,
|
||||
StateSchema,
|
||||
Workflow,
|
||||
)
|
||||
from .runtime import RuntimeContext, WorkflowExecutionError, execute_workflow
|
||||
from .validate import ValidationIssue, ValidationReport, validate_workflow
|
||||
|
||||
__all__ = [
|
||||
"ConditionNode",
|
||||
"Edge",
|
||||
"ForeachNode",
|
||||
"JoinNode",
|
||||
"NodeDef",
|
||||
"NodeResult",
|
||||
"NodeUse",
|
||||
"StateField",
|
||||
"StateSchema",
|
||||
"RuntimeContext",
|
||||
"ValidationIssue",
|
||||
"ValidationReport",
|
||||
"Workflow",
|
||||
"WorkflowExecutionError",
|
||||
"execute_workflow",
|
||||
"validate_workflow",
|
||||
]
|
||||
@@ -0,0 +1,146 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Annotated, Any, Literal
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
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(None, ge=0)
|
||||
timeout_seconds: int | None = Field(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(None, ge=0)
|
||||
timeout_seconds: int | None = Field(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"),
|
||||
]
|
||||
|
||||
|
||||
class ConditionNode(BaseModel):
|
||||
id: str
|
||||
type: Literal["condition"]
|
||||
check: Condition
|
||||
|
||||
|
||||
class ForeachNode(BaseModel):
|
||||
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"]
|
||||
|
||||
|
||||
Step = Annotated[
|
||||
NodeUse | ConditionNode | ForeachNode | JoinNode,
|
||||
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):
|
||||
from .validate import validate_workflow
|
||||
|
||||
return 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
|
||||
|
||||
print(json.dumps(Workflow.model_json_schema(), indent=2))
|
||||
@@ -0,0 +1,339 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
from .model import (
|
||||
BinaryCondition,
|
||||
Condition,
|
||||
ConditionNode,
|
||||
ExistsCondition,
|
||||
ForeachNode,
|
||||
JoinNode,
|
||||
LiteralOperand,
|
||||
NodeDef,
|
||||
NodeResult,
|
||||
NodeUse,
|
||||
NotCondition,
|
||||
PathOperand,
|
||||
VariadicCondition,
|
||||
Workflow,
|
||||
)
|
||||
|
||||
|
||||
class WorkflowExecutionError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class RuntimeContext:
|
||||
current_node_id: str
|
||||
retry_count: int = 0
|
||||
prior_outcome: str | None = None
|
||||
activated_incoming_edge: str | None = None
|
||||
|
||||
|
||||
NodeHandler = Callable[[dict[str, Any], RuntimeContext], NodeResult | dict[str, Any]]
|
||||
|
||||
|
||||
def execute_workflow(
|
||||
workflow: Workflow,
|
||||
workflow_input: dict[str, Any],
|
||||
registry: dict[str, NodeHandler],
|
||||
) -> dict[str, Any]:
|
||||
report = workflow.validate_structure()
|
||||
report.raise_for_errors()
|
||||
|
||||
_validate_payload_against_schema(workflow.input_schema, workflow_input, "workflow input")
|
||||
|
||||
node_defs = {node_def.name: node_def for node_def in workflow.node_defs}
|
||||
nodes_by_id = {node.id: node for node in workflow.nodes}
|
||||
edge_map = {(edge.from_, edge.outcome): edge.to for edge in workflow.edges}
|
||||
|
||||
state = dict(workflow_input)
|
||||
current_node_id = workflow.start
|
||||
prior_outcome: str | None = None
|
||||
activated_incoming_edge: str | None = None
|
||||
|
||||
while current_node_id != "__end__":
|
||||
step = nodes_by_id[current_node_id]
|
||||
|
||||
if isinstance(step, NodeUse):
|
||||
node_def = node_defs[step.node]
|
||||
outcome = _execute_node_use(
|
||||
step,
|
||||
node_def,
|
||||
state,
|
||||
workflow_input,
|
||||
registry,
|
||||
prior_outcome,
|
||||
activated_incoming_edge,
|
||||
workflow,
|
||||
)
|
||||
elif isinstance(step, ConditionNode):
|
||||
predicate = _eval_condition(step.check, state, workflow_input, prior_outcome)
|
||||
outcome = "true" if predicate else "false"
|
||||
elif isinstance(step, JoinNode):
|
||||
outcome = "done"
|
||||
elif isinstance(step, ForeachNode):
|
||||
raise WorkflowExecutionError("foreach execution is not implemented yet")
|
||||
else:
|
||||
raise WorkflowExecutionError(f"unsupported step type {step.type!r}")
|
||||
|
||||
next_node_id = edge_map.get((current_node_id, outcome))
|
||||
if next_node_id is None:
|
||||
raise WorkflowExecutionError(
|
||||
f"no edge found for node {current_node_id!r} and outcome {outcome!r}"
|
||||
)
|
||||
|
||||
prior_outcome = outcome
|
||||
activated_incoming_edge = current_node_id
|
||||
current_node_id = next_node_id
|
||||
|
||||
final_output = _project_output(workflow, state)
|
||||
_validate_payload_against_schema(workflow.output_schema, final_output, "workflow output")
|
||||
return {"state": state, "output": final_output}
|
||||
|
||||
|
||||
def _execute_node_use(
|
||||
node: NodeUse,
|
||||
node_def: NodeDef,
|
||||
state: dict[str, Any],
|
||||
workflow_input: dict[str, Any],
|
||||
registry: dict[str, NodeHandler],
|
||||
prior_outcome: str | None,
|
||||
activated_incoming_edge: str | None,
|
||||
workflow: Workflow,
|
||||
) -> str:
|
||||
handler = registry.get(node.node)
|
||||
if handler is None:
|
||||
raise WorkflowExecutionError(f"no handler registered for node def {node.node!r}")
|
||||
|
||||
resolved_input = {
|
||||
destination_field: _resolve_path(source_path, state, workflow_input, {})
|
||||
for source_path, destination_field in node.in_map.items()
|
||||
}
|
||||
_validate_payload_against_schema(
|
||||
node_def.input_schema, resolved_input, f"node input for {node.id}"
|
||||
)
|
||||
|
||||
context = RuntimeContext(
|
||||
current_node_id=node.id,
|
||||
prior_outcome=prior_outcome,
|
||||
activated_incoming_edge=activated_incoming_edge,
|
||||
)
|
||||
raw_result = handler(resolved_input, context)
|
||||
result = _coerce_node_result(raw_result)
|
||||
|
||||
if result.outcome not in node_def.outcomes:
|
||||
raise WorkflowExecutionError(
|
||||
f"node {node.id!r} returned undeclared outcome {result.outcome!r}"
|
||||
)
|
||||
|
||||
_validate_payload_against_schema(
|
||||
node_def.output_schema, result.output, f"node output for {node.id}"
|
||||
)
|
||||
_apply_output_map(workflow, node, result.output, state)
|
||||
return result.outcome
|
||||
|
||||
|
||||
def _coerce_node_result(raw_result: NodeResult | dict[str, Any]) -> NodeResult:
|
||||
if isinstance(raw_result, NodeResult):
|
||||
return raw_result
|
||||
if "outcome" in raw_result and "output" in raw_result:
|
||||
return NodeResult.model_validate(raw_result)
|
||||
return NodeResult(outcome="ok", output=raw_result)
|
||||
|
||||
|
||||
def _apply_output_map(
|
||||
workflow: Workflow,
|
||||
node: NodeUse,
|
||||
node_output: dict[str, Any],
|
||||
state: dict[str, Any],
|
||||
) -> None:
|
||||
for source_field, destination_path in node.out_map.items():
|
||||
if source_field not in node_output:
|
||||
raise WorkflowExecutionError(
|
||||
f"node {node.id!r} did not return required mapped field {source_field!r}"
|
||||
)
|
||||
value = node_output[source_field]
|
||||
_write_state_value(workflow, state, destination_path, value)
|
||||
|
||||
|
||||
def _write_state_value(
|
||||
workflow: Workflow, state: dict[str, Any], destination_path: str, value: Any
|
||||
) -> None:
|
||||
root, field_name, *rest = destination_path.split(".")
|
||||
if root != "state":
|
||||
raise WorkflowExecutionError(
|
||||
f"executor only supports writes into state.*, got {destination_path!r}"
|
||||
)
|
||||
declared_field = workflow.state_schema.fields.get(field_name)
|
||||
merge_strategy = declared_field.merge_strategy if declared_field else "replace"
|
||||
key_path = [field_name, *rest]
|
||||
|
||||
if merge_strategy == "replace":
|
||||
_set_nested_value(state, key_path, value)
|
||||
return
|
||||
|
||||
current_value = _get_nested_value(state, key_path)
|
||||
if merge_strategy == "append":
|
||||
if current_value is None:
|
||||
_set_nested_value(state, key_path, [value] if not isinstance(value, list) else value)
|
||||
return
|
||||
if not isinstance(current_value, list):
|
||||
raise WorkflowExecutionError(
|
||||
f"cannot append into non-list state path {destination_path!r}"
|
||||
)
|
||||
if isinstance(value, list):
|
||||
current_value.extend(value)
|
||||
else:
|
||||
current_value.append(value)
|
||||
return
|
||||
|
||||
if merge_strategy == "merge_object":
|
||||
if current_value is None:
|
||||
if not isinstance(value, dict):
|
||||
raise WorkflowExecutionError(
|
||||
f"cannot merge non-object value into {destination_path!r}"
|
||||
)
|
||||
_set_nested_value(state, key_path, dict(value))
|
||||
return
|
||||
if not isinstance(current_value, dict) or not isinstance(value, dict):
|
||||
raise WorkflowExecutionError(
|
||||
f"merge_object requires dict values at {destination_path!r}"
|
||||
)
|
||||
current_value.update(value)
|
||||
return
|
||||
|
||||
raise WorkflowExecutionError(f"unknown merge strategy {merge_strategy!r}")
|
||||
|
||||
|
||||
def _project_output(workflow: Workflow, state: dict[str, Any]) -> dict[str, Any]:
|
||||
return {
|
||||
key: state[key]
|
||||
for key in workflow.output_schema.properties
|
||||
if key in state
|
||||
}
|
||||
|
||||
|
||||
def _validate_payload_against_schema(schema: Any, payload: Any, label: str) -> None:
|
||||
if schema.type == "object":
|
||||
if not isinstance(payload, dict):
|
||||
raise WorkflowExecutionError(f"{label} must be an object")
|
||||
for required_key in schema.required:
|
||||
if required_key not in payload:
|
||||
raise WorkflowExecutionError(
|
||||
f"{label} is missing required field {required_key!r}"
|
||||
)
|
||||
|
||||
|
||||
def _eval_condition(
|
||||
condition: Condition,
|
||||
state: dict[str, Any],
|
||||
workflow_input: dict[str, Any],
|
||||
context_data: str | None,
|
||||
) -> bool:
|
||||
if isinstance(condition, ExistsCondition):
|
||||
return _path_exists(condition.path, state, workflow_input, context_data)
|
||||
if isinstance(condition, NotCondition):
|
||||
return not _eval_condition(condition.arg, state, workflow_input, context_data)
|
||||
if isinstance(condition, VariadicCondition):
|
||||
values = [
|
||||
_eval_condition(arg, state, workflow_input, context_data)
|
||||
for arg in condition.args
|
||||
]
|
||||
return all(values) if condition.op == "and" else any(values)
|
||||
if isinstance(condition, BinaryCondition):
|
||||
left = _resolve_operand(condition.left, state, workflow_input, context_data)
|
||||
right = _resolve_operand(condition.right, state, workflow_input, context_data)
|
||||
if condition.op == "eq":
|
||||
return left == right
|
||||
if condition.op == "ne":
|
||||
return left != right
|
||||
if condition.op == "gt":
|
||||
return left > right
|
||||
if condition.op == "lt":
|
||||
return left < right
|
||||
raise WorkflowExecutionError(f"unsupported condition operator {condition.op!r}")
|
||||
|
||||
|
||||
def _resolve_operand(
|
||||
operand: PathOperand | LiteralOperand,
|
||||
state: dict[str, Any],
|
||||
workflow_input: dict[str, Any],
|
||||
context_data: str | None,
|
||||
) -> Any:
|
||||
if isinstance(operand, LiteralOperand):
|
||||
return operand.value
|
||||
return _resolve_path(
|
||||
operand.path,
|
||||
state,
|
||||
workflow_input,
|
||||
{"prior_outcome": context_data},
|
||||
)
|
||||
|
||||
|
||||
def _path_exists(
|
||||
path: str,
|
||||
state: dict[str, Any],
|
||||
workflow_input: dict[str, Any],
|
||||
context_data: str | None,
|
||||
) -> bool:
|
||||
try:
|
||||
_resolve_path(path, state, workflow_input, {"prior_outcome": context_data})
|
||||
except WorkflowExecutionError:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _resolve_path(
|
||||
path: str,
|
||||
state: dict[str, Any],
|
||||
workflow_input: dict[str, Any],
|
||||
context: dict[str, Any],
|
||||
) -> Any:
|
||||
root, *parts = path.split(".")
|
||||
if not parts:
|
||||
raise WorkflowExecutionError(f"invalid path {path!r}")
|
||||
|
||||
if root == "state":
|
||||
source = state
|
||||
elif root == "input":
|
||||
source = workflow_input
|
||||
elif root == "context":
|
||||
source = context
|
||||
else:
|
||||
raise WorkflowExecutionError(f"unknown path root {root!r}")
|
||||
|
||||
current: Any = source
|
||||
for part in parts:
|
||||
if not isinstance(current, dict) or part not in current:
|
||||
raise WorkflowExecutionError(f"path {path!r} could not be resolved")
|
||||
current = current[part]
|
||||
return current
|
||||
|
||||
|
||||
def _get_nested_value(state: dict[str, Any], path_parts: list[str]) -> Any:
|
||||
current: Any = state
|
||||
for part in path_parts:
|
||||
if not isinstance(current, dict) or part not in current:
|
||||
return None
|
||||
current = current[part]
|
||||
return current
|
||||
|
||||
|
||||
def _set_nested_value(state: dict[str, Any], path_parts: list[str], value: Any) -> None:
|
||||
current = state
|
||||
for part in path_parts[:-1]:
|
||||
next_value = current.get(part)
|
||||
if next_value is None:
|
||||
next_value = {}
|
||||
current[part] = next_value
|
||||
if not isinstance(next_value, dict):
|
||||
raise WorkflowExecutionError(
|
||||
f"cannot descend into non-object state field {part!r}"
|
||||
)
|
||||
current = next_value
|
||||
current[path_parts[-1]] = value
|
||||
@@ -0,0 +1,333 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from .model import (
|
||||
BinaryCondition,
|
||||
Condition,
|
||||
ConditionNode,
|
||||
ExistsCondition,
|
||||
ForeachNode,
|
||||
LiteralOperand,
|
||||
NodeDef,
|
||||
NodeUse,
|
||||
NotCondition,
|
||||
PathOperand,
|
||||
Step,
|
||||
VariadicCondition,
|
||||
Workflow,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class ValidationIssue:
|
||||
path: str
|
||||
message: str
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class ValidationReport:
|
||||
errors: list[ValidationIssue] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def ok(self) -> bool:
|
||||
return not self.errors
|
||||
|
||||
def add(self, path: str, message: str) -> None:
|
||||
self.errors.append(ValidationIssue(path=path, message=message))
|
||||
|
||||
def raise_for_errors(self) -> None:
|
||||
if not self.errors:
|
||||
return
|
||||
rendered = "\n".join(
|
||||
f"- {issue.path}: {issue.message}" for issue in self.errors
|
||||
)
|
||||
raise ValueError(f"Workflow validation failed:\n{rendered}")
|
||||
|
||||
|
||||
def validate_workflow(workflow: Workflow) -> ValidationReport:
|
||||
report = ValidationReport()
|
||||
|
||||
node_defs: dict[str, NodeDef] = {}
|
||||
for index, node_def in enumerate(workflow.node_defs):
|
||||
if node_def.name in node_defs:
|
||||
report.add(
|
||||
f"node_defs[{index}].name", f"duplicate node def name {node_def.name!r}"
|
||||
)
|
||||
else:
|
||||
node_defs[node_def.name] = node_def
|
||||
|
||||
nodes_by_id = {}
|
||||
state_root_fields = set(workflow.state_schema.fields)
|
||||
input_root_fields = set(workflow.input_schema.properties)
|
||||
|
||||
for index, node in enumerate(workflow.nodes):
|
||||
if node.id in nodes_by_id:
|
||||
report.add(f"nodes[{index}].id", f"duplicate node id {node.id!r}")
|
||||
else:
|
||||
nodes_by_id[node.id] = node
|
||||
|
||||
if isinstance(node, NodeUse):
|
||||
_validate_node_use(node, index, node_defs, workflow, report)
|
||||
elif isinstance(node, ConditionNode):
|
||||
_validate_condition_node(
|
||||
node, index, report, state_root_fields, input_root_fields
|
||||
)
|
||||
elif isinstance(node, ForeachNode):
|
||||
_validate_foreach_node(
|
||||
node, index, report, state_root_fields, input_root_fields
|
||||
)
|
||||
|
||||
if workflow.start not in nodes_by_id:
|
||||
report.add("start", f"unknown start node {workflow.start!r}")
|
||||
|
||||
outgoing: dict[str, set[str]] = {}
|
||||
edge_keys: set[tuple[str, str]] = set()
|
||||
|
||||
for index, edge in enumerate(workflow.edges):
|
||||
edge_key = (edge.from_, edge.outcome)
|
||||
if edge_key in edge_keys:
|
||||
report.add(
|
||||
f"edges[{index}]",
|
||||
f"duplicate edge for source {edge.from_!r} and outcome {edge.outcome!r}",
|
||||
)
|
||||
else:
|
||||
edge_keys.add(edge_key)
|
||||
|
||||
source = nodes_by_id.get(edge.from_)
|
||||
if source is None:
|
||||
report.add(f"edges[{index}].from", f"unknown source node {edge.from_!r}")
|
||||
else:
|
||||
allowed = _declared_outcomes_for_step(source, node_defs)
|
||||
if edge.outcome not in allowed:
|
||||
report.add(
|
||||
f"edges[{index}].outcome",
|
||||
f"outcome {edge.outcome!r} is not declared by node {edge.from_!r}",
|
||||
)
|
||||
outgoing.setdefault(edge.from_, set()).add(edge.outcome)
|
||||
|
||||
if edge.to != "__end__":
|
||||
if edge.to not in nodes_by_id:
|
||||
report.add(
|
||||
f"edges[{index}].to", f"unknown destination node {edge.to!r}"
|
||||
)
|
||||
|
||||
reachable = _reachable_node_ids(workflow.start, workflow.edges, nodes_by_id)
|
||||
|
||||
for node_id in reachable:
|
||||
node = nodes_by_id[node_id]
|
||||
declared_outcomes = _declared_outcomes_for_step(node, node_defs)
|
||||
wired = outgoing.get(node_id, set())
|
||||
missing = declared_outcomes - wired
|
||||
if missing:
|
||||
report.add(
|
||||
f"nodes[{node_id}]",
|
||||
f"reachable node is missing edges for outcomes {sorted(missing)!r}",
|
||||
)
|
||||
|
||||
return report
|
||||
|
||||
|
||||
def _validate_node_use(
|
||||
node: NodeUse,
|
||||
index: int,
|
||||
node_defs: dict[str, NodeDef],
|
||||
workflow: Workflow,
|
||||
report: ValidationReport,
|
||||
) -> None:
|
||||
node_def = node_defs.get(node.node)
|
||||
if node_def is None:
|
||||
report.add(f"nodes[{index}].node", f"unknown node def {node.node!r}")
|
||||
return
|
||||
|
||||
input_fields = set(node_def.input_schema.properties)
|
||||
output_fields = set(node_def.output_schema.properties)
|
||||
state_fields = set(workflow.state_schema.fields)
|
||||
input_root_fields = set(workflow.input_schema.properties)
|
||||
|
||||
for source_path, destination_field in node.in_map.items():
|
||||
if destination_field not in input_fields:
|
||||
report.add(
|
||||
f"nodes[{index}].in_map[{source_path!r}]",
|
||||
f"destination field {destination_field!r} is not declared in node input schema",
|
||||
)
|
||||
if not _is_valid_source_path(source_path, state_fields, input_root_fields):
|
||||
report.add(
|
||||
f"nodes[{index}].in_map[{source_path!r}]",
|
||||
"source path must start with input. or state. and reference a declared root field",
|
||||
)
|
||||
|
||||
for source_field, destination_path in node.out_map.items():
|
||||
if source_field not in output_fields:
|
||||
report.add(
|
||||
f"nodes[{index}].out_map[{source_field!r}]",
|
||||
f"source field {source_field!r} is not declared in node output schema",
|
||||
)
|
||||
if not _is_valid_destination_path(destination_path):
|
||||
report.add(
|
||||
f"nodes[{index}].out_map[{source_field!r}]",
|
||||
"destination path must start with state.",
|
||||
)
|
||||
|
||||
|
||||
def _validate_condition_node(
|
||||
node: ConditionNode,
|
||||
index: int,
|
||||
report: ValidationReport,
|
||||
state_root_fields: set[str],
|
||||
input_root_fields: set[str],
|
||||
) -> None:
|
||||
if isinstance(node.check, VariadicCondition) and not node.check.args:
|
||||
report.add(f"nodes[{index}].check.args", "condition args must not be empty")
|
||||
_validate_condition_expr(
|
||||
node.check,
|
||||
f"nodes[{index}].check",
|
||||
report,
|
||||
state_root_fields,
|
||||
input_root_fields,
|
||||
)
|
||||
|
||||
|
||||
def _validate_foreach_node(
|
||||
node: ForeachNode,
|
||||
index: int,
|
||||
report: ValidationReport,
|
||||
state_root_fields: set[str],
|
||||
input_root_fields: set[str],
|
||||
) -> None:
|
||||
if not _is_valid_source_path(node.over, state_root_fields, input_root_fields):
|
||||
report.add(
|
||||
f"nodes[{index}].over",
|
||||
"foreach source path must start with input. or state. and reference a declared root field",
|
||||
)
|
||||
|
||||
|
||||
def _validate_condition_expr(
|
||||
condition: Condition,
|
||||
path: str,
|
||||
report: ValidationReport,
|
||||
state_root_fields: set[str],
|
||||
input_root_fields: set[str],
|
||||
) -> None:
|
||||
if isinstance(condition, ExistsCondition):
|
||||
if not _is_valid_condition_path(
|
||||
condition.path, state_root_fields, input_root_fields
|
||||
):
|
||||
report.add(path, f"invalid condition path {condition.path!r}")
|
||||
return
|
||||
|
||||
if isinstance(condition, NotCondition):
|
||||
_validate_condition_expr(
|
||||
condition.arg,
|
||||
f"{path}.arg",
|
||||
report,
|
||||
state_root_fields,
|
||||
input_root_fields,
|
||||
)
|
||||
return
|
||||
|
||||
if isinstance(condition, VariadicCondition):
|
||||
for index, arg in enumerate(condition.args):
|
||||
_validate_condition_expr(
|
||||
arg,
|
||||
f"{path}.args[{index}]",
|
||||
report,
|
||||
state_root_fields,
|
||||
input_root_fields,
|
||||
)
|
||||
return
|
||||
|
||||
if isinstance(condition, BinaryCondition):
|
||||
_validate_operand(
|
||||
condition.left,
|
||||
f"{path}.left",
|
||||
report,
|
||||
state_root_fields,
|
||||
input_root_fields,
|
||||
)
|
||||
_validate_operand(
|
||||
condition.right,
|
||||
f"{path}.right",
|
||||
report,
|
||||
state_root_fields,
|
||||
input_root_fields,
|
||||
)
|
||||
|
||||
|
||||
def _validate_operand(
|
||||
operand: PathOperand | LiteralOperand,
|
||||
path: str,
|
||||
report: ValidationReport,
|
||||
state_root_fields: set[str],
|
||||
input_root_fields: set[str],
|
||||
) -> None:
|
||||
if isinstance(operand, LiteralOperand):
|
||||
return
|
||||
if not _is_valid_condition_path(operand.path, state_root_fields, input_root_fields):
|
||||
report.add(path, f"invalid operand path {operand.path!r}")
|
||||
|
||||
|
||||
def _declared_outcomes_for_step(step: Step, node_defs: dict[str, NodeDef]) -> set[str]:
|
||||
if isinstance(step, NodeUse):
|
||||
node_def = node_defs.get(step.node)
|
||||
return set(node_def.outcomes) if node_def else set()
|
||||
if step.type == "condition":
|
||||
return {"true", "false"}
|
||||
if step.type == "foreach":
|
||||
return {"done"}
|
||||
if step.type == "join":
|
||||
return {"done"}
|
||||
return set()
|
||||
|
||||
|
||||
def _reachable_node_ids(start: str, edges, nodes_by_id: dict[str, Step]) -> set[str]:
|
||||
if start not in nodes_by_id:
|
||||
return set()
|
||||
|
||||
adjacency: dict[str, list[str]] = {}
|
||||
for edge in edges:
|
||||
if edge.to == "__end__":
|
||||
continue
|
||||
adjacency.setdefault(edge.from_, []).append(edge.to)
|
||||
|
||||
seen = set()
|
||||
stack = [start]
|
||||
while stack:
|
||||
node_id = stack.pop()
|
||||
if node_id in seen:
|
||||
continue
|
||||
seen.add(node_id)
|
||||
stack.extend(adjacency.get(node_id, []))
|
||||
return seen
|
||||
|
||||
|
||||
def _is_valid_source_path(
|
||||
path: str, state_root_fields: set[str], input_root_fields: set[str]
|
||||
) -> bool:
|
||||
if "." not in path:
|
||||
return False
|
||||
root, field_name, *_ = path.split(".")
|
||||
if root == "state":
|
||||
return field_name in state_root_fields
|
||||
if root == "input":
|
||||
return field_name in input_root_fields
|
||||
return False
|
||||
|
||||
|
||||
def _is_valid_condition_path(
|
||||
path: str, state_root_fields: set[str], input_root_fields: set[str]
|
||||
) -> bool:
|
||||
if "." not in path:
|
||||
return False
|
||||
root, field_name, *_ = path.split(".")
|
||||
if root == "context":
|
||||
return True
|
||||
if root == "state":
|
||||
return field_name in state_root_fields
|
||||
if root == "input":
|
||||
return field_name in input_root_fields
|
||||
return False
|
||||
|
||||
|
||||
def _is_valid_destination_path(path: str) -> bool:
|
||||
return path.startswith("state.") and len(path.split(".")) >= 2
|
||||
Reference in New Issue
Block a user