Note

Go to the end to download the full example code.

Monitor external events as a task

Introduction

The monitor decorator is designed for tasks that need to poll a specific state at regular intervals until a success criterion is met. Some possible use cases include:

  • Time-based events: Start a task at a specified time

  • File-based events: Execute a task when a particular file exists

  • Monitor a task: Observe the state of another task and act based on certain conditions

  • Cross-workGraph dependencies: Check the state of a task in a different workgraph

While polling, the task sleeps for a specified interval (default 1.0 second), allowing the workgraph engine to manage other tasks.

The monitor task has two built-in parameters:

  • interval: The time interval between each poll

  • timeout: The maximum time to wait for the success criterion to be met

In the following sections, we will walk through examples of how to use the monitor task decorator for these scenarios.

We’ll temporarily set the AiiDA log level to REPORT, so that we can inspect the execution order of the workgraph.

from aiida_workgraph import task
from aiida import load_profile

load_profile()

Profile<uuid='2c6af82ef2b241408a4c27c7c32d5490' name='presto'>

Time-based events

Here we design a workgraph that waits until 5 seconds have passed.

import datetime


@task.monitor
def monitor_time(time: datetime.datetime):
    return datetime.datetime.now() > time

Note

The monitor task can also be implemented using asynchronous function. For more details, please refer to the section on Run async functions as tasks. For example, the above monitor task can be implemented as:

@task
async def monitor_time_async(time: datetime.datetime, interval=1, timeout=None):
    start = datetime.datetime.now()
    while True:
        if datetime.datetime.now() > time:
            break
        await asyncio.sleep(interval)
        if timeout is not None and datetime.datetime.now() - start > timeout:
            raise TimeoutError(f'Timeout after {timeout} seconds')

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


@task.graph
def TimeMonitor(time, x, y):
    monitor_time(time) >> (the_sum := add(x, y).result)  # wait, THEN (>>) add
    return the_sum


wg = TimeMonitor.build(
    time=datetime.datetime.now() + datetime.timedelta(seconds=5),
    x=1,
    y=2,
)
wg



wg.run()

06/29/2026 09:10:11 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [297|WorkGraphEngine|continue_workgraph]: tasks ready to run: monitor_time
06/29/2026 09:10:11 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [297|WorkGraphEngine|on_wait]: Process status: Waiting for child processes: 300
06/29/2026 09:10:17 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [297|WorkGraphEngine|update_task_state]: Task: monitor_time, type: MONITOR, finished.
06/29/2026 09:10:17 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [297|WorkGraphEngine|continue_workgraph]: tasks ready to run: add
06/29/2026 09:10:17 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [297|WorkGraphEngine|update_task_state]: Task: add, type: PYFUNCTION, finished.
06/29/2026 09:10:17 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [297|WorkGraphEngine|continue_workgraph]: tasks ready to run:
06/29/2026 09:10:17 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [297|WorkGraphEngine|finalize]: Finalize workgraph.

{'result': <Int: uuid: 1d2d8212-339f-441e-9ff7-07498ed0c2c5 (pk: 303) value: 3>}

Note the time difference between the monitor task and the next (~5 seconds)

File-based events

Here we design a workgraph that waits until a file exists before it proceeds. We create the file asynchronously after 5 seconds. During this time, the monitor task will poll for the file’s existence. Once the file is found, the workgraph will proceed to discard the file and add two numbers (independently), then finally multiply the sum by a factor.

import os
import asyncio
from aiida_workgraph.collection import group


# Create a file asynchronously, so it does not block the event loop
@task
async def sleep_create_file(filepath, content):
    await asyncio.sleep(5)
    with open(filepath, 'w') as f:
        f.write(content)


@task.monitor
def monitor_file(filepath):
    return os.path.exists(filepath)


@task
def discard_file(filepath):
    if os.path.exists(filepath):
        os.remove(filepath)
    else:
        raise FileNotFoundError(f'File {filepath} does not exist.')


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


@task.graph
def FileMonitor(x, y, z):
    # Asynchronously create a file after 5 seconds
    sleep_create_file(
        filepath='/tmp/monitor_test.txt',
        content='This is a test file',
    )

    # While above is running, monitor for the file
    # Once done (`>>` means wait on left operant), discard the file and add two numbers
    monitor_file(
        '/tmp/monitor_test.txt',
        interval=1,
        timeout=10,
    ) >> group(
        discard_file('/tmp/monitor_test.txt'),
        the_sum := add(x, y).result,
    )

    # Finally, multiply the sum by a factor
    return multiply(the_sum, z).result


wg = FileMonitor.build(x=1, y=2, z=3)
wg



wg.run()

06/29/2026 09:10:18 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [313|WorkGraphEngine|continue_workgraph]: tasks ready to run: sleep_create_file,monitor_file
06/29/2026 09:10:18 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [313|WorkGraphEngine|on_wait]: Process status: Waiting for child processes: 314, 317
06/29/2026 09:10:24 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [313|WorkGraphEngine|update_task_state]: Task: sleep_create_file, type: PYFUNCTION, finished.
06/29/2026 09:10:24 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [313|WorkGraphEngine|continue_workgraph]: tasks ready to run:
06/29/2026 09:10:24 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [313|WorkGraphEngine|on_wait]: Process status: Waiting for child processes: 317
06/29/2026 09:10:24 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [313|WorkGraphEngine|update_task_state]: Task: monitor_file, type: MONITOR, finished.
06/29/2026 09:10:24 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [313|WorkGraphEngine|continue_workgraph]: tasks ready to run: discard_file,add
06/29/2026 09:10:24 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [313|WorkGraphEngine|update_task_state]: Task: discard_file, type: PYFUNCTION, finished.
06/29/2026 09:10:24 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [313|WorkGraphEngine|continue_workgraph]: tasks ready to run: add
06/29/2026 09:10:25 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [313|WorkGraphEngine|update_task_state]: Task: add, type: PYFUNCTION, finished.
06/29/2026 09:10:25 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [313|WorkGraphEngine|continue_workgraph]: tasks ready to run: multiply
06/29/2026 09:10:25 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [313|WorkGraphEngine|update_task_state]: Task: multiply, type: PYFUNCTION, finished.
06/29/2026 09:10:25 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [313|WorkGraphEngine|continue_workgraph]: tasks ready to run:
06/29/2026 09:10:25 AM <2622> aiida.orm.nodes.process.workflow.workchain.WorkChainNode: [REPORT] [313|WorkGraphEngine|finalize]: Finalize workgraph.

{'result': <Int: uuid: 76fdea3e-1601-4f8b-b66d-43f645ff42e8 (pk: 323) value: 9>}

You can inspect the process reports above to verify the order of events.

Kill a monitor task

One can kill a running monitor task by using the following command:

verdi process kill <pk>

A killed task will has the status KILLED and the following task will not be executed.

Built-in monitors

WorkGraph provides the the above time and file monitors as built-in tasks. You can use them directly without having to define them.

from aiida_workgraph.tasks.monitors import monitor_time, monitor_file

Summary

You have learned how to use the monitor decorator to create tasks that poll for specific conditions, such as time-based events, file-based events, and task monitoring. You also learned how to kill a monitor task and about the built-in monitor tasks provided by WorkGraph.

Total running time of the script: (0 minutes 15.471 seconds)

Gallery generated by Sphinx-Gallery