### Install aiida-workgraph using pip

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/installation.rst

Installs the latest stable version of aiida-workgraph and its dependencies from PyPI using the pip package manager. This is the recommended installation method.

```console
pip install aiida-workgraph
```

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

### Launch AiiDA-WorkGraph GUI

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/README.md

Provides instructions to launch the AiiDA-WorkGraph interactive graphical user interface (GUI). This involves installing an additional package (`aiida-gui-workgraph`) and then running a command to start the GUI server. The GUI allows for visualization, monitoring, and debugging of workflows.

```console
pip install aiida-gui-workgraph
aiida-gui start
```

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

### Setup AiiDA Environment

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/README.md

Sets up the AiiDA environment, which is a prerequisite for using AiiDA-WorkGraph. 'verdi presto' is a quick setup command, while 'verdi quicksetup' provides a more detailed interactive setup.

```console
verdi presto  # Or 'verdi quicksetup' for a detailed setup
```

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

### Install aiida-gui-workgraph using pip

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/installation.rst

Installs the graphical user interface (GUI) for aiida-workgraph. This GUI is an experimental feature currently under active development and should be used with caution in production environments.

```console
pip install aiida-gui-workgraph
```

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

### Install aiida-workgraph from source using pip

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/installation.rst

Installs the aiida-workgraph package from its source code repository cloned from GitHub. This method is useful for development or when needing the latest unreleased features. The '-e' flag installs in editable mode.

```console
git clone https://github.com/aiidateam/aiida-workgraph
cd aiida-workgraph
pip install -e .
```

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

### Example Help String for 'workgraph graph kill'

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/howto/cli.rst

Provides an example of the help string output for the 'workgraph graph kill' command. This includes the command's usage, a brief description, and a list of available options with their descriptions and default values.

```text
Usage: workgraph graph kill [OPTIONS] [PROCESSES]...

    Kill running processes.

Options:
    -t, --timeout FLOAT  Time in seconds to wait for a response before timing
                         out.  [default: 5.0]
    --wait / --no-wait   Wait for the action to be completed otherwise return as
                         soon as it's scheduled.
    -h, --help           Show this message and exit.
```

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

### Install Dependencies and Pre-commit Hooks

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/development/contributing.rst

Installs the project in editable mode with test and pre-commit dependencies, and then installs the pre-commit hooks to ensure code quality before commits.

```console
pip install -e .[tests, pre-commit]
pre-commit install
```

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

### Install Documentation Dependencies and Build Docs

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/development/contributing.rst

Installs Sphinx documentation requirements and builds the HTML documentation. This process involves installing the 'docs' extra dependency and the requirements from docs/requirements.txt, then using make to build the HTML output.

```console
pip install .[docs]
pip install -r docs/requirements.txt
make -C docs html
<YOUR-BROWSER> docs/build/html/index.html
```

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

### Accessing Help Strings in AiiDA-WorkGraph CLI

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/howto/cli.rst

Demonstrates how to retrieve help information for any 'workgraph' subcommand by appending the '--help' option. This is useful for understanding available commands, options, and their usage. It shows an example for 'workgraph graph kill'.

```bash
workgraph graph kill --help
```

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

### Run a Workflow and Print Results (Python)

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/README.md

Shows how to execute a composed AiiDA-WorkGraph workflow and retrieve its final result. This involves loading an AiiDA profile, calling the `.run()` method on the graph function with specific inputs, and then printing the returned result. The example demonstrates a simple addition and multiplication workflow.

```python
from aiida import load_profile

# Load your AiiDA profile
load_profile()

# Build and run the workflow
results = add_multiply.run(x=2, y=3, z=4)

# Print the final result
print(f"✅ Result: {results}")
# Expected output: ✅ Result: 20
```

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

### Run a Daemon Worker in aiida-workgraph

Source: https://github.com/aiidateam/aiida-workgraph/wiki/WorkGraph-scheduler

This Python code defines a command-line interface command to start a single daemon worker. It requires the database environment and a running broker. The worker is started in the foreground.

```python
import click
from aiida.cmdline.params import decorators
from aiida.common.folders import defaults


@click.command()
@decorators.with_dbenv()
@decorators.requires_broker
def worker():
    """Run a single daemon worker in the current interpreter."""
    from aiida.engine.daemon.worker import start_daemon_worker

    start_daemon_worker(foreground=True)
```

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

### Parallel Execution: Scatter-Gather Pattern

Source: https://context7.com/aiidateam/aiida-workgraph/llms.txt

Shows how to perform parallel execution using the scatter-gather pattern. Tasks without explicit dependencies run in parallel. This example squares numbers in parallel and then sums the results using a `gather_sum` task. Imports include `typing`, `aiida.load_profile`, `aiida_workgraph.task`, `namespace`, and `dynamic`.

