Workflow States

# Workflow state schema — shared across all workflow nodes
from dataclasses import dataclass, field
from typing import Any, Optional
from datetime import datetime

@dataclass
class WorkflowState:
    """State that flows through the entire workflow."""

    # Identity
    workflow_id: str
    workflow_name: str
    started_at: str = field(default_factory=lambda: datetime.now().isoformat())

    # Input (set at the start, never changed)
    input: dict = field(default_factory=dict)

    # Intermediate data (written by each node)
    data: dict = field(default_factory=dict)

    # Execution tracking
    current_node: str = "START"
    completed_nodes: list = field(default_factory=list)
    status: str = "running"  # running | waiting | complete | failed | escalated

    # Error tracking
    errors: list = field(default_factory=list)

    # Output (set when workflow completes)
    output: Optional[Any] = None

    def write(self, key: str, value: Any):
        """Store a result from a completed node."""
        self.data[key] = value

    def read(self, key: str, default=None):
        """Read data written by a previous node."""
        return self.data.get(key, default)

    def mark_node_complete(self, node_id: str):
        self.completed_nodes.append(node_id)
        self.current_node = node_id

# Create and use workflow state
state = WorkflowState(
    workflow_id="WF-001",
    workflow_name="refund_processing",
    input={"order_id": "ORD-123", "reason": "product damaged"}
)

state.write("order_details", {"id": "ORD-123", "total": 49.99})
state.mark_node_complete("lookup_order")
print(f"Current: {state.current_node}")
print(f"Order total: {state.read('order_details')['total']}")