### Install and Sync Dependencies with uv

Source: https://github.com/jacksonvieiracs/twpm/blob/master/CLAUDE.md

Commands to manage Python dependencies using uv. 'uv sync' installs all dependencies, while 'uv sync --no-dev' installs only production dependencies.

```bash
uv sync

uv sync --no-dev
```

--------------------------------

### Run Twpm CLI Example (Shell)

Source: https://github.com/jacksonvieiracs/twpm/blob/master/README.md

Demonstrates how to clone the Twpm repository and run an example CLI application using uv. It shows two variations: a standard CLI workflow and a quiz workflow.

```sh
git clone https://github.com/jacksonvieiracs/twpm
cd twpm
```

```sh
uv run python3 examples/cli/main.py
```

```sh
uv run python3 examples/cli/main.py --quiz
```

--------------------------------

### Install Twpm Package (Shell)

Source: https://github.com/jacksonvieiracs/twpm/blob/master/README.md

Provides commands to install the Twpm library using either uv or pip package managers. These commands are executed in a shell environment.

```sh
uv add twpm
```

```sh
pip install twpm
```

--------------------------------

### Control Orchestrator Workflow Lifecycle in Python

Source: https://context7.com/jacksonvieiracs/twpm/llms.txt

Illustrates managing the workflow execution lifecycle using the Orchestrator's `start`, `process`, and `reset` methods, along with state inspection properties like `current_node` and `is_ended`. This example uses the `Chain` builder to construct the workflow and demonstrates how to handle user input and inspect collected data.

```python
import asyncio
from twpm import Chain, Orchestrator
from twpm.core.base import NodeStatus
from twpm.core.container import Container, ServiceScope
from twpm.core.primitives.question import QuestionNode
from twpm.core.primitives.display_message import DisplayMessageNode
from twpm.core.depedencies import Output


class ConsoleOutput:
    async def send_text(self, text: str):
        print(text, end=" ")


async def main():
    workflow = (
        Chain()
        .add(DisplayMessageNode("Start workflow", key="start"))
        .add(QuestionNode("Your name", key="name"))
        .add(QuestionNode("Your age", key="age"))
        .add(DisplayMessageNode("Complete!", key="end"))
        .build()
    )

    container = Container()
    container.register(Output, lambda: ConsoleOutput(), ServiceScope.SINGLETON)

    orchestrator = Orchestrator(container)

    # Start workflow
    orchestrator.start(workflow)
    print(f"[INFO] Started: {orchestrator.current_node.key}")

    # Process until input needed
    await orchestrator.process()

    # Inspect current state
    current = orchestrator.current_node
    print(f"\n[INFO] Current node: {current.key}")
    print(f"[INFO] Status: {current.status.value}")
    print(f"[INFO] Ended: {orchestrator.is_ended}")

    # Continue with user input
    while not orchestrator.is_ended:
        user_input = await asyncio.to_thread(input, ">> ")
        await orchestrator.process(input=user_input.strip())

        if orchestrator.current_node:
            print(f"\n[INFO] Now at: {orchestrator.current_node.key}")

    print("\n[INFO] Workflow completed")
    print(f"[INFO] Collected data: {orchestrator._data.data}")

    # Reset to beginning
    orchestrator.reset()
    print(f"[INFO] Reset to: {orchestrator.current_node.key}")
    print(f"[INFO] Ended status: {orchestrator.is_ended}")


asyncio.run(main())
```

--------------------------------

### Orchestrator Core Methods

Source: https://github.com/jacksonvieiracs/twpm/blob/master/CLAUDE.md

Demonstrates key methods of the Orchestrator class, responsible for managing workflow execution. Includes starting the workflow, processing nodes, and resetting the orchestrator.

```python
class Orchestrator:
    def __init__(self):
        self.current_node = None
        self.shared_data = ListData()
        self.state = "DEFAULT"

    def start(self, start_node: Node):
        self.current_node = start_node

    async def process(self, input=None):
        # ... execution logic ...
        pass

    def reset(self):
        self.current_node = None
        self.shared_data.clear()
        self.state = "DEFAULT"
```