```python
import typing as t
from aiida import load_profile
from aiida_workgraph import task, namespace, dynamic

load_profile()

@task
def square(x: int) -> int:
    """Square a number."""
    return x * x

@task
def gather_sum(data: t.Annotated[dict, dynamic(int)]) -> int:
    """Sum all values in a dictionary."""
    return sum(data.values())

@task.graph
def parallel_square_sum(numbers: list) -> int:
    """Square numbers in parallel and sum results."""
    # Scatter: create parallel square tasks
    squares = {}
    for i, num in enumerate(numbers):
        squares[f'num_{i}'] = square(x=num).result

    # Gather: sum all results
    return gather_sum(data=squares).result

# Execute parallel workflow
wg = parallel_square_sum.build(numbers=[1, 2, 3, 4, 5])
wg.run()

print(f"Sum of squares: {wg.outputs.result.value}")  # Output: 55
print(f"Number of tasks: {len(wg.tasks)}")
```

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

### Launch and Continue Aiida Process from Checkpoint

Source: https://github.com/aiidateam/aiida-workgraph/wiki/Control-an-AiiDA-process-precisely

This Python code snippet shows how to continue a previously saved Aiida process using its primary key (pk). It utilizes the Aiida manager to get the process controller and then calls the continue_process method. It is crucial not to call continue_process multiple times for the same process.

```python
from aiida.manage import get_manager
process_controller = get_manager().get_process_controller()
pk = 4070
process_controller.continue_process(pk)
```

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

### Visual Graph Construction with If Zone

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/overview.rst

Demonstrates building a workflow using explicit zones for control flow. This example uses an 'If' zone to conditionally execute tasks based on a condition (x > 0). It showcases how to integrate tasks within these zones.

```python
from aiida_workgraph import If

with If(x > 0) as if_zone:
    add1 = add(x=x, y=y).result
    add2 = add(x=add1, y=z).result
```

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

### Shell Job: Execute Shell Commands

Source: https://context7.com/aiidateam/aiida-workgraph/llms.txt

Demonstrates how to execute shell commands as workflow tasks using the `shelljob` function. This example calculates `(x + y) * z` by chaining two `expr` shell commands. It includes a custom `result_parser` function to extract data from stdout and uses `aiida.orm` for data types. Requires `aiida.load_profile`, `aiida_workgraph.task`, `shelljob`, and `dynamic`.

```python
import typing as t
from aiida import load_profile, orm
from aiida_workgraph import task, shelljob, dynamic

load_profile()

def result_parser(dirpath):
    """Parse stdout to extract numeric result."""
    return {'result': orm.Int((dirpath / 'stdout').read_text().strip())}

@task.graph
def shell_calculation(x: int, y: int, z: int) -> int:
    """Calculate (x + y) * z using shell commands."""
    # First shell job: addition
    sum_result = shelljob(
        command='expr',
        arguments=['{x}', '+', '{y}'],
        nodes={'x': x, 'y': y},
        parser=result_parser,
        parser_outputs=['result']
    ).result

    # Second shell job: multiplication
    product = shelljob(
        command='expr',
        arguments=['{sum}', '*', '{z}'],
        nodes={'sum': sum_result, 'z': z},
        parser=result_parser,
        parser_outputs=['result']
    ).result

    return product

# Execute shell-based workflow
wg = shell_calculation.build(x=5, y=3, z=2)
wg.run()

print(f"Shell calculation result: {wg.outputs.result.value}")  # Output: 16
print(f"Workflow state: {wg.state}")
```

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

### Error Handling in Workflows (Console Output)

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/overview.rst

Presents a console output example demonstrating a workflow's error handling capabilities. It shows a task failing with a specific exit code (410) but the workflow recovering and completing successfully due to a handler.

```console
WorkGraph<handling_error_negative_number><968> Finished [0]
    ├── ArithmeticAddCalculation<971> Finished [410]
    └── ArithmeticAddCalculation<977> Finished [0]
```

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

### Event-Driven Task Monitoring

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/overview.rst

Shows an example of event-driven execution using a time-based monitor. The '@task.monitor' decorator allows tasks to be triggered by external events, integrating with real-time data streams. The monitor checks if a specified time has passed.

```python
@task.monitor
def time_monitor(time):
    """This task waits until a specified time has passed."""
    import datetime
    return datetime.datetime.now() > time.value
```

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

### Initialize Aiida Workgraph Node Editor

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/_static/first_workflow.html

