Skip to main content

Debugging User Code Errors

When task execution fails or workflows behave unexpectedly, errors in flyte-sdk fall into structured classifications. User-originated problems—such as type mismatch between return annotations and outputs, syntax or dependency failures during module loading, improper reference task invocations, and calls made outside a task execution context—raise subclasses of RuntimeUserError.

Understanding how flyte-sdk classifies errors, formats stack traces, and validates data allows you to quickly locate and fix issues in your task and workflow code.

Error Classification and Stack Trace Filtering​

flyte-sdk categorizes execution failures using BaseRuntimeError subclasses:

  • RuntimeUserError: Represents errors in user code, configuration, or data contracts (has kind="user").
  • RuntimeSystemError: Represents infrastructure, controller, or system-level communication failures (has kind="system").
  • RuntimeUnknownError: Represents unexpected exceptions that cannot be categorized (has kind="unknown").

When an unhandled Python exception occurs during task execution in _task.py, flyte-sdk wraps the exception into a RuntimeUserError:

try:
    # Task execution
    ...
except RuntimeSystemError:
    raise
except RuntimeUserError:
    raise
except Exception as e:
    raise RuntimeUserError(type(e).__name__, str(e)) from e

Revealing Full Tracebacks with Debug Logging​

By default, flyte-sdk registers a custom exception hook (sys.excepthook = custom_excepthook in _excepthook.py) that filters out internal engine frames matching _internal, syncify, and _code_bundle. This keeps tracebacks focused on your task code.

If you need to inspect full internal stack traces when debugging complex errors, set the logger level to DEBUG:

import logging
from flyte._logging import logger

logger.setLevel(logging.DEBUG)

When logger.getEffectiveLevel() <= logging.DEBUG, custom_excepthook bypasses filtering and delegates directly to Python's original exception handler.

Data Serialization and Type Validation Errors (RuntimeDataValidationError)​

flyte-sdk validates task outputs against their type annotations during output serialization in _internal/runtime/convert.py via convert_from_native_to_outputs. If validation or conversion fails, flyte-sdk raises RuntimeDataValidationError (error code "DataValiationError").

Issue 1: Missing Return Type Annotation​

If a task returns a value but does not define a return type annotation on the function signature (or annotates the return type as None), flyte-sdk expects zero outputs. Returning a value triggers a validation error.

Symptom​

RuntimeDataValidationError: In task calculate_sum variable o0, failed to serialize/deserialize because of Expected no outputs but got (15,),did you miss a return type annotation?

Problematic Code​

import flyte

@flyte.task
def calculate_sum(a: int, b: int):
    # Missing return type annotation -> interface.outputs is empty
    return a + b

Solution​

Add the return type annotation to the task definition:

import flyte

@flyte.task
def calculate_sum(a: int, b: int) -> int:
    return a + b

Issue 2: Return Value Incompatible with Type Engine​

When the returned value cannot be transformed into a Flyte literal matching the annotated type (for instance, returning an incompatible data structure or a type unsupported by TypeEngine), TypeEngine.to_literal raises TypeTransformerFailedError, which is wrapped in a RuntimeDataValidationError.

Problematic Code​

import flyte

@flyte.task
def fetch_count() -> int:
    # Returns a string where an integer is expected
    return "not_a_number"  # type: ignore

Solution​

Ensure returned values match the annotated types:

import flyte

@flyte.task
def fetch_count() -> int:
    return 42

For multiple outputs, specify a tuple with matching return types:

import flyte

@flyte.task
def compute_metrics(val: int) -> tuple[int, str]:
    return val * 2, f"Processed {val}"

Module Loading and Deployment Errors (ModuleLoadError)​

When deploying workflows via the CLI or loading tasks dynamically, flyte-sdk uses _utils/module_loader.py to discover and import Python files using importlib. If an import fails due to syntax errors, missing packages, or exceptions executing top-level module code, flyte-sdk raises ModuleLoadError (error code "ModuleLoadError").

Symptom​

ModuleLoadError: Failed to load module from /path/to/my_workflow.py: No module named 'missing_package'

Common Causes and Fixes​

  1. Top-Level Code Execution: Avoid executing heavy initialization, database connections, or network requests at the module's top level. Place setup logic inside task bodies or initialization functions.
  2. Missing Dependencies: Ensure all third-party libraries imported by the task files are installed in the deployment environment or specified in the container image requirements.
  3. Ignoring Unrelated Load Failures During Deployment: When running flyte deploy, flyte-sdk scans all Python files in the target directory by default. To skip files that fail to import (such as test utilities or uninstalled optional dependencies), pass the --ignore-load-errors flag:
flyte deploy --ignore-load-errors

Remote Reference Task Errors (ReferenceTaskError)​

ReferenceTaskError (error code "ReferenceTaskUsageError") occurs when interacting with remote tasks via flyte.remote.Task or TaskDetails.

Issue 1: Task Not Found When Resolving Remote Version​

When retrieving a remote task with Task.get(name="...", auto_version="latest"), flyte-sdk queries the remote control plane for the latest task version. If the task does not exist in the specified project and domain, flyte-sdk raises ReferenceTaskError.

Problematic Code​

import flyte

# Raises ReferenceTaskError if 'non_existent_task' is not deployed
remote_task = flyte.remote.Task.get(
    name="non_existent_task",
    project="flytesnacks",
    domain="development",
    auto_version="latest",
)

Solution​

Verify that the task name, project, and domain exist on the remote control plane, or deploy the task before referencing it:

import flyte

remote_task = flyte.remote.Task.get(
    name="my_deployed_task",
    project="flytesnacks",
    domain="development",
    version="v1.0.0",  # Or auto_version="latest" once deployed
)

Issue 2: Passing Positional Arguments to Reference Tasks​

Reference tasks do not accept positional arguments. Calling a reference task with positional arguments raises ReferenceTaskError.

Problematic Code​

# Raises ReferenceTaskError: Reference task my_deployed_task does not support positional arguments currently.
res = await remote_task(10, "value")

Solution​

Pass all inputs as keyword arguments:

res = await remote_task(x=10, name="value")

Issue 3: Executing Reference Tasks in Local Controller​

Reference tasks represent remote task definitions and cannot be executed locally under LocalController. Invoking submit_task_ref locally raises ReferenceTaskError("Reference tasks cannot be executed locally, only remotely.").

Solution​

Run workflows containing reference tasks remotely or mock the reference task during local development and testing.

Task Context Misuse (NotInTaskContextError)​

NotInTaskContextError is raised when calling context-dependent operations that require an active TaskContext outside of an executing task.

In _internal/controllers/_local_controller.py, methods such as get_action_outputs and record_trace verify that internal_ctx().data.task_context is initialized:

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

Similarly, calling Task.get(..., auto_version="current") requires an active task context to resolve the version of the currently executing task. If called outside a task, it raises a ValueError.

Problematic Code​

from flyte.remote import Task

# Attempting to resolve "current" auto_version outside of an active task context
task_ref = Task.get(
    name="sub_task",
    project="flytesnacks",
    domain="development",
    auto_version="current",  # Fails: no active task context exists
)

Solution​

Only request auto_version="current" inside the body of an active task where flyte.ctx() contains an initialized TaskContext:

import flyte
from flyte.remote import Task

@flyte.task
async def parent_task(x: int) -> int:
    # Valid: auto_version="current" resolves the version from the executing task context
    sub_task = await Task.get(
        name="sub_task",
        project="flytesnacks",
        domain="development",
        auto_version="current",
    )
    return await sub_task(val=x)

For code executing outside tasks (such as setup scripts or workflow definition scripts), pass explicit versions or use auto_version="latest".