--------------------------------

### Node Base Class Example

Source: https://github.com/jacksonvieiracs/twpm/blob/master/CLAUDE.md

Illustrates the abstract base class for workflow nodes in twpm. Nodes form a linked list and implement an asynchronous execute method, wrapped by a safe execution decorator.

```python
class Node:
    def __init__(self):
        self.next = None
        self.previous = None

    @safe_execute()
    async def execute(self, data: ListData) -> NodeResult:
        raise NotImplementedError
```

--------------------------------

### Implement Conditional Branching in Python Workflows

Source: https://context7.com/jacksonvieiracs/twpm/llms.txt

Shows how to create dynamic workflows that branch based on runtime data using `ConditionalNode`. This involves defining a condition function that evaluates input data and then configuring the `ConditionalNode` to direct the workflow to different branches (represented by `DisplayMessageNode` in this example) based on the condition's result.

```python
import asyncio
from twpm import Chain, Orchestrator, ConditionalNode
from twpm.core.base import ListData
from twpm.core.container import Container, ServiceScope
from twpm.core.primitives.display_message import DisplayMessageNode
from twpm.core.primitives.question import QuestionNode
from twpm.core.primitives.pool import PoolNode
from twpm.core.depedencies import Output


class ConsoleOutput:
    async def send_text(self, text: str):
        print(text, end=" ")


async def main():
    # Create branch nodes
    premium_message = DisplayMessageNode(
        "🌟 Premium features unlocked! Enjoy advanced analytics.",
        key="premium_path"
    )

    basic_message = DisplayMessageNode(
        "📊 Basic features available. Upgrade for more!",
        key="basic_path"
    )

    # Create conditional node
    condition_node = ConditionalNode()

    # Define condition function
    def is_premium_user(data: ListData) -> bool:
        """Return True for premium path, False for basic path"""
        plan = data.get("plan", "")
        return plan == "Premium"

    # Configure condition node with branches
    condition_node.set_condition(
        condition_func=is_premium_user,
        true_node=premium_message,   # Shown if premium
        false_node=basic_message      # Shown if not premium
    )

    # Build workflow
    workflow = (
        Chain()
        .add(DisplayMessageNode("=== Account Setup ===", key="welcome"))
        .add(QuestionNode("Enter your name", key="name"))
        .add(PoolNode(
            question="Select your plan",
            options=["Basic", "Premium"],
            key="plan"
        ))
        .add(condition_node)
        .add(DisplayMessageNode("\n✅ Setup complete!", key="complete"))
        .build()
    )

    container = Container()
    container.register(Output, lambda: ConsoleOutput(), ServiceScope.SINGLETON)

    orchestrator = Orchestrator(container)
    orchestrator.start(workflow)
    await orchestrator.process()

    while not orchestrator.is_ended:
        user_input = await asyncio.to_thread(input, ">> ")
        await orchestrator.process(input=user_input.strip())


asyncio.run(main())

```

--------------------------------

### Build Package Distributions with uv

Source: https://github.com/jacksonvieiracs/twpm/blob/master/CLAUDE.md

Bash commands to build Python package distributions using uv. Supports building for distribution and building only the wheel.

```bash
uv build

uv build --wheel
```

--------------------------------

### Build and Run a Basic Twpm Workflow (Python)

Source: https://github.com/jacksonvieiracs/twpm/blob/master/README.md

Demonstrates how to construct a simple workflow using Twpm's Chain, Orchestrator, and primitive nodes (DisplayMessageNode, QuestionNode, SummaryNode). It includes setting up a container for dependency injection and running the workflow asynchronously.

```python
import asyncio
from twpm.core import Chain, Orchestrator
from twpm.core.container import Container, ServiceScope
from twpm.core.primitives import DisplayMessageNode, QuestionNode, SummaryNode
from twpm.core.depedencies import Output

workflow = (
    Chain()
    .add(DisplayMessageNode("Welcome!", key="welcome"))
    .add(QuestionNode("What's your name?", key="name"))
    .add(QuestionNode("What's your email?", key="email"))
    .add(SummaryNode(
        title="Thank you!",
        fields=[("Name", "name"), ("Email", "email")],
        key="summary"
    ))
    .build()
)

class ConsoleOutput:
    async def send_text(self, text: str):
        print(text, end=" ")

async def main():
    container = Container()
    container.registry(Output, lambda: ConsoleOutput(), ServiceScope.SINGLETON)
    orchestrator = Orchestrator(container)
    
    orchestrator.start(workflow)
    await orchestrator.process()

asyncio.run(main())
```