Sets up the Rete.js node editor instance with various plugins for rendering, area management, connections, and auto-arrangement. It defines sockets and presets for a classic node editor appearance. Returns the editor instance, area, and layout function.

```javascript
async function createEditor(container) {
  const socket = new ClassicPreset.Socket("socket");
  const editor = new NodeEditor(Schemes);
  const area = new AreaPlugin(container);
  const connection = new ConnectionPlugin();
  const render = new ReactPlugin({ createRoot });
  const arrange = new AutoArrangePlugin();
  AreaExtensions.selectableNodes(area, AreaExtensions.selector(), {
    accumulating: AreaExtensions.accumulateOnCtrl(),
  });
  render.addPreset(Presets.classic.setup());
  connection.addPreset(ConnectionPresets.classic.setup());
  const applier = new ArrangeAppliers.TransitionApplier({
    duration: 500,
    timingFunction: (t) => t,
    async onTick() {
      await AreaExtensions.zoomAt(area, editor.getNodes());
    }
  });
  arrange.addPreset(ArrangePresets.classic.setup());
  editor.use(area);
  editor.use(render);
  editor.use(arrange);
  AreaExtensions.simpleNodesOrder(area);
  async function layout(animate) {
    await arrange.layout({ applier: animate ? applier : undefined });
    AreaExtensions.zoomAt(area, editor.getNodes());
  }
  const nodeMap = {};
  editor.nodeMap = nodeMap;
  return { editor: editor, area: area, layout: layout, destroy: () => area.destroy() };
}
```

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

### Initialize Rete.js Editor with Plugins (Vanilla JS)

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/_static/first_workflow.html

Initializes the Rete.js editor with essential plugins like AreaPlugin, ConnectionPlugin, ReactPlugin, and AutoArrangePlugin. It sets up the rendering context and prepares the editor for dynamic node and connection management.

```javascript
const { useState, useRef, useEffect } = React;
const { createRoot } = ReactDOM;
const { NodeEditor, ClassicPreset } = Rete;
const { AreaPlugin, AreaExtensions } = ReteAreaPlugin;
const { ConnectionPlugin, Presets: ConnectionPresets } = ReteConnectionPlugin;
const { ReactPlugin, Presets } = ReteReactPlugin;
const { AutoArrangePlugin, Presets: ArrangePresets,
ArrangeAppliers } = ReteAutoArrangePlugin;
const { RenderUtils } = ReteRenderUtils;
const styled = window.styled;

// Define Schemes to use in vanilla JS
const Schemes = {
    Node: ClassicPreset.Node,
    Connection: ClassicPreset.Connection
};

class Node extends ClassicPreset.Node {
    width = 180;
    height = 100;
}

class Connection extends ClassicPreset.Connection {}
```

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

### Run Workflow with Success and Failure/Recovery Cases (Python)

Source: https://context7.com/aiidateam/aiida-workgraph/llms.txt

Demonstrates building and running a resilient workflow in AiiDA-WorkGraph. It shows a successful execution path and a path with failure and subsequent recovery, illustrating how outputs are handled in both scenarios. Assumes `resilient_workflow` is a pre-defined AiiDA-WorkGraph workflow.

```python
wg_success = resilient_workflow.build(input_value=10, fail_flag=False)
wg_success.run()
print(f"Success case: {wg_success.outputs.result.value}")

wg_fail = resilient_workflow.build(input_value=10, fail_flag=True)
wg_fail.run()
print(f"Recovery case: {wg_fail.outputs.result.value}")
```

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

### Create and Save Process Node Checkpoint in Aiida

Source: https://github.com/aiidateam/aiida-workgraph/wiki/Control-an-AiiDA-process-precisely

This Python code snippet demonstrates how to create a task, specifically a MultiplyAddWorkChain, and save its checkpoint. It requires the Aiida library and loads a profile and code. The inputs for the work chain are defined, and then the checkpoint is saved.

```python
from aiida import load_profile
from aiida.workflows.arithmetic.multiply_add import MultiplyAddWorkChain
from aiida.orm import Int, load_code

load_profile()
code = load_code("add@localhost")
p = MultiplyAddWorkChain(inputs={"x":Int(1), "y": Int(2), "z": Int(3), "code": code})
p.runner.persister.save_checkpoint(p)
```

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

### Submit and Manage Workflows Asynchronously

Source: https://context7.com/aiidateam/aiida-workgraph/llms.txt

Shows how to submit workflows for asynchronous execution using `wg.submit()` and manage their lifecycle. It covers waiting for completion with timeouts and reloading workflows from the database using `WorkGraph.load()`. Dependencies are aiida, aiida_workgraph, and time.

