App Module

app

Public application construction API.

This package provides the entry points for constructing and configuring a FunctualizeApp instance, including preset factory functions for common deployment strategies and grouped configuration objects.

Usage

from functualize.app import FunctualizeApp, JobSources, ConfigSources from functualize.app import classic, twelve_factor

app = FunctualizeApp( "myapp", job_sources=JobSources(directories=["./jobs"]), config_sources=twelve_factor(dotenv=True), )

__all__ = ['FunctualizeApp', 'FallbackCommand', 'DiscoveryConfig', 'JobSources', 'ConfigSources', 'PluginSources', 'ExecutionConfig', 'classic', 'twelve_factor', 'env_only', 'remote_first', 'get_perf_timeline'] module-attribute

ConfigSources(file_pattern='^config\\.(\\w+)\\.(\\w+)$', config_resolution_chain=None, dotenv=True, dotenv_path=None, remote=False, vault_max_age=None) dataclass

Configuration resolution settings.

Controls how the application discovers and resolves configuration: - file_pattern: regex for matching config files (default: config..) - config_resolution_chain: explicit resolution chain (skips file discovery) - dotenv: whether to load .env files (default True) - dotenv_path: explicit path to .env file (None = auto-discover)

When config_resolution_chain is None (default), the boot path builds the classic chain [CliSource, EnvSource, FileSource, DefaultSource] using file discovery. When set to an explicit ResolutionChain (e.g., from twelve_factor()), that chain is used directly without file discovery.

The default pattern requires a <slot> segment but does not pin the extension. Which extensions count is decided by the registered format providers, so a plugin that registers .yaml makes config.prod.yaml discoverable without anyone editing this regex. The pattern used to spell (ini|toml) inline, which meant it silently disagreed with the file reader in both directions: an extension some provider handled could not anchor a directory unless the regex happened to name it.

Since ADR-007 the only extension registered by default is .toml, so config.prod.ini neither anchors nor resolves unless a plugin registers IniFormatProvider. That is a change in the provider set, not in this rule.

file_pattern = '^config\\.(\\w+)\\.(\\w+)$' class-attribute instance-attribute

config_resolution_chain = None class-attribute instance-attribute

dotenv = True class-attribute instance-attribute

dotenv_path = None class-attribute instance-attribute

remote = False class-attribute instance-attribute

Whether to resolve declared remote annotations from the local vault.

Set by remote_first(). It exists because a bare None chain cannot distinguish "build the classic chain" from "build the remote chain" -- and that ambiguity is exactly how remote_first() came to resolve silently as classic() for its whole shipped life (ADR-016). The intent is now a fact in the data rather than an inference from absence.

vault_max_age = None class-attribute instance-attribute

How old the vault may be before every run warns, e.g. "7d".

None means unconfigured, which resolves to the "24h" default. The distinction matters: $FUNCTUALIZE_VAULT_MAX_AGE outranks this field, and a field that defaulted to "24h" could not be told apart from an author who wrote max_age="24h" on purpose.

Only consulted when :attr:remote is set. Exceeding it warns and the run continues -- offline work stays possible (ADR-016).

The literal default lives in _config.vault.DEFAULT_MAX_AGE, not here, and is deliberately not imported: _config.vault pulls in cryptography, and this module is on the cold boot path for every app including the ones that never open a vault. tests/config/ test_vault_staleness.py asserts the two agree, so the duplication cannot drift.

DiscoveryConfig(exclude_patterns=(), extra_directories=(), require_file_prefix=None, require_file_postfix=None, require_file_import=None, require_file_marker=None, require_job_decorators=None, require_job_prefix=None, require_job_postfix=None, pre_filter=None) dataclass

All discovery-related settings for job discovery filtering.

When all require_* fields are None, baseline convention mode applies: all public functions in qualifying files become jobs.

Fields are composable via AND logic — each set field adds a constraint. Uses tuples for immutability and hashability. None means "not configured" (no constraint); empty tuple/string has different semantics.

exclude_patterns = () class-attribute instance-attribute

extra_directories = () class-attribute instance-attribute

require_file_prefix = None class-attribute instance-attribute

require_file_postfix = None class-attribute instance-attribute

require_file_import = None class-attribute instance-attribute

require_file_marker = None class-attribute instance-attribute

require_job_decorators = None class-attribute instance-attribute

require_job_prefix = None class-attribute instance-attribute

require_job_postfix = None class-attribute instance-attribute

pre_filter = None class-attribute instance-attribute

ExecutionConfig(max_invoke_depth=10) dataclass

Execution parameters.

Controls runtime behavior: - max_invoke_depth: maximum nested job invocation depth (prevents infinite recursion)

max_invoke_depth = 10 class-attribute instance-attribute

JobSources(directories=None, functions=None, job_providers=None, children=None, children_glob=None, lazy=True) dataclass

All job source configuration.