--------------------------------

### Build Basic Interactive Workflow with twpm

Source: https://context7.com/jacksonvieiracs/twpm/llms.txt

Demonstrates creating a multi-step user input workflow using twpm primitives. It includes welcome messages, questions with keys for data retrieval, and a summary node to display collected information. Requires 'twpm' library and an Output implementation like ConsoleOutput.

```python
import asyncio
from twpm import Chain, Orchestrator
from twpm.core.container import Container, ServiceScope
from twpm.core.primitives.display_message import DisplayMessageNode
from twpm.core.primitives.question import QuestionNode
from twpm.core.primitives.summary import SummaryNode
from twpm.core.depedencies import Output


# Define output handler (required for node I/O)
class ConsoleOutput:
    async def send_text(self, text: str):
        print(text, end=" ")


async def main():
    # Build workflow chain
    workflow = (
        Chain()
        .add(DisplayMessageNode("=== Welcome to Registration ===", key="welcome"))
        .add(QuestionNode("What's your name?", key="name"))
        .add(QuestionNode("What's your email?", key="email"))
        .add(QuestionNode("What's your company?", key="company"))
        .add(SummaryNode(
            title="Registration Complete!",
            fields=[
                ("Name", "name"),
                ("Email", "email"),
                ("Company", "company")
            ],
            key="summary"
        ))
        .build()
    )

    # Setup dependency injection
    container = Container()
    container.register(Output, lambda: ConsoleOutput(), ServiceScope.SINGLETON)

    # Initialize and run orchestrator
    orchestrator = Orchestrator(container)
    orchestrator.start(workflow)
    await orchestrator.process()

    # Interactive input loop
    while not orchestrator.is_ended:
        user_input = await asyncio.to_thread(input, ">> ")
        await orchestrator.process(input=user_input.strip())


asyncio.run(main())

```

--------------------------------

### Lint and Format Code with ruff using uv

Source: https://github.com/jacksonvieiracs/twpm/blob/master/CLAUDE.md

Commands to check code quality and formatting using ruff via uv. Includes options for checking, auto-fixing linting issues, and formatting code.

```bash
uv run ruff check .

uv run ruff check --fix .

uv run ruff format .

uv run ruff format --check .
```

--------------------------------

### Define Custom Services in Python

Source: https://context7.com/jacksonvieiracs/twpm/llms.txt

Demonstrates defining and registering custom services like Database and EmailService within a dependency injection container. Services can have different scopes (SINGLETON, TRANSIENT) affecting their instantiation. This snippet shows how to build a workflow and use the orchestrator to process it, injecting dependencies from the container.

```python
import asyncio


class Database:
    async def save(self, data: dict):
        print(f"\n[DB] Saving: {data}")


class EmailService:
    async def send(self, to: str, message: str):
        print(f"\n[EMAIL] To: {to}, Message: {message}")


async def main():
    # Create and configure container
    container = Container()

    # Register services with different scopes
    container.register(
        Output,
        lambda: ConsoleOutput(),
        ServiceScope.SINGLETON  # Single instance shared across workflow
    )

    container.register(
        Database,
        lambda: Database(),
        ServiceScope.SINGLETON
    )

    container.register(
        EmailService,
        lambda: EmailService(),
        ServiceScope.TRANSIENT  # New instance each time
    )

    # Build workflow
    workflow = (
        Chain()
        .add(DisplayMessageNode("=== Service Demo ===", key="welcome"))
        .add(QuestionNode("Your name", key="name"))
        .add(QuestionNode("Your email", key="email"))
        .build()
    )

    # Orchestrator uses container to inject dependencies
    orchestrator = Orchestrator(container)
    orchestrator.start(workflow)
    await orchestrator.process()

    while not orchestrator.is_ended:
        user_input = await asyncio.to_thread(input, ">> ")
        await orchestrator.process(input=user_input.strip())

    # Access services after workflow completion
    db = container.resolve(Database)
    email = container.resolve(EmailService)

    await db.save(orchestrator._data.data)
    await email.send(
        orchestrator._data.get("email", ""),
        f"Welcome, {orchestrator._data.get('name', '')}!"
    )


asyncio.run(main())

```