```python
from aiida import load_profile
from aiida_workgraph import task, WorkGraph

load_profile()

@task
def long_running_task(duration):
    import time
    time.sleep(duration)
    return f"Completed after {duration} seconds"

@task.graph
def long_workflow(num_tasks, task_duration):
    """Create workflow with multiple long tasks."""
    results = []
    for i in range(num_tasks):
        result = long_running_task(duration=task_duration).result
        results.append(result)
    return results

# Submit workflow for async execution
wg = long_workflow.build(num_tasks=3, task_duration=2)
wg.submit(wait=False)  # Submit and return immediately

print(f"Workflow submitted with PK: {wg.pk}")
print(f"Initial state: {wg.state}")

# Wait for completion with timeout
try:
    wg.wait(timeout=30, interval=2)
    print(f"Final state: {wg.state}")
    print(f"Results: {wg.outputs.result.value}")
except TimeoutError as e:
    print(f"Workflow timeout: {e}")

# Reload workflow from database
wg_loaded = WorkGraph.load(wg.pk)
print(f"Loaded workflow state: {wg_loaded.state}")
```

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

### Conditional Calculation with Different Operations

Source: https://context7.com/aiidateam/aiida-workgraph/llms.txt

Demonstrates building and running a conditional calculation workflow that can perform either addition or multiplication based on the 'operation' parameter. It utilizes the `conditional_calculation.build` function and `.run()` method. Outputs are accessed via `.outputs.result.value`.

```python
wg_add = conditional_calculation.build(operation="add", x=10, y=5)
wg_add.run()
print(f"Addition result: {wg_add.outputs.result.value}")  # Output: 15

wg_mult = conditional_calculation.build(operation="multiply", x=10, y=5)
wg_mult.run()
print(f"Multiplication result: {wg_mult.outputs.result.value}")  # Output: 50
```

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

### Define CalcFunctions and WorkFunctions with @task Decorators

Source: https://context7.com/aiidateam/aiida-workgraph/llms.txt

Illustrates the use of @task.calcfunction and @task.workfunction decorators to define AiiDA calculation and workflow functions. These decorators provide enhanced provenance tracking for computations and orchestrations. Dependencies include aiida and aiida_workgraph.

```python
from aiida import load_profile
from aiida_workgraph import task

load_profile()

@task.calcfunction
def expensive_calculation(x, y):
    """CPU-intensive calculation tracked as CalcFunction."""
    result = 0
    for i in range(x):
        result += i * y
    return result

@task.workfunction
def orchestrate_calculations(data_list):
    """Coordinate multiple calculations as WorkFunction."""
    results = []
    for item in data_list:
        calc_result = expensive_calculation(x=item, y=2).result
        results.append(calc_result)
    return sum(results)

@task.graph
def computation_pipeline(inputs):
    """Pipeline combining calc and work functions."""
    intermediate = expensive_calculation(x=inputs['a'], y=inputs['b']).result
    final = orchestrate_calculations(data_list=[intermediate, 10, 20]).result
    return final

wg = computation_pipeline.build(inputs={'a': 5, 'b': 3})
wg.run()

print(f"Pipeline result: {wg.outputs.result.value}")
print(f"Process type: {wg.process.process_type}")
```

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

### Low-Level Node-Graph Programming

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/overview.rst

Illustrates constructing a workflow programmatically by defining tasks and linking their inputs and outputs. This method provides maximum control for advanced use cases like programmatically generating graph structures.

```python
from aiida_workgraph import WorkGraph

wg = WorkGraph("test_add_multiply")
wg.add_task(add, name="add1")
wg.add_task(multiply, name="multiply1")
wg.add_link(wg.tasks.add1.outputs.result, wg.tasks.multiply1.inputs.x)
```

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

### Run All Tests with Pytest

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/development/contributing.rst

Executes all the tests defined in the project using the pytest framework. This is a standard command to verify code integrity.

```console
pytest
```

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

### Dynamic Workflow with Python If Statement

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/overview.rst

Demonstrates a dynamic workflow using a Python 'if' statement to control the execution path based on input data. This allows for adaptive workflow logic at runtime. Requires the 'task.graph' decorator.

```python
# 'if' condition example
@task.graph()
def conditional_workflow(x):
    if x > 0:
        return add(x, 10).result
    else:
        return multiply(x, 10).result
```

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

### Implement Resilient Workflows with Error Handling

Source: https://context7.com/aiidateam/aiida-workgraph/llms.txt

Details how to implement error handling and recovery mechanisms in workflows using try-except blocks and dedicated recovery tasks. This allows workflows to gracefully handle task failures and continue execution. Dependencies include aiida and aiida_workgraph.

