whole ass global ass runstate over here

Hopefully i can generalize this soontm T bound BaseModel

wf_core/__init__.py

@@ -11,11 +11,14 @@ from .model import (
     Workflow,
 )
 from .runtime import (
-    RuntimeContext,
-    TraceEntry,
+    NodeHandler,
     WorkflowExecutionError,
+    coerce_node_result,
     execute_workflow,
+    resume_workflow,
+    step_workflow,
 )
+from .run_state import RunState, RunStatus, RuntimeContext, TraceEntry
 from .tokens import END, START
 from .validate import (
     ValidationIssue,
@@ -34,6 +37,9 @@ __all__ = [
     "NodeUse",
     "StateField",
     "StateSchema",
+    "NodeHandler",
+    "RunState",
+    "RunStatus",
     "RuntimeContext",
     "TraceEntry",
     "START",
@@ -43,6 +49,9 @@ __all__ = [
     "ValidationReport",
     "Workflow",
     "WorkflowExecutionError",
+    "coerce_node_result",
     "execute_workflow",
+    "resume_workflow",
+    "step_workflow",
     "validate_workflow",
 ]

wf_core/conditions.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from .errors import WorkflowExecutionError
+from .model import (
+    BinaryCondition,
+    Condition,
+    ExistsCondition,
+    LiteralOperand,
+    NotCondition,
+    PathOperand,
+    VariadicCondition,
+)
+from .paths import PathResolutionError, path_exists, resolve_graph_path
+
+
+def eval_condition(
+    condition: Condition,
+    state: Mapping[str, Any],
+    workflow_input: Mapping[str, Any],
+    context_data: str | None,
+) -> bool:
+    if isinstance(condition, ExistsCondition):
+        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):
+        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: Mapping[str, Any],
+    workflow_input: Mapping[str, Any],
+    context_data: str | None,
+) -> Any:
+    if isinstance(operand, LiteralOperand):
+        return operand.value
+    return safe_resolve_path(
+        operand.path,
+        state=state,
+        workflow_input=workflow_input,
+        context={"prior_outcome": context_data},
+    )
+
+
+def safe_resolve_path(
+    path: str,
+    *,
+    state: Mapping[str, Any],
+    workflow_input: Mapping[str, Any],
+    context: Mapping[str, Any],
+) -> Any:
+    try:
+        return resolve_graph_path(
+            path,
+            state=state,
+            workflow_input=workflow_input,
+            context=context,
+        )
+    except PathResolutionError as exc:
+        raise WorkflowExecutionError(str(exc)) from exc

wf_core/errors.py

@@ -0,0 +1,5 @@
+class WorkflowExecutionError(RuntimeError):
+    pass
+
+
+__all__ = ["WorkflowExecutionError"]

wf_core/run_state.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass, field
+from enum import StrEnum
+from typing import Any
+
+
+class RunStatus(StrEnum):
+    PENDING = "pending"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    INTERRUPTED = "interrupted"
+
+
+@dataclass(slots=True)
+class RuntimeContext:
+    current_node_id: str
+    retry_count: int = 0
+    prior_outcome: str | None = None
+    activated_incoming_edge: str | None = None
+
+
+@dataclass(slots=True)
+class TraceEntry:
+    node_id: str
+    step_type: str
+    resolved_input: dict[str, Any]
+    outcome: str
+    next_node_id: str
+    output: dict[str, Any] = field(default_factory=dict)
+    state_changes: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(slots=True)
+class RunState:
+    workflow_name: str
+    status: RunStatus
+    workflow_input: dict[str, Any]
+    state: dict[str, Any]
+    output: dict[str, Any] = field(default_factory=dict)
+    trace: list[TraceEntry] = field(default_factory=list)
+    current_node_id: str | None = None
+    prior_outcome: str | None = None
+    activated_incoming_edge: str | None = None
+    error: str | None = None
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)

wf_core/runtime.py

@@ -1,59 +1,17 @@
 from __future__ import annotations
 
-from collections.abc import Callable, Mapping
-from dataclasses import asdict, dataclass, field
+from collections.abc import Callable
 from typing import Any
 
-from .model import (
-    BinaryCondition,
-    Condition,
-    ConditionNode,
-    ExistsCondition,
-    ForeachNode,
-    JoinNode,
-    LiteralOperand,
-    NodeDef,
-    NodeResult,
-    NodeUse,
-    NotCondition,
-    PathOperand,
-    VariadicCondition,
-    Workflow,
-)
-from .paths import (
-    PathResolutionError,
-    get_nested_value,
-    path_exists,
-    resolve_graph_path,
-    set_nested_value,
-    split_graph_path,
-)
+from .conditions import eval_condition, safe_resolve_path
+from .errors import WorkflowExecutionError
+from .model import ConditionNode, ForeachNode, JoinNode, NodeDef, NodeResult, NodeUse, Workflow
+from .run_state import RunState, RunStatus, RuntimeContext, TraceEntry
+from .schema_tools import validate_payload_against_schema
+from .state_ops import apply_output_map, project_output
 from .tokens import END
 
 
