Managing Executions (Runs)

Retrieve a run and inspect its root action

After initializing the Flyte client and its common organization, project, and domain configuration, retrieve a remote execution with Run.get. A Run is the run-level handle; it eagerly creates an Action for the run's root action.

from flyte.remote import Run


async def inspect_run(run_name: str) -> None:
    run = await Run.get.aio(name=run_name)
    print(run.name, run.phase, run.url)

    details = await run.details.aio()
    print(details.name, details.task_name)
    print(details.action_details.action_id)

Run.get is implemented in remote/_run.py. It obtains RunDetails, then reconstructs a lightweight Run from the returned action identifier, metadata, and status. Run.details() fetches details only the first time and caches the resulting RunDetails in _details. RunDetails similarly constructs an ActionDetails wrapper for its root action.

Remote retrieval and listing require an initialized client. The implementation calls ensure_client() and uses get_common_config() to scope requests. Initialize the client with flyte.init(...), or use the CLI configuration setup before calling these APIs.

List runs or actions

Run.listall is also available synchronously and asynchronously. It lists runs for the configured project and domain, defaults to created_at ascending, follows service pagination tokens, and stops after limit results.

from flyte.remote import Action, Run


runs = Run.listall(limit=20)
actions = Action.listall(for_run_name="my-run")

async def list_remote_runs() -> None:
    async for run in Run.listall.aio(sort_by=("created_at", "desc"), limit=20):
        print(run.name, run.phase)

    async for action in Action.listall.aio(for_run_name="my-run"):
        print(action.run_name, action.name, action.phase)

Action.listall(for_run_name=...) lists the actions belonging to one run through the run service. Action.get(run_name=..., name=...) retrieves one action and creates an Action handle containing the server's identifier, metadata, and status.

Most remote APIs decorated with syncify expose both a synchronous call and an .aio(...) form. RunDetails.inputs() and RunDetails.outputs() are exceptions: they are plain asynchronous methods and must be awaited directly once details have been obtained.

Monitor status and completion

Use the handle's phase properties for a snapshot, and wait or watch when the execution is changing.

from flyte.remote import Run


async def wait_for_run(run_name: str) -> None:
    run = await Run.get.aio(name=run_name)

    print("phase:", run.phase)          # for example, "PHASE_RUNNING"
    print("raw phase:", run.raw_phase)  # protobuf enum value
    print("done:", run.done())

    await run.wait.aio(quiet=True)
    details = await run.details.aio()
    print("final phase:", details.action_details.phase)
    print("attempts:", details.action_details.attempts)
    print("runtime:", details.action_details.runtime)

Run.phase delegates to the root Action.phase, which returns the protobuf phase name as a string. raw_phase returns the protobuf enum value. ActionDetails adds is_running, attempts, runtime, error_info, and abort_info. done() is true only for the terminal phases used by the action terminal-state check: failed, succeeded, aborted, and timed out; a phase that is merely not running is not automatically considered done.

Run.wait delegates to Action.wait. With the default wait_for="terminal", it watches status updates until completion and displays a Rich progress display unless quiet=True. The action implementation also supports the running and logs-ready milestones when called directly:

from flyte.remote import Action


async def wait_for_readiness(run_name: str, action_name: str) -> None:
    action = await Action.get.aio(run_name=run_name, name=action_name)
    await action.wait(wait_for="running")
    await action.wait(wait_for="logs-ready")

For each streamed update, use Action.watch. It updates the action's cached details and stops at the requested milestone or when the action reaches a terminal phase.

from flyte.remote import Action


async def print_updates(run_name: str, action_name: str) -> None:
    action = await Action.get.aio(run_name=run_name, name=action_name)
    async for update in action.watch(cache_data_on_done=True):
        print(update.phase, update.runtime, update.attempts)
        if update.error_info is not None:
            print(update.error_info)

When cache_data_on_done=True, Action.watch requests outputs after a terminal update. ActionDetails.watch is the lower-level class method: it calls WatchActionDetails, yields ActionDetails instances, and stops when one is terminal. ActionDetails.watch_updates additionally replaces the instance's protobuf with the terminal update.

Read inputs and outputs

Fetch a detailed run or action, then retrieve its data asynchronously:

from flyte.remote import ActionDetails, RunDetails


async def read_io(run_name: str, action_name: str | None = None) -> None:
    if action_name:
        details = await ActionDetails.get.aio(run_name=run_name, name=action_name)
    else:
        details = await RunDetails.get.aio(name=run_name)

    inputs = await details.inputs()
    print(inputs["input_name"])

    try:
        outputs = await details.outputs()
    except RuntimeError as error:
        print(error)
        return

    print(outputs[0])

ActionDetails.inputs() and outputs() lazily call _cache_data(). That method invokes GetActionData and uses the task or trace interface to convert protobuf literals to native Python values. If no task or trace interface is present, conversion falls back to an empty dictionary or tuple, while the serialized protobuf remains available through pb2.