--------------------------------

### Run Tests with pytest using uv

Source: https://github.com/jacksonvieiracs/twpm/blob/master/CLAUDE.md

Bash commands to execute tests using pytest via uv. Supports running all tests, verbose output, specific files, or single test cases.

```bash
uv run pytest

uv run pytest -v

uv run pytest tests/core/test_orchestrator.py

uv run pytest tests/core/test_orchestrator.py::TestOrchestrator::test_single_node_execution
```

--------------------------------

### Dependency Injection with TWPM Container

Source: https://context7.com/jacksonvieiracs/twpm/llms.txt

Illustrates how to leverage the TWPM Container for managing dependencies via dependency injection. It shows registering different implementations of the Output service (ConsoleOutput and LogFileOutput) and how they can be resolved.

```python
import asyncio
from twpm import Chain, Orchestrator
from twpm.core.container import Container, ServiceScope
from twpm.core.primitives.question import QuestionNode
from twpm.core.primitives.display_message import DisplayMessageNode
from twpm.core.depedencies import Output


# Define different output implementations
class ConsoleOutput:
    async def send_text(self, text: str):
        print(f"[CONSOLE] {text}", end=" ")


class LogFileOutput:
    def __init__(self):
        self.logs = []

    async def send_text(self, text: str):
        self.logs.append(text)
        print(f"[LOGGED] {text}", end=" ")

```

--------------------------------

### Build Simple Linear Workflows with chain() in Python

Source: https://context7.com/jacksonvieiracs/twpm/llms.txt

Demonstrates using the convenience `chain()` function to quickly create linear workflows without the builder pattern. This is useful for straightforward, sequential tasks. It requires importing necessary components from the twpm library and defining a basic output handler.

```python
import asyncio
from twpm import chain, Orchestrator
from twpm.core.container import Container, ServiceScope
from twpm.core.primitives.display_message import DisplayMessageNode
from twpm.core.primitives.question import QuestionNode
from twpm.core.primitives.summary import SummaryNode
from twpm.core.depedencies import Output


class ConsoleOutput:
    async def send_text(self, text: str):
        print(text, end=" ")


async def main():
    # Use chain() function for simple linear workflows
    workflow = chain(
        DisplayMessageNode("Quick survey!", key="start"),
        QuestionNode("Your name", key="name"),
        QuestionNode("Your city", key="city"),
        SummaryNode(
            title="Thank you!",
            fields=[("Name", "name"), ("City", "city")],
            key="end"
        )
    )

    container = Container()
    container.register(Output, lambda: ConsoleOutput(), ServiceScope.SINGLETON)

    orchestrator = Orchestrator(container)
    orchestrator.start(workflow)
    await orchestrator.process()

    while not orchestrator.is_ended:
        user_input = await asyncio.to_thread(input, ">> ")
        await orchestrator.process(input=user_input.strip())


asyncio.run(main())
```

--------------------------------

### Using Multiple Choice Questions in Python with TWPM

Source: https://context7.com/jacksonvieiracs/twpm/llms.txt

Demonstrates how to use PoolNode in Python with the TWPM library to present users with enumerated options for validated selection. It requires importing necessary components from twpm and asyncio, defining a custom output class, and building a workflow chain with DisplayMessageNode, PoolNode, and SummaryNode.

