Understanding Run Actions

Actions represent the fundamental units of execution within a larger run. Each action corresponds to the execution of a specific task and manages its lifecycle, state, and interactions with the remote API. Understanding actions is crucial for monitoring, debugging, and programmatically interacting with your running workflows.

Core Concepts

The system defines several core classes to manage and interact with actions:

• Action: This class serves as the primary client-side interface for an action. It provides methods to retrieve action details, monitor its progress, and access logs. An Action object represents a specific instance of a task's execution within a run.
• ActionDetails: This class holds comprehensive, real-time information about an action's state, metadata, status, and execution attempts. It is a more detailed view of an action, often retrieved when deeper inspection is required.
• ActionInputs: This class encapsulates the inputs provided to an action. It allows programmatic access to the data that initiated the action's execution.
• ActionOutputs: This class holds the results produced by a successfully completed action. It provides access to the data generated by the task.

Managing Actions

Listing Actions for a Run

To retrieve all actions associated with a specific run, use the Action.listall class method. This method returns an iterator of Action objects, allowing you to process actions efficiently, especially for runs with many tasks.

from your_package_name import Action

async def get_run_actions(run_name: str):
    actions = []
    async for action in Action.listall(for_run_name=run_name):
        actions.append(action)
    return actions

# Example usage:
# all_actions = await get_run_actions("my_workflow_run_123")
# for action in all_actions:
#     print(f"Action Name: {action.name}, Phase: {action.phase}")

The listall method supports filtering and sorting to refine the list of returned actions.

Retrieving a Specific Action

You can retrieve a specific Action or ActionDetails object by its name within a given run.

To get an Action object:

from your_package_name import Action

async def get_single_action(run_name: str, action_name: str):
    action = await Action.get(run_name=run_name, name=action_name)
    return action

# Example usage:
# my_action = await get_single_action("my_workflow_run_123", "my_task_action")
# print(f"Retrieved action: {my_action.name} in phase {my_action.phase}")

To get an ActionDetails object, which provides a richer set of information:

from your_package_name import ActionDetails

async def get_action_details(run_name: str, action_name: str):
    details = await ActionDetails.get(run_name=run_name, name=action_name)
    return details

# Example usage:
# details = await get_action_details("my_workflow_run_123", "my_task_action")
# print(f"Action {details.name} has {details.attempts} attempts.")

The ActionDetails.get_details method can also be used directly if you already have the ActionIdentifier protobuf object.

Monitoring Action Status

Actions progress through various phases during their execution. Monitoring these phases is essential for understanding the state of your run.

Checking the Current Phase

The phase property on both Action and ActionDetails objects provides the current execution phase as a human-readable string. The raw_phase property provides the underlying protobuf enum value.

# Assuming 'action' is an Action object
current_phase = action.phase
print(f"Action '{action.name}' is currently in phase: {current_phase}")

# Assuming 'details' is an ActionDetails object
is_running = details.is_running
print(f"Action '{details.name}' is running: {is_running}")

Watching for Updates

For real-time monitoring, use the watch method on an Action object. This method returns an asynchronous generator that yields ActionDetails objects as the action's state changes. This is particularly useful for building interactive dashboards or logging systems.

from your_package_name import Action

async def watch_action_progress(action: Action):
    print(f"Watching action '{action.name}'...")
    async for details in action.watch():
        print(f"  Current Phase: {details.phase}, Runtime: {details.runtime.total_seconds():.2f}s, Attempts: {details.attempts}")
        if details.done():
            print(f"Action '{action.name}' finished.")
            break

# Example usage:
# my_action = await Action.get(run_name="my_workflow_run_123", name="my_task_action")
# await watch_action_progress(my_action)

The watch method can be configured with wait_for parameters (e.g., "running", "logs-ready", "terminal") to break the watch loop once a specific state is reached.

Waiting for Completion

To block execution until an action reaches a terminal state (succeeded, failed, aborted), use the wait method. This method provides a rich progress display in the console, showing status transitions and elapsed time.

from your_package_name import Action

async def wait_for_action_completion(run_name: str, action_name: str):
    action = await Action.get(run_name=run_name, name=action_name)
    print(f"Waiting for action '{action.name}' in run '{action.run_name}' to complete...")
    await action.wait()
    print(f"Action '{action.name}' has completed with phase: {action.phase}")

# Example usage:
# await wait_for_action_completion("my_workflow_run_123", "my_task_action")

