and thats a server we can use

src/wf_core/__init__.py

@@ -0,0 +1,82 @@
+from .model import (
+    ConditionNode,
+    Edge,
+    ForeachNode,
+    InterruptNode,
+    JoinNode,
+    NodeDef,
+    NodeResult,
+    NodeUse,
+    SchemaRef,
+    StateField,
+    StateSchema,
+    Workflow,
+)
+from .runtime import (
+    AsyncNodeHandler,
+    NodeHandler,
+    WorkflowExecutionError,
+    coerce_node_result,
+    execute_workflow_async,
+    execute_workflow,
+    resume_workflow_async,
+    resume_workflow,
+    step_workflow_async,
+    step_workflow,
+)
+from .run_state import (
+    ExecutionFrame,
+    FrameStatus,
+    InterruptRequest,
+    RunState,
+    RunStatus,
+    RuntimeContext,
+    StepExecutionResult,
+    TraceEntry,
+)
+from .tokens import END, START
+from .validate import (
+    ValidationIssue,
+    ValidationIssueCode,
+    ValidationReport,
+    validate_workflow,
+)
+
+__all__ = [
+    "ConditionNode",
+    "Edge",
+    "ForeachNode",
+    "InterruptNode",
+    "JoinNode",
+    "NodeDef",
+    "NodeResult",
+    "NodeUse",
+    "SchemaRef",
+    "StateField",
+    "StateSchema",
+    "AsyncNodeHandler",
+    "NodeHandler",
+    "ExecutionFrame",
+    "FrameStatus",
+    "RunState",
+    "RunStatus",
+    "RuntimeContext",
+    "StepExecutionResult",
+    "TraceEntry",
+    "InterruptRequest",
+    "START",
+    "END",
+    "ValidationIssue",
+    "ValidationIssueCode",
+    "ValidationReport",
+    "Workflow",
+    "WorkflowExecutionError",
+    "coerce_node_result",
+    "execute_workflow_async",
+    "execute_workflow",
+    "resume_workflow_async",
+    "resume_workflow",
+    "step_workflow_async",
+    "step_workflow",
+    "validate_workflow",
+]

src/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

src/wf_core/demo_workflow.py