```python
import asyncio
from twpm import Chain, Orchestrator
from twpm.core.container import Container, ServiceScope
from twpm.core.primitives.pool import PoolNode
from twpm.core.primitives.display_message import DisplayMessageNode
from twpm.core.primitives.summary import SummaryNode
from twpm.core.depedencies import Output


class ConsoleOutput:
    async def send_text(self, text: str):
        print(text, end=" ")


async def main():
    workflow = (
        Chain()
        .add(DisplayMessageNode("=== Product Survey ===\n", key="welcome")),
        .add(PoolNode(
            question="What's your favorite color?",
            options=["Red", "Blue", "Green", "Yellow"],
            key="color"
        ))
        .add(PoolNode(
            question="Which size do you prefer?",
            options=["Small", "Medium", "Large", "Extra Large"],
            key="size"
        ))
        .add(PoolNode(
            question="How often do you shop?",
            options=["Daily", "Weekly", "Monthly", "Rarely"],
            key="frequency"
        ))
        .add(SummaryNode(
            title="\n=== Your Preferences ===",
            fields=[
                ("Color", "color"),
                ("Size", "size"),
                ("Shopping Frequency", "frequency")
            ],
            key="summary"
        ))
        .build()
    )

    container = Container()
    container.register(Output, lambda: ConsoleOutput(), ServiceScope.SINGLETON)

    orchestrator = Orchestrator(container)
    orchestrator.start(workflow)
    await orchestrator.process()

    while not orchestrator.is_ended:
        user_input = await asyncio.to_thread(input, ">> ")
        await orchestrator.process(input=user_input.strip())


asyncio.run(main())

```

--------------------------------

### TaskNode and ConditionalNode Implementations

Source: https://github.com/jacksonvieiracs/twpm/blob/master/CLAUDE.md

Shows basic implementations of TaskNode and ConditionalNode. TaskNode executes async functions, while ConditionalNode uses Cursor.insert() for dynamic branching based on a condition.

```python
class TaskNode(Node):
    def __init__(self, task_func: Callable):
        super().__init__()
        self.task_func = task_func

    async def execute(self, data: ListData) -> NodeResult:
        return await self.task_func(data)

class ConditionalNode(Node):
    def __init__(self):
        super().__init__()
        self.condition_func = None
        self.true_node = None
        self.false_node = None

    def set_condition(self, condition_func, true_node, false_node):
        self.condition_func = condition_func
        self.true_node = true_node
        self.false_node = false_node

    async def execute(self, data: ListData) -> NodeResult:
        if self.condition_func(data):
            Cursor.insert(self, self.true_node)
        else:
            Cursor.insert(self, self.false_node)
        return NodeResult(success=True)
```

--------------------------------

### Define Custom Async Task Functions in Python

Source: https://context7.com/jacksonvieiracs/twpm/llms.txt

Demonstrates how to define asynchronous functions for common tasks such as validating email addresses, saving data to a database, and sending welcome emails. These functions are designed to be integrated into TWPM workflows. They utilize Python's `async`/`await` syntax and simulate I/O operations with `asyncio.sleep`.

```python
import asyncio
from typing import Dict, Any

# Assuming ListData is a type alias for Dict[str, Any]
ListData = Dict[str, Any]

async def validate_email(data: ListData) -> bool:
    """Task returns True for success, False for failure"""
    email = data.get("email", "")
    if "@" in email and "." in email:
        print(f"✅ Email validated: {email}")
        return True
    print(f"❌ Invalid email: {email}")
    return False


async def save_to_database(data: ListData) -> bool:
    """Simulate database save operation"""
    await asyncio.sleep(0.5)  # Simulate async I/O
    print(f"💾 Saved to database: {data.data}")
    return True


async def send_welcome_email(data: ListData) -> bool:
    """Simulate email sending"""
    email = data.get("email", "")
    name = data.get("name", "User")
    print(f"📧 Sending welcome email to {email} (Dear {name})")
    await asyncio.sleep(0.3)
    return True


class ConsoleOutput:
    async def send_text(self, text: str):
        print(text, end=" ")


async def main():
    from twpm.core.primitives.question import QuestionNode
    from twpm.core.node import TaskNode
    from twpm.core.chain import Chain
    from twpm.core.container import Container, ServiceScope
    from twpm.core.orchestrator import Orchestrator
    from twpm.core.depedencies import Output

    # Build workflow with task nodes
    workflow = (
        Chain()
        .add(QuestionNode("Enter your name", key="name"))
        .add(QuestionNode("Enter your email", key="email"))
        .add(TaskNode(validate_email))
        .add(TaskNode(save_to_database))
        .add(TaskNode(send_welcome_email))
        .build()
    )

    container = Container()
    container.register(Output, lambda: ConsoleOutput(), ServiceScope.SINGLETON)

    orchestrator = Orchestrator(container)
    orchestrator.start(workflow)
    await orchestrator.process()

    while not orchestrator.is_ended:
        user_input = await asyncio.to_thread(input, ">> ")
        await orchestrator.process(input=user_input.strip())


asyncio.run(main())

```

