Skip to main content

Interacting with Task Runs

When you trigger task executions remotely or need to track asynchronous jobs across projects and domains, you must query their status, stream runtime logs, retrieve produced artifacts, or terminate runaway jobs. In flyte-sdk, the flyte.remote.Run, flyte.remote.RunDetails, and flyte.remote.Action classes provide synchronous and asynchronous interfaces to manage the full lifecycle of remote executions.

Managing Task Runs​

Use flyte.remote.Run to inspect, track, and control executions. All major methods on Run are decorated with @syncify, enabling them to be invoked directly in synchronous scripts or asynchronously using the .aio accessor.

import flyte
from flyte.remote import Run, RunDetails

# Initialize client connection context
flyte.init(project="my-project", domain="development")

# 1. Retrieve an existing run by name
run = Run.get(name="f-abcd1234ef567890")
print(f"Run {run.name} phase: {run.phase}")
print(f"Console URL: {run.url}")

# 2. Wait for completion with progress feedback
run.wait(quiet=False)

# 3. Access execution details and outputs
if run.done():
    details: RunDetails = run.details()
    outputs = run.details().outputs()
    print("Action Outputs:", outputs)

The underlying client connects to Flyte's backend services through get_client().run_service. When initialized via flyte.init(), the project and domain configuration (CommonInit) context is automatically applied to all run queries.

Listing and Retrieving Runs​

To query multiple runs or fetch a specific run instance, use Run.listall() and Run.get().

Listing Runs with Pagination and Sorting​

Run.listall() returns a generator yielding Run objects from the current project and domain. It accepts custom sort orders and maximum limits:

import flyte
from flyte.remote import Run

flyte.init(project="my-project", domain="development")

# List up to 50 most recent runs in descending order
for r in Run.listall(sort_by=("created_at", "desc"), limit=50):
    print(f"Name: {r.name} | Phase: {r.phase} | URL: {r.url}")

Equivalent CLI command:

flyte get run --limit 50

Retrieving a Run by Name​

To retrieve a single run, pass its name to Run.get():

from flyte.remote import Run

# Synchronous lookup
run = Run.get(name="f-abcdef1234567890")

# Asynchronous lookup
# run = await Run.get.aio(name="f-abcdef1234567890")

Monitoring Progress and Phases​

Run and its underlying Action expose execution phases and status properties.

Inspecting Phases​

  • run.phase: String representation of the phase (e.g., "PHASE_SUCCEEDED", "PHASE_RUNNING", "PHASE_FAILED").
  • run.raw_phase: Raw protobuf enum value from run_definition_pb2.Phase.
  • run.done(): Boolean indicating if the execution has reached a terminal phase (PHASE_SUCCEEDED, PHASE_FAILED, PHASE_ABORTED, PHASE_TIMED_OUT).
if run.done():
    print(f"Execution finished with status: {run.phase}")
else:
    print(f"Execution in progress: {run.phase}")

Waiting for Status Transitions​

run.wait() creates a Rich-based progress indicator displaying runtime and state transitions until the target state is reached:

# Wait until run reaches a terminal state (default)
run.wait(quiet=False, wait_for="terminal")

# Wait until run transitions from queued to running
run.wait(quiet=False, wait_for="running")

Streaming State Updates with watch()​

To subscribe to granular state updates programmatically, use run.watch(), which yields ActionDetails updates as they occur:

async def stream_run_status(run: Run):
    async for action_detail in run.watch(cache_data_on_done=True):
        print(f"State update: {action_detail.phase}, elapsed: {action_detail.runtime}s")
        if action_detail.done():
            break

Streaming and Viewing Logs​

Run.show_logs() streams execution logs for the root action of the run.

# Display the last 50 lines with timestamps
run.show_logs(max_lines=50, show_ts=True, raw=False, filter_system=True)

# Asynchronously display logs for a specific attempt (1-based index)
await run.show_logs.aio(attempt=1, max_lines=100, show_ts=True)

Viewing Logs via the CLI​

# Stream raw logs in terminal
flyte get logs <run_name>

# Display auto-scrolling boxed viewer showing last 50 lines with timestamps
flyte get logs <run_name> --pretty --lines 50 --show-ts

# View logs for a specific action within a run
flyte get logs <run_name> <action_name> --filter-system

Inspecting Inputs, Outputs, and Metadata​

Execution parameters, specs, inputs, and outputs are accessible through RunDetails.

import flyte
from flyte.remote import Run, RunDetails

flyte.init(project="my-project", domain="development")

run = Run.get("f-abcdef1234567890")
details: RunDetails = run.details()

# Inspect execution metadata
print(f"Task Name: {details.task_name}")
print(f"Action ID: {details.action_id}")

Fetching Inputs and Outputs​

Inputs and outputs are fetched asynchronously through ActionInputs and ActionOutputs:

async def fetch_io(run: Run):
    details: RunDetails = await run.details.aio()

    # Inputs are accessible during or after execution
    inputs = await details.inputs()
    print("Run Inputs:", inputs)

    # Outputs are only available once the run is done
    if details.done():
        outputs = await details.outputs()
        print("Run Outputs:", outputs)

Equivalent CLI command:

# View both inputs and outputs
flyte get io <run_name>

# View only outputs
flyte get io <run_name> --outputs-only

Terminating a Run​

To cancel an ongoing task execution, call run.abort().

from flyte.remote import Run

run = Run.get("f-abcdef1234567890")
if not run.done():
    run.abort()
    print(f"Run {run.name} has been aborted.")

Equivalent CLI command:

flyte abort run <run_name>

run.abort() sends a gRPC AbortRun request to the backend. If the run no longer exists on the backend (grpc.StatusCode.NOT_FOUND), the exception is handled silently and the method returns without error.

Working with Individual Actions​

A Run contains an underlying Action representing its execution unit. In complex workflows where multiple actions are present, use flyte.remote.Action directly:

from flyte.remote import Action

# List all actions associated with a specific run
for action in Action.listall(for_run_name="f-abcdef1234567890"):
    print(f"Action: {action.name} | Task: {action.task_name} | Phase: {action.phase}")

# Fetch a specific action
action = Action.get(run_name="f-abcdef1234567890", name="a0")
action.wait()

Equivalent CLI command:

flyte get action <run_name>
flyte get action <run_name> <action_name>

Troubleshooting and Common Gotchas​

RuntimeError: Action is not in a terminal state, outputs are not available​

Calling await details.outputs() or await action_details.outputs() before a task finishes executing raises a RuntimeError.

  • Solution: Always wait for completion using run.wait() or verify if run.done(): before calling .outputs().

ValueError: Attempt number must be greater than 0​

Execution attempts are 1-indexed. Passing attempt=0 to show_logs() raises a ValueError.

  • Solution: Use positive integers starting from 1 for the attempt parameter, or omit it to default to the latest attempt.

AssertionError: Cannot call sync method from within an async event loop​

Invoking synchronous methods like run.wait() from inside an active asyncio event loop thread managed by syncify causes an assertion failure to prevent deadlocks.

  • Solution: When working in async coroutines, call the asynchronous method with .aio (e.g., await run.wait.aio(), await run.details.aio(), await run.show_logs.aio()).

RuntimeError: Run does not have an action​

When instantiating Run(pb2=...) directly without an embedded action field in the protobuf payload, Run.__post_init__ raises a RuntimeError.

  • Solution: Always instantiate Run objects using Run.get(name) or Run.listall().