Skip to main content

Understanding the Execution Model: Runs and Actions

When you trigger an execution on a remote Flyte or Union cluster, track its execution state, or query previously executed runs, you interact with three core abstractions: Tasks, Runs, and Actions.

In flyte-sdk, these objects define the remote execution lifecycle and data flow:

  • A Task represents a deployed, reusable task template registered on the control plane.
  • A Run represents a top-level execution instance initiated on the control plane.
  • An Action represents the actual unit of execution within a run. A run contains a root action, and complex runs can spawn child actions forming an execution tree.

The Execution Model Hierarchy​

The relationship between Tasks, Runs, and Actions follows a structured hierarchy:

+-------------------------------------------------------------+
| Task (remote._task.Task / TaskDetails)                      |
|  - Registered definition: interface, resources, cache, etc. |
+-------------------------------------------------------------+
                              |
                     Trigger Execution
                              v
+-------------------------------------------------------------+
| Run (remote._run.Run / RunDetails)                          |
|  - Top-level execution session                             |
|  - Tracks overall execution status and web console URL      |
|  - Contains root Action: self.action = Action(...)         |
+-------------------------------------------------------------+
                              |
                     Executes as Action Tree
                              v
+-------------------------------------------------------------+
| Action (remote._action.Action / ActionDetails)              |
|  - Root Action (ID: \{run_name\})                             |
|  - Sub-actions / Child Tasks spawned during execution        |
|  - Manages attempts, runtime, logs, inputs & outputs        |
+-------------------------------------------------------------+

When you initialize a Run in remote._run.Run, it validates that the protobuf definition includes an action:

if not self.pb2.HasField("action"):
    raise RuntimeError("Run does not have an action")
self.action = Action(self.pb2.action)

Every Run instance encapsulates a root Action instance (self.action) and delegates phase checks, waiting logic, log viewers, and state streaming to it.

Managing Remote Executions with Run and RunDetails​

When you trigger a remote run with flyte.run(...) or fetch an existing execution by name, you work with Run and RunDetails.

Fetching and Inspecting Runs​

To fetch a run by its unique execution name and inspect its state:

import flyte.remote as remote

# Retrieve a remote run by name
run = remote.Run.get("a1b2c3d4e5f6g7h8")

print(f"Run Name: \{run.name\}")
print(f"Current Phase: \{run.phase\}")
print(f"Console URL: \{run.url\}")
print(f"Is Finished: \{run.done()\}")

To list runs within the currently configured organization, project, and domain:

# List the latest 10 runs sorted descending by creation time
for r in remote.Run.listall(sort_by=("created_at", "desc"), limit=10):
    print(f"Run \{r.name\} -> \{r.phase\}")

Waiting, Streaming Logs, and Aborting​

Run provides user-facing methods for monitoring and control:

# Wait for the run to reach a terminal state with a Rich terminal progress bar
run.wait()

# Or wait only until the execution transitions into the RUNNING state
run.wait(wait_for="running")

# View terminal logs for the run
run.show_logs(max_lines=50, show_ts=True)

# Terminate / abort an in-flight run
run.abort()

Under the hood:

  • run.wait() creates a rich.progress.Progress spinner and streams updates from self.watch(cache_data_on_done=True).
  • run.show_logs(...) delegates to self.action.show_logs(...), which initializes a log viewer via Logs.create_viewer(...).
  • run.abort() sends an AbortRunRequest to the gRPC run_service. If the run is not found (grpc.StatusCode.NOT_FOUND), the exception is caught and ignored so aborting is idempotent.

Detailed Run Metadata with RunDetails​

While Run contains basic execution metadata, calling run.details() or remote.RunDetails.get(name) fetches the full RunDetails protobuf containing execution specifications (run_spec):

details = run.details()

print(f"Labels: \{details.pb2.run_spec.labels\}")
print(f"Annotations: \{details.pb2.run_spec.annotations\}")
print(f"Interruptible: \{details.pb2.run_spec.interruptible\}")
print(f"Cache Overwrite: \{details.pb2.run_spec.overwrite_cache\}")

Execution Units: Action and ActionDetails​

An Action represents a specific execution step. For single-task executions, the run consists of a single root action. For workflow executions or sub-task launches, child actions execute under the parent run.

Inspecting Actions within a Run​

To list all child actions associated with a specific run:

from flyte.remote import Action, ActionDetails

# List all actions under a given run
for action in Action.listall(for_run_name="a1b2c3d4e5f6g7h8"):
    print(f"Action: \{action.name\}, Task: \{action.task_name\}, Phase: \{action.phase\}")

To fetch a specific action directly:

# Fetch an action by run_name and action name
action_details = ActionDetails.get(run_name="a1b2c3d4e5f6g7h8", name="a0")
print(f"Action Status: \{action_details.phase\}")
print(f"Attempts: \{action_details.attempts\}")
print(f"Runtime Duration: \{action_details.runtime\}")

Action Phases and Lifecycle​