--------------------------------

### Create Custom Task Nodes with twpm

Source: https://context7.com/jacksonvieiracs/twpm/llms.txt

Shows how to create and use custom task nodes within a twpm workflow. These nodes execute asynchronous functions and can return success or failure statuses, integrating custom logic into the workflow. Requires 'twpm' library and a suitable Output implementation.

```python
import asyncio
from twpm import Chain, Orchestrator, TaskNode
from twpm.core.base import ListData
from twpm.core.container import Container, ServiceScope
from twpm.core.depedencies import Output



```

--------------------------------

### Adding Progress Tracking in Python with TWPM

Source: https://context7.com/jacksonvieiracs/twpm/llms.txt

Illustrates how to automatically inject progress tracking nodes after specific node types (QuestionNode, PoolNode) in Python using TWPM. This feature helps display workflow completion status. It involves defining progress fields, building a workflow with a `.with_progress()` method, and includes QuestionNode, PoolNode, and SummaryNode.

```python
import asyncio
from twpm import Chain, Orchestrator
from twpm.core.container import Container, ServiceScope
from twpm.core.primitives.question import QuestionNode
from twpm.core.primitives.pool import PoolNode
from twpm.core.primitives.display_message import DisplayMessageNode
from twpm.core.primitives.summary import SummaryNode
from twpm.core.depedencies import Output


class ConsoleOutput:
    async def send_text(self, text: str):
        print(text, end=" ")


async def main():
    # Define progress tracking fields
    progress_fields = [
        ("Full Name", "name"),
        ("Email Address", "email"),
        ("Phone Number", "phone"),
        ("Company Name", "company"),
        ("Job Title", "title")
    ]

    workflow = (
        Chain()
        .add(DisplayMessageNode("=== Professional Registration ===\n", key="welcome")),
        .add(QuestionNode("Full name", key="name")),
        .add(QuestionNode("Email address", key="email")),
        .add(QuestionNode("Phone number", key="phone")),
        .add(QuestionNode("Company name", key="company")),
        .add(PoolNode(
            question="Job title",
            options=["Engineer", "Manager", "Director", "Executive", "Other"],
            key="title"
        ))
        # Automatically insert progress tracking after each question/pool
        .with_progress(
            fields=progress_fields,
            after_each=(QuestionNode, PoolNode)  # Track after these node types
        ),
        .add(SummaryNode(
            title="\n=== Registration Complete ===",
            fields=progress_fields,
            key="final_summary"
        ))
        .build()
    )

    container = Container()
    container.register(Output, lambda: ConsoleOutput(), ServiceScope.SINGLETON)

    orchestrator = Orchestrator(container)
    orchestrator.start(workflow)
    await orchestrator.process()

    while not orchestrator.is_ended:
        user_input = await asyncio.to_thread(input, ">> ")
        await orchestrator.process(input=user_input.strip())


asyncio.run(main())

```

--------------------------------

### Safe Execute Decorator

Source: https://github.com/jacksonvieiracs/twpm/blob/master/CLAUDE.md

Demonstrates the @safe_execute() decorator. It wraps node execution, catching exceptions and converting them into NodeResult objects to prevent workflow interruption.

```python
from functools import wraps

def safe_execute():
    def decorator(func):
        @wraps(func)
        async def wrapper(self, data, *args, **kwargs):
            try:
                result = await func(self, data, *args, **kwargs)
                if not isinstance(result, NodeResult):
                    # Default to success if not NodeResult and no exception
                    return NodeResult(success=True, data=result if isinstance(result, dict) else {})
                return result
            except Exception as e:
                # Log error here
                print(f"Error in node execution: {e}")
                return NodeResult(success=False, message=str(e))
        return wrapper
    return decorator
```

