Skip to main content

Inspect Execution State, Inputs, Outputs, and Timing

Object model and retrieval​

A remote Flyte execution exposes two useful detail views. RunDetails wraps a run_definition_pb2.RunDetails protobuf and, in __post_init__, creates an embedded ActionDetails from pb2.action. ActionDetails is the detail view for one action and is backed by run_definition_pb2.ActionDetails. Consequently, run-level input/output access delegates to the action detail object: RunDetails.inputs() calls self.action_details.inputs(), and RunDetails.outputs() calls self.action_details.outputs().

Name-based retrieval uses the initialized Flyte client configuration. Initialize Flyte with an endpoint or API key and configured org, project, and domain before calling these methods. ActionDetails.get(run_name=..., name=...) constructs an action identifier from those configured values; RunDetails.get(name=...) constructs a run identifier in the same way. Both classes call ensure_client() before making their remote-service request.

import flyte
from flyte.remote import ActionDetails, RunDetails

flyte.init_from_config()

action_details = ActionDetails.get(
    run_name="my-run",
    name="a0",
)
run_details = RunDetails.get(name="my-run")

ActionDetails.get_details(action_id) and RunDetails.get_details(run_id) are the identifier-based alternatives. The retrieval methods are decorated with syncify, so they have synchronous forms and .aio() forms. In an asynchronous application, use the async form and await the I/O methods:

import flyte
from flyte.remote import RunDetails


async def inspect_run(run_name: str) -> None:
    flyte.init_from_config()
    details = await RunDetails.get.aio(name=run_name)
    inputs = await details.inputs()
    print(details.name, inputs)

When an existing flyte.remote.Run is available, its details() method supplies the same RunDetails view; asynchronous callers use await run.details.aio(). The run object also delegates inputs() and outputs() to its details object.

For live action state, ActionDetails.watch(action_id) yields successive ActionDetails snapshots and returns after a snapshot satisfies done(). The instance method watch_updates(cache_data_on_done=False) watches using the current action ID, updates self.pb2 with the terminal snapshot, and can cache inputs and outputs after completion when cache_data_on_done=True.

State, identity, errors, and runtime​

ActionDetails exposes both convenient and protobuf-level state. phase converts the stored phase to an ActionPhase; raw_phase returns the phase_pb2.ActionPhase value. is_running compares the raw value with ACTION_PHASE_RUNNING. Identity and metadata are available through name, run_name, task_name, action_id, metadata, and status. task_name is None when the action metadata does not contain a task ID.

done() checks the raw phase with the action terminal-state helper. The terminal phases recognized by that check are FAILED, SUCCEEDED, ABORTED, and TIMED_OUT. Terminal means that execution has reached one of those states; it does not guarantee that output data exists. In particular, failed, aborted, or timed-out actions can be terminal without having outputs.

The optional failure and interruption records are exposed without manufacturing a value when the protobuf field is absent:

if details.error_info is not None:
    print(details.error_info)
if details.abort_info is not None:
    print(details.abort_info)

print("phase:", details.phase)
print("running:", details.is_running)
print("attempts:", details.attempts)
print("runtime:", details.runtime)

runtime converts the status start timestamp to UTC. If the status has an end timestamp, it returns end_time - start_time; otherwise it returns the elapsed time from the start timestamp to the current UTC time. attempts is the attempt count in status.attempts. Per-attempt log availability can be checked with logs_available(attempt=None), where the default selects the latest attempt.

The original protobuf remains available as details.pb2. This is useful when a high-level property does not expose the detail needed by an integration. For example, the Dask plugin example waits for the run and then reads the latest attempt's log metadata for the root action:

run.wait()

action_details = flyte.remote.ActionDetails.get(run_name=run.name, name="a0")
for log in action_details.pb2.attempts[-1].log_info:
    print(f"{log.name}: {log.uri}")

The debugger helper similarly iterates details.pb2.attempts and returns a log URI only for a ready IDE link. These uses inspect the serialized attempt data directly rather than the convenience status properties.

Converted inputs and outputs​

Inputs and outputs are lazy. ActionDetails._cache_data() calls the remote data proxy, determines the native interface from the action's task template (or trace interface), and uses Flyte's runtime conversion functions to turn protobuf literals into native Python values. The resulting wrappers retain both the converted values and the original protobuf data. Once cached, later calls return the cached wrapper.

ActionInputs is a UserDict, so its data contains converted values and ordinary dictionary operations work directly:

