Execution Contexts and Runtime State

Two context layers

When a task needs a temporary file, a logger, an execution identifier, or task-specific runtime objects, use flytekit.current_context(). That public function returns the ExecutionParameters stored inside the active execution state; it does not return the internal FlyteContext:

from flytekit import task, current_context

@task
def process_data(input_data: str) -> str:
    ctx = current_context()
    ctx.logging.info(f"Processing data for execution ID: {ctx.execution_id.name}")

    temp_file_path = ctx.working_directory + "/temp_output.txt"
    with open(temp_file_path, "w") as f:
        f.write(f"Processed: {input_data}")

    ctx.default_deck.append(f"## Data Processed\nProcessed input: `{input_data}`")

    if ctx.has_attr("SPARK_SESSION"):
        spark = ctx.get("SPARK_SESSION")
        ctx.logging.info(f"Spark session available: {spark}")

    return f"Processed {input_data} successfully!"

The layers are arranged as follows:

FlyteContextManager
    active stack of FlyteContext objects
        FlyteContext
            compilation_state       (CompilationState, when compiling)
            execution_state          (ExecutionState, when executing)
                user_space_params    (ExecutionParameters, for task code)
            file_access
            flyte_client
            serialization_settings
            worker_queue
            output_metadata_tracker

FlyteContextManager scopes the internal FlyteContext. FlyteContext.execution_state carries the execution mode, directories, branch state, and user parameters. Its user_space_params property simply delegates to execution_state.user_space_params. By contrast, ExecutionParameters is the user-centric object documented as accessible from every task through flytekit.current_context().

Compilation uses the same internal context mechanism, but carries a CompilationState instead of—or in addition to—an execution state. CompilationState stores the node list built while walking a workflow, a node-name prefix, a mode, and a task resolver. FlyteContext.new_compilation_state() creates a default state with flytekit's default task resolver; derived contexts attach it with with_compilation_state(...).

Execution modes are behavioral switches

Refer to modes as ExecutionState.Mode.<value>. The implementation's Mode enum is nested in ExecutionState; there is no separate top-level mode declaration to use at call sites.

Mode	Behavior represented by the source comments and runtime paths
TASK_EXECUTION	Task execution that mimics the runtime environment. Dynamic tasks use this kind of runtime-like path to extract a runtime specification.
LOCAL_WORKFLOW_EXECUTION	A workflow is run locally; task results are handled as node outputs rather than simply calling the task function as ordinary Python.
LOCAL_TASK_EXECUTION	A task function runs purely locally, without a container or Propeller.
DYNAMIC_TASK_EXECUTION	A dynamic task executes through the dynamic-task path and can compile a generated workflow.
EAGER_EXECUTION	Eager calls use the context's worker queue.
EAGER_LOCAL_EXECUTION	Local eager execution.
LOCAL_DYNAMIC_TASK_EXECUTION	A generated dynamic workflow is executed locally.

The distinction between TASK_EXECUTION and LOCAL_TASK_EXECUTION matters. In the dynamic-task implementation, runtime-like dynamic execution compiles a generated workflow, while local dynamic execution invokes the generated workflow locally. Changing the mode therefore changes dispatch behavior; it is not just descriptive metadata.

ExecutionState.is_local_execution() returns True only for LOCAL_TASK_EXECUTION, LOCAL_WORKFLOW_EXECUTION, EAGER_LOCAL_EXECUTION, and LOCAL_DYNAMIC_TASK_EXECUTION. TASK_EXECUTION, DYNAMIC_TASK_EXECUTION, and EAGER_EXECUTION are not included. Several runtime decisions use this predicate, including local error handling and local secret lookup.

Working and engine directories

Construct an ExecutionState with a truthy working directory. Construction immediately creates an engine directory:

from flytekit.core.context_manager import ExecutionState

state = ExecutionState(
    working_dir="/tmp/flyte-work",
    mode=ExecutionState.Mode.LOCAL_TASK_EXECUTION,
)

ExecutionState stores working_dir as supplied. If engine_dir is omitted, it uses <working_dir>/engine_dir; if it is supplied, that path is used instead. In either case, pathlib.Path(self.engine_dir).mkdir(parents=True, exist_ok=True) runs during construction. Omitting or passing an empty working_dir raises ValueError("Working directory is needed").

The state also contains branch_eval_mode and an optional user_space_params. Use with_params() to derive another state while retaining values that you do not replace:

child_state = state.with_params(
    engine_dir="/tmp/flyte-work/custom-engine",
    mode=ExecutionState.Mode.TASK_EXECUTION,
)