```python
from aiida import load_profile
from aiida_workgraph import task

load_profile()

@task
def potentially_failing_task(value, should_fail=False):
    """Task that may fail based on input."""
    if should_fail:
        raise ValueError("Intentional failure for testing")
    return value * 2

@task
def recovery_task(value):
    """Fallback task when primary task fails."""
    return value  # Return original value unchanged

@task.graph
def resilient_workflow(input_value, fail_flag):
    """Workflow with error handling logic."""
    try:
        # Try primary task
        result = potentially_failing_task(
            value=input_value,
            should_fail=fail_flag
        ).result
    except Exception:
        # Fallback to recovery task
        result = recovery_task(value=input_value).result

    return result
```

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

### Remote Task Execution with Shell Job

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/overview.rst

Shows how to execute a shell command on a remote computer. It utilizes the `shelljob` function and requires a configured AiiDA computer. The output of the command is captured.

```python
from aiida_workgraph import shelljob

remote_computer = orm.load_computer("remote_computer_label")

outputs = shelljob(
    command="date", # The command to execute
    metadata={"computer": remote_computer},
)
```

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

### React App Component for Aiida Workgraph Editor

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/_static/first_workflow.html

The main React component that initializes and renders the node editor. It uses `useState` and `useRef` hooks to manage the editor instance and the container element. It also handles loading workflow data and applying layout. Includes a button to manually trigger the layout.

```javascript
function App() {
  const [editor, setEditor] = useState(null);
  const containerRef = useRef(null);

  useEffect(() => {
    if (containerRef.current && !editor) {
      createEditor(containerRef.current).then((editor) => {
        setEditor(editor);
        loadJSON(editor.editor, editor.area, editor.layout, workgraphData).then(() => {
          editor?.layout(false).then(() => editor?.layout(true));
        });
      });
    }
    if (document.getElementById('fullscreen-btn')) {
      document.getElementById('fullscreen-btn').addEventListener('click', toggleFullScreen);
    }
    return () => {
      if (editor) {
        editor.destroy();
      }
    };
  }, [containerRef, editor]);

  return (
    <div className="App">
      <div>
        <button onClick={() => editor?.layout(true)}>Arrange</button>
      </div>
      <div ref={containerRef} className="rete" style={{ height: "100vh", width: "100%" }}></div>
    </div>
  );
}
```

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

### Iterative Workflows: For and While Loops

Source: https://context7.com/aiidateam/aiida-workgraph/llms.txt

Illustrates how to implement iterative workflows using Python's for and while loops within aiida-workgraph tasks. The `@task.graph` decorator is used to define graph tasks. Provenance is tracked for each iteration. Requires `aiida.load_profile` and `@task` decorator.

```python
from aiida_workgraph import task
from aiida import load_profile

load_profile()

@task
def increment(x):
    return x + 1

# For loop example
@task.graph
def count_up(start, iterations):
    """Increment a value multiple times."""
    result = start
    for _ in range(iterations):
        result = increment(result).result
    return result

wg_for = count_up.build(start=0, iterations=5)
wg_for.run()
print(f"Count up result: {wg_for.outputs.result.value}")  # Output: 5

# While loop via recursion
@task.graph
def count_until(current, target):
    """Count up until reaching target value."""
    if current >= target:
        return current
    next_val = increment(current).result
    return count_until(current=next_val, target=target)

wg_while = count_until.build(current=0, target=10)
wg_while.run()
print(f"Count until result: {wg_while.outputs.result.value}")  # Output: 10
```

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

### Reusable Workflow using Sub-workflows

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/overview.rst

Illustrates how to create reusable workflows by encapsulating common routines as sub-workflows. This promotes modularity and maintainability in complex pipelines. Uses the 'task.graph' decorator.

```python
@task.graph()
def main_workflow(x, y):
    sum1 = add(x, y).result
    # Call the reusable add_multiply workflow
    result1 = add_multiply(sum1, 2, 3).result
    return result1
```

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

### Create Daemon Runner for Workers and Testing in aiida-workgraph

Source: https://github.com/aiidateam/aiida-workgraph/wiki/WorkGraph-scheduler

This Python method creates and returns a new daemon runner. It is used by workers when the daemon is running and also in testing scenarios. The runner is configured to work within the daemon's environment, listening for launch requests.