Controls where the application discovers job functions: - directories: filesystem paths to scan for job modules - functions: pre-imported callables or Job definitions (static wiring) - job_providers: custom JobProvider instances with optional transforms - children: named child project mappings (namespace → directory) - children_glob: glob pattern for discovering child projects - lazy: whether to use cache-first discovery (default True)

directories = None class-attribute instance-attribute

functions = None class-attribute instance-attribute

job_providers = None class-attribute instance-attribute

Providers to add to the resolution pipeline, in declaration order.

Each entry is a JobProvider or a (provider, [transform, ...]) pair. They are added after whatever the boot path derives from directories and functions, so the pipeline order matches the order these fields are declared in.

The type was list[Any] for as long as the field was read by nothing: boot_static and boot_standard both ignored it, so a caller who declared a provider here got an empty job list and no diagnostic. It is now honoured on both paths by _app.boot.wire_declared_job_providers, and the annotation says what the docstring always promised.

app.extensions.add_job_provider() remains the imperative equivalent -- the path a plugin uses from inside its __call__(app), where there is no JobSources left to declare into.

children = None class-attribute instance-attribute

children_glob = None class-attribute instance-attribute

lazy = True class-attribute instance-attribute

PluginSources(entry_point_group=PLUGINS, explicit_plugins=None, disabled=None, ambient_directory=True) dataclass

Plugin discovery settings.

Controls how plugins are found and loaded: - entry_point_group: entry point group name for plugin discovery - explicit_plugins: list of pre-instantiated plugin objects - disabled: list of plugin names to skip during discovery - ambient_directory: whether the .functualize/plugins/ convention directory in the working directory is loaded

ambient_directory draws the same line adjacent-defects/T14 drew for job discovery: a directory the caller declared is read, and the one the working directory supplies implicitly is not, when the caller asked for one file rather than for a project. [tool.functualize] plugins_directories is declared and is unaffected; only the convention fallback is refused.

It defaults to True, because a project app in its own directory is exactly who that convention is for. func <file>.py <job> sets it False: the cwd there is wherever the user's shell happened to be, and a plugin module's top level runs during app construction — so a stray file under ./.functualize/plugins/ could take over an invocation that named a different program entirely.

entry_point_group = PLUGINS class-attribute instance-attribute

explicit_plugins = None class-attribute instance-attribute

disabled = None class-attribute instance-attribute

ambient_directory = True class-attribute instance-attribute

FunctualizeApp(name, *, job_sources=None, config_sources=None, plugin_sources=None, execution=None, discovery_config=None)

Application kernel — delivery-agnostic.

Contains DI registry, execution engine, lifecycle management, discovery pipeline, config resolution, and RunContext construction. Adapters (CLI, HTTP, Lambda) connect via the facade methods.

Constructor accepts grouped frozen dataclass configs:

app = FunctualizeApp(
    "myapp",
    job_sources=JobSources(directories=["./jobs"]),
    config_sources=ConfigSources(dotenv=False),
    plugin_sources=PluginSources(entry_point_group="myapp.plugins"),
    execution=ExecutionConfig(max_invoke_depth=5),
)

Parameters:

Name	Type	Description	Default
name	str	Application name (used for app dir fallback and logging).	required
job_sources	JobSources | None	Grouped job source configuration (JobSources()).	None
config_sources	ConfigSources | None	Configuration resolution settings (ConfigSources()).	None
plugin_sources	PluginSources | None	Plugin discovery settings (PluginSources()).	None
execution	ExecutionConfig | None	Execution parameters (ExecutionConfig()).	None

Source code in src/functualize/app/core.py