--------------------------------

### Manipulate Linked Lists with Cursor in Python

Source: https://context7.com/jacksonvieiracs/twpm/llms.txt

Illustrates using Cursor utilities for advanced linked-list operations within TWPM workflows. This includes inserting nodes dynamically, retrieving ranges of nodes, finding nodes by type, and adding nodes after specific criteria. These operations allow for dynamic workflow modification during runtime.

```python
import asyncio
from twpm import Chain, Orchestrator, Node
from twpm.core.cursor import Cursor
from twpm.core.container import Container, ServiceScope
from twpm.core.primitives.display_message import DisplayMessageNode
from twpm.core.primitives.question import QuestionNode
from twpm.core.depedencies import Output


class ConsoleOutput:
    async def send_text(self, text: str):
        print(text, end=" ")


async def main():
    # Build initial workflow
    workflow = (
        Chain()
        .add(DisplayMessageNode("=== Step 1 ===", key="step1"))
        .add(QuestionNode("Your name", key="name"))
        .add(DisplayMessageNode("=== Step 3 ===", key="step3"))
        .build()
    )

    # Use Cursor to dynamically insert node
    step2_node = DisplayMessageNode("=== Step 2 (inserted) ===", key="step2")
    name_node = workflow.next  # Second node in chain
    Cursor.insert(name_node, step2_node)  # Insert after name question

    # Get range of nodes
    all_nodes = Cursor.get_range(begin=workflow, end=workflow.next.next.next)
    print(f"\n[INFO] Workflow has {len(all_nodes)} nodes")

    # Find specific node types
    questions = Cursor.find_by_type(
        begin=workflow,
        end=workflow.next.next.next,
        node_type=QuestionNode
    )
    print(f"[INFO] Found {len(questions)} question nodes\n")

    # Add nodes after each question dynamically
    def create_confirmation(node: Node) -> Node:
        return DisplayMessageNode("✅ Answer recorded", key=f"confirm_{node.key}")

    last_node = workflow
    while last_node.next:
        last_node = last_node.next

    Cursor.add_after_each(
        begin=workflow,
        end=last_node,
        node_factory=create_confirmation,
        filter_fn=lambda n: isinstance(n, QuestionNode)
    )

    # Execute workflow
    container = Container()
    container.register(Output, lambda: ConsoleOutput(), ServiceScope.SINGLETON)

    orchestrator = Orchestrator(container)
    orchestrator.start(workflow)
    await orchestrator.process()

    while not orchestrator.is_ended:
        user_input = await asyncio.to_thread(input, ">> ")
        await orchestrator.process(input=user_input.strip())


asyncio.run(main())

```

--------------------------------

### Create Custom Nodes: Email Validation and Calculation

Source: https://context7.com/jacksonvieiracs/twpm/llms.txt

Demonstrates how to extend the base Node class to create custom workflow components. The ValidatedEmailNode validates email format and retries on failure, while CalculationNode performs arithmetic operations on workflow data. Both utilize the @safe_execute decorator for error handling.