An action transitions through phases defined in run_definition_pb2.Phase:

  • PHASE_UNDEFINED
  • PHASE_QUEUED
  • PHASE_INITIALIZING
  • PHASE_RUNNING
  • PHASE_SUCCEEDED (terminal)
  • PHASE_FAILED (terminal)
  • PHASE_ABORTED (terminal)
  • PHASE_TIMED_OUT (terminal)

The action.done() and action_details.done() methods determine whether an action has reached a terminal state.

Streaming Live Action Updates​

ActionDetails.watch(action_id) opens a server-streaming gRPC call (WatchActionDetails) to receive real-time updates:

import asyncio
from flyte.remote import Action

async def stream_action(action: Action):
    async for details in action.watch(cache_data_on_done=True):
        print(f"Action \{details.name\} is now in phase: \{details.phase\} (Attempt \{details.attempts\})")
        if details.error_info:
            print(f"Failure reason: \{details.error_info\}")

Working with Inputs and Outputs: ActionInputs and ActionOutputs​

When a task or action executes, its inputs and outputs are managed remotely and converted into native Python objects by ActionDetails.

Accessing Inputs and Outputs​

You can retrieve inputs and outputs from either RunDetails or ActionDetails:

details = run.details()

# Retrieve inputs as ActionInputs (a UserDict subclass containing deserialized values)
inputs = details.inputs()
for key, value in inputs.items():
    print(f"Input '\{key\}': \{value\}")

# Retrieve outputs as ActionOutputs (a tuple subclass containing deserialized values)
outputs = details.outputs()
for item in outputs:
    print(f"Output: \{item\}")

Output Availability Constraints​

Calling .outputs() triggers an internal call to _cache_data(), which requests action data from run_service.GetActionData and converts protobuf literals into native Python types using flyte._internal.runtime.convert.

If the action has not yet reached a terminal state, calling outputs() raises a RuntimeError:

try:
    outputs = details.outputs()
except RuntimeError as e:
    # "Action is not in a terminal state, outputs are not available. Please wait for the action to complete."
    print("Execution in progress; outputs cannot be fetched yet.")

Referencing and Overriding Remote Tasks: Task and TaskDetails​

The remote._task.Task class allows you to fetch, inspect, and invoke pre-deployed tasks registered on the Flyte/Union control plane.

Fetching Remote Tasks​

Task.get(...) returns a LazyEntity that defers loading the full task definition until it is explicitly called or fetched:

from flyte.remote import Task

# Fetch by exact version
lazy_task = Task.get(name="my_project.workflows.data_prep", version="v1.0.0")

# Fetch by latest version
latest_task = Task.get(name="my_project.workflows.data_prep", auto_version="latest")

# Fetch matching the current execution context (only valid inside an active task)
context_task = Task.get(name="my_project.workflows.data_prep", auto_version="current")

auto_version="current" can only be used when executing inside an active task context where flyte.ctx().version is set. If used outside a task context, it raises a ValueError.

Overriding Task Specifications​

You can override runtime resources, retries, timeouts, environment variables, or secrets on a reference task before execution:

import flyte
from flyte.remote import Task

task_ref = Task.get(name="my_model_trainer", auto_version="latest")

# Override execution attributes
task_ref.override(
    resources=flyte.Resources(cpu="4", memory="16Gi"),
    retries=3,
    timeout=flyte.Timeout(hours=2),
    env_vars=\{"ENVIRONMENT": "production"\},
)

TaskDetails.override() modifies the underlying task_definition_pb2.TaskDetails protobuf template:

  • resources updates container.resources.requests and limits.
  • env_vars updates container.env.
  • retries updates metadata.retries.
  • timeout updates metadata.timeout.
  • secrets updates security_context.secrets.

Calling Remote Tasks inside Workflows or Dynamic Tasks​

When a LazyEntity or TaskDetails is invoked within a task execution context, it forwards keyword arguments to the controller to schedule a sub-action:

import flyte
from flyte.remote import Task

remote_step = Task.get(name="data_processing_step", auto_version="latest")

@flyte.task
def parent_task(raw_data_uri: str) -> str:
    # Invoking a reference task requires keyword arguments
    result = remote_step(input_path=raw_data_uri)
    return result

Note: Reference tasks only support keyword arguments. Passing positional arguments raises flyte.errors.ReferenceTaskError.

Dual Sync and Async APIs with @syncify​

Methods across Run, RunDetails, Action, ActionDetails, and Task are decorated with @syncify. This decorator enables seamless use in both synchronous blocking scripts and asynchronous event loops.

Synchronous Usage​

In standard Python scripts, call methods directly without await:

import flyte.remote as remote

# Synchronous execution
run = remote.Run.get("a1b2c3d4e5f6g7h8")
run.wait()
details = run.details()
outputs = details.outputs()

Asynchronous Usage​

Inside an async function or coroutine, access the underlying async implementation using the .aio attribute:

import asyncio
import flyte.remote as remote

async def main():
    # Asynchronous execution using .aio
    run = await remote.Run.get.aio("a1b2c3d4e5f6g7h8")
    await run.wait.aio()
    details = await run.details.aio()
    outputs = await details.outputs()

asyncio.run(main())

Calling await run.wait() directly without .aio inside an async context will not work; use .aio to access the coroutine version.