```python
def create_daemon_runner(self, loop: Optional['asyncio.AbstractEventLoop'] = None) -> 'Runner':
        """Create and return a new daemon runner.

        This is used by workers when the daemon is running and in testing.

        :param loop: the (optional) asyncio event loop to use

        :return: a runner configured to work in the daemon configuration

        """
        from plumpy.persistence import LoadSaveContext

        from aiida.engine import persistence
        from aiida.engine.processes.launcher import ProcessLauncher

        runner = self.create_runner(broker_submit=True, loop=loop)
        runner_loop = runner.loop

        # Listen for incoming launch requests
        task_receiver = ProcessLauncher(
            loop=runner_loop,
            persister=self.get_persister(),
            load_context=LoadSaveContext(runner=runner),
            loader=persistence.get_object_loader(),
        )

        assert runner.communicator is not None, 'communicator not set for runner'
        runner.communicator.add_task_subscriber(task_receiver)

        return runner
```

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

### Control AiiDA WorkGraph tasks via command line

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/howto/control.rst

Manage AiiDA WorkGraph tasks using the 'workgraph' command-line tool. This enables pausing, resuming, and terminating specific tasks within a workgraph. Requires the workgraph primary key (pk) and task name as arguments.

```bash
# pause the task
workgraph task pause <workgraph_pk> <task_name>
# play the task
workgraph task play <workgraph_pk> <task_name>
# kill the task
workgraph task kill <workgraph_pk> <task_name>
```

```bash
verdi process list
```

```bash
workgraph task list <workgraph_pk>
```

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

### Compose Workflows with @task.graph Decorator in Python

Source: https://context7.com/aiidateam/aiida-workgraph/llms.txt

The `@task.graph` decorator facilitates the creation of composite workflows by visually linking tasks. It allows for Pythonic workflow composition, where data flows seamlessly from task outputs to inputs, simplifying the construction of complex computational graphs.

```python
from aiida_workgraph import task
from aiida import load_profile

load_profile()

@task
def square(x):
    return x * x

@task
def sum_numbers(numbers):
    return sum(numbers)

@task.graph
def sum_of_squares(numbers):
    """Calculate sum of squares for a list of numbers."""
    squared = [square(x).result for x in numbers]
    return sum_numbers(squared).result

# Build and run the workflow
wg = sum_of_squares.build(numbers=[1, 2, 3, 4])
wg.run()

print(f"Sum of squares: {wg.outputs.result.value}")  # Output: 30
print(f"WorkGraph state: {wg.state}")  # Output: FINISHED
print(f"Process PK: {wg.pk}")  # AiiDA process node primary key
```

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

### Create Dynamic Rete.js Node from JSON Data (Vanilla JS)

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/_static/first_workflow.html

Dynamically creates a Rete.js Node instance based on provided JSON data, defining its inputs, outputs, and dimensions. This function ensures that nodes are sized appropriately to accommodate socket names and calculates dimensions based on the number of inputs and outputs.

```javascript
function createDynamicNode(nodeData) {
    const node = new Node(nodeData.label);

    // resize the node based on the max length of the input/output names
    let maxSocketNameLength = 0;
    nodeData.inputs.forEach((input) => {
        let socket = new ClassicPreset.Socket(input.name);
        if (!node.inputs.hasOwnProperty(input.name)) {
            node.addInput(input.name, new ClassicPreset.Input(socket, input.name));
            maxSocketNameLength = Math.max(maxSocketNameLength, input.name.length);
        }
    });

    nodeData.outputs.forEach((output) => {
        let socket = new ClassicPreset.Socket(output.name);
        if (!node.outputs.hasOwnProperty(output.name)) {
            node.addOutput(output.name, new ClassicPreset.Output(socket, output.name));
            maxSocketNameLength = Math.max(maxSocketNameLength, output.name.length);
        }
    });

    node.height = Math.max(140, node.height + (nodeData.inputs.length + nodeData.outputs.length) * 35)
    node.width += maxSocketNameLength * 5;

    return node;
}
```

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

### Add and Position Dynamic Node in Rete.js Editor (Vanilla JS)

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/_static/first_workflow.html

Adds a dynamically created Rete.js node to the editor and positions it according to the data. It also stores a reference to the node in the editor's map for easy access.

```javascript
async function addNode(editor, area, nodeData) {
    console.log("Adding node", nodeData);
    const node = createDynamicNode(nodeData);
    await editor.addNode(node);
    editor.nodeMap[nodeData.label] = node; // Assuming each nodeData has a unique ID
    await area.translate(node.id, { x: nodeData.position[0], y: nodeData.position[1] });
}
```

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

### Manage Workflows with WorkGraph Class in Python

Source: https://context7.com/aiidateam/aiida-workgraph/llms.txt

The `WorkGraph` class offers comprehensive methods for building, executing, submitting, and managing workflows. It centralizes process lifecycle control, task coordination, and state management, providing a robust framework for workflow orchestration.