def __init__(
    self,
    name: str,
    *,
    job_sources: JobSources | None = None,
    config_sources: ConfigSources | None = None,
    plugin_sources: PluginSources | None = None,
    execution: ExecutionConfig | None = None,
    discovery_config: DiscoveryConfig | None = None,
):
    from functualize._events.perf import perf_timeline

    perf_timeline.mark("boot.total.start")

    # Phase: app_init — covers top-level imports and config dataclass resolution
    perf_timeline.mark("boot.app_init.start")

    from functualize._app.boot import boot_standard, boot_static

    self.name = name

    # --- Resolve grouped configs ---
    self._job_sources = job_sources if job_sources is not None else JobSources()
    self._config_sources = (
        config_sources if config_sources is not None else ConfigSources()
    )
    self._plugin_sources = (
        plugin_sources if plugin_sources is not None else PluginSources()
    )
    self._execution_config = (
        execution if execution is not None else ExecutionConfig()
    )
    self._discovery_config = discovery_config

    # Where this project's derived run state (fingerprints, history,
    # workflow scopes) lives. Read here, once, at the delivery boundary —
    # the kernel asks *this* object rather than the operating system, which
    # is what stops three call sites from answering the same question
    # differently (see `contributor/architecture/run-model/05-engine-seal.md` §C).
    self._state_root = Path.cwd()
    #: Installed by a plugin at boot; None means the filesystem
    #: default. See the :attr:`substrate` property.
    self._substrate: StoreSubstrate | None = None
    #: Every storage claim registration made — installs and offers alike —
    #: for boot step 6.5 to settle. More than one is a refusal.
    self._substrate_claims: list[tuple[str, SubstrateOffer | None]] = []
    #: Set by boot_standard once `general.max_invoke_depth` resolves.
    self._resolved_max_invoke_depth: int | None = None

    # Extract effective values from resolved configs
    self._jobs_directories = self._job_sources.directories or []
    self._children = self._job_sources.children
    self._children_glob = self._job_sources.children_glob
    self._lazy_boot = self._job_sources.lazy
    # Set by boot_standard when lazy boot wires a CachedDirectoryScanProvider
    self._cached_provider: Any = None
    self._config_file_regex = self._config_sources.file_pattern

    # Detect static wiring fast path: all sources are explicit, zero I/O
    self._static_wiring = self._is_fully_explicit()
    #: Facades. This class is the composition root's public face, and a
    #: 71-member flat surface cannot be shrunk by moving bodies — only by
    #: grouping names (T9). Each is built on first use, so an app that
    #: registers no hooks never constructs one.
    self._hooks_facade: HooksFacade | None = None
    self._configuration_facade: ConfigurationFacade | None = None
    self._extensions_facade: ExtensionsFacade | None = None
    self._workflows_facade: WorkflowScopeFacade | None = None
    self._di_facade: DependencyFacade | None = None
    self._gates_facade: GatesFacade | None = None

    perf_timeline.mark("boot.app_init.end")

    if self._static_wiring:
        boot_static(self, perf_timeline)
    else:
        boot_standard(self, perf_timeline)

job_registry instance-attribute

plugin_loader instance-attribute

config_registry instance-attribute

name = name instance-attribute

event_bus property

The central event bus for structured event emission and subscription.

middleware property

Per-operation-point middleware registry for observability.

hook_registry property

Access to the hook system.

perf_timeline property

The global PerfTimeline singleton instance.

execution_engine property

The job execution engine (read-only property).

substrate property

The storage in effect for this app — boot's one selection.

Never None on a booted app: boot step 6.5 resolved the answer (the override a plugin installed, otherwise the filesystem default) and built the engine with it, so there is exactly one decision and this reads it (FUN-17/T11, T12). That is the whole reason this name moved — it used to mean the install slot, which is None in the ordinary case, and ten call sites wanting the storage in effect had to say app.execution_engine.substrate to get it. Two meanings under one name, and the two objects were verifiably not the same one.

The install slot is now :attr:substrate_override; the write door is :meth:install_substrate.

substrate_override property

The override a plugin installed, or None when nothing did.

The :class:~functualize._types.protocols.EngineHost member boot reads (store-substrate/T5, renamed by plugin-host-protocol/T3): step 6.5 takes this or the filesystem default and hands the result to the engine. Until FUN-17/T12 the engine read it, lazily on first use, which made an ordinary read of :attr:substrate decide where documents live and made the winner depend on hook order.

None is not an error and not a missing feature — it is what an app with no storage plugin has.

fresh_root property

Where this project's derived run state lives.

One answer to a question three places in the kernel used to answer for themselves by asking the operating system — and one of them answered it differently, which is why a run's fingerprints could land somewhere other than the run's own project. Recorded when the app is constructed, so a later chdir cannot move a run's state out from under it.

max_invoke_depth property

The deepest chain of nested invoke() calls allowed.

The config-resolved value when boot found one, else the constructor's ExecutionConfig. The engine reads this rather than the value it was built with, so the resolution order (ExecutionConfig at construction, then general.max_invoke_depth from config) is this object's business and no boot step has to know the engine exists to apply it.

domain_registry property

The domain SDK registry (discovered at boot time).

cli_command deletable property writable

The CLI command tree — a click.Group, lazily built.

Holds discovered jobs (nested one group per dotted segment), plugin command namespaces, and the reserved builtin subtree.

run_log property

The run-log subscriber (EngineHost.run_log), or None.

None before init_observability has run — a kernel constructed but not booted — which the engine reads as "nothing is collecting", not as an error.

gates property

app.gates — who answers a gate, and in what order.

di property

app.di — register what jobs can ask for by type or name.

workflows property

app.workflows — create or fetch a workflow scope.

extensions property

app.extensions — what a plugin registers: commands, providers, surfaces, constructs.

configuration property

app.configuration — read what configuration resolved to.

hooks property

app.hooks — every hook and middleware registration point.

Fifteen members that were fifteen four-line properties on this class. Grouping them is what let it come off a 71-member flat surface (T9); the names underneath are unchanged.

context property

The observability context module (PropagationContext API).

get_descriptor(name)

The descriptor for name, or None when nothing is registered.

Source code in src/functualize/app/core.py

def get_descriptor(self, name: str) -> JobDescriptor | None:
    """The descriptor for ``name``, or None when nothing is registered."""
    try:
        return self.job_registry.get_descriptor(name)
    except KeyError:
        return None