ActionInputs is a UserDict carrying both data and the original protobuf pb2; it therefore has mapping semantics. Its representation is produced from the protobuf literals using flyte.types.literal_string_repr. ActionOutputs is an immutable tuple of native values, not a mapping. Use positional access for converted values and outputs.pb2 when the serialized output literal is needed.

Outputs are not available until the action is terminal and its data can be cached. ActionDetails.outputs() raises a RuntimeError with that explanation when data is not yet available. Inputs can still be read before completion. The CLI's flyte get io command follows the same pattern: it reads inputs, attempts outputs, and displays not yet available if output retrieval fails.

Stream execution logs

For the root action, use Run.show_logs; for a named action, use Action.show_logs. The run wrapper defaults max_lines to 100, while the action and log viewer defaults are 30.

from flyte.remote import Run


async def follow_logs(run_name: str) -> None:
    run = await Run.get.aio(name=run_name)
    await run.show_logs.aio(
        attempt=None,       # use the latest attempt
        max_lines=50,
        show_ts=True,
        raw=True,
        filter_system=True,
    )

Action.show_logs obtains details first. If the action is neither running nor terminal, it waits for logs-ready; if no attempt is supplied, it selects details.attempts. It then calls Logs.create_viewer in remote/_logs.py.

Logs.create_viewer rejects attempts below 1. In raw mode it consumes Logs.tail and prints formatted lines immediately. Otherwise it creates an AsyncLogViewer, which retains a bounded recent window (max_lines) in a deque and renders a live Rich view. In IPython, the method automatically falls back to raw console output when widgets are unavailable. show_ts adds timestamps, and filter_system=True suppresses system-originated lines and lines containing [flyte] unless they contain flyte.errors.

For direct access to the internal stream, Logs.tail yields individual protobuf LogLine values after flattening the log-service responses:

from flyte.remote import Action
from flyte.remote._logs import Logs


async def consume_log_lines(run_name: str, action_name: str) -> None:
    action = await Action.get.aio(run_name=run_name, name=action_name)
    async for line in Logs.tail.aio(action_id=action.action_id, attempt=1):
        print(line)

Logs.tail retries RPC failures using its retry limit and a two-second delay. If the service continues to return NOT_FOUND after the retry threshold, it raises LogsNotYetAvailableError. A caller using the lower-level API can handle that condition explicitly:

from flyte.errors import LogsNotYetAvailableError
from flyte.remote import Action
from flyte.remote._logs import Logs


async def try_logs(run_name: str, action_name: str) -> None:
    action = await Action.get.aio(run_name=run_name, name=action_name)
    try:
        async for line in Logs.tail.aio(action_id=action.action_id):
            print(line)
    except LogsNotYetAvailableError:
        print("Logs are not available yet")

Abort a run

Retrieve the run and call abort after client configuration has been initialized:

from flyte.remote import Run


async def abort_run(run_name: str) -> None:
    run = await Run.get.aio(name=run_name)
    await run.abort.aio()

Run.abort sends AbortRun with the run identifier. A NOT_FOUND gRPC response is treated as already absent and returns without raising; other gRPC errors are re-raised.

The CLI implementation in cli/_abort.py performs the same operation synchronously:

flyte abort run my-run

It initializes the selected project and domain, calls Run.get(name=run_name), then calls r.abort().

CLI equivalents

The CLI uses the same remote wrappers for common inspection workflows:

flyte get run
flyte get run my-run
flyte get action my-run
flyte get action my-run my-action
flyte get logs my-run
flyte get logs my-run my-action --pretty --lines 50 --show-ts --attempt 2
flyte get io my-run
flyte get io my-run my-action
flyte abort run my-run

flyte get run calls Run.listall, while a named run is rendered from RunDetails.get. flyte get action selects Action.listall or Action.get. flyte get logs selects a Run when only the run name is supplied and an Action when an action name is supplied; its --pretty option selects the bounded live viewer, while the default path passes raw output. flyte get io selects RunDetails for the root action or ActionDetails for a named action.

Caveats to account for

• Run.sync() and Action.sync() are explicit placeholders that return the existing object; they do not refresh remote state. Use details() or watch() for current details.
• Run.watch is declared async def but returns self.action.watch(...) rather than yielding updates itself. For an async update stream, prefer Action.watch or ActionDetails.watch directly.
• ActionDetails.get accepts a positional uri parameter, but its current implementation validates that either uri or both names are present and then constructs the identifier from run_name and name; the URI itself is not used to build the request.
• Run.__post_init__ requires the wrapped protobuf to contain an action field and raises RuntimeError("Run does not have an action") otherwise.
• Logs.tail has an implementation default of attempt 1, despite a docstring saying default 0. Action.show_logs normally chooses the latest attempt from action details.
• The RunDetails and ActionDetails data methods are asynchronous even when the lookup methods have synchronous wrappers. Call await details.inputs() and await details.outputs().

A production integration in _internal/imagebuild/remote_builder.py combines these operations: it submits a remote task, waits quietly with await run.wait.aio(quiet=True), fetches details, compares run_details.action_details.raw_phase with run_definition_pb2.PHASE_SUCCEEDED, and then reads typed outputs. This is the same lifecycle to use when a remote build or another caller must not consume outputs until completion.