```python
import asyncio
from typing import override
from twpm import Chain, Orchestrator, Node
from twpm.core.base import ListData, NodeResult
from twpm.core.decorators import safe_execute
from twpm.core.container import Container, ServiceScope
from twpm.core.primitives.question import QuestionNode
from twpm.core.depedencies import Output


class ValidatedEmailNode(Node):
    """Custom node that validates email format and retries on failure"""

    def __init__(self, prompt: str, key: str):
        super().__init__()
        self.prompt = prompt
        self.key = key
        self._waiting_for_input = True

    @override
    @safe_execute()  # Required: wraps execution with error handling
    async def execute(self, data: ListData, output: Output) -> NodeResult:
        if self._waiting_for_input:
            await output.send_text(f"\n? {self.prompt}: ")
            self._waiting_for_input = False
            return NodeResult(
                success=True,
                data={},
                message="Awaiting email input",
                is_awaiting_input=True
            )

        # Validate email
        email = data.get("_user_input", "")
        if "@" in email and "." in email.split("@")[-1]:
            data[self.key] = email
            await output.send_text("✅ Email validated\n")
            return NodeResult(
                success=True,
                data={},
                message=f"Email validated: {email}"
            )

        # Invalid email - request again
        await output.send_text("❌ Invalid email format. Try again: ")
        return NodeResult(
            success=True,
            data={},
            message="Invalid email, awaiting retry",
            is_awaiting_input=True
        )


class CalculationNode(Node):
    """Custom node that performs calculations on workflow data"""

    def __init__(self, key: str):
        super().__init__()
        self.key = key

    @override
    @safe_execute()
    async def execute(self, data: ListData, output: Output) -> NodeResult:
        # Calculate total from collected data
        price = float(data.get("price", 0))
        quantity = int(data.get("quantity", 1))
        total = price * quantity

        data[self.key] = str(total)
        await output.send_text(f"\n💰 Total calculated: ${total:.2f}\n")

        return NodeResult(
            success=True,
            data={self.key: str(total)},
            message=f"Calculated total: {total}"
        )


class ConsoleOutput:
    async def send_text(self, text: str):
        print(text, end=" ")


async def main():
    workflow = (
        Chain()
        .add(QuestionNode("Product name", key="product"))
        .add(QuestionNode("Price per unit", key="price"))
        .add(QuestionNode("Quantity", key="quantity"))
        .add(CalculationNode(key="total"))
        .add(ValidatedEmailNode("Receipt email", key="email")),
    )
    .build()

    container = Container()
    container.register(Output, lambda: ConsoleOutput(), ServiceScope.SINGLETON)

    orchestrator = Orchestrator(container)
    orchestrator.start(workflow)
    await orchestrator.process()

    while not orchestrator.is_ended:
        user_input = await asyncio.to_thread(input, ">> ")
        await orchestrator.process(input=user_input.strip())


asyncio.run(main())

```

--------------------------------

### Workflow Pause and Resume Pattern

Source: https://github.com/jacksonvieiracs/twpm/blob/master/CLAUDE.md

Illustrates how nodes can pause workflow execution by returning `NodeResult(is_awaiting_input=True)`. The orchestrator handles pausing and resuming, passing user input via `ListData`.

```python
# In a Node's execute method:
if condition_for_input:
    return NodeResult(is_awaiting_input=True, message="Please provide input")

# In the Orchestrator's process method:
if node_result.is_awaiting_input:
    self.state = "AWAITING_INPUT"
    return

# When processing input:
if self.state == "AWAITING_INPUT":
    user_input = self.shared_data.get("_user_input")
    # ... node processes user_input ...
    self.state = "DEFAULT" # Reset state after processing input
```

--------------------------------

### Cursor Linked List Manipulation

Source: https://github.com/jacksonvieiracs/twpm/blob/master/CLAUDE.md

Provides the Cursor utility for modifying the linked list structure at runtime. The 'insert' method allows for dynamic insertion of nodes, crucial for conditional branching.

```python
class Cursor:
    @staticmethod
    def insert(target_node: Node, new_node: Node):
        new_node.next = target_node.next
        new_node.previous = target_node
        if target_node.next:
            target_node.next.previous = new_node
        target_node.next = new_node
```

--------------------------------

### ListData and NodeResult Models

Source: https://github.com/jacksonvieiracs/twpm/blob/master/CLAUDE.md

Defines the data structures used for passing information between nodes. ListData holds shared data, and NodeResult encapsulates the outcome of a node's execution.

```python
class ListData(dict):
    def get(self, key, default=None):
        return super().get(key, default)

    def update(self, data: dict):
        super().update(data)

    def has(self, key):
        return key in self

class NodeResult:
    def __init__(self, success: bool, data: dict = None, is_awaiting_input: bool = False, message: str = ""):
        self.success = success
        self.data = data if data is not None else {}
        self.is_awaiting_input = is_awaiting_input
        self.message = message
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.