registered_jobs()

Every registered job, as a read-only mapping.

A view, not a copy. The engine reads this where it needs the whole set, and what it used to be handed instead was the registry's private dict by reference — two objects sharing mutable state with no contract between them. Read-only is what makes the sharing unnecessary, and it costs nothing: a proxy over a mapping already in memory.

Source code in src/functualize/app/core.py

def registered_jobs(self) -> Mapping[str, RegisteredJob]:
    """Every registered job, as a read-only mapping.

    A view, not a copy. The engine reads this where it needs the whole set,
    and what it used to be handed instead was the registry's *private* dict
    by reference — two objects sharing mutable state with no contract
    between them. Read-only is what makes the sharing unnecessary, and it
    costs nothing: a proxy over a mapping already in memory.
    """
    return MappingProxyType(self.job_registry._registered_jobs)

replace_job(current, replacement)

Swap current for replacement in the job registry.

The engine calls this when it materializes a lazily-registered entry: the placeholder that carries the deferred import is replaced by one carrying the real function. The engine keeps its own copy for resolution; this is what keeps the registry's from diverging from it.

Source code in src/functualize/app/core.py

def replace_job(self, current: RegisteredJob, replacement: RegisteredJob) -> None:
    """Swap ``current`` for ``replacement`` in the job registry.

    The engine calls this when it materializes a lazily-registered entry:
    the placeholder that carries the deferred import is replaced by one
    carrying the real function. The engine keeps its own copy for
    resolution; this is what keeps the registry's from diverging from it.
    """
    jobs = self.job_registry._registered_jobs
    if jobs.get(current.name) is current:
        jobs[current.name] = replacement

install_substrate(substrate)

Install a backend. Before boot selects a store — refused after.

A method rather than a setter because installing is an event with an ordering rule, not an assignment: once step 6.5 has selected the store the engine was built with (FUN-17/T12), a second substrate would leave some of a run's documents in one backend and some in the other. The guard that refuses it lives in _app/impl.py — real logic, and this class has a line budget.

Source code in src/functualize/app/core.py

def install_substrate(self, substrate: StoreSubstrate) -> None:
    """Install a backend. **Before boot selects a store** — refused after.

    A method rather than a setter because installing is an *event* with an
    ordering rule, not an assignment: once step 6.5 has selected the store
    the engine was built with (FUN-17/T12), a second substrate would leave
    some of a run's documents in one backend and some in the other. The
    guard that refuses it lives in ``_app/impl.py`` — real logic, and this
    class has a line budget.
    """
    from functualize._app.impl import install_substrate

    install_substrate(self, substrate)

offer_substrate(offer)

Offer a backend boot asks for at step 6.5, once config has resolved.

The door for a storage plugin whose choice reads its own configuration (FUN-17/T12): registration on the standard path runs before config resolves, so :meth:install_substrate there reads nothing. Recorded here, invoked by _app/boot._select_runtime_store, refused once that step has run — the same window and the same guard as an install.

Source code in src/functualize/app/core.py

def offer_substrate(self, offer: SubstrateOffer) -> None:
    """Offer a backend boot asks for at step 6.5, once config has resolved.

    The door for a storage plugin whose choice reads its own configuration
    (FUN-17/T12): registration on the standard path runs before config
    resolves, so :meth:`install_substrate` there reads nothing. Recorded
    here, invoked by ``_app/boot._select_runtime_store``, refused once that
    step has run — the same window and the same guard as an install.
    """
    from functualize._app.impl import offer_substrate

    offer_substrate(self, offer)

live_zone()

The surface that should host Live constructs, or None.

Top of the pushed stack wins, then the first registered live-capable surface. None is the kernel's answer, where Live no-ops.

Source code in src/functualize/app/core.py

def live_zone(self) -> Any | None:
    """The surface that should host ``Live`` constructs, or None.

    Top of the pushed stack wins, then the first registered live-capable
    surface. None is the kernel's answer, where ``Live`` no-ops.
    """
    from functualize._engine.surface_routing import active_live_zone

    return active_live_zone(self)

collector()

The one surface that should answer a prompt, or None.

None is not an error: it is what turns a would-be hang into a typed InputNotAvailable at the job's call site.

Source code in src/functualize/app/core.py

def collector(self) -> Any | None:
    """The one surface that should answer a prompt, or None.

    None is not an error: it is what turns a would-be hang into a typed
    ``InputNotAvailable`` at the job's call site.
    """
    from functualize._engine.surface_routing import active_collector

    return active_collector(self)

get_jobs()

Return all discovered job descriptors (Layer 2 memoized).

Source code in src/functualize/app/core.py

def get_jobs(self) -> list[JobDescriptor]:
    """Return all discovered job descriptors (Layer 2 memoized)."""
    if self._jobs_memo is not None:
        return self._jobs_memo
    result = self.job_registry.get_descriptors()
    self._jobs_memo = result
    return result

