Skip to main content

System and Context Errors

When you run tasks, interact with the remote control plane, or fetch outputs and execution logs, runtime failures can stem from platform infrastructure, uninitialized state, or missing execution context. Understanding how flyte-sdk categorizes and raises system and context errors enables you to handle platform anomalies cleanly and fix configuration or context misuse early.

Error Hierarchy and Taxonomy​

flyte-sdk defines its core runtime exceptions in errors.py. The base class for platform and runtime failures is BaseRuntimeError, which extends Python's standard RuntimeError with structured metadata:

from typing import Literal

ErrorKind = Literal["system", "unknown", "user"]

class BaseRuntimeError(RuntimeError):
    def __init__(self, code: str, kind: ErrorKind, root_cause_message: str, worker: str | None = None):
        super().__init__(root_cause_message)
        self.code = code
        self.kind = kind
        self.worker = worker

The error hierarchy splits platform-level failures from user-level bugs through the kind discriminator:

RuntimeError (built-in)
├── BaseRuntimeError (errors.py: code, kind, worker)
│   ├── InitializationError (kind defaults to user or system at call sites)
│   ├── LogsNotYetAvailableError (code="LogsNotYetAvailable", kind="system")
│   ├── RuntimeSystemError (kind="system")
│   │   └── UnionRpcError (backend gRPC/network failures)
│   └── RuntimeUserError (kind="user")
│       ├── NotInTaskContextError
│       └── ReferenceTaskError (code="ReferenceTaskUsageError")
└── ActionNotFoundError (direct subclass of built-in RuntimeError)

System Errors vs. User Errors​

  • RuntimeSystemError (errors.py): Raised when execution fails due to platform infrastructure, remote scheduling issues, or backend unavailability. It sets kind="system".
  • UnionRpcError (errors.py): Subclasses RuntimeSystemError to represent communication failures against the backend gRPC services (such as endpoint connectivity loss or network timeouts).
  • ActionNotFoundError (errors.py): Raised when attempting to access an action ID, sub-action, or output that does not exist. Unlike other errors in flyte-sdk, ActionNotFoundError directly subclasses Python's standard RuntimeError rather than BaseRuntimeError, meaning it does not carry .code, .kind, or .worker attributes.

Initialization and Context Guards​

Handling InitializationError​

If you attempt to invoke remote execution launchers, access cloud object storage, or call client APIs before configuring the SDK session, flyte-sdk raises InitializationError.

import flyte

# Incorrect: Interacting with remote entities or storage before initialization
# Raises flyte.errors.InitializationError: Client has not been initialized.
run = flyte.get_run("run-12345")

To resolve this, initialize the client session using flyte.init() with your target endpoint or storage parameters:

import flyte

# Correct: Initialize the session with connection settings
flyte.init(
    endpoint="localhost:8080",
    insecure=True,
    project="flytesnacks",
    domain="development",
)

# Now remote lookups and run submissions succeed
run = flyte.get_run("run-12345")

Internally, modules such as _initialize.py, storage/_storage.py, and remote/_data.py protect resources using guard functions and decorators:

import functools
import typing
from flyte.errors import InitializationError

T = typing.TypeVar("T", bound=typing.Callable)

def ensure_client():
    if _get_init_config() is None or _get_init_config().client is None:
        raise InitializationError(
            "ClientNotInitializedError",
            "user",
            "Client has not been initialized. Call flyte.init() with a valid endpoint"
            " or api-key before using this function.",
        )

def requires_storage(func: T) -> T:
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        if _get_init_config() is None or _get_init_config().storage is None:
            raise InitializationError(
                "StorageNotInitializedError",
                "user",
                f"Function '{func.__name__}' requires storage to be initialized. "
                f"Call flyte.init() with a valid storage configuration before using this function.",
            )
        return func(*args, **kwargs)
    return typing.cast(T, wrapper)

Because InitializationError inherits __init__ directly from BaseRuntimeError, call sites pass (code, kind, root_cause_message, worker=None).

Handling NotInTaskContextError​

NotInTaskContextError is raised when internal controllers or user code attempt to access task runtime data (such as action IDs, output paths, or raw task contexts) outside of an active task execution.

In _internal/controllers/_local_controller.py, context lookup verifies the active task state:

import flyte.errors