@@ -0,0 +1,272 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import cast
+
+from .model import Workflow
+from .run_state import RuntimeContext
+from .tokens import END
+
+DemoHandler = Callable[[dict[str, object], RuntimeContext], dict[str, object]]
+
+
+def build_demo_workflow() -> Workflow:
+    return Workflow.model_validate(
+        {
+            "name": "drive_summary_demo",
+            "input_schema": {
+                "type": "object",
+                "properties": {
+                    "folder_id": {"type": "string"},
+                    "should_email": {"type": "boolean"},
+                },
+                "required": ["folder_id", "should_email"],
+            },
+            "state_schema": {
+                "fields": {
+                    "folder_id": {"type": "string"},
+                    "should_email": {"type": "boolean"},
+                    "documents": {"type": "array", "merge_strategy": "replace"},
+                    "item_summaries": {"type": "array", "merge_strategy": "append"},
+                    "summary": {"type": "string", "merge_strategy": "replace"},
+                    "approved": {"type": "boolean", "merge_strategy": "replace"},
+                    "approval_comment": {
+                        "type": "string",
+                        "merge_strategy": "replace",
+                    },
+                    "email_status": {"type": "string", "merge_strategy": "replace"},
+                }
+            },
+            "output_schema": {
+                "type": "object",
+                "properties": {
+                    "summary": {"type": "string"},
+                    "email_status": {"type": "string"},
+                },
+                "required": ["summary", "email_status"],
+            },
+            "node_defs": [
+                {
+                    "name": "drive_list_files",
+                    "input_schema": {
+                        "type": "object",
+                        "properties": {"folder_id": {"type": "string"}},
+                        "required": ["folder_id"],
+                    },
+                    "output_schema": {
+                        "type": "object",
+                        "properties": {"documents": {"type": "array"}},
+                        "required": ["documents"],
+                    },
+                    "outcomes": ["ok"],
+                },
+                {
+                    "name": "summarize_document",
+                    "input_schema": {
+                        "type": "object",
+                        "properties": {"document": {"type": "string"}},
+                        "required": ["document"],
+                    },
+                    "output_schema": {
+                        "type": "object",
+                        "properties": {"item_summary": {"type": "string"}},
+                        "required": ["item_summary"],
+                    },
+                    "outcomes": ["ok"],
+                },
+                {
+                    "name": "combine_summaries",
+                    "input_schema": {
+                        "type": "object",
+                        "properties": {"item_summaries": {"type": "array"}},
+                        "required": ["item_summaries"],
+                    },
+                    "output_schema": {
+                        "type": "object",
+                        "properties": {"summary": {"type": "string"}},
+                        "required": ["summary"],
+                    },
+                    "outcomes": ["ok"],
+                },
+                {
+                    "name": "send_email",
+                    "input_schema": {
+                        "type": "object",
+                        "properties": {"summary": {"type": "string"}},
+                        "required": ["summary"],
+                    },
+                    "output_schema": {
+                        "type": "object",
+                        "properties": {"email_status": {"type": "string"}},
+                        "required": ["email_status"],
+                    },
+                    "outcomes": ["sent"],
+                },
+                {
+                    "name": "mark_email_skipped",
+                    "input_schema": {
+                        "type": "object",
+                        "properties": {},
+                    },
+                    "output_schema": {
+                        "type": "object",
+                        "properties": {"email_status": {"type": "string"}},
+                        "required": ["email_status"],
+                    },
+                    "outcomes": ["ok"],
+                },
+            ],
+            "start": "list_files",
+            "nodes": [
+                {
+                    "id": "list_files",
+                    "type": "node",
+                    "node": "drive_list_files",
+                    "desc": "List files from a Google Drive folder",
+                    "in_map": {"input.folder_id": "folder_id"},
+                    "out_map": {"documents": "state.documents"},
+                },
+                {
+                    "id": "summarize_each",
+                    "type": "foreach",
+                    "over": "state.documents",
+                    "as": "document",
+                    "mode": "serial",
+                    "on_item_error": "fail",
+                },
+                {
+                    "id": "summarize_one",
+                    "type": "node",
+                    "node": "summarize_document",
+                    "desc": "Summarize one document",
+                    "in_map": {"context.document": "document"},
+                    "out_map": {"item_summary": "state.item_summaries"},
+                },
+                {
+                    "id": "combine_summaries",
+                    "type": "node",
+                    "node": "combine_summaries",
+                    "desc": "Combine item summaries into one final summary",
+                    "in_map": {"state.item_summaries": "item_summaries"},
+                    "out_map": {"summary": "state.summary"},
+                },
+                {
+                    "id": "should_email",
+                    "type": "condition",
+                    "check": {
+                        "op": "eq",
+                        "left": {"path": "state.should_email"},
+                        "right": {"value": True},
+                    },
+                },
+                {
+                    "id": "send_email",
+                    "type": "node",
+                    "node": "send_email",
+                    "desc": "Send the summary by email",
+                    "in_map": {"state.summary": "summary"},
+                    "out_map": {"email_status": "state.email_status"},
+                },
+                {
+                    "id": "approve_email",
+                    "type": "interrupt",
+                    "kind": "approval",
+                    "request_map": {
+                        "state.summary": "summary",
+                        "input.folder_id": "folder_id",
+                    },
+                    "out_map": {
+                        "approved": "state.approved",
+                        "comment": "state.approval_comment",
+                    },
+                    "outcomes": ["submitted", "cancelled"],
+                },
+                {
+                    "id": "skip_email",
+                    "type": "node",
+                    "node": "mark_email_skipped",
+                    "desc": "Record that email delivery was skipped",
+                    "out_map": {"email_status": "state.email_status"},
+                },
+            ],
+            "edges": [
+                {"from": "list_files", "outcome": "ok", "to": "summarize_each"},
+                {"from": "summarize_each", "outcome": "loop", "to": "summarize_one"},
+                {
+                    "from": "summarize_each",
+                    "outcome": "done",
+                    "to": "combine_summaries",
+                },
+                {"from": "summarize_one", "outcome": "ok", "to": END},
+                {"from": "combine_summaries", "outcome": "ok", "to": "should_email"},
+                {"from": "should_email", "outcome": "true", "to": "approve_email"},
+                {"from": "should_email", "outcome": "false", "to": "skip_email"},
+                {"from": "approve_email", "outcome": "submitted", "to": "send_email"},
+                {"from": "approve_email", "outcome": "cancelled", "to": "skip_email"},
+                {"from": "send_email", "outcome": "sent", "to": END},
+                {"from": "skip_email", "outcome": "ok", "to": END},
+            ],
+        }
+    )
+
+
+def drive_list_files(
+    payload: dict[str, object], ctx: RuntimeContext
+) -> dict[str, object]:
+    folder_id = payload["folder_id"]
+    return {
+        "outcome": "ok",
+        "output": {
+            "documents": [
+                f"{folder_id}/meeting-notes.md",
+                f"{folder_id}/weekly-report.md",
+            ]
+        },
+    }
+
+
+def summarize_documents(
+    payload: dict[str, object], ctx: RuntimeContext
+) -> dict[str, object]:
+    document = payload["document"]
+    return {
+        "outcome": "ok",
+        "output": {"item_summary": f"Summary of {document}"},
+    }
+
+
+def combine_summaries(
+    payload: dict[str, object], ctx: RuntimeContext
+) -> dict[str, object]:
+    raw_item_summaries = cast(list[object], payload["item_summaries"])
+    item_summaries = [str(item) for item in raw_item_summaries]
+    return {
+        "outcome": "ok",
+        "output": {"summary": " | ".join(item_summaries)},
+    }
+
+
+def send_email(payload: dict[str, object], ctx: RuntimeContext) -> dict[str, object]:
+    return {
+        "outcome": "sent",
+        "output": {"email_status": f"sent: {payload['summary']}"},
+    }
+
+
+def mark_email_skipped(
+    payload: dict[str, object], ctx: RuntimeContext
+) -> dict[str, object]:
+    return {
+        "outcome": "ok",
+        "output": {"email_status": "skipped"},
+    }
+
+
+def build_demo_registry() -> dict[str, DemoHandler]:
+    return {
+        "drive_list_files": drive_list_files,
+        "summarize_document": summarize_documents,
+        "combine_summaries": combine_summaries,
+        "send_email": send_email,
+        "mark_email_skipped": mark_email_skipped,
+    }

src/wf_core/errors.py

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