get_job(name)

Retrieve a single job descriptor by name.

Source code in src/functualize/app/core.py

def get_job(self, name: str) -> JobDescriptor | None:
    """Retrieve a single job descriptor by name."""
    from functualize._app.impl import get_job

    return get_job(self, name)

resolution_chain()

Return the config resolution chain [CLI → Env → Files → Defaults].

The sanctioned way to read provenance — which source supplied a value and in what precedence order. Long-lived consumers (TUI provenance panels, MCP introspection) must use this rather than reaching for the private _resolution_chain attribute.

Returns:

Type	Description
ResolutionChain	The active ResolutionChain. Never None on a booted app.

Source code in src/functualize/app/core.py

def resolution_chain(self) -> ResolutionChain:
    """Return the config resolution chain [CLI → Env → Files → Defaults].

    The sanctioned way to read provenance — which source supplied a value
    and in what precedence order. Long-lived consumers (TUI provenance
    panels, MCP introspection) must use this rather than reaching for the
    private ``_resolution_chain`` attribute.

    Returns:
        The active ResolutionChain. Never None on a booted app.
    """
    return self._resolution_chain

refresh()

Re-read the project from disk: discovery and config resolution.

Source code in src/functualize/app/core.py

def refresh(self) -> None:
    """Re-read the project from disk: discovery and config resolution."""
    from functualize._app.impl import refresh

    return refresh(self)

scope_for(scope_id)

The scope named by scope_id, created and announced if new.

The engine's seam onto scope creation (EngineHost.scope_for). It goes through create_workflow_scope, so the ON_SCOPE_CREATED hook fires and the registry is populated exactly as it did when execute() minted scopes itself — the difference is that every door now reaches it, including the CLI, which calls engine.run() directly and therefore never ran a line of execute().

Source code in src/functualize/app/core.py

def scope_for(self, scope_id: str) -> Any:
    """The scope named by ``scope_id``, created and announced if new.

    The engine's seam onto scope creation (`EngineHost.scope_for`). It goes
    through `create_workflow_scope`, so the `ON_SCOPE_CREATED` hook fires
    and the registry is populated exactly as it did when `execute()` minted
    scopes itself — the difference is that *every* door now reaches it,
    including the CLI, which calls `engine.run()` directly and therefore
    never ran a line of `execute()`.
    """
    existing = self._scope_registry.get(scope_id)
    if existing is not None:
        return existing
    return self.workflows.create_workflow_scope(scope_id)

execute(request)

Execute a job — the single surface-facing entry.

Takes a :class:RunRequest and nothing else. Every door builds one, so a run's origin is carried rather than reconstructed, and the delivery inputs travel with the run instead of being read off this object.

Why there is no (job_name, **kwargs) form. There was one until run-request/T15, and it was the accidental control channel of spec §1.6a: a wire surface splatting a caller's payload into execute(name, **body) meant a JSON body of {"scope_id": "abc"} did not arrive as an argument called scope_id — it chose the scope the run joined. group_option_values leaked the same way. Wave 3 stopped every door from splatting; deleting the parameters is what makes it unrepresentable rather than merely unpractised.

For the short programmatic spelling use :func:request_for: app.execute(request_for("build", target="x")). It puts every keyword in kwargs, where a job argument belongs. A caller who genuinely means a control input constructs the request and names the field, so the intent is visible in their source instead of hiding in a dict key.

A WorkflowScope is created for each top-level execution, grouping related invocations under one traceable context: the request's workflow_scope_id when it names one (reused if it already exists), otherwise a generated <job>-<hex>.

Parameters:

Name	Type	Description	Default
request	RunRequest	The run to perform.	required

Returns:

Type	Description
JobResult	JobResult with status, duration, return value, and metadata.

Source code in src/functualize/app/core.py

def execute(self, request: RunRequest) -> JobResult:
    """Execute a job — the single surface-facing entry.

    Takes a :class:`RunRequest` and nothing else. Every door builds one, so
    a run's origin is carried rather than reconstructed, and the delivery
    inputs travel with the run instead of being read off this object.

    **Why there is no ``(job_name, **kwargs)`` form.** There was one until
    run-request/T15, and it was the *accidental control channel* of spec
    §1.6a: a wire surface splatting a caller's payload into
    ``execute(name, **body)`` meant a JSON body of ``{"scope_id": "abc"}``
    did not arrive as an argument called ``scope_id`` — it chose **the scope
    the run joined**. ``group_option_values`` leaked the same way. Wave 3
    stopped every door from splatting; deleting the parameters is what makes
    it unrepresentable rather than merely unpractised.

    For the short programmatic spelling use
    :func:`request_for`: ``app.execute(request_for("build", target="x"))``.
    It puts every keyword in ``kwargs``, where a job argument belongs. A
    caller who genuinely means a control input constructs the request and
    names the field, so the intent is visible in their source instead of
    hiding in a dict key.

    A ``WorkflowScope`` is created for each top-level execution, grouping
    related invocations under one traceable context: the request's
    ``workflow_scope_id`` when it names one (reused if it already exists),
    otherwise a generated ``<job>-<hex>``.

    Args:
        request: The run to perform.

    Returns:
        JobResult with status, duration, return value, and metadata.
    """
    group_option_values = (
        dict(request.group_option_values)
        if request.group_option_values is not None
        else None
    )
    kwargs = dict(request.kwargs)

    # **No scope minted here.** `engine.run()` does it, because that is
    # where every run passes and this method is not: the CLI calls
    # `execution_engine.run()` directly, so a job launched from the command
    # line got no scope at all. That was invisible while `rc.state` handed
    # out a private dict and became a hard failure the moment state was made
    # durable — the same entrypoint divergence, found again, in the one
    # place still doing work the engine should own.
    return self._execution_engine.run(
        request.replace(
            kwargs=kwargs,
            group_option_values=group_option_values,
        )
    )