-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
-
-
-@dataclass(slots=True)
-class TraceEntry:
-    node_id: str
-    step_type: str
-    resolved_input: dict[str, Any]
-    outcome: str
-    next_node_id: str
-    output: dict[str, Any] = field(default_factory=dict)
-    state_changes: dict[str, Any] = field(default_factory=dict)
-
-
 NodeHandler = Callable[[dict[str, Any], RuntimeContext], NodeResult | dict[str, Any]]
 
 
@@ -61,104 +19,135 @@ 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"
+) -> RunState:
+    run = RunState(
+        workflow_name=workflow.name,
+        status=RunStatus.PENDING,
+        workflow_input=dict(workflow_input),
+        state=dict(workflow_input),
+        current_node_id=workflow.start,
     )
 
+    try:
+        workflow.validate_structure().raise_for_errors()
+        validate_payload_against_schema(
+            workflow.input_schema, workflow_input, "workflow input"
+        )
+        return resume_workflow(workflow, run, registry)
+    except Exception as exc:
+        run.status = RunStatus.FAILED
+        run.error = str(exc)
+        raise
+
+
+def resume_workflow(
+    workflow: Workflow,
+    run: RunState,
+    registry: dict[str, NodeHandler],
+) -> RunState:
+    if run.workflow_name != workflow.name:
+        raise WorkflowExecutionError(
+            f"run state belongs to workflow {run.workflow_name!r}, not {workflow.name!r}"
+        )
+
+    if run.current_node_id is None:
+        raise WorkflowExecutionError("run has no current node")
+
+    if run.status == RunStatus.COMPLETED:
+        return run
+
+    run.status = RunStatus.RUNNING
+    run.error = None
     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)
-    trace: list[TraceEntry] = []
-    current_node_id = workflow.start
-    prior_outcome: str | None = None
-    activated_incoming_edge: str | None = None
+    while run.current_node_id != END:
+        step_workflow(workflow, run, registry, node_defs=node_defs, nodes_by_id=nodes_by_id, edge_map=edge_map)
 
-    while current_node_id != END:
-        step = nodes_by_id[current_node_id]
+    run.output = project_output(workflow, run.state)
+    validate_payload_against_schema(
+        workflow.output_schema, run.output, "workflow output"
+    )
+    run.status = RunStatus.COMPLETED
+    run.current_node_id = END
+    return run
 
-        if isinstance(step, NodeUse):
-            node_def = node_defs[step.node]
-            node_result = _execute_node_use(
-                step,
-                node_def,
-                state,
-                workflow_input,
-                registry,
-                prior_outcome,
-                activated_incoming_edge,
-                workflow,
-            )
-            outcome = node_result["outcome"]
-        elif isinstance(step, ConditionNode):
-            predicate = _eval_condition(
-                step.check, state, workflow_input, prior_outcome
-            )
-            outcome = "true" if predicate else "false"
-            node_result = {
-                "resolved_input": {},
-                "output": {"predicate": predicate},
-                "state_changes": {},
-            }
-        elif isinstance(step, JoinNode):
-            outcome = "done"
-            node_result = {
-                "resolved_input": {},
-                "output": {},
-                "state_changes": {},
-            }
-        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}"
-            )
+def step_workflow(
+    workflow: Workflow,
+    run: RunState,
+    registry: dict[str, NodeHandler],
+    *,
+    node_defs: dict[str, NodeDef] | None = None,
+    nodes_by_id: dict[str, Any] | None = None,
+    edge_map: dict[tuple[str, str], str] | None = None,
+) -> RunState:
+    if run.current_node_id is None or run.current_node_id == END:
+        return run
 