The wait method also accepts a wait_for parameter, allowing you to wait until the action is running, logs are available, or it reaches a terminal state. The quiet parameter can suppress the progress display.

Checking Completion Status

The done method on both Action and ActionDetails objects indicates whether the action has reached a terminal state (succeeded, failed, or aborted).

# Assuming 'action' is an Action object
if action.done():
    print(f"Action '{action.name}' is complete.")
else:
    print(f"Action '{action.name}' is still in progress.")

Accessing Action Data

Retrieving Logs

To view the execution logs for an action, use the show_logs method. This method provides a viewer for the logs, optionally filtering system logs, showing timestamps, and limiting the number of lines.

from your_package_name import Action

async def display_action_logs(run_name: str, action_name: str):
    action = await Action.get(run_name=run_name, name=action_name)
    print(f"Displaying logs for action '{action.name}' (latest attempt):")
    log_viewer = await action.show_logs(max_lines=50, show_ts=True)
    # The log_viewer object can then be used to display logs, e.g., by printing it
    # or iterating over its content if it's an iterable.
    # For simplicity, assuming it prints directly or has a display method:
    print(log_viewer)

# Example usage:
# await display_action_logs("my_workflow_run_123", "my_task_action")

The logs_available method on ActionDetails can be used to check if logs exist for a given attempt before attempting to retrieve them.

Accessing Inputs

The inputs provided to an action can be retrieved using the inputs method on an ActionDetails object. This method returns an ActionInputs object, which behaves like a dictionary, allowing access to input parameters by name.

from your_package_name import ActionDetails

async def get_action_inputs(run_name: str, action_name: str):
    details = await ActionDetails.get(run_name=run_name, name=action_name)
    inputs = await details.inputs()
    print(f"Inputs for action '{details.name}':")
    for key, value in inputs.items():
        print(f"  {key}: {value}")

# Example usage:
# await get_action_inputs("my_workflow_run_123", "my_task_action")

Accessing Outputs

Once an action has successfully completed, its outputs can be retrieved using the outputs method on an ActionDetails object. This method returns an ActionOutputs object, which behaves like a tuple, containing the results in the order they were defined.

from your_package_name import ActionDetails

async def get_action_outputs(run_name: str, action_name: str):
    details = await ActionDetails.get(run_name=run_name, name=action_name)
    if not details.done() or details.phase != "SUCCEEDED":
        print(f"Action '{details.name}' is not in a succeeded state. Outputs may not be available.")
        return

    outputs = await details.outputs()
    print(f"Outputs for action '{details.name}':")
    for i, output_value in enumerate(outputs):
        print(f"  Output {i}: {output_value}")

# Example usage:
# await get_action_outputs("my_workflow_run_123", "my_task_action")

Attempting to retrieve outputs for an action that is not in a terminal state will raise a RuntimeError. It is best practice to check details.done() and details.phase before requesting outputs.

Key Properties

Both Action and ActionDetails objects expose several useful properties:

• name: The unique name of the action within its run.
• run_name: The name of the run to which this action belongs.
• task_name: The name of the underlying task definition executed by this action.
• action_id: A protobuf identifier object for the action, including organization, project, domain, run name, and action name.
• phase: The current execution phase (e.g., "QUEUED", "RUNNING", "SUCCEEDED", "FAILED").
• runtime (ActionDetails only): The duration of the action's execution as a timedelta object.
• attempts (ActionDetails only): The number of times the action has been attempted.
• error_info (ActionDetails only): Detailed error information if the action failed.
• abort_info (ActionDetails only): Information if the action was aborted.

Integration and Best Practices

• Asynchronous Operations: Most methods for interacting with actions are asynchronous (async def). The @syncify decorator allows these methods to be called synchronously as well, but for optimal performance and responsiveness in applications, prefer awaiting the asynchronous versions.
• Error Handling: When an action fails, ActionDetails.error_info provides structured details about the failure. Implement robust error handling by checking the action's phase and inspecting error_info when done() is true and the phase is not SUCCEEDED.
• Data Caching: The watch method on Action and watch_updates on ActionDetails include a cache_data_on_done parameter. Setting this to True ensures that inputs and outputs are fetched and cached locally once the action completes, making subsequent calls to inputs() and outputs() faster.
• Client Configuration: All interactions with the remote API implicitly rely on a configured client. Ensure the client is properly initialized (e.g., via ensure_client() and get_common_config()) before making API calls.