ctx = internal_ctx()
tctx = ctx.data.task_context
if not tctx:
    raise flyte.errors.NotInTaskContextError("BadContext", "Task context not initialized")

If you need execution context values (like raw action IDs or execution IDs), ensure your code runs within a decorated @flyte.task function during runtime rather than at the top-level script scope.

Task Lifecycle and Platform Service Errors​

Backend Service Failures and Run Creation​

When launching executions via flyte.run(), flyte-sdk converts gRPC communication errors into structured RuntimeSystemError or RuntimeUserError exceptions based on the gRPC status code returned by the server:

import grpc
import flyte.errors

try:
    pass
except grpc.aio.AioRpcError as e:
    if e.code() == grpc.StatusCode.UNAVAILABLE:
        raise flyte.errors.RuntimeSystemError(
            "SystemUnavailableError",
            "Flyte system is currently unavailable. check your configuration, or the service status.",
        ) from e
    elif e.code() == grpc.StatusCode.INVALID_ARGUMENT:
        raise flyte.errors.RuntimeUserError("InvalidArgumentError", e.details())
    elif e.code() == grpc.StatusCode.ALREADY_EXISTS:
        raise flyte.errors.RuntimeUserError(
            "RunAlreadyExistsError",
            f"A run with the name '{run_name}' already exists. Please choose a different name.",
        )
    else:
        raise flyte.errors.RuntimeSystemError(
            "RunCreationError",
            f"Failed to create run: {e.details()}",
        ) from e

Catch RuntimeSystemError to handle transient infrastructure outages or backend connection loss:

import flyte
from flyte.errors import RuntimeSystemError

try:
    run = flyte.run(my_task, x=10)
except RuntimeSystemError as e:
    if e.code == "SystemUnavailableError":
        print("Backend service unreachable. Check network endpoint and cluster status.")
    else:
        print(f"System error: {e}")

Reference Task Constraints and ReferenceTaskError​

ReferenceTaskError signals invalid usage or resolution failures when working with remote reference tasks (remote/_task.py). Two main constraints enforce valid reference task invocation:

  1. Remote Execution Only: Reference tasks cannot be executed inside the local controller. In _internal/controllers/_local_controller.py:
from typing import Any
import flyte.errors

async def submit_task_ref(
    self, _task: Any, max_inline_io_bytes: int, *args, **kwargs
) -> Any:
    raise flyte.errors.ReferenceTaskError("Reference tasks cannot be executed locally, only remotely.")
  1. Keyword Arguments Requirement: When calling a reference task proxy, all arguments must be supplied as keyword arguments:
import flyte.errors

if len(args) > 0:
    raise flyte.errors.ReferenceTaskError(
        f"Reference task {task_name} does not support positional arguments"
        f"currently. Please use keyword arguments."
    )

To invoke a reference task correctly, pass parameters by keyword within a remote run workflow:

import flyte

ref_task = flyte.Task.get(name="data_prep_task", version="v1")

@flyte.task
def workflow_task():
    # Correct: Pass arguments as keyword arguments
    ref_task(dataset_path="s3://my-bucket/data.csv", batch_size=64)

Log Streaming and LogsNotYetAvailableError​

When tailing execution logs from the logs service using Logs.tail() in remote/_logs.py, flyte-sdk polls the backend gRPC logs endpoint. If the container is still initializing or logs have not yet been flushed, the service returns grpc.StatusCode.NOT_FOUND.

After exhausting retries (defaulting to 5 attempts with 2-second sleep intervals), flyte-sdk raises LogsNotYetAvailableError:

import asyncio
import grpc
from flyte.errors import LogsNotYetAvailableError

retries = 0
retry = 5

try:
    pass
except grpc.aio.AioRpcError as e:
    retries += 1
    if retries >= retry:
        if e.code() == grpc.StatusCode.NOT_FOUND:
            raise LogsNotYetAvailableError(
                f"Log stream not available for action {action_id.name} in run {action_id.run.name}."
            )
    else:
        await asyncio.sleep(2)

LogsNotYetAvailableError initializes with fixed attributes code="LogsNotYetAvailable" and kind="system". When interacting with the log streaming API or terminal UI viewers, catch LogsNotYetAvailableError to handle delayed pod initialization gracefully:

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

try:
    for line in Logs.tail(action_id=action.id):
        print(line.message)
except LogsNotYetAvailableError:
    print("Action logs are not yet available. Wait for the pod container to start.")