```python
from aiida import load_profile
from aiida_workgraph import WorkGraph, task

load_profile()

@task
def process_data(data, factor):
    return data * factor

# Create a WorkGraph instance
wg = WorkGraph(name="DataProcessing")

# Add tasks programmatically
task1 = wg.add_task(process_data, name="process1", x=10, factor=2)
task2 = wg.add_task(process_data, name="process2", factor=3)

# Link tasks: output of task1 -> input of task2
wg.add_link(task1.outputs.result, task2.inputs.data)

# Submit for asynchronous execution
wg.submit(wait=True, timeout=300)

print(f"Workflow state: {wg.state}")
print(f"Final result: {wg.tasks['process2'].outputs.result.value}")

# Load a previously executed workflow
wg_loaded = WorkGraph.load(wg.pk)
print(f"Loaded workflow result: {wg_loaded.outputs.result.value}")
```

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

### Load Aiida Workgraph Data from JSON

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/_static/first_workflow.html

Loads workflow nodes and links from a JSON object into the Rete.js editor. It iterates through the nodes and links provided in the `workgraphData` and uses `addNode` and `addLink` functions to populate the editor. Requires `addNode` and `addLink` to be defined.

```javascript
async function loadJSON(editor, area, layout, workgraphData) {
  for (const nodeId in workgraphData.nodes) {
    const nodeData = workgraphData.nodes[nodeId];
    await addNode(editor, area, nodeData);
  }
  workgraphData.links.forEach(async (link) => {
    await addLink(editor, area, layout, link);
  });
}
```

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

### Monitor External Events with @task.monitor

Source: https://context7.com/aiidateam/aiida-workgraph/llms.txt

Demonstrates using the @task.monitor decorator to create polling tasks that wait for specific conditions, such as a target time being reached or a file existing. These tasks block workflow execution until the condition is met. Dependencies include datetime, asyncio, and aiida.

```python
import datetime
import asyncio
from aiida import load_profile
from aiida_workgraph import task

load_profile()

@task.monitor
def monitor_time(target_time: datetime.datetime):
    """Wait until specified time is reached."""
    return datetime.datetime.now() > target_time

@task.async_monitor
def create_file_delayed(filepath, content, delay):
    """Create a file after a delay."""
    await asyncio.sleep(delay)
    with open(filepath, 'w') as f:
        f.write(content)

@task.monitor
def monitor_file(filepath):
    """Wait until file exists."""
    import os
    return os.path.exists(filepath)

@task
def process_when_ready(value):
    """Process after monitoring completes."""
    return value * 2

@task.graph
def monitored_workflow(delay_seconds):
    """Wait for time, then process."""
    target = datetime.datetime.now() + datetime.timedelta(seconds=delay_seconds)

    # Monitor task blocks execution until condition is met
    monitor_time(time=target, interval=1, timeout=30)

    # This runs only after monitor completes
    result = process_when_ready(value=42).result
    return result

wg = monitored_workflow.build(delay_seconds=3)
wg.run()
print(f"Result after monitoring: {wg.outputs.result.value}")  # Output: 84
```

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

### Parallel Task Execution

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/overview.rst

Defines a workflow that launches multiple instances of a task in parallel. The `@task.graph()` decorator is used, and a loop generates tasks, with their results collected into a dictionary.

```python
@task.graph()
def parallel_add(x, N):
    results = {}
    # Launch N parallel tasks to add x with each index
    for i in range(N):
        results[f"add_{i}"] = add(x, i).result
    return results
```

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

### Pythonic Workflow Definition and Execution

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/overview.rst

Defines individual tasks using the @task decorator and composes them into a workflow using the @task.graph decorator. This allows for rapid prototyping and integration into larger Python applications. The workflow can then be run with specified parameters.

```python
from aiida_workgraph import task

# Define individual tasks with the @task decorator
@task
def add(x, y):
    return x + y

@task
def multiply(x, y):
    return x * y

# Compose tasks into a workflow using the @task.graph decorator
@task.graph()
def add_multiply(x, y, z):
    if x > 0:
        return add(x, y).result
    else:
        return multiply(x, y).result

# Run the workflow
add_multiply.run(x=2, y=3, z=4)
```

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

### Implement Conditional Logic in Workflows with Python Control Flow

Source: https://context7.com/aiidateam/aiida-workgraph/llms.txt

AiiDA-WorkGraph allows native Python conditionals and loops within `@task.graph` decorated functions. This enables the creation of dynamic workflows with branching logic, allowing different operations to be executed based on input parameters or intermediate results.