src/wf_core/flow_ops.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+from typing import Any
+
+from .model import Workflow
+from .run_state import (
+    ExecutionFrame,
+    FrameStatus,
+    RunState,
+    RunStatus,
+    StepExecutionResult,
+    TraceEntry,
+)
+from .schema_tools import validate_payload_against_schema
+from .state_ops import project_output
+from .tokens import END
+
+
+def append_trace(
+    run: RunState,
+    *,
+    frame_id: str,
+    node_id: str,
+    step_type: str,
+    resolved_input: dict[str, Any],
+    outcome: str,
+    next_node_id: str,
+    output: dict[str, Any],
+    state_changes: dict[str, Any],
+) -> None:
+    run.trace.append(
+        TraceEntry(
+            frame_id=frame_id,
+            node_id=node_id,
+            step_type=step_type,
+            resolved_input=resolved_input,
+            outcome=outcome,
+            next_node_id=next_node_id,
+            output=output,
+            state_changes=state_changes,
+        )
+    )
+
+
+def append_step_result_trace(
+    run: RunState,
+    *,
+    frame_id: str,
+    node_id: str,
+    step_type: str,
+    next_node_id: str,
+    result: StepExecutionResult,
+) -> None:
+    append_trace(
+        run,
+        frame_id=frame_id,
+        node_id=node_id,
+        step_type=step_type,
+        resolved_input=result.resolved_input,
+        outcome=result.outcome,
+        next_node_id=next_node_id,
+        output=result.output,
+        state_changes=result.state_changes,
+    )
+
+
+def advance_frame(
+    run: RunState,
+    frame: ExecutionFrame,
+    *,
+    outcome: str,
+    next_node_id: str,
+) -> None:
+    frame.prior_outcome = outcome
+    frame.activated_incoming_edge = frame.node_id
+    frame.node_id = next_node_id
+    if next_node_id == END:
+        frame.status = FrameStatus.COMPLETED
+        frame.finished_at_node_id = END
+    else:
+        frame.finished_at_node_id = None
+    run.sync_from_current_frame()
+
+
+def finalize_run(workflow: Workflow, run: RunState) -> RunState:
+    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

src/wf_core/foreach_ops.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+from .conditions import safe_resolve_path
+from .errors import WorkflowExecutionError
+from .flow_ops import advance_frame, append_step_result_trace
+from .frame_ops import frame_context_values
+from .model import ForeachNode, Workflow
+from .run_state import ExecutionFrame, FrameStatus, RunState, StepExecutionResult
+from .workflow_index import WorkflowIndex
+
+
+def step_foreach(
+    workflow: Workflow,
+    run: RunState,
+    step: ForeachNode,
+    index: WorkflowIndex,
+) -> RunState:
+    if step.mode != "serial":
+        raise WorkflowExecutionError(
+            "parallel foreach execution is not implemented yet"
+        )
+
+    frame = run.current_frame()
+    progress_map = frame.metadata.setdefault("foreach_progress", {})
+    progress = progress_map.setdefault(step.id, {"index": 0})
+
+    iterable = safe_resolve_path(
+        step.over,
+        state=run.state,
+        workflow_input=run.workflow_input,
+        context=frame_context_values(frame),
+    )
+    if not isinstance(iterable, list):
+        raise WorkflowExecutionError(
+            f"foreach source {step.over!r} must resolve to a list"
+        )
+
+    loop_index = progress["index"]
+    if loop_index >= len(iterable):
+        outcome = "done"
+        next_node_id = index.next_node_id(frame.node_id, outcome)
+        append_step_result_trace(
+            run,
+            frame_id=frame.id,
+            node_id=frame.node_id,
+            step_type=step.type,
+            next_node_id=next_node_id,
+            result=StepExecutionResult(
+                outcome=outcome,
+                resolved_input={"count": len(iterable), "index": loop_index},
+                output={},
+                state_changes={},
+            ),
+        )
+        advance_frame(run, frame, outcome=outcome, next_node_id=next_node_id)
+        return run
+
+    loop_start = index.next_node_id(frame.node_id, "loop")
+
+    item = iterable[loop_index]
+    progress["index"] = loop_index + 1
+    child_id = f"{frame.id}:{step.id}:{loop_index}"
+    child_metadata = {
+        "foreach_node_id": step.id,
+        "loop_index": loop_index,
+        "loop_item": item,
+        "loop_alias": step.as_,
+    }
+    run.frames[child_id] = ExecutionFrame(
+        id=child_id,
+        kind="foreach_iteration",
+        node_id=loop_start,
+        status=FrameStatus.PENDING,
+        parent_frame_id=frame.id,
+        metadata=child_metadata,
+    )
+    append_step_result_trace(
+        run,
+        frame_id=frame.id,
+        node_id=frame.node_id,
+        step_type=step.type,
+        next_node_id=loop_start,
+        result=StepExecutionResult(
+            outcome="loop",
+            resolved_input={"item": item, "index": loop_index},
+            output={},
+            state_changes={},
+        ),
+    )
+    run.current_frame_id = child_id
+    run.sync_from_current_frame()
+    return run

src/wf_core/frame_ops.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from .run_state import ExecutionFrame, FrameStatus, RunState
+from .tokens import END
+
+
+def collapse_completed_frames(run: RunState) -> None:
+    while run.current_frame_id is not None:
+        frame = run.current_frame()
+        if frame.node_id == END and frame.status != FrameStatus.COMPLETED:
+            frame.status = FrameStatus.COMPLETED
+            frame.finished_at_node_id = END
+        if frame.status != FrameStatus.COMPLETED or frame.parent_frame_id is None:
+            run.sync_from_current_frame()
+            return
+        run.current_frame_id = frame.parent_frame_id
+        parent = run.current_frame()
+        if parent.status == FrameStatus.PENDING:
+            parent.status = FrameStatus.RUNNING
+        run.sync_from_current_frame()
+
+
+def frame_context_values(frame: ExecutionFrame) -> dict[str, object | None]:
+    context: dict[str, object | None] = {
+        "prior_outcome": frame.prior_outcome,
+        "activated_incoming_edge": frame.activated_incoming_edge,
+    }
+    if frame.kind == "foreach_iteration":
+        loop_item = frame.metadata.get("loop_item")
+        loop_index = frame.metadata.get("loop_index")
+        loop_alias = frame.metadata.get("loop_alias")
+        context["loop_item"] = loop_item
+        context["loop_index"] = loop_index
+        if isinstance(loop_alias, str) and loop_alias:
+            context[loop_alias] = loop_item
+    return context