-        trace.append(
-            TraceEntry(
-                node_id=current_node_id,
-                step_type=step.type,
-                resolved_input=node_result["resolved_input"],
-                outcome=outcome,
-                next_node_id=next_node_id,
-                output=node_result["output"],
-                state_changes=node_result["state_changes"],
-            )
+    node_defs = node_defs or {node_def.name: node_def for node_def in workflow.node_defs}
+    nodes_by_id = nodes_by_id or {node.id: node for node in workflow.nodes}
+    edge_map = edge_map or {(edge.from_, edge.outcome): edge.to for edge in workflow.edges}
+
+    step = nodes_by_id[run.current_node_id]
+
+    if isinstance(step, NodeUse):
+        node_def = node_defs[step.node]
+        step_result = _execute_node_use(workflow, run, step, node_def, registry)
+        outcome = step_result["outcome"]
+    elif isinstance(step, ConditionNode):
+        predicate = eval_condition(
+            step.check, run.state, run.workflow_input, run.prior_outcome
+        )
+        outcome = "true" if predicate else "false"
+        step_result = {
+            "resolved_input": {},
+            "output": {"predicate": predicate},
+            "state_changes": {},
+        }
+    elif isinstance(step, JoinNode):
+        outcome = "done"
+        step_result = {
+            "resolved_input": {},
+            "output": {},
+            "state_changes": {},
+        }
+    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((run.current_node_id, outcome))
+    if next_node_id is None:
+        raise WorkflowExecutionError(
+            f"no edge found for node {run.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"
+    run.trace.append(
+        TraceEntry(
+            node_id=run.current_node_id,
+            step_type=step.type,
+            resolved_input=step_result["resolved_input"],
+            outcome=outcome,
+            next_node_id=next_node_id,
+            output=step_result["output"],
+            state_changes=step_result["state_changes"],
+        )
     )
-    return {
-        "state": state,
-        "output": final_output,
-        "trace": [asdict(entry) for entry in trace],
-    }
+
+    run.prior_outcome = outcome
+    run.activated_incoming_edge = run.current_node_id
+    run.current_node_id = next_node_id
+    return run
 
 
 def _execute_node_use(
+    workflow: Workflow,
+    run: RunState,
     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,
 ) -> dict[str, Any]:
     handler = registry.get(node.node)
     if handler is None:
@@ -167,35 +156,35 @@ def _execute_node_use(
         )
 
     resolved_input = {
-        destination_field: _safe_resolve_path(
+        destination_field: safe_resolve_path(
             source_path,
-            state=state,
-            workflow_input=workflow_input,
+            state=run.state,
+            workflow_input=run.workflow_input,
             context={},
         )
         for source_path, destination_field in node.in_map.items()
     }
-    _validate_payload_against_schema(
+    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,
+        prior_outcome=run.prior_outcome,
+        activated_incoming_edge=run.activated_incoming_edge,
     )
     raw_result = handler(resolved_input, context)
-    result = _coerce_node_result(raw_result)
+    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(
+    validate_payload_against_schema(
         node_def.output_schema, result.output, f"node output for {node.id}"
     )
-    state_changes = _apply_output_map(workflow, node, result.output, state)
+    state_changes = apply_output_map(workflow, node, result.output, run.state)
     return {
         "outcome": result.outcome,
         "resolved_input": resolved_input,
@@ -204,179 +193,9 @@ def _execute_node_use(
     }
 
 
-def _coerce_node_result(raw_result: NodeResult | dict[str, Any]) -> NodeResult:
+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],
-) -> dict[str, Any]:
-    state_changes: dict[str, Any] = {}
-    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)
-        state_changes[destination_path] = value
-    return state_changes
-
-
-def _write_state_value(
-    workflow: Workflow, state: dict[str, Any], destination_path: str, value: Any
-) -> None:
-    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 = parts
-
-    if merge_strategy == "replace":
-        _safe_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:
-            _safe_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}"
-                )
-            _safe_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=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):
-        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 _safe_resolve_path(
-        operand.path,
-        state=state,
-        workflow_input=workflow_input,
-        context={"prior_outcome": context_data},
-    )
-
-
-def _safe_resolve_path(
-    path: str,
-    *,
-    state: Mapping[str, Any],
-    workflow_input: Mapping[str, Any],
-    context: Mapping[str, Any],
-) -> Any:
-    try:
-        return resolve_graph_path(
-            path,
-            state=state,
-            workflow_input=workflow_input,
-            context=context,
-        )
-    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/schema_tools.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+from typing import Any
+
+from .errors import WorkflowExecutionError
+
+
+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}"
+                )

wf_core/state_ops.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+from typing import Any
+
+from .errors import WorkflowExecutionError
+from .model import NodeUse, Workflow
+from .paths import PathResolutionError, get_nested_value, set_nested_value, split_graph_path
+
+
+def apply_output_map(
+    workflow: Workflow,
+    node: NodeUse,
+    node_output: dict[str, Any],
+    state: dict[str, Any],
+) -> dict[str, Any]:
+    state_changes: dict[str, Any] = {}
+    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)
+        state_changes[destination_path] = value
+    return state_changes
+
+
+def write_state_value(
+    workflow: Workflow, state: dict[str, Any], destination_path: str, value: Any
+) -> None:
+    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 = parts
+
+    if merge_strategy == "replace":
+        safe_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:
+            safe_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}"
+                )
+            safe_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 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