with_params() constructs a new ExecutionState, inheriting the current working directory, engine directory, branch state, and user parameters when replacement arguments are not supplied. Its replacement logic uses truthiness checks, so it is intended for replacing valid, non-empty values—not for clearing a field by passing None or an empty value.

For contexts created from an existing FlyteContext, prefer the context builder methods. ctx.with_execution_state(state) returns a builder, and FlyteContextManager.with_context(...) scopes the resulting context instead of changing the active base context:

ctx = FlyteContextManager.current_context()
with FlyteContextManager.with_context(
    ctx.with_execution_state(
        ctx.execution_state.with_params(
            mode=ExecutionState.Mode.TASK_EXECUTION
        )
    )
):
    # Calls here observe the derived execution state.
    pass

Local conditional state

BranchEvalMode has two enum values: BRANCH_ACTIVE and BRANCH_SKIPPED. None means that the execution state is not currently inside a conditional section. This state is used for local execution.

ExecutionState.take_branch() sets branch_eval_mode to BRANCH_ACTIVE; branch_complete() sets it to BRANCH_SKIPPED. The methods mutate the current state with object.__setattr__, so do not assume that every execution-state transition is immutable.

FlyteContext.Builder.enter_conditional_section() creates the conditional scope. For local execution, the first conditional section is initialized as skipped. The local conditional implementation then evaluates each case and activates the selected one:

class LocalExecutedConditionalSection(ConditionalSection):
    def start_branch(self, c: Case, last_case: bool = False) -> Case:
        added_case = super().start_branch(c, last_case)
        ctx = FlyteContextManager.current_context()
        if self._selected_case is None:
            if c.expr is None or c.expr.eval() or last_case:
                ctx.execution_state.take_branch()
                self._selected_case = added_case
        return added_case

    def end_branch(self) -> Union[Condition, Promise]:
        ctx = FlyteContextManager.current_context()
        ctx.execution_state.branch_complete()
        if self._last_case and self._selected_case:
            FlyteContextManager.pop_context()
            # Return the selected branch's promise or report its error.

A skipped local branch is consequently represented in the active execution state rather than executed as an ordinary task. At the end of the conditional, end_branch() marks the branch complete and pops the conditional context after the last selected case. Nested conditional handling in enter_conditional_section() preserves this state so a skipped nested section remains skipped.

User parameters, sandboxes, and task attributes

ExecutionParameters holds the runtime values intended for task code:

• working_directory is the user working directory.
• logging and stats expose the configured logging and tagged-stats handlers.
• execution_id, execution_date, and task_id identify the workflow/task runtime when those values are available.
• raw_output_prefix and output_metadata_prefix carry output locations.
• secrets provides a SecretsManager.
• checkpoint provides the configured checkpoint handle.
• decks, default_deck, timeline_deck, and enable_deck expose deck state.

The source documentation explicitly says not to use execution_date or execution_id to drive production logic. They are consistent workflow-run values intended as debugging or output tags. task_id can be absent outside production runtime.

Derive parameters through a builder when you need to preserve the existing runtime values while changing one part:

params = current_context()
params = params.with_enable_deck(True).build()

ExecutionParameters.Builder copies the current stats, date, working directory, execution ID, logger, checkpoint, decks, attributes, output prefixes, task ID, and deck setting. Its build() creates a normal working directory if the selected path does not already exist. As with ExecutionState.with_params(), builder replacement uses truthiness in several places.

For local task isolation and checkpointing, use the task-sandbox derivation used by task dispatch:

params = current_context()
sandbox_params = params.with_task_sandbox().build()

with_task_sandbox() creates a tempfile.mkdtemp() directory, creates __cp below it, installs a SyncCheckpoint whose destination is that directory, and returns a builder pointing at the sandbox. The local task execution path replaces the execution state's user parameters with this built object before dispatching the task. Conversely, accessing params.checkpoint without a configured checkpoint raises NotImplementedError.

Task-specific values are passed as keyword attributes to ExecutionParameters. Lookup uppercases the requested name, so these forms address the same stored key:

if ctx.has_attr("SPARK_SESSION"):
    spark = ctx.get("SPARK_SESSION")
    # Attribute lookup also uppercases the requested name.
    spark_again = ctx.SPARK_SESSION

If the attribute is unavailable, __getattr__ raises an assertion identifying the missing uppercase parameter and suggesting that the task type may be wrong. This is how integrations such as Spark expose task-specific runtime objects without adding them to the common parameter list.

Scoping and lifecycle

FlyteContextManager owns a singleton-style stack stored in the module's ContextVar. The module initializes a default local context at import time. current_context() returns the top stack entry; if the current logical context has no stack—such as in a new thread—it calls initialize() first.