src/wf_core/interrupt_ops.py

@@ -0,0 +1,88 @@
+from __future__ import annotations
+
+from typing import Any
+
+from .conditions import safe_resolve_path
+from .errors import WorkflowExecutionError
+from .flow_ops import advance_frame, append_step_result_trace
+from .model import InterruptNode, Workflow
+from .run_state import InterruptRequest, RunState, StepExecutionResult
+from .state_ops import apply_mapped_state
+from .workflow_index import WorkflowIndex
+
+
+def build_interrupt_request(
+    node: InterruptNode,
+    *,
+    frame_id: str,
+    state: dict[str, Any],
+    workflow_input: dict[str, Any],
+    context: dict[str, Any],
+) -> InterruptRequest:
+    payload = {
+        payload_field: safe_resolve_path(
+            source_path,
+            state=state,
+            workflow_input=workflow_input,
+            context=context,
+        )
+        for source_path, payload_field in node.request_map.items()
+    }
+    return InterruptRequest(
+        id=f"interrupt:{node.id}",
+        frame_id=frame_id,
+        node_id=node.id,
+        kind=node.kind,
+        payload=payload,
+    )
+
+
+def resume_interrupt(
+    workflow: Workflow,
+    run: RunState,
+    *,
+    index: WorkflowIndex,
+    resume_payload: dict[str, Any],
+    resume_outcome: str,
+) -> None:
+    if run.current_frame_id is None:
+        raise WorkflowExecutionError("interrupted run has no current frame")
+    if run.current_node_id is None:
+        raise WorkflowExecutionError("interrupted run has no current node")
+    if run.interrupt is None:
+        raise WorkflowExecutionError("run is interrupted but has no interrupt request")
+
+    frame = run.current_frame()
+    step = index.nodes_by_id[frame.node_id]
+    if not isinstance(step, InterruptNode):
+        raise WorkflowExecutionError(
+            f"interrupted run expected interrupt node, got {step.type!r}"
+        )
+    if resume_outcome not in step.outcomes:
+        raise WorkflowExecutionError(
+            f"interrupt node {step.id!r} does not declare resume outcome {resume_outcome!r}"
+        )
+
+    state_changes = apply_mapped_state(
+        workflow,
+        resume_payload,
+        step.out_map,
+        run.state,
+        missing_field_message="interrupt resume payload is missing required field {field}",
+    )
+    next_node_id = index.next_node_id(frame.node_id, resume_outcome)
+    append_step_result_trace(
+        run,
+        frame_id=frame.id,
+        node_id=frame.node_id,
+        step_type=step.type,
+        next_node_id=next_node_id,
+        result=StepExecutionResult(
+            outcome=resume_outcome,
+            resolved_input=resume_payload,
+            output=resume_payload,
+            state_changes=state_changes,
+        ),
+    )
+    run.interrupt = None
+    advance_frame(run, frame, outcome=resume_outcome, next_node_id=next_node_id)

src/wf_core/model.py

@@ -0,0 +1,157 @@
+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(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"),
+]
+
+
+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):
+        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))

src/wf_core/node_exec.py

