Debugging and Introspection

Trace a task back to its source file

When a task needs debugging or deployment tracing, inspect the source_file property on the task object created by @env.task:

import flyte

env = flyte.TaskEnvironment("example")

@env.task
async def example_task(x: int, y: str) -> str:
    return f"{x} {y}"

print(example_task.source_file)

if __name__ == "__main__":
    flyte.run(example_task, 1, y="hello")

TaskEnvironment.task normally creates an AsyncFunctionTaskTemplate for a decorated function and stores the resulting template in the environment's task registry. The template keeps the original callable in func. source_file checks whether that callable has a truthy __code__ object and, when it does, returns func.__code__.co_filename; otherwise it returns None.

source_file is a property, so access it as example_task.source_file, not example_task.source_file(). The value is the filename recorded by the function's code object. The property does not call inspect.getsourcefile or unwrap decorators, so callable objects, builtins, and wrappers without a usable __code__ can return None.

Use the same metadata in deployment failures

Deployment error handling reads task.source_file while constructing a flyte.errors.DeploymentError:

raise flyte.errors.DeploymentError(
    f"Failed to deploy task {task.name} file{task.source_file} with image {image_uri}, Error: {e!s}"
) from e

Consequently, a deployment error for an AsyncFunctionTaskTemplate includes the source filename when the wrapped function exposes a code object. A missing source filename is represented by None rather than by a path discovered through a separate source-inspection mechanism.

Understand which path is being debugged

The same template has different invocation paths depending on the execution context. This matters when a local debugging call does not exercise the runtime lifecycle.

Local invocation calls the function directly

Outside a task context, inherited TaskTemplate.__call__ calls AsyncFunctionTaskTemplate.forward. forward invokes self.func(*args, **kwargs) without awaiting it:

result = example_task(1, y="hello")

For the asynchronous example_task above, result is the coroutine returned by the wrapped function; the caller is responsible for awaiting it. This direct local path does not run pre, replace the task context, or run post.

Invocation from a task context submits to the controller

Inside a task context, TaskTemplate.__call__ obtains the controller and submits the template instead of calling forward. An asynchronous template uses controller.submit(self, *args, **kwargs). If the wrapped function is synchronous, AsyncFunctionTaskTemplate.__post_init__ sets _call_as_synchronous to True; the inherited call path then uses controller.submit_sync and returns the completed future's result.

This means inspecting source_file is safe and local, but invoking the task from another running task can involve controller submission rather than immediate Python execution. If the task context has no initialized controller, the call raises a RuntimeSystemError with the BadContext code.

Runtime execution runs lifecycle hooks

The runtime task runner calls task.execute(**inputs):

@log
async def run_task(
    tctx: TaskContext, controller: Controller, task: TaskTemplate, inputs: Dict[str, Any]
) -> Tuple[Any, Optional[Exception]]:
    try:
        logger.info(f"Parent task executing {tctx.action}")
        outputs = await task.execute(**inputs)
        logger.info(f"Parent task completed successfully, {tctx.action}")
        return outputs, None
    except RuntimeSystemError as e:
        logger.exception(f"Task failed with error: {e}")
        return {}, e
    except RuntimeUnknownError as e:
        logger.exception(f"Task failed with error: {e}")
        return {}, e
    except RuntimeUserError as e:
        logger.exception(f"Task failed with error: {e}")
        return {}, e
    except Exception as e:
        logger.exception(f"Task failed with error: {e}")
        return {}, CustomError.from_exception(e)

AsyncFunctionTaskTemplate.execute requires an active internal_ctx().data.task_context; it asserts if that context is absent. With the context present, it:

1. Awaits pre(*args, **kwargs).
2. Replaces the task context with the data returned by pre.
3. Awaits the wrapped function when func is a coroutine function, or calls it synchronously otherwise.
4. Awaits post(v).
5. Returns v.

The value returned by post is not used: execute returns the original v. Use ordinary task invocation for local function debugging; execute is the runtime path used by the task runner and is not a general-purpose direct-call API.

Inspect module identity when resolution matters

The wrapped func is also used by extract_task_module in _internal/resolvers/_task_module.py. For an AsyncFunctionTaskTemplate, the resolver obtains the module with inspect.getmodule(task.func), reads that module's __file__, and computes a dotted module path relative to the supplied source_dir. It uses task.func.__name__ as the entity name.

This gives a second useful debugging check: a task can have a source_file while still failing module resolution if its module cannot be inspected, its __file__ is unavailable, or its file is outside the resolver's source_dir. The relative-path operation raises ValueError when the module file is not under that directory. For a task loaded as __main__, resolution substitutes the executing script's filename stem; an interactive __main__ without __file__ cannot provide that fallback.

Account for synchronous functions and extensions

Despite its name, AsyncFunctionTaskTemplate can wrap a synchronous function. The decorator passes either kind of function to the same template when no plugin configuration is supplied:

import flyte

env = flyte.TaskEnvironment("example")

@env.task
def my_legacy_task(x: int) -> int:
    return x

print(my_legacy_task.source_file)

For this function, __post_init__ marks the template as synchronous. Local calls go through forward and return the integer directly; calls from a task context use the controller's synchronous submission path. The .aio method inherited from TaskTemplate is the awaitable path for calling a synchronous task from asynchronous task code.

A reusable environment with concurrency greater than one rejects a synchronous function during decoration. Use an async function or configure reusable concurrency to one. If TaskEnvironment.plugin_config is set, the decorator does not select the default AsyncFunctionTaskTemplate; it looks up a template class in flyte.extend.TaskPluginRegistry using the exact type of that configuration object. Missing registration raises ValueError during decoration. A plugin template can therefore change the concrete template used for introspection, so verify the registered class before relying on subclass-specific behavior.