execute_parallel(job_names, *, timeout=None, observer=None)

Execute jobs concurrently, returning results in input order (T40).

Source code in src/functualize/app/core.py

def execute_parallel(
    self,
    job_names: Sequence[str],
    *,
    timeout: float | None = None,
    observer: Any | None = None,
) -> list[JobResult]:
    """Execute jobs concurrently, returning results in input order (T40)."""
    from functualize._app.impl import execute_parallel

    return execute_parallel(self, job_names, timeout=timeout, observer=observer)

explain(job_name)

Render why job_name would or would not run (§D.6).

The prose half of :meth:explain_verdicts. Two forms of one answer, derived from one set of verdicts, so func builtin why and func builtin why --json cannot disagree.

Source code in src/functualize/app/core.py

def explain(self, job_name: str) -> str:
    """Render why ``job_name`` would or would not run (§D.6).

    The prose half of :meth:`explain_verdicts`. Two forms of one answer,
    derived from one set of verdicts, so `func builtin why` and
    `func builtin why --json` cannot disagree.
    """
    from functualize._app.impl import explain

    return explain(self, job_name)

explain_verdicts(job_name)

The raw material behind func builtin why.

Returns (target_verdict, [(dep_name, dep_verdict), …], note, error). Evaluates the same pre-flight pipeline the executor consults, fresh rather than from a cache — a verdict is a function of the world now, and a stored explanation goes stale exactly when someone asks.

Stays on the app because func builtin why, the JSON form and the TUI all need it and none of them may import the engine directly; the ~115 executable lines behind it live in _app/impl.py (T9).

Source code in src/functualize/app/core.py

def explain_verdicts(self, job_name: str) -> tuple[Any, list[Any], str, str | None]:
    """The raw material behind `func builtin why`.

    Returns ``(target_verdict, [(dep_name, dep_verdict), …], note, error)``.
    Evaluates the same pre-flight pipeline the executor consults, fresh
    rather than from a cache — a verdict is a function of the world *now*,
    and a stored explanation goes stale exactly when someone asks.

    Stays on the app because `func builtin why`, the JSON form and the TUI
    all need it and none of them may import the engine directly; the
    ~115 executable lines behind it live in `_app/impl.py` (T9).
    """
    from functualize._app.impl import explain_verdicts

    return explain_verdicts(self, job_name)

explain_data(job_name)

The machine-readable half of :meth:explain, off the same verdicts.

Source code in src/functualize/app/core.py

def explain_data(self, job_name: str) -> dict[str, Any]:
    """The machine-readable half of :meth:`explain`, off the same verdicts."""
    from functualize._app.impl import explain_data

    return explain_data(self, job_name)

cache_stats()

Return statistics about the job discovery cache.

Returns a CacheInfo dataclass with entry_count, stale_count, file_size_bytes, and cache_path.

Source code in src/functualize/app/core.py

def cache_stats(self) -> CacheInfo:
    """Return statistics about the job discovery cache.

    Returns a CacheInfo dataclass with entry_count, stale_count,
    file_size_bytes, and cache_path.
    """
    from functualize._app.impl import get_cache_stats

    return get_cache_stats(self)

push_surface(surface)

Push a phase-scoped surface onto the surface stack.

Used by TTY.run and the orchestrator to make a job-owned app the active surface for the duration of a phase. Always pair with :meth:pop_surface in a finally so a crashing phase still unwinds before the shell resumes. Top-of-stack answers prompts, and while a terminal-owning surface is on the stack the fan-out skips other terminal surfaces (see _engine/surface_routing).

Source code in src/functualize/app/core.py

def push_surface(self, surface: Any) -> None:
    """Push a phase-scoped surface onto the surface stack.

    Used by ``TTY.run`` and the orchestrator to make a job-owned app the
    active surface for the duration of a phase. Always pair with
    :meth:`pop_surface` in a ``finally`` so a crashing phase still unwinds
    before the shell resumes. Top-of-stack answers prompts, and while a
    terminal-owning surface is on the stack the fan-out skips other
    terminal surfaces (see ``_engine/surface_routing``).
    """
    if not hasattr(self, "_surface_stack"):
        self._surface_stack = []
    self._surface_stack.append(surface)