@@ -0,0 +1,150 @@
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable, Mapping
+from typing import Any, cast
+
+from .conditions import safe_resolve_path
+from .errors import WorkflowExecutionError
+from .frame_ops import frame_context_values
+from .model import NodeDef, NodeResult, NodeUse, Workflow
+from .run_state import RunState, RuntimeContext, StepExecutionResult
+from .schema_tools import validate_payload_against_schema
+from .state_ops import apply_output_map
+
+NodeHandler = Callable[[dict[str, Any], RuntimeContext], NodeResult | dict[str, Any]]
+AsyncNodeHandler = Callable[
+    [dict[str, Any], RuntimeContext],
+    Awaitable[NodeResult | dict[str, Any]] | NodeResult | dict[str, Any],
+]
+
+
+def _resolve_node_execution(
+    *,
+    workflow: Workflow,
+    run: RunState,
+    node: NodeUse,
+    node_def: NodeDef,
+) -> tuple[dict[str, Any], RuntimeContext]:
+    frame = run.current_frame()
+    context_values = frame_context_values(frame)
+    resolved_input = {
+        destination_field: safe_resolve_path(
+            source_path,
+            state=run.state,
+            workflow_input=run.workflow_input,
+            context=context_values,
+        )
+        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,
+        frame_id=frame.id,
+        prior_outcome=frame.prior_outcome,
+        activated_incoming_edge=frame.activated_incoming_edge,
+        metadata=dict(frame.metadata),
+    )
+    return resolved_input, context
+
+
+def _finalize_node_execution(
+    *,
+    workflow: Workflow,
+    run: RunState,
+    node: NodeUse,
+    node_def: NodeDef,
+    resolved_input: dict[str, Any],
+    raw_result: NodeResult | dict[str, Any],
+) -> StepExecutionResult:
+    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}"
+    )
+    state_changes = apply_output_map(workflow, node, result.output, run.state)
+    return StepExecutionResult(
+        outcome=result.outcome,
+        resolved_input=resolved_input,
+        output=result.output,
+        state_changes=state_changes,
+    )
+
+
+def execute_node_use(
+    workflow: Workflow,
+    run: RunState,
+    node: NodeUse,
+    node_def: NodeDef,
+    registry: Mapping[str, NodeHandler],
+) -> StepExecutionResult:
+    handler = registry.get(node.node)
+    if handler is None:
+        raise WorkflowExecutionError(
+            f"no handler registered for node def {node.node!r}"
+        )
+
+    resolved_input, context = _resolve_node_execution(
+        workflow=workflow,
+        run=run,
+        node=node,
+        node_def=node_def,
+    )
+    raw_result = handler(resolved_input, context)
+    return _finalize_node_execution(
+        workflow=workflow,
+        run=run,
+        node=node,
+        node_def=node_def,
+        resolved_input=resolved_input,
+        raw_result=raw_result,
+    )
+
+
+async def execute_node_use_async(
+    workflow: Workflow,
+    run: RunState,
+    node: NodeUse,
+    node_def: NodeDef,
+    registry: Mapping[str, AsyncNodeHandler],
+) -> StepExecutionResult:
+    handler = registry.get(node.node)
+    if handler is None:
+        raise WorkflowExecutionError(
+            f"no handler registered for node def {node.node!r}"
+        )
+
+    resolved_input, context = _resolve_node_execution(
+        workflow=workflow,
+        run=run,
+        node=node,
+        node_def=node_def,
+    )
+    raw_or_awaitable = handler(resolved_input, context)
+    if isinstance(raw_or_awaitable, Awaitable):
+        raw_result = await raw_or_awaitable
+    else:
+        raw_result = raw_or_awaitable
+    return _finalize_node_execution(
+        workflow=workflow,
+        run=run,
+        node=node,
+        node_def=node_def,
+        resolved_input=resolved_input,
+        raw_result=cast(NodeResult | dict[str, Any], raw_result),
+    )
+
+
+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)

src/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

src/wf_core/run_factory.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from .model import Workflow
+from .run_state import ExecutionFrame, FrameStatus, RunState, RunStatus
+
+
+def create_run_state(workflow: Workflow, workflow_input: dict[str, object]) -> RunState:
+    run = RunState(
+        workflow_name=workflow.name,
+        status=RunStatus.PENDING,
+        workflow_input=dict(workflow_input),
+        state=dict(workflow_input),
+        frames={
+            "root": ExecutionFrame(
+                id="root",
+                kind="workflow",
+                node_id=workflow.start,
+                status=FrameStatus.PENDING,
+            )
+        },
+        current_frame_id="root",
+        current_node_id=workflow.start,
+    )
+    run.sync_from_current_frame()
+    return run

src/wf_core/run_state.py

@@ -0,0 +1,105 @@
+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"
+
+
+class FrameStatus(StrEnum):
+    PENDING = "pending"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    INTERRUPTED = "interrupted"
+
+
+@dataclass(slots=True)
+class ExecutionFrame:
+    id: str
+    kind: str
+    node_id: str
+    status: FrameStatus = FrameStatus.PENDING
+    parent_frame_id: str | None = None
+    prior_outcome: str | None = None
+    activated_incoming_edge: str | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+    finished_at_node_id: str | None = None
+
+
+@dataclass(slots=True)
+class RuntimeContext:
+    current_node_id: str
+    frame_id: str = "root"
+    retry_count: int = 0
+    prior_outcome: str | None = None
+    activated_incoming_edge: str | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(slots=True)
+class TraceEntry:
+    frame_id: str
+    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 StepExecutionResult:
+    outcome: str
+    resolved_input: dict[str, Any] = field(default_factory=dict)
+    output: dict[str, Any] = field(default_factory=dict)
+    state_changes: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(slots=True)
+class InterruptRequest:
+    id: str
+    frame_id: str
+    node_id: str
+    kind: str
+    payload: dict[str, Any] = field(default_factory=dict)
+    resumable: bool = True
+
+
+@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)
+    frames: dict[str, ExecutionFrame] = field(default_factory=dict)
+    current_frame_id: str | None = None
+    current_node_id: str | None = None
+    prior_outcome: str | None = None
+    activated_incoming_edge: str | None = None
+    error: str | None = None
+    interrupt: InterruptRequest | None = None
+
+    def current_frame(self) -> ExecutionFrame:
+        if self.current_frame_id is None:
+            raise ValueError("run has no current frame")
+        return self.frames[self.current_frame_id]
+
+    def sync_from_current_frame(self) -> None:
+        frame = self.current_frame()
+        self.current_node_id = frame.node_id
+        self.prior_outcome = frame.prior_outcome
+        self.activated_incoming_edge = frame.activated_incoming_edge
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)

src/wf_core/runtime.py