inputs = await details.inputs()
value = inputs["x"]
all_inputs = dict(inputs)

It also has pb2, containing the original common_pb2.Inputs. Its __repr__ uses Flyte's literal representation helper on that protobuf, so the displayed representation need not be the same as dict(inputs).

ActionOutputs is a tuple subclass. Positional indexing, iteration, unpacking, and slicing therefore operate on converted native values. The wrapper creates generated field names such as o0, o1, and o2 when no names are supplied, exposes those values as attributes, and provides the named_outputs dictionary:

outputs = await details.outputs()
first = outputs[0]
first, second = outputs
by_name = outputs.named_outputs

The tuple retains the serialized common_pb2.Outputs as outputs.pb2. This distinction matters because named_outputs and tuple access expose converted values, while protobuf-oriented code can inspect pb2. The remote image builder, for example, extracts a fully qualified image name from outputs.pb2.literals[0].value.scalar.primitive.string_value.

Outputs have an explicit terminal-state restriction. If they are not already cached, ActionDetails.outputs() calls _cache_data(); when that call reports that the action is not terminal, it raises:

RuntimeError: Action is not in a terminal state, outputs are not available. Please wait for the action to complete.

Wait for a run or verify done() before requesting outputs, and still handle missing output data for terminal failures. The CLI follows this behavior: it always retrieves inputs, attempts outputs in a try block, and displays "not yet available" if output retrieval fails. The webhook integration first checks run.done(), then catches output failures because a failed run may have no outputs.

A run detail has the same behavior because RunDetails.outputs() delegates to its embedded ActionDetails. A production-style serialization path can therefore use the wrappers without reaching into protobuf fields:

from flyte.remote import Run

run = await Run.get.aio(name=name)
inputs = await run.inputs.aio()
outputs = None
if run.done():
    try:
        outputs = await run.outputs.aio()
    except Exception:
        pass

inputs_dict = dict(inputs) if inputs is not None else None
outputs_dict = outputs.named_outputs if outputs is not None else None

Attempts and phase timing​

ActionDetails.get_phase_transitions(attempt=None) provides a more granular view than the current phase. It converts each protobuf phase transition in an action attempt into a PhaseTransitionInfo, whose fields are an ActionPhase, a timezone-aware start_time, and an optional end_time. PhaseTransitionInfo is defined in flyte.remote._action; it is returned by the public ActionDetails method even though it is not included in flyte.remote.__all__.

Attempt selection is 1-indexed. With no argument, the method uses self.pb2.status.attempts, which represents the latest attempt. Passing attempt=1 selects the first element in self.pb2.attempts, while attempt=2 selects the second. If there are no attempt records or the requested index is out of range, the method returns an empty list.

transitions = details.get_phase_transitions()       # latest attempt
first_attempt = details.get_phase_transitions(attempt=1)

for transition in transitions:
    print(
        transition.phase,
        transition.start_time,
        transition.end_time,
        transition.duration,
    )

PhaseTransitionInfo.duration uses the end timestamp when one is present and returns abs(end_time - start_time). An ongoing transition has end_time=None; its duration is calculated from the start timestamp to the current UTC time each time the property is read, so it changes while the phase continues.

For the latest attempt, phase_durations turns the transition list into a dictionary keyed by ActionPhase. The convenience properties queued_time, waiting_for_resources_time, initializing_time, and running_time read the corresponding dictionary entries and return None if that phase did not occur:

print(details.queued_time)
print(details.waiting_for_resources_time)
print(details.initializing_time)
print(details.running_time)

The phase properties can account for a latest attempt's lifecycle breakdown. The focused phase-transition tests construct values of 10 seconds queued, 15 seconds waiting for resources, 30 seconds initializing, and 145 seconds running; their sum is 200 seconds. This is a breakdown of the selected attempt, not necessarily the same measure as runtime, which is calculated from the action status's overall start and end timestamps.

There is an important dictionary tradeoff: phase_durations is implemented as {t.phase: t.duration for t in transitions}. If one attempt contains the same phase more than once, a later transition replaces the earlier value for that phase rather than being added to it. Use get_phase_transitions() when the individual intervals—and their order—matter. phase_durations and the convenience properties are best suited to the one-value-per-phase view.

The timing and raw attempt APIs complement rather than replace one another. Use phase, runtime, and done() for current lifecycle state; use get_phase_transitions(attempt=...) for retry-specific timing; and use pb2.attempts when an integration needs log metadata or other fields that are not modeled by the convenience properties.