pop_surface(surface=None)

Pop the top surface (or surface if given) off the stack.

Tolerant of an already-empty stack and of a mismatched argument so a finally-guaranteed unwind never raises over the original error.

Source code in src/functualize/app/core.py

def pop_surface(self, surface: Any = None) -> None:
    """Pop the top surface (or ``surface`` if given) off the stack.

    Tolerant of an already-empty stack and of a mismatched argument so a
    ``finally``-guaranteed unwind never raises over the original error.
    """
    stack = getattr(self, "_surface_stack", None)
    if not stack:
        return
    if surface is None or stack[-1] is surface:
        stack.pop()
    elif surface in stack:
        stack.remove(surface)

register_dynamic_job(name, function, config_class=None, group=None)

Register a callable as an executable job at runtime.

Source code in src/functualize/app/core.py

def register_dynamic_job(
    self,
    name: str,
    function: Callable[..., Any],
    config_class: Any | None = None,
    group: str | None = None,
) -> None:
    """Register a callable as an executable job at runtime."""
    from functualize._app.impl import register_dynamic_job

    register_dynamic_job(self, name, function, config_class, group)

run()

Entry point — delegates to the active adapter.

Source code in src/functualize/app/core.py

def run(self) -> None:
    """Entry point — delegates to the active adapter."""
    from functualize._app.impl import shutdown_plugins

    try:
        self.cli_command()
    finally:
        shutdown_plugins(self.plugin_loader, self)

FallbackCommand

Bases: Protocol

Handler for CLI commands that don't match any registered Click command.

Fallbacks are tried in order; first match wins. If no fallback matches, CliAdapter shows a "command not found" error with suggestions.

matches(args, app)

Return True if this fallback can handle the given arguments.

Source code in src/functualize/app/fallback.py

def matches(self, args: list[str], app: FunctualizeApp) -> bool:
    """Return True if this fallback can handle the given arguments."""
    ...

execute(args, app)

Execute the fallback handler. Returns exit code (0 = success).

Source code in src/functualize/app/fallback.py

def execute(self, args: list[str], app: FunctualizeApp) -> int:
    """Execute the fallback handler. Returns exit code (0 = success)."""
    ...

classic(*, file_pattern='^config\\.(\\w+)\\.(\\w+)$', dotenv=True)

CLI → Env → Files (upward search) → Defaults.

Leaves config_resolution_chain as None so that the boot path performs file discovery and builds the full chain at startup.

Parameters:

Name	Type	Description	Default
file_pattern	str	Regex for matching config files during discovery.	'^config\\.(\\w+)\\.(\\w+)$'
dotenv	bool	Whether to load .env files.	True

Returns:

Type	Description
ConfigSources	A ConfigSources instance configured for classic file-based resolution.

Source code in src/functualize/app/presets.py

def classic(
    *,
    file_pattern: str = r"^config\.(\w+)\.(\w+)$",
    dotenv: bool = True,
) -> ConfigSources:
    """CLI → Env → Files (upward search) → Defaults.

    Leaves ``config_resolution_chain`` as None so that the boot path
    performs file discovery and builds the full chain at startup.

    Args:
        file_pattern: Regex for matching config files during discovery.
        dotenv: Whether to load .env files.

    Returns:
        A ConfigSources instance configured for classic file-based resolution.
    """
    return ConfigSources(
        file_pattern=file_pattern,
        dotenv=dotenv,
        config_resolution_chain=None,
    )

env_only(*, dotenv=True, dotenv_path=None)

CLI → Env → Defaults. Minimal configuration.

Like twelve_factor but with dotenv enabled by default for local development convenience.

Parameters:

Name	Type	Description	Default
dotenv	bool	Whether to load .env files (default True).	True
dotenv_path	str | None	Explicit path to .env file (None = auto-discover).	None

Returns:

Type	Description
ConfigSources	A ConfigSources instance configured for environment-only resolution.

Source code in src/functualize/app/presets.py

def env_only(*, dotenv: bool = True, dotenv_path: str | None = None) -> ConfigSources:
    """CLI → Env → Defaults. Minimal configuration.

    Like twelve_factor but with dotenv enabled by default for local
    development convenience.

    Args:
        dotenv: Whether to load .env files (default True).
        dotenv_path: Explicit path to .env file (None = auto-discover).

    Returns:
        A ConfigSources instance configured for environment-only resolution.
    """
    chain = ResolutionChain([CliSource({}), EnvSource(), DefaultSource({})])
    return ConfigSources(
        dotenv=dotenv,
        dotenv_path=dotenv_path,
        config_resolution_chain=chain,
    )

remote_first(*, file_pattern='config.*', dotenv=False, max_age=None)