```python
from aiida_workgraph import task
from aiida import load_profile

load_profile()

@task
def add(x, y):
    return x + y

@task
def multiply(x, y):
    return x * y

@task
def divide(x, y):
    return x / y

@task.graph
def conditional_calculation(operation, x, y):
    """Execute different operations based on input parameter."""
    if operation == "add":
        return add(x, y).result
    elif operation == "multiply":
        return multiply(x, y).result
    elif operation == "divide":
        return divide(x, y).result
    else:
        return None

```

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

### Add Connection Between Rete.js Nodes (Vanilla JS)

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/_static/first_workflow.html

Establishes a connection between two nodes in the Rete.js editor based on provided link data. This function dynamically adds input/output sockets if they don't exist and updates node dimensions accordingly.

```javascript
async function addLink(editor, area, layout, linkData) {
    const fromNode = editor.nodeMap[linkData.from_node];
    const toNode = editor.nodeMap[linkData.to_node];
    console.log("fromNode", fromNode, "toNode", toNode);

    let socket;
    if (fromNode && toNode) {
        socket = new ClassicPreset.Socket(linkData.from_socket);
        if (!fromNode.outputs.hasOwnProperty(linkData.from_socket)) {
            fromNode.addOutput(linkData.from_socket, new ClassicPreset.Output(socket, linkData.from_socket));
            fromNode.height += 25;
            area.update('node', fromNode.id);
        }

        socket = new ClassicPreset.Socket(linkData.to_socket);
        if (!toNode.inputs.hasOwnProperty(linkData.to_socket)) {
            toNode.addInput(linkData.to_socket, new ClassicPreset.Input(socket, linkData.to_socket));
            toNode.height += 25;
            area.update('node', toNode.id);
        }

        const link = new Schemes.Connection(fromNode.outputs[linkData.from_socket], toNode.inputs[linkData.to_socket]);
        await editor.connect(link);
    }
}
```

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

### Compose Workflow with @task.graph Decorator (Python)

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/README.md

Illustrates composing a workflow by linking tasks defined with the @task decorator. The @task.graph decorator is used to define the overall workflow structure, where the output of one task (e.g., `add`) becomes the input for another (e.g., `multiply`). This allows for natural data flow and the creation of directed acyclic graphs (DAGs) for workflow execution.

```python
@task.graph
def add_multiply(x, y, z):
    """A workflow to add two numbers and then multiply by a third."""
    sum_result = add(x, y).result
    product_result = multiply(x=sum_result, y=z).result
    return product_result
```

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

### Configure Python Executable for PythonJob Tests

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/development/contributing.rst

Sets the PYTEST_PYTHONJOB_PYTHON_EXEC_PATH environment variable to specify a custom Python executable for running pythonjob calcjobs in tests. This is useful for testing with different Python environments.

```console
PYTEST_PYTHONJOB_PYTHON_EXEC_PATH=/home/user/pyvenv/workgraph-dev/bin/python pytest tests/test_python.py
```

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

### Define Workflow Tasks with @task Decorator in Python

Source: https://context7.com/aiidateam/aiida-workgraph/llms.txt

The `@task` decorator transforms Python functions into AiiDA-WorkGraph tasks. It manages input/output sockets, data serialization, and integrates with AiiDA's provenance system. This enables the creation of reusable computational units within workflows.

```python
from aiida import load_profile
from aiida_workgraph import task

load_profile()

@task
def add(x, y):
    """Adds two numbers with full provenance tracking."""
    return x + y

@task
def multiply(x, y):
    """Multiplies two numbers."""
    return x * y

# Use tasks to build a simple calculation
@task.graph
def calculate(x, y, z):
    sum_result = add(x, y).result
    return multiply(sum_result, z).result

# Execute the workflow
result = calculate.run(x=2, y=3, z=4)
print(f"Result: {result}")  # Output: 20
```

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

### Define Workflow Tasks with @task Decorator (Python)

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/README.md

Demonstrates how to define individual tasks within an AiiDA-WorkGraph workflow. The @task decorator transforms standard Python functions into workflow components, enabling them to be orchestrated within a larger workflow. These functions take inputs and return outputs that can be passed to subsequent tasks.

```python
from aiida_workgraph import task

@task
def add(x, y):
    """Adds two numbers."""
    return x + y

@task
def multiply(x, y):
    """Multiplies two numbers."""
    return x * y
```

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

### Toggle Full-Screen Mode

Source: https://github.com/aiidateam/aiida-workgraph/blob/main/docs/source/_static/first_workflow.html

Toggles the browser's full-screen mode for the application. It checks if the document is already in full-screen and either requests full-screen or exits it. This function is typically attached to a button click event.

```javascript
function toggleFullScreen() {
  if (!document.fullscreenElement) {
    document.documentElement.requestFullscreen();
  } else if (document.exitFullscreen) {
    document.exitFullscreen();
  }
}
```