@@ -0,0 +1,343 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from .errors import WorkflowExecutionError
+from .foreach_ops import step_foreach
+from .flow_ops import advance_frame, append_step_result_trace, finalize_run
+from .frame_ops import collapse_completed_frames
+from .interrupt_ops import resume_interrupt
+from .model import (
+    ConditionNode,
+    ForeachNode,
+    InterruptNode,
+    JoinNode,
+    NodeUse,
+    Workflow,
+)
+from .node_exec import (
+    AsyncNodeHandler,
+    NodeHandler,
+    coerce_node_result,
+    execute_node_use,
+    execute_node_use_async,
+)
+from .run_factory import create_run_state
+from .run_state import (
+    FrameStatus,
+    RunState,
+    RunStatus,
+)
+from .schema_tools import validate_payload_against_schema
+from .step_handlers import (
+    handle_condition_step,
+    handle_interrupt_step,
+    handle_join_step,
+)
+from .tokens import END
+from .workflow_index import WorkflowIndex, build_workflow_index
+
+__all__ = [
+    "AsyncNodeHandler",
+    "NodeHandler",
+    "coerce_node_result",
+    "execute_workflow_async",
+    "execute_workflow",
+    "resume_workflow_async",
+    "resume_workflow",
+    "step_workflow_async",
+    "step_workflow",
+]
+
+
+def _prepare_new_run(workflow: Workflow, workflow_input: dict[str, Any]) -> RunState:
+    run = create_run_state(workflow, workflow_input)
+    workflow.validate_structure().raise_for_errors()
+    validate_payload_against_schema(
+        workflow.input_schema, workflow_input, "workflow input"
+    )
+    return run
+
+
+def _prepare_resume(
+    workflow: Workflow,
+    run: RunState,
+    *,
+    resume_payload: dict[str, Any] | None,
+    resume_outcome: str,
+) -> WorkflowIndex | None:
+    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_frame_id is None:
+        raise WorkflowExecutionError("run has no current frame")
+    collapse_completed_frames(run)
+
+    if run.current_node_id is None:
+        raise WorkflowExecutionError("run has no current node")
+
+    if run.status == RunStatus.COMPLETED:
+        return None
+
+    index = build_workflow_index(workflow)
+
+    if run.status == RunStatus.INTERRUPTED:
+        if resume_payload is None:
+            return None
+        resume_interrupt(
+            workflow,
+            run,
+            index=index,
+            resume_payload=resume_payload,
+            resume_outcome=resume_outcome,
+        )
+        collapse_completed_frames(run)
+        if run.current_node_id == END:
+            return None
+
+    run.status = RunStatus.RUNNING
+    run.error = None
+    run.current_frame().status = FrameStatus.RUNNING
+    return index
+
+
+def _prepare_step(
+    workflow: Workflow,
+    run: RunState,
+    index: WorkflowIndex | None,
+) -> tuple[WorkflowIndex, object] | None:
+    if run.current_frame_id is None:
+        raise WorkflowExecutionError("run has no current frame")
+
+    collapse_completed_frames(run)
+    if run.current_node_id is None or run.current_node_id == END:
+        return None
+    if run.status == RunStatus.INTERRUPTED:
+        return None
+
+    if run.status == RunStatus.PENDING:
+        run.status = RunStatus.RUNNING
+    run.error = None
+
+    resolved_index = index or build_workflow_index(workflow)
+    frame = run.current_frame()
+    if frame.status == FrameStatus.PENDING:
+        frame.status = FrameStatus.RUNNING
+    step = resolved_index.nodes_by_id[frame.node_id]
+    return resolved_index, step
+
+
+def _complete_step(
+    *,
+    run: RunState,
+    index: WorkflowIndex,
+    outcome: str,
+    frame_id: str,
+    node_id: str,
+    step_type: str,
+    step_result: Any,
+) -> RunState:
+    next_node_id = index.next_node_id(node_id, outcome)
+
+    append_step_result_trace(
+        run,
+        frame_id=frame_id,
+        node_id=node_id,
+        step_type=step_type,
+        next_node_id=next_node_id,
+        result=step_result,
+    )
+    advance_frame(
+        run,
+        run.frames[frame_id],
+        outcome=outcome,
+        next_node_id=next_node_id,
+    )
+    return run
+
+
+def execute_workflow(
+    workflow: Workflow,
+    workflow_input: dict[str, Any],
+    registry: Mapping[str, NodeHandler],
+) -> RunState:
+    run = create_run_state(workflow, workflow_input)
+
+    try:
+        run = _prepare_new_run(workflow, workflow_input)
+        return resume_workflow(workflow, run, registry)
+    except Exception as exc:
+        run.status = RunStatus.FAILED
+        run.error = str(exc)
+        raise
+
+
+async def execute_workflow_async(
+    workflow: Workflow,
+    workflow_input: dict[str, Any],
+    registry: Mapping[str, AsyncNodeHandler],
+) -> RunState:
+    run = create_run_state(workflow, workflow_input)
+
+    try:
+        run = _prepare_new_run(workflow, workflow_input)
+        return await resume_workflow_async(workflow, run, registry)
+    except Exception as exc:
+        run.status = RunStatus.FAILED
+        run.error = str(exc)
+        raise
+
+
+def resume_workflow(
+    workflow: Workflow,
+    run: RunState,
+    registry: Mapping[str, NodeHandler],
+    *,
+    resume_payload: dict[str, Any] | None = None,
+    resume_outcome: str = "submitted",
+) -> RunState:
+    index = _prepare_resume(
+        workflow,
+        run,
+        resume_payload=resume_payload,
+        resume_outcome=resume_outcome,
+    )
+    if index is None:
+        if run.current_node_id == END:
+            return finalize_run(workflow, run)
+        return run
+
+    while True:
+        collapse_completed_frames(run)
+        if run.current_node_id == END:
+            break
+        step_workflow(
+            workflow,
+            run,
+            registry,
+            index=index,
+        )
+        if run.status == RunStatus.INTERRUPTED:
+            return run
+
+    return finalize_run(workflow, run)
+
+
+async def resume_workflow_async(
+    workflow: Workflow,
+    run: RunState,
+    registry: Mapping[str, AsyncNodeHandler],
+    *,
+    resume_payload: dict[str, Any] | None = None,
+    resume_outcome: str = "submitted",
+) -> RunState:
+    index = _prepare_resume(
+        workflow,
+        run,
+        resume_payload=resume_payload,
+        resume_outcome=resume_outcome,
+    )
+    if index is None:
+        if run.current_node_id == END:
+            return finalize_run(workflow, run)
+        return run
+
+    while True:
+        collapse_completed_frames(run)
+        if run.current_node_id == END:
+            break
+        await step_workflow_async(
+            workflow,
+            run,
+            registry,
+            index=index,
+        )
+        if run.status == RunStatus.INTERRUPTED:
+            return run
+
+    return finalize_run(workflow, run)
+
+
+def step_workflow(
+    workflow: Workflow,
+    run: RunState,
+    registry: Mapping[str, NodeHandler],
+    *,
+    index: WorkflowIndex | None = None,
+) -> RunState:
+    prepared = _prepare_step(workflow, run, index)
+    if prepared is None:
+        return run
+    index, step = prepared
+    frame = run.current_frame()
+
+    if isinstance(step, NodeUse):
+        node_def = index.node_defs[step.node]
+        step_result = execute_node_use(workflow, run, step, node_def, registry)
+    elif isinstance(step, ConditionNode):
+        step_result = handle_condition_step(run, step)
+    elif isinstance(step, JoinNode):
+        step_result = handle_join_step()
+    elif isinstance(step, InterruptNode):
+        return handle_interrupt_step(run, step)
+    elif isinstance(step, ForeachNode):
+        return step_foreach(workflow, run, step, index)
+    else:
+        raise WorkflowExecutionError(
+            f"unsupported step type {getattr(step, 'type', type(step).__name__)!r}"
+        )
+
+    return _complete_step(
+        run=run,
+        index=index,
+        outcome=step_result.outcome,
+        frame_id=frame.id,
+        node_id=frame.node_id,
+        step_type=step.type,
+        step_result=step_result,
+    )
+
+
+async def step_workflow_async(
+    workflow: Workflow,
+    run: RunState,
+    registry: Mapping[str, AsyncNodeHandler],
+    *,
+    index: WorkflowIndex | None = None,
+) -> RunState:
+    prepared = _prepare_step(workflow, run, index)
+    if prepared is None:
+        return run
+    index, step = prepared
+    frame = run.current_frame()
+
+    if isinstance(step, NodeUse):
+        node_def = index.node_defs[step.node]
+        step_result = await execute_node_use_async(
+            workflow, run, step, node_def, registry
+        )
+    elif isinstance(step, ConditionNode):
+        step_result = handle_condition_step(run, step)
+    elif isinstance(step, JoinNode):
+        step_result = handle_join_step()
+    elif isinstance(step, InterruptNode):
+        return handle_interrupt_step(run, step)
+    elif isinstance(step, ForeachNode):
+        return step_foreach(workflow, run, step, index)
+    else:
+        raise WorkflowExecutionError(
+            f"unsupported step type {getattr(step, 'type', type(step).__name__)!r}"
+        )
+
+    return _complete_step(
+        run=run,
+        index=index,
+        outcome=step_result.outcome,
+        frame_id=frame.id,
+        node_id=frame.node_id,
+        step_type=step.type,
+        step_result=step_result,
+    )