initialize() replaces the current stack and creates local defaults from Config.auto(). It creates <local_sandbox_path>/user_space, constructs local execution identifiers, uses MockStats and user_space_logger, and creates default ExecutionParameters with that user-space directory and the default file-access raw-output prefix. It also installs the SIGINT handler only when called from the main thread.

Use with_context(...) for scoped changes:

ctx = FlyteContextManager.current_context()
builder = ctx.new_builder()
builder.with_new_compilation_state()

with FlyteContextManager.with_context(builder) as scoped:
    print(f"Current context level: {scoped.level}")
    print(f"Is in compilation mode: {scoped.compilation_state is not None}")

with_context() builds and pushes the context, records the stack depth, and restores that depth in finally. The cleanup loop intentionally pops any contexts leaked above the entry depth; conditional syntax can leave a conditional context on the stack when evaluation or compilation fails. Manual push_context() and pop_context() exist, but the class documentation recommends the scoped form. FlyteContextManager is documented as not thread-safe, even though the stack is held in a ContextVar.

Builders preserve unrelated state when nesting. The context-manager test changes a client in an inner context, then derives a new compilation state from the outer context and verifies that the outer client remains available:

ctx = FlyteContextManager.current_context()
b = ctx.new_builder()
b.flyte_client = SampleTestClass(value=1)
with FlyteContextManager.with_context(b) as outer:
    b = outer.new_builder()
    b.flyte_client = SampleTestClass(value=2)
    with FlyteContextManager.with_context(b) as inner:
        assert inner.flyte_client.value == 2

    with FlyteContextManager.with_context(outer.with_new_compilation_state()) as compiled:
        assert compiled.flyte_client.value == 1

Clients, serialization, and worker queues

FlyteContext is the carrier for runtime objects that should be shared by the current scoped operation. Its builder supports with_file_access, with_client, with_serialization_settings, with_worker_queue, with_output_metadata_tracker, with_compilation_state, and with_execution_state.

Production task context

The production entrypoint constructs ExecutionParameters from backend execution and task identifiers, timestamps, stats, logging, user workspace paths, output prefixes, checkpoint configuration, and task ID. It attaches those parameters to an ExecutionState in TASK_EXECUTION, adds output metadata tracking, and scopes the resulting context. This is why a task's current_context() exposes backend-derived identifiers and output settings rather than only the local defaults.

The entrypoint also reads the serialized context environment variable _F_SS_C and reconstructs SerializationSettings with SerializationSettings.from_transport(...). The resulting settings are attached to the FlyteContext with with_serialization_settings(...), allowing serialization and registration code to observe the same settings through the scoped context.

Remote clients and file access

Remote operations can derive a context with a control-plane file-access provider and a synchronous client:

with FlyteContextManager.with_context(
    ctx.with_file_access(provider).with_client(remote.client)
):
    res = loop_manager.run_sync(parent_wf.run_with_backend, a=1, b=100)

The remote implementation creates a dedicated FileAccessProvider below its configured local sandbox's control_plane_metadata directory. Adding the client with with_client(...) and the provider with with_file_access(...) scopes those remote dependencies to the operation.

Eager worker queues

Eager execution stores a Controller in FlyteContext.worker_queue. The central entity call handler requires this queue for EAGER_EXECUTION; without it, the eager path raises an assertion. Eager setup creates the controller with remote and serialization settings, registers signal handlers, and places it in a derived context. Calls then add tasks to the queue.

The worker queue also propagates the eager root name through _F_EE_ROOT. Its polling thread temporarily pushes the remote context while publishing a deck and pops it afterward. Thus the queue is not merely a collection of pending tasks: it is another runtime dependency carried through the active FlyteContext.

Practical boundaries and failure cases

• Use ExecutionState.Mode.<value>, not a presumed standalone Mode class.
• Always provide a truthy working_dir when constructing ExecutionState; construction creates engine_dir immediately.
• Prefer with_context() to manual stack operations. Popping the base context leaves no valid context and raises an assertion.
• Do not expect with_params() or the parameter builder to clear values with falsey replacements; their inheritance checks use truthiness.
• Checkpointing is optional. ExecutionParameters.checkpoint raises NotImplementedError unless the runtime or with_task_sandbox() supplied one.
• Eager execution requires FlyteContext.worker_queue.
• Local secret resolution differs by mode: SecretsManager checks the configured prefix and, for local execution, also checks an unprefixed environment variable. In task execution, the configured prefix is required by that lookup path.
• current_context() is the user-facing API, but it assumes an execution state with user parameters. Internal compilation-only contexts may not satisfy that assumption; use the internal FlyteContextManager.current_context() when working with compilation or runtime infrastructure.