CLI → Vault → Env → Files → Defaults.

Values declared as provider://reference annotations resolve from the project's encrypted local vault, which func builtin vault sync fills from the registered remote providers. Reads never touch the network (ADR-016).

Requires at least one remote provider to be registered through the functualize.remote_providers entry-point group — install functualize-aws or another provider plugin. An app selecting this preset with none registered raises at construction rather than quietly resolving from local files.

.. note:: Before 0.2.4 this preset silently behaved as :func:classic: it returned config_resolution_chain=None and no boot path built a remote source, so an operator choosing it for AWS Secrets Manager got local files and environment variables with no warning. remote=True is what makes the intent legible to boot.

Parameters:

Name	Type	Description	Default
file_pattern	str	Glob pattern for config file matching.	'config.*'
dotenv	bool	Whether to load .env files.	False
max_age	str | None	How old the vault may be before every run warns, e.g. "7d". Default "24h". Exceeding it never fails a run; $FUNCTUALIZE_VAULT_MAX_AGE overrides it.	None

Returns:

Type	Description
ConfigSources	A ConfigSources instance configured for remote-first resolution.

Source code in src/functualize/app/presets.py

def remote_first(
    *,
    file_pattern: str = "config.*",
    dotenv: bool = False,
    max_age: str | None = None,
) -> ConfigSources:
    """CLI → Vault → Env → Files → Defaults.

    Values declared as ``provider://reference`` annotations resolve from the
    project's encrypted local vault, which ``func builtin vault sync`` fills
    from the registered remote providers. Reads never touch the network
    (ADR-016).

    Requires at least one remote provider to be registered through the
    ``functualize.remote_providers`` entry-point group — install
    ``functualize-aws`` or another provider plugin. An app selecting this
    preset with none registered raises at construction rather than quietly
    resolving from local files.

    .. note::
        Before 0.2.4 this preset **silently behaved as** :func:`classic`: it
        returned ``config_resolution_chain=None`` and no boot path built a
        remote source, so an operator choosing it for AWS Secrets Manager got
        local files and environment variables with no warning. ``remote=True``
        is what makes the intent legible to boot.

    Args:
        file_pattern: Glob pattern for config file matching.
        dotenv: Whether to load .env files.
        max_age: How old the vault may be before every run warns, e.g.
            ``"7d"``. Default ``"24h"``. Exceeding it never fails a run;
            ``$FUNCTUALIZE_VAULT_MAX_AGE`` overrides it.

    Returns:
        A ConfigSources instance configured for remote-first resolution.
    """
    return ConfigSources(
        file_pattern=file_pattern,
        dotenv=dotenv,
        config_resolution_chain=None,
        remote=True,
        vault_max_age=max_age,
    )

twelve_factor(*, dotenv=False)

CLI → Env → Defaults. No file discovery.

Sets an explicit resolution chain that skips FileSource entirely. Environment variables are the primary configuration source.

Parameters:

Name	Type	Description	Default
dotenv	bool	Whether to load .env files (default False for pure 12-factor).	False

Returns:

Type	Description
ConfigSources	A ConfigSources instance configured for twelve-factor apps.

Source code in src/functualize/app/presets.py

def twelve_factor(*, dotenv: bool = False) -> ConfigSources:
    """CLI → Env → Defaults. No file discovery.

    Sets an explicit resolution chain that skips FileSource entirely.
    Environment variables are the primary configuration source.

    Args:
        dotenv: Whether to load .env files (default False for pure 12-factor).

    Returns:
        A ConfigSources instance configured for twelve-factor apps.
    """
    chain = ResolutionChain([CliSource({}), EnvSource(), DefaultSource({})])
    return ConfigSources(
        dotenv=dotenv,
        config_resolution_chain=chain,
    )

get_perf_timeline()

Return the global performance timeline singleton.

This provides public access to the framework-level PerfTimeline instance that records startup and runtime performance marks. The timeline exists as a module-level singleton before any FunctualizeApp is constructed, making it suitable for pre-boot instrumentation in the CLI layer.

Returns:

Type	Description
PerfTimeline	The global PerfTimeline instance.

Source code in src/functualize/app/__init__.py

def get_perf_timeline() -> PerfTimeline:
    """Return the global performance timeline singleton.

    This provides public access to the framework-level PerfTimeline instance
    that records startup and runtime performance marks. The timeline exists
    as a module-level singleton before any FunctualizeApp is constructed,
    making it suitable for pre-boot instrumentation in the CLI layer.

    Returns:
        The global PerfTimeline instance.
    """
    return _perf_timeline

---

Overview

The functualize.app module is the primary entry point for constructing and configuring a Functualize application.

Module location: src/functualize/app/

Public API

from functualize.app import (
    FunctualizeApp,
    JobSources,
    ConfigSources,
    PluginSources,
    ExecutionConfig,
    classic,
    twelve_factor,
    env_only,
    remote_first,
)

See the Architecture Guide for how this module relates to other public packages.