general improvements to modularity

main.py

@@ -1,7 +1,7 @@
 import json
 import sys
 
-from wf_core import RuntimeContext, Workflow, execute_workflow
+from wf_core import END, RuntimeContext, Workflow, execute_workflow
 
 
 workflow = Workflow.model_validate(
@@ -137,8 +137,8 @@ workflow = Workflow.model_validate(
             {"from": "summarize", "outcome": "ok", "to": "should_email"},
             {"from": "should_email", "outcome": "true", "to": "send_email"},
             {"from": "should_email", "outcome": "false", "to": "skip_email"},
-            {"from": "send_email", "outcome": "sent", "to": "__end__"},
-            {"from": "skip_email", "outcome": "ok", "to": "__end__"},
+            {"from": "send_email", "outcome": "sent", "to": END},
+            {"from": "skip_email", "outcome": "ok", "to": END},
         ],
     }
 )

wf_core/__init__.py

@@ -16,7 +16,13 @@ from .runtime import (
     WorkflowExecutionError,
     execute_workflow,
 )
-from .validate import ValidationIssue, ValidationReport, validate_workflow
+from .tokens import END, START
+from .validate import (
+    ValidationIssue,
+    ValidationIssueCode,
+    ValidationReport,
+    validate_workflow,
+)
 
 __all__ = [
     "ConditionNode",
@@ -30,7 +36,10 @@ __all__ = [
     "StateSchema",
     "RuntimeContext",
     "TraceEntry",
+    "START",
+    "END",
     "ValidationIssue",
+    "ValidationIssueCode",
     "ValidationReport",
     "Workflow",
     "WorkflowExecutionError",

wf_core/paths.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, MutableMapping
+from typing import Any
+
+
+class PathResolutionError(ValueError):
+    pass
+
+
+def split_graph_path(path: str) -> tuple[str, list[str]]:
+    root, *parts = path.split(".")
+    if not root or not parts:
+        raise PathResolutionError(f"invalid path {path!r}")
+    return root, parts
+
+
+def is_valid_source_path(
+    path: str,
+    state_root_fields: set[str],
+    input_root_fields: set[str],
+    *,
+    allow_context: bool = False,
+) -> bool:
+    try:
+        root, parts = split_graph_path(path)
+    except PathResolutionError:
+        return False
+
+    field_name = parts[0]
+    if allow_context and 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:
+    try:
+        root, parts = split_graph_path(path)
+    except PathResolutionError:
+        return False
+    return root == "state" and bool(parts)
+
+
+def resolve_graph_path(
+    path: str,
+    *,
+    state: Mapping[str, Any],
+    workflow_input: Mapping[str, Any],
+    context: Mapping[str, Any],
+) -> Any:
+    root, parts = split_graph_path(path)
+
+    if root == "state":
+        source: Mapping[str, Any] = state
+    elif root == "input":
+        source = workflow_input
+    elif root == "context":
+        source = context
+    else:
+        raise PathResolutionError(f"unknown path root {root!r}")
+
+    current: Any = source
+    for part in parts:
+        if not isinstance(current, Mapping) or part not in current:
+            raise PathResolutionError(f"path {path!r} could not be resolved")
+        current = current[part]
+    return current
+
+
+def path_exists(
+    path: str,
+    *,
+    state: Mapping[str, Any],
+    workflow_input: Mapping[str, Any],
+    context: Mapping[str, Any],
+) -> bool:
+    try:
+        resolve_graph_path(
+            path, state=state, workflow_input=workflow_input, context=context
+        )
+    except PathResolutionError:
+        return False
+    return True
+
+
+def get_nested_value(state: Mapping[str, Any], path_parts: list[str]) -> Any:
+    current: Any = state
+    for part in path_parts:
+        if not isinstance(current, Mapping) or part not in current:
+            return None
+        current = current[part]
+    return current
+
+
+def set_nested_value(
+    state: MutableMapping[str, Any], path_parts: list[str], value: Any
+) -> None:
+    current: MutableMapping[str, Any] = 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, MutableMapping):
+            raise PathResolutionError(
+                f"cannot descend into non-object state field {part!r}"
+            )
+        current = next_value
+    current[path_parts[-1]] = value

wf_core/runtime.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
 from dataclasses import asdict, dataclass, field
 from typing import Any
 
@@ -20,6 +20,15 @@ from .model import (
     VariadicCondition,
     Workflow,
 )
+from .paths import (
+    PathResolutionError,
+    get_nested_value,
+    path_exists,
+    resolve_graph_path,
+    set_nested_value,
+    split_graph_path,
+)
+from .tokens import END
 
 
 class WorkflowExecutionError(RuntimeError):
@@ -70,7 +79,7 @@ def execute_workflow(
     prior_outcome: str | None = None
     activated_incoming_edge: str | None = None
 
-    while current_node_id != "__end__":
+    while current_node_id != END:
         step = nodes_by_id[current_node_id]
 
         if isinstance(step, NodeUse):
@@ -158,7 +167,12 @@ def _execute_node_use(
         )
 
     resolved_input = {
-        destination_field: _resolve_path(source_path, state, workflow_input, {})
+        destination_field: _safe_resolve_path(
+            source_path,
+            state=state,
+            workflow_input=workflow_input,
+            context={},
+        )
         for source_path, destination_field in node.in_map.items()
     }
     _validate_payload_against_schema(
@@ -219,23 +233,29 @@ def _apply_output_map(
 def _write_state_value(
     workflow: Workflow, state: dict[str, Any], destination_path: str, value: Any
 ) -> None:
-    root, field_name, *rest = destination_path.split(".")
+    try:
+        root, parts = split_graph_path(destination_path)
+    except PathResolutionError as exc:
+        raise WorkflowExecutionError(str(exc)) from exc
+
     if root != "state":
         raise WorkflowExecutionError(
             f"executor only supports writes into state.*, got {destination_path!r}"
         )
+
+    field_name = parts[0]
     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]
+    key_path = parts
 
     if merge_strategy == "replace":
-        _set_nested_value(state, key_path, value)
+        _safe_set_nested_value(state, key_path, value)
         return
 
-    current_value = _get_nested_value(state, key_path)
+    current_value = get_nested_value(state, key_path)
     if merge_strategy == "append":
         if current_value is None:
-            _set_nested_value(
+            _safe_set_nested_value(
                 state, key_path, [value] if not isinstance(value, list) else value
             )
             return
@@ -255,7 +275,7 @@ def _write_state_value(
                 raise WorkflowExecutionError(
                     f"cannot merge non-object value into {destination_path!r}"
                 )
-            _set_nested_value(state, key_path, dict(value))
+            _safe_set_nested_value(state, key_path, dict(value))
             return
         if not isinstance(current_value, dict) or not isinstance(value, dict):
             raise WorkflowExecutionError(
@@ -291,7 +311,12 @@ def _eval_condition(
     context_data: str | None,
 ) -> bool:
     if isinstance(condition, ExistsCondition):
-        return _path_exists(condition.path, state, workflow_input, context_data)
+        return path_exists(
+            condition.path,
+            state=state,
+            workflow_input=workflow_input,
+            context={"prior_outcome": context_data},
+        )
     if isinstance(condition, NotCondition):
         return not _eval_condition(condition.arg, state, workflow_input, context_data)
     if isinstance(condition, VariadicCondition):
@@ -322,73 +347,36 @@ def _resolve_operand(
 ) -> Any:
     if isinstance(operand, LiteralOperand):
         return operand.value
-    return _resolve_path(
+    return _safe_resolve_path(
         operand.path,
-        state,
-        workflow_input,
-        {"prior_outcome": context_data},
+        state=state,
+        workflow_input=workflow_input,
+        context={"prior_outcome": context_data},
     )
 
 
-def _path_exists(
+def _safe_resolve_path(
     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],
+    *,
+    state: Mapping[str, Any],
+    workflow_input: Mapping[str, Any],
+    context: Mapping[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}"
+    try:
+        return resolve_graph_path(
+            path,
+            state=state,
+            workflow_input=workflow_input,
+            context=context,
         )
-        current = next_value
-    current[path_parts[-1]] = value
+    except PathResolutionError as exc:
+        raise WorkflowExecutionError(str(exc)) from exc
+
+
+def _safe_set_nested_value(
+    state: dict[str, Any], path_parts: list[str], value: Any
+) -> None:
+    try:
+        set_nested_value(state, path_parts, value)
+    except PathResolutionError as exc:
+        raise WorkflowExecutionError(str(exc)) from exc

wf_core/tokens.py

@@ -0,0 +1,4 @@
+START = "__start__"
+END = "__end__"
+
+__all__ = ["START", "END"]

wf_core/validate.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
+from enum import StrEnum
+from typing import Any
 
 from .model import (
     BinaryCondition,
     Condition,
     ConditionNode,
+    Edge,
     ExistsCondition,
     ForeachNode,
     LiteralOperand,
@@ -17,10 +20,32 @@ from .model import (
     VariadicCondition,
     Workflow,
 )
+from .paths import is_valid_destination_path, is_valid_source_path
+from .tokens import END
+
+
+class ValidationIssueCode(StrEnum):
+    DUPLICATE_NODE_DEF = "duplicate_node_def"
+    DUPLICATE_NODE_ID = "duplicate_node_id"
+    UNKNOWN_START = "unknown_start"
+    DUPLICATE_EDGE = "duplicate_edge"
+    UNKNOWN_EDGE_SOURCE = "unknown_edge_source"
+    UNKNOWN_EDGE_DESTINATION = "unknown_edge_destination"
+    UNDECLARED_EDGE_OUTCOME = "undeclared_edge_outcome"
+    MISSING_OUTCOME_EDGE = "missing_outcome_edge"
+    UNKNOWN_NODE_DEF = "unknown_node_def"
+    INVALID_NODE_INPUT_FIELD = "invalid_node_input_field"
+    INVALID_SOURCE_PATH = "invalid_source_path"
+    INVALID_NODE_OUTPUT_FIELD = "invalid_node_output_field"
+    INVALID_DESTINATION_PATH = "invalid_destination_path"
+    EMPTY_CONDITION_ARGS = "empty_condition_args"
+    INVALID_CONDITION_PATH = "invalid_condition_path"
+    INVALID_FOREACH_SOURCE = "invalid_foreach_source"
 
 
 @dataclass(slots=True)
 class ValidationIssue:
+    code: ValidationIssueCode
     path: str
     message: str
 
@@ -33,14 +58,14 @@ class ValidationReport:
     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 add(self, code: ValidationIssueCode, path: str, message: str) -> None:
+        self.errors.append(ValidationIssue(code=code, 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
+            f"- [{issue.code}] {issue.path}: {issue.message}" for issue in self.errors
         )
         raise ValueError(f"Workflow validation failed:\n{rendered}")
 
@@ -52,18 +77,24 @@ def validate_workflow(workflow: Workflow) -> ValidationReport:
     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}"
+                ValidationIssueCode.DUPLICATE_NODE_DEF,
+                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 = {}
+    nodes_by_id: dict[str, Step] = {}
     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}")
+            report.add(
+                ValidationIssueCode.DUPLICATE_NODE_ID,
+                f"nodes[{index}].id",
+                f"duplicate node id {node.id!r}",
+            )
         else:
             nodes_by_id[node.id] = node
 
@@ -79,7 +110,11 @@ def validate_workflow(workflow: Workflow) -> ValidationReport:
             )
 
     if workflow.start not in nodes_by_id:
-        report.add("start", f"unknown start node {workflow.start!r}")
+        report.add(
+            ValidationIssueCode.UNKNOWN_START,
+            "start",
+            f"unknown start node {workflow.start!r}",
+        )
 
     outgoing: dict[str, set[str]] = {}
     edge_keys: set[tuple[str, str]] = set()
@@ -88,6 +123,7 @@ def validate_workflow(workflow: Workflow) -> ValidationReport:
         edge_key = (edge.from_, edge.outcome)
         if edge_key in edge_keys:
             report.add(
+                ValidationIssueCode.DUPLICATE_EDGE,
                 f"edges[{index}]",
                 f"duplicate edge for source {edge.from_!r} and outcome {edge.outcome!r}",
             )
@@ -96,20 +132,26 @@ def validate_workflow(workflow: Workflow) -> ValidationReport:
 
         source = nodes_by_id.get(edge.from_)
         if source is None:
-            report.add(f"edges[{index}].from", f"unknown source node {edge.from_!r}")
+            report.add(
+                ValidationIssueCode.UNKNOWN_EDGE_SOURCE,
+                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(
+                    ValidationIssueCode.UNDECLARED_EDGE_OUTCOME,
                     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:
+        if edge.to != END and edge.to not in nodes_by_id:
             report.add(
-                    f"edges[{index}].to", f"unknown destination node {edge.to!r}"
+                ValidationIssueCode.UNKNOWN_EDGE_DESTINATION,
+                f"edges[{index}].to",
+                f"unknown destination node {edge.to!r}",
             )
 
     reachable = _reachable_node_ids(workflow.start, workflow.edges, nodes_by_id)
@@ -121,6 +163,7 @@ def validate_workflow(workflow: Workflow) -> ValidationReport:
         missing = declared_outcomes - wired
         if missing:
             report.add(
+                ValidationIssueCode.MISSING_OUTCOME_EDGE,
                 f"nodes[{node_id}]",
                 f"reachable node is missing edges for outcomes {sorted(missing)!r}",
             )
@@ -137,7 +180,11 @@ def _validate_node_use(
 ) -> 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}")
+        report.add(
+            ValidationIssueCode.UNKNOWN_NODE_DEF,
+            f"nodes[{index}].node",
+            f"unknown node def {node.node!r}",
+        )
         return
 
     input_fields = set(node_def.input_schema.properties)
@@ -148,11 +195,13 @@ def _validate_node_use(
     for source_path, destination_field in node.in_map.items():
         if destination_field not in input_fields:
             report.add(
+                ValidationIssueCode.INVALID_NODE_INPUT_FIELD,
                 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):
+        if not is_valid_source_path(source_path, state_fields, input_root_fields):
             report.add(
+                ValidationIssueCode.INVALID_SOURCE_PATH,
                 f"nodes[{index}].in_map[{source_path!r}]",
                 "source path must start with input. or state. and reference a declared root field",
             )
@@ -160,11 +209,13 @@ def _validate_node_use(
     for source_field, destination_path in node.out_map.items():
         if source_field not in output_fields:
             report.add(
+                ValidationIssueCode.INVALID_NODE_OUTPUT_FIELD,
                 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):
+        if not is_valid_destination_path(destination_path):
             report.add(
+                ValidationIssueCode.INVALID_DESTINATION_PATH,
                 f"nodes[{index}].out_map[{source_field!r}]",
                 "destination path must start with state.",
             )
@@ -178,7 +229,11 @@ def _validate_condition_node(
     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")
+        report.add(
+            ValidationIssueCode.EMPTY_CONDITION_ARGS,
+            f"nodes[{index}].check.args",
+            "condition args must not be empty",
+        )
     _validate_condition_expr(
         node.check,
         f"nodes[{index}].check",
@@ -195,8 +250,9 @@ def _validate_foreach_node(
     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):
+    if not is_valid_source_path(node.over, state_root_fields, input_root_fields):
         report.add(
+            ValidationIssueCode.INVALID_FOREACH_SOURCE,
             f"nodes[{index}].over",
             "foreach source path must start with input. or state. and reference a declared root field",
         )
@@ -210,10 +266,17 @@ def _validate_condition_expr(
     input_root_fields: set[str],
 ) -> None:
     if isinstance(condition, ExistsCondition):
-        if not _is_valid_condition_path(
-            condition.path, state_root_fields, input_root_fields
+        if not is_valid_source_path(
+            condition.path,
+            state_root_fields,
+            input_root_fields,
+            allow_context=True,
         ):
-            report.add(path, f"invalid condition path {condition.path!r}")
+            report.add(
+                ValidationIssueCode.INVALID_CONDITION_PATH,
+                path,
+                f"invalid condition path {condition.path!r}",
+            )
         return
 
     if isinstance(condition, NotCondition):
@@ -263,8 +326,14 @@ def _validate_operand(
 ) -> 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}")
+    if not is_valid_source_path(
+        operand.path, state_root_fields, input_root_fields, allow_context=True
+    ):
+        report.add(
+            ValidationIssueCode.INVALID_CONDITION_PATH,
+            path,
+            f"invalid operand path {operand.path!r}",
+        )
 
 
 def _declared_outcomes_for_step(step: Step, node_defs: dict[str, NodeDef]) -> set[str]:
@@ -280,17 +349,19 @@ def _declared_outcomes_for_step(step: Step, node_defs: dict[str, NodeDef]) -> se
     return set()
 
 
-def _reachable_node_ids(start: str, edges, 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()
 
     adjacency: dict[str, list[str]] = {}
     for edge in edges:
-        if edge.to == "__end__":
+        if edge.to == END:
             continue
         adjacency.setdefault(edge.from_, []).append(edge.to)
 
-    seen = set()
+    seen: set[str] = set()
     stack = [start]
     while stack:
         node_id = stack.pop()
@@ -299,35 +370,3 @@ def _reachable_node_ids(start: str, edges, nodes_by_id: dict[str, Step]) -> set[
         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