src/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}"
+                )

src/wf_core/state_ops.py

@@ -0,0 +1,119 @@
+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]:
+    return apply_mapped_state(
+        workflow,
+        node_output,
+        node.out_map,
+        state,
+        missing_field_message=f"node {node.id!r} did not return required mapped field {{field}}",
+    )
+
+
+def apply_mapped_state(
+    workflow: Workflow,
+    source_data: dict[str, Any],
+    mapping: dict[str, str],
+    state: dict[str, Any],
+    *,
+    missing_field_message: str,
+) -> dict[str, Any]:
+    state_changes: dict[str, Any] = {}
+    for source_field, destination_path in mapping.items():
+        if source_field not in source_data:
+            raise WorkflowExecutionError(
+                missing_field_message.format(field=repr(source_field))
+            )
+        value = source_data[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

src/wf_core/step_handlers.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+from .conditions import eval_condition
+from .flow_ops import append_trace
+from .frame_ops import frame_context_values
+from .interrupt_ops import build_interrupt_request
+from .model import ConditionNode, InterruptNode
+from .run_state import FrameStatus, RunState, RunStatus, StepExecutionResult
+
+
+def handle_condition_step(
+    run: RunState,
+    step: ConditionNode,
+) -> StepExecutionResult:
+    frame = run.current_frame()
+    predicate = eval_condition(
+        step.check,
+        run.state,
+        run.workflow_input,
+        frame.prior_outcome,
+    )
+    outcome = "true" if predicate else "false"
+    return StepExecutionResult(
+        outcome=outcome,
+        resolved_input={},
+        output={"predicate": predicate},
+        state_changes={},
+    )
+
+
+def handle_join_step() -> StepExecutionResult:
+    return StepExecutionResult(
+        outcome="done",
+        resolved_input={},
+        output={},
+        state_changes={},
+    )
+
+
+def handle_interrupt_step(
+    run: RunState,
+    step: InterruptNode,
+) -> RunState:
+    frame = run.current_frame()
+    interrupt_request = build_interrupt_request(
+        step,
+        frame_id=frame.id,
+        state=run.state,
+        workflow_input=run.workflow_input,
+        context=frame_context_values(frame),
+    )
+    run.interrupt = interrupt_request
+    run.status = RunStatus.INTERRUPTED
+    frame.status = FrameStatus.INTERRUPTED
+    append_trace(
+        run,
+        frame_id=frame.id,
+        node_id=frame.node_id,
+        step_type=step.type,
+        resolved_input=interrupt_request.payload,
+        outcome="interrupt",
+        next_node_id=frame.node_id,
+        output=interrupt_request.payload,
+        state_changes={},
+    )
+    return run

src/wf_core/tokens.py

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

src/wf_core/validate.py

@@ -0,0 +1,418 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from enum import StrEnum
+
+from .model import (
+    BinaryCondition,
+    Condition,
+    ConditionNode,
+    Edge,
+    ExistsCondition,
+    ForeachNode,
+    InterruptNode,
+    LiteralOperand,
+    NodeDef,
+    NodeUse,
+    NotCondition,
+    PathOperand,
+    Step,
+    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"
+    INVALID_INTERRUPT_SOURCE = "invalid_interrupt_source"
+    INVALID_INTERRUPT_DESTINATION = "invalid_interrupt_destination"
+
+
+@dataclass(slots=True)
+class ValidationIssue:
+    code: ValidationIssueCode
+    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, 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.code}] {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(
+                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: 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(
+                ValidationIssueCode.DUPLICATE_NODE_ID,
+                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
+            )
+        elif isinstance(node, InterruptNode):
+            _validate_interrupt_node(
+                node, index, report, state_root_fields, input_root_fields
+            )
+
+    if workflow.start not in nodes_by_id:
+        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()
+
+    for index, edge in enumerate(workflow.edges):
+        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}",
+            )
+        else:
+            edge_keys.add(edge_key)
+
+        source = nodes_by_id.get(edge.from_)
+        if source is None:
+            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 and edge.to not in nodes_by_id:
+            report.add(
+                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)
+
+    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(
+                ValidationIssueCode.MISSING_OUTCOME_EDGE,
+                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(
+            ValidationIssueCode.UNKNOWN_NODE_DEF,
+            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(
+                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, allow_context=True
+        ):
+            report.add(
+                ValidationIssueCode.INVALID_SOURCE_PATH,
+                f"nodes[{index}].in_map[{source_path!r}]",
+                "source path must start with input., state., or context. and reference a declared root field when applicable",
+            )
+
+    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):
+            report.add(
+                ValidationIssueCode.INVALID_DESTINATION_PATH,
+                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(
+            ValidationIssueCode.EMPTY_CONDITION_ARGS,
+            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(
+            ValidationIssueCode.INVALID_FOREACH_SOURCE,
+            f"nodes[{index}].over",
+            "foreach source path must start with input. or state. and reference a declared root field",
+        )
+
+
+def _validate_interrupt_node(
+    node: InterruptNode,
+    index: int,
+    report: ValidationReport,
+    state_root_fields: set[str],
+    input_root_fields: set[str],
+) -> None:
+    for source_path, payload_field in node.request_map.items():
+        if not payload_field:
+            report.add(
+                ValidationIssueCode.INVALID_INTERRUPT_SOURCE,
+                f"nodes[{index}].request_map[{source_path!r}]",
+                "interrupt request payload field must not be empty",
+            )
+        if not is_valid_source_path(source_path, state_root_fields, input_root_fields):
+            report.add(
+                ValidationIssueCode.INVALID_INTERRUPT_SOURCE,
+                f"nodes[{index}].request_map[{source_path!r}]",
+                "interrupt request source must start with input. or state. and reference a declared root field",
+            )
+
+    for resume_field, destination_path in node.out_map.items():
+        if not resume_field:
+            report.add(
+                ValidationIssueCode.INVALID_INTERRUPT_DESTINATION,
+                f"nodes[{index}].out_map[{resume_field!r}]",
+                "interrupt resume field must not be empty",
+            )
+        if not is_valid_destination_path(destination_path):
+            report.add(
+                ValidationIssueCode.INVALID_INTERRUPT_DESTINATION,
+                f"nodes[{index}].out_map[{resume_field!r}]",
+                "interrupt resume destination must start with state.",
+            )
+
+
+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_source_path(
+            condition.path,
+            state_root_fields,
+            input_root_fields,
+            allow_context=True,
+        ):
+            report.add(
+                ValidationIssueCode.INVALID_CONDITION_PATH,
+                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_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]:
+    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 {"loop", "done"}
+    if step.type == "join":
+        return {"done"}
+    if isinstance(step, InterruptNode):
+        return set(step.outcomes)
+    return set()
+
+
+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:
+            continue
+        adjacency.setdefault(edge.from_, []).append(edge.to)
+
+    seen: set[str] = 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

src/wf_core/workflow_index.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from .errors import WorkflowExecutionError
+from .model import NodeDef, Workflow
+
+
+@dataclass(slots=True)
+class WorkflowIndex:
+    node_defs: dict[str, NodeDef]
+    nodes_by_id: dict[str, Any]
+    edge_map: dict[tuple[str, str], str]
+
+    def next_node_id(self, node_id: str, outcome: str) -> str:
+        next_node_id = self.edge_map.get((node_id, outcome))
+        if next_node_id is None:
+            raise WorkflowExecutionError(
+                f"no edge found for node {node_id!r} and outcome {outcome!r}"
+            )
+        return next_node_id
+
+
+def build_workflow_index(workflow: Workflow) -> WorkflowIndex:
+    return WorkflowIndex(
+        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},
+    )