Plugin Module

plugin

Public plugin author API for functualize.

This module re-exports symbols that plugin authors need to build functualize plugins: event infrastructure, job provider protocols, adapter protocols, plugin metadata, and TUI extension protocols.

Usage

from functualize.plugin import EventBus, JobProvider, AdapterPlugin, PluginMetadata from functualize.plugin import DisplayProvider, PanelProvider, ThemeProvider

DEFAULT_SIGIL = '' module-attribute

__all__ = ['EventBus', 'HookEvent', 'StructuredEvent', 'JobProvider', 'JobTransform', 'Job', 'StaticProvider', 'ModulePreFilter', 'AdapterPlugin', 'SubstrateInstallError', 'PluginHost', 'AppSettingsSchema', 'CommandNode', 'CommandProvider', 'DEFAULT_SIGIL', 'InputMode', 'InputModeRegistry', 'Setting', 'SettingsSources', 'PromptCollector', 'Surface', 'LiveConstruct', 'PromptRequest', 'PromptResponse', 'PromptIntent', 'PromptSeverity', 'PromptChoice', 'PluginMetadata', 'PluginWithShutdown', 'Source', 'FormatProvider', 'VaultKeyInitializer', 'VaultKeyProvider', 'AgentStepExecutor', 'AgentCapability', 'AgentStepContext', 'AgentStepResult', 'discover_domains', 'scan_domain_providers', 'BarRenderer', 'DisplayProvider', 'HeaderItemProvider', 'InteractiveContent', 'PanelProvider', 'PostRunStampProvider', 'SessionState', 'SignatureProvider', 'StatusBarItemProvider', 'ThemeProvider', 'validate_extension_id'] module-attribute

Job(function, name=None, group=None) dataclass

Explicit job definition with overrides.

Used with StaticProvider to provide metadata overrides for a callable. When name is None, the function's name is used.

Attributes:

Name	Type	Description
function	Callable[..., Any]	The callable to wrap as a job.
name	str | None	Override for the job name. Defaults to function.name.
group	str | None	Override for the job group. Defaults to None.

function instance-attribute

name = None class-attribute instance-attribute

group = None class-attribute instance-attribute

StaticProvider(functions)

Wraps pre-imported callables as JobDescriptors with zero I/O.

Accepts a list of plain callables or Job dataclass instances. Plain callables derive job name from function.name. Job instances allow overriding name and group.

Satisfies the JobProvider Protocol via structural typing.

Parameters:

Name	Type	Description	Default
functions	list[Callable[..., Any] | Job]	List of callables or Job dataclass instances.	required

Source code in src/functualize/_discovery/providers.py

def __init__(self, functions: list[Callable[..., Any] | Job]) -> None:
    self._descriptors: list[JobDescriptor] = []
    self._by_name: dict[str, JobDescriptor] = {}

    for item in functions:
        descriptor = self._build_descriptor(item)
        self._descriptors.append(descriptor)
        self._by_name[descriptor.name] = descriptor

list_jobs()

Return all job descriptors from this source.

Source code in src/functualize/_discovery/providers.py

def list_jobs(self) -> Sequence[JobDescriptor]:
    """Return all job descriptors from this source."""
    return self._descriptors

get_job(name)

Retrieve a specific job by name. None if not found.

Source code in src/functualize/_discovery/providers.py

def get_job(self, name: str) -> JobDescriptor | None:
    """Retrieve a specific job by name. None if not found."""
    found = self._by_name.get(name)
    if found is None:
        canonical = normalize_name(name) or name
        found = self._by_name.get(canonical) if canonical != name else None
    return found

EventBus()

Central event emission and subscriber routing.

The EventBus is the primary interface for emitting structured events and registering subscribers. It delegates pattern matching to an internal TrieRouter and maintains an EventCatalog for introspection.

Zero-cost guarantee: when no subscribers are registered, emit() returns after a single self._router.has_subscribers check — no event object construction, no string formatting, no time.time() call.

Usage::

bus = EventBus()
handle = bus.subscribe("config.file.*", my_callback)
bus.emit("config.file.parse.end", resource="/path/to/file.toml", duration=0.05)
bus.unsubscribe(handle)

Source code in src/functualize/_events/bus.py

def __init__(self) -> None:
    self._router = TrieRouter()
    self._catalog = EventCatalog()

has_subscribers property

True if any subscriber is registered on the bus.

subscribe(pattern, callback)

Subscribe to events matching the given pattern.

Parameters:

Name	Type	Description	Default
pattern	str	One of: - Exact event name ("config.file.parse.end") - Prefix wildcard ("config.") - Global wildcard ("")	required
callback	SubscriberCallback	Callable receiving a StructuredEvent.	required

Returns:

Type	Description
SubscriptionHandle	SubscriptionHandle for later unsubscription.

Source code in src/functualize/_events/bus.py

def subscribe(
    self, pattern: str, callback: SubscriberCallback
) -> SubscriptionHandle:
    """Subscribe to events matching the given pattern.

    Args:
        pattern: One of:
            - Exact event name ("config.file.parse.end")
            - Prefix wildcard ("config.*")
            - Global wildcard ("*")
        callback: Callable receiving a StructuredEvent.

    Returns:
        SubscriptionHandle for later unsubscription.
    """
    return self._router.subscribe(pattern, callback)

unsubscribe(handle)

Remove a previously registered subscriber.

Parameters:

Name	Type	Description	Default
handle	SubscriptionHandle	The SubscriptionHandle returned by subscribe().	required

Source code in src/functualize/_events/bus.py

def unsubscribe(self, handle: SubscriptionHandle) -> None:
    """Remove a previously registered subscriber.

    Args:
        handle: The SubscriptionHandle returned by subscribe().
    """
    self._router.unsubscribe(handle)

emit(event_name, resource='', related=None, **payload)

Emit a structured event to all matching subscribers.

Implements a zero-cost bypass path: 1. If no subscribers exist at all → return immediately. 2. If no subscribers match this specific event → return. 3. Validate event name format. 4. Attach PropagationContext (trace_id, span_id). 5. Construct StructuredEvent. 6. Dispatch to matching callbacks in registration order.

If a subscriber raises an exception, it is logged at ERROR level and dispatch continues to remaining subscribers.

Parameters:

Name	Type	Description	Default
event_name	str	Must match the grammar {domain}.{resource}.{action} (at least 3 dot-separated segments of lowercase alphanumeric/underscores).	required
resource	str	Primary resource identifier (e.g., file path, job name).	''
related	list[str] | None	Optional list of associated resource identifiers.	None
**payload	Any	Arbitrary event-specific key-value data.	{}

Source code in src/functualize/_events/bus.py

def emit(
    self,
    event_name: str,
    resource: str = "",
    related: list[str] | None = None,
    **payload: Any,
) -> None:
    """Emit a structured event to all matching subscribers.

    Implements a zero-cost bypass path:
    1. If no subscribers exist at all → return immediately.
    2. If no subscribers match this specific event → return.
    3. Validate event name format.
    4. Attach PropagationContext (trace_id, span_id).
    5. Construct StructuredEvent.
    6. Dispatch to matching callbacks in registration order.

    If a subscriber raises an exception, it is logged at ERROR level
    and dispatch continues to remaining subscribers.

    Args:
        event_name: Must match the grammar
            ``{domain}.{resource}.{action}`` (at least 3 dot-separated
            segments of lowercase alphanumeric/underscores).
        resource: Primary resource identifier (e.g., file path, job name).
        related: Optional list of associated resource identifiers.
        **payload: Arbitrary event-specific key-value data.
    """
    # ZERO-COST CHECK 1: bail if no subscribers at all
    if not self._router.has_subscribers:
        return

    # ZERO-COST CHECK 2: bail if no subscribers match this event
    if not self._router.has_subscribers_for(event_name):
        return

    # Fast-path: known events in catalog skip regex validation
    if not self._catalog.contains(event_name) and not _EVENT_NAME_RE.match(
        event_name
    ):
        logger.warning(
            f"Invalid event name format: {event_name!r}. "
            f"Expected {{domain}}.{{resource}}.{{action}}."
        )
        return

    # Attach propagation context automatically
    ctx = current_context()
    event = StructuredEvent(
        event_name=event_name,
        resource=resource,
        related=related or [],
        payload=payload,
        trace_id=ctx.trace_id,
        span_id=ctx.span_id,
    )

    # Dispatch to subscribers synchronously in registration order
    callbacks = self._router.match(event_name)
    for callback in callbacks:
        try:
            callback(event)
        except Exception as exc:
            logger.error(
                f"Subscriber {callback!r} raised during event "
                f"'{event_name}': {exc}",
                exc_info=True,
            )

catalog()

Return the full event catalog for plugin introspection.

Returns:

Type	Description
dict[str, EventMetadata]	Mapping of event names to their EventMetadata.

Source code in src/functualize/_events/bus.py

def catalog(self) -> dict[str, EventMetadata]:
    """Return the full event catalog for plugin introspection.

    Returns:
        Mapping of event names to their EventMetadata.
    """
    return self._catalog.all()

register_event_metadata(metadata)

Register event metadata in the catalog.

Allows plugins to register custom event metadata alongside framework-defined events.

Parameters:

Name	Type	Description	Default
metadata	EventMetadata	The EventMetadata to register.	required

Source code in src/functualize/_events/bus.py

def register_event_metadata(self, metadata: EventMetadata) -> None:
    """Register event metadata in the catalog.

    Allows plugins to register custom event metadata alongside
    framework-defined events.

    Args:
        metadata: The EventMetadata to register.
    """
    self._catalog.register(metadata)

StructuredEvent(event_name, resource, related=list(), payload=dict(), timestamp=time.time(), trace_id=None, span_id=None) dataclass

Immutable structured event emitted through the EventBus.

Attributes:

Name	Type	Description
event_name	str	Hierarchical name following {domain}.{resource}.{action} grammar.
resource	str	Primary resource identifier (e.g., file path, job name).
related	list[str]	Associated resource identifiers.
payload	dict[str, Any]	Event-specific data dictionary.
timestamp	float	Seconds since epoch (from time.time()).
trace_id	str | None	Active trace ID (auto-attached from PropagationContext).
span_id	str | None	Active span ID (auto-attached from PropagationContext).

event_name instance-attribute

resource instance-attribute

related = field(default_factory=list) class-attribute instance-attribute

payload = field(default_factory=dict) class-attribute instance-attribute

timestamp = field(default_factory=(time.time)) class-attribute instance-attribute

trace_id = None class-attribute instance-attribute

span_id = None class-attribute instance-attribute

HookEvent

Constants for lifecycle hook events.

BEFORE_JOB = 'before_job' class-attribute instance-attribute

AFTER_SUCCESS = 'after_success' class-attribute instance-attribute

AFTER_FAILURE = 'after_failure' class-attribute instance-attribute

ON_TEARDOWN = 'on_teardown' class-attribute instance-attribute

JOB_REGISTERED = 'job_registered' class-attribute instance-attribute

Fired when a job is registered with the framework.

PRE_EXECUTE = 'pre_execute' class-attribute instance-attribute

Fired after config resolution and before the job function is called.

APP_READY = 'app_ready' class-attribute instance-attribute

Fired after the application has fully booted.

INVOKE_START = 'invoke_start' class-attribute instance-attribute

Fired before a nested rc.invoke() child job begins execution.

INVOKE_END = 'invoke_end' class-attribute instance-attribute

Fired after a nested rc.invoke() child job completes execution.

INVOKE_FAILURE = 'invoke_failure' class-attribute instance-attribute

Fired after a nested rc.invoke() child job completes with FAILURE status.

ON_PHASE_START = 'on_phase_start' class-attribute instance-attribute

Fired when track_phase() creates a NEW phase.

ON_PHASE_FAILURE = 'on_phase_failure' class-attribute instance-attribute

Fired when track_phase() transitions a phase to FAILURE status.

ON_PHASE_COMPLETE = 'on_phase_complete' class-attribute instance-attribute

Fired when track_phase() transitions a phase to SUCCESS status.

ON_SCOPE_CREATED = 'on_scope_created' class-attribute instance-attribute

Fired when a WorkflowScope is created.

TUI_STARTED = 'tui_started' class-attribute instance-attribute

Fired when the TUI application launches.

PluginMetadata

Bases: Protocol

Protocol that plugins must satisfy to be loaded.

Attributes:

Name	Type	Description
name	str	Plugin name, maximum 64 characters.
version	str	Plugin version conforming to PEP 440.
description	str	Plugin description, maximum 256 characters.

name instance-attribute

version instance-attribute

description instance-attribute

CommandNode

Bases: Protocol

One runnable-or-navigable position in the shell's command tree.

A node may be runnable (execute), navigable (children), or both — the duality case the trie already models (a deploy job that also has deploy web beneath it). Nothing here distinguishes a job from a builtin; that is the point.

Attributes:

Name	Type	Description
name	str	The node's own segment as typed, not its full path.
help_text	str	One-line description for listings and completion.
needs_terminal	bool	True when running this node takes over the controlling terminal, so a TUI front-end must suspend itself around it rather than capture its output. Read by the orchestrator handoff, and by nothing else.

Decision — needs_terminal is a plain bool, not a predicate over args. It exists today in two incompatible shapes: a method on _cli/builtins.BuiltinCommand (needs_terminal(args) -> bool, computed as any(arg in self.terminal_subcommands for arg in args)) and a bool field on _cli/tui/panel_live_zone's surface. The method form exists only because BuiltinCommand models a whole command family: config carries terminal_subcommands=("edit",), so the answer depends on which subcommand was typed.

A CommandNode is not a family — the tree splits config into distinct config edit and config show nodes, and at that granularity the answer is static. So the bool wins, and the args-dependence is resolved once, at node construction: a provider wrapping BuiltinCommand evaluates needs_terminal([segment]) for each child it emits (this is ClickCommandProvider's job). No information is lost and the consumer — the orchestrator handoff — stops re-deriving it per invocation.

Distinct from a descriptor's requires_tty, which is a job author's declaration about the job function; needs_terminal is a property of the command node the shell is about to run.

name property

The node's own segment as typed, not its full path.

help_text property

One-line description for listings and completion.

needs_terminal property

True when running this node takes over the controlling terminal.

children()

Direct children, for drill-down. Empty for a leaf.

Backed by GroupTrie.children() for job nodes; by the click group's registered subcommands for builtin nodes.

Source code in src/functualize/_types/commands.py

def children(self) -> list[CommandNode]:
    """Direct children, for drill-down. Empty for a leaf.

    Backed by ``GroupTrie.children()`` for job nodes; by the click group's
    registered subcommands for builtin nodes.
    """
    ...

params()

The node's CLI-facing parameters.

Deliberately the existing FieldDescriptor — the same type build_click_params consumes — rather than a parallel param model invented for the shell. There is one description of a job's parameters and every surface reads it.

May be expensive: for a lazily-cached job this can force materialization (one module import). Callers that must stay import-free — completion of a command name, drill-down listing — must not call it.

Source code in src/functualize/_types/commands.py

def params(self) -> list[FieldDescriptor]:
    """The node's CLI-facing parameters.

    Deliberately the **existing** ``FieldDescriptor`` — the same type
    ``build_click_params`` consumes — rather than a parallel param model
    invented for the shell. There is one description of a job's parameters
    and every surface reads it.

    May be expensive: for a lazily-cached job this can force materialization
    (one module import). Callers that must stay import-free — completion of
    a *command name*, drill-down listing — must not call it.
    """
    ...

execute(args)

Run this node with args; return a process-style exit code.

Source code in src/functualize/_types/commands.py

def execute(self, args: Sequence[str]) -> int:
    """Run this node with ``args``; return a process-style exit code."""
    ...

CommandProvider

Bases: Protocol

A source of top-level :class:CommandNode s the shell composes into one tree.

nodes()

This provider's top-level nodes, in display order.

Source code in src/functualize/_types/commands.py

def nodes(self) -> list[CommandNode]:
    """This provider's top-level nodes, in display order."""
    ...

SubstrateInstallError

Bases: Exception

A plugin could not install the project's configured substrate.

Substrate installation is the one APP_READY failure that aborts boot: continuing would silently fall back to a different backend and split the project's persisted documents across substrates.

PluginHost

Bases: Protocol

What a plugin needs from the application, and nothing else.

Peer of :class:~functualize._types.protocols.EngineHost for the plugin boundary. Members ask; none of them lends — the rule that port inherits. A member that hands a plugin a mutable internal, or the machinery to fire lifecycle events at the rest of the app, is not a narrowing of app: Any; it is the same reach with a type on it.

Twelve members. What is absent is as deliberate as what is present:

hook_registry Four plugin clients, and refused. Seven public methods, of which four are invoke* — the firing half. On the port, every plugin could fire arbitrary lifecycle events at every other. T6 migrated all four clients to hooks.on_ready, which registers and cannot fire. execution_engine Five apparent clients, one real: four were the chain app.execution_engine.substrate, collapsed to substrate by T4. The survivor wants materialize_job (functualize-mcp/…/_workflow_tools.py:447). One call site does not buy handing plugins the whole engine. run One client, -> None, and it is the CLI entrypoint. A plugin calling it re-enters delivery from inside delivery. workflows Zero plugin clients. A port lists what is needed, not what exists. di.resolve Zero clients after T1 and T2. Its only consumer was inside a permanently dead block, which is why this port has no read side for DI.

di property

Register what jobs can ask for, by type or by name.

extensions property

Hang commands, surfaces and constructs off the application.

configuration property

Read this project's resolved configuration.

gates property

Register gate strategies and presets.

hooks property

Register lifecycle callbacks — for a plugin, on_ready.

substrate property

The storage in effect — never None, boot's one selection.

Renamed from the install slot by T3, which is what lets this member be declared without | None. The slot is substrate_override and FUN-17/T12 settled who reads it: boot, at step 6.5, which hands the answer to the engine — so the slot is absent here.

fresh_root property

Where this project's derived run state lives.

The other single-client member, for the same reason — and already a declared member of the sibling port EngineHost.

get_jobs()

Every discovered job descriptor.

Source code in src/functualize/_types/host.py

def get_jobs(self) -> list[JobDescriptor]:
    """Every discovered job descriptor."""
    ...

get_job(name)

One descriptor by name, or None when nothing is registered.

Source code in src/functualize/_types/host.py

def get_job(self, name: str) -> JobDescriptor | None:
    """One descriptor by name, or None when nothing is registered."""
    ...

execute(request)

Run a job. Takes a :class:RunRequest and nothing else.

Source code in src/functualize/_types/host.py

def execute(self, request: RunRequest) -> JobResult:
    """Run a job. Takes a :class:`RunRequest` and nothing else."""
    ...

install_substrate(substrate)

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

For a plugin whose choice needs no configuration: it can build its substrate at registration and hand it over there. No shipped plugin is such a plugin since FUN-17/T12 moved functualize-substrate-sqlite to :meth:offer_substrate, so this member has no plugin client — kept for the config-free case and because the app's own door is this one, and recorded as a deviation from "sized to measured use" in plan.md → Surviving smells. An install and an offer are both storage claims; two claims refuse at selection rather than one winning.

Source code in src/functualize/_types/host.py

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

    For a plugin whose choice needs **no configuration**: it can build its
    substrate at registration and hand it over there. No shipped plugin is
    such a plugin since FUN-17/T12 moved ``functualize-substrate-sqlite`` to
    :meth:`offer_substrate`, so this member has no plugin client — kept for
    the config-free case and because the app's own door is this one, and
    recorded as a deviation from "sized to measured use" in ``plan.md`` →
    *Surviving smells*. An install and an offer are both storage *claims*;
    two claims refuse at selection rather than one winning.
    """
    ...

offer_substrate(offer)

Offer a backend that boot asks for once configuration has resolved.

Called from a plugin's registration call. Boot invokes offer inside step 6.5 — after the configuration chain exists, before the store is selected — and uses what it returns; the plugin never names a boot step. That is the window a config-driven substrate plugin needs and :meth:install_substrate cannot give it, since on the standard path registration runs before configuration resolves.

A storage member, not a lifecycle hook, and so not on :class:HooksView: it registers a question boot will ask, and nothing but that one selection site ever invokes it. A failure inside offer aborts boot — step 6.5 is deliberately uncaught — and more than one claim (offers and installs alike) is refused with SubstrateInstallError naming every claimant.

Source code in src/functualize/_types/host.py

def offer_substrate(self, offer: SubstrateOffer) -> None:
    """Offer a backend that boot asks for once configuration has resolved.

    Called from a plugin's registration call. Boot invokes ``offer`` inside
    step 6.5 — after the configuration chain exists, before the store is
    selected — and uses what it returns; the plugin never names a boot
    step. That is the window a *config-driven* substrate plugin needs and
    :meth:`install_substrate` cannot give it, since on the standard path
    registration runs before configuration resolves.

    A storage member, not a lifecycle hook, and so not on
    :class:`HooksView`: it registers a question boot will ask, and nothing
    but that one selection site ever invokes it. A failure inside ``offer``
    aborts boot — step 6.5 is deliberately uncaught — and more than one
    claim (offers and installs alike) is refused with
    ``SubstrateInstallError`` naming every claimant.
    """
    ...

InputMode(sigil, name, candidate_source, is_ready, submit, history_namespace) dataclass

One thing the input bar can be doing.

Attributes:

Name	Type	Description
sigil	str	First character that selects this mode ("!", "?"), or :data:DEFAULT_SIGIL for the fallback command mode.
name	str	Short identifier, used in logs and history namespacing.
candidate_source	Callable[[str, int], list[Any]]	Returns completion candidates for (text, cursor). The text excludes the sigil, and the cursor offset is relative to that stripped text. Completion is cursor-sensitive in every mode — a command mode needs to know which token is being edited, a shell mode needs it for path completion — so it is part of the contract rather than something each mode smuggles in via instance state.
is_ready	Callable[[str], bool]	Readiness rule for the input FSM — may this be submitted?
submit	Callable[[str], None]	Runs the input. Also receives text without the sigil.
history_namespace	str	Where this mode's history is recorded. Distinct namespaces stop !ls from polluting job-argument history.

sigil instance-attribute

name instance-attribute

candidate_source instance-attribute

is_ready instance-attribute

submit instance-attribute

history_namespace instance-attribute

strip_sigil(text)

text with this mode's sigil removed, if present.

Source code in src/functualize/_types/input_modes.py

def strip_sigil(self, text: str) -> str:
    """``text`` with this mode's sigil removed, if present."""
    if self.sigil and text.startswith(self.sigil):
        return text[len(self.sigil) :]
    return text

InputModeRegistry()

Sigil -> mode. One default mode, any number of sigil modes.

Registration is explicit and collision-checked: two modes claiming ! would make dispatch order decide behavior, which is exactly the kind of silent precedence the convergence exists to remove.

Source code in src/functualize/_types/input_modes.py

def __init__(self) -> None:
    self._modes: dict[str, InputMode] = {}

__slots__ = ('_modes',) class-attribute instance-attribute

sigils property

Registered sigils, default first.

register(mode)

Add mode.

Raises:

Type	Description
ValueError	another mode already claims this sigil, or the sigil is longer than one character (dispatch reads exactly one).

Source code in src/functualize/_types/input_modes.py

def register(self, mode: InputMode) -> None:
    """Add ``mode``.

    Raises:
        ValueError: another mode already claims this sigil, or the sigil is
            longer than one character (dispatch reads exactly one).
    """
    if len(mode.sigil) > 1:
        raise ValueError(
            f"sigil {mode.sigil!r} must be a single character "
            f"(or empty for the default mode)"
        )
    existing = self._modes.get(mode.sigil)
    if existing is not None:
        raise ValueError(
            f"sigil {mode.sigil!r} is already registered to "
            f"{existing.name!r}; cannot also register {mode.name!r}"
        )
    self._modes[mode.sigil] = mode

resolve(text)

The mode that owns text, by its first character.

Falls back to the default mode. Returns None only when no default has been registered — a shell always registers one, so that is a programming error rather than a user-visible state.

Source code in src/functualize/_types/input_modes.py

def resolve(self, text: str) -> InputMode | None:
    """The mode that owns ``text``, by its first character.

    Falls back to the default mode. Returns None only when no default has
    been registered — a shell always registers one, so that is a
    programming error rather than a user-visible state.
    """
    if text:
        mode = self._modes.get(text[0])
        if mode is not None:
            return mode
    return self._modes.get(DEFAULT_SIGIL)

get(sigil)

The mode registered for sigil, if any.

Source code in src/functualize/_types/input_modes.py

def get(self, sigil: str) -> InputMode | None:
    """The mode registered for ``sigil``, if any."""
    return self._modes.get(sigil)

__contains__(sigil)

Source code in src/functualize/_types/input_modes.py

def __contains__(self, sigil: object) -> bool:
    return sigil in self._modes

__len__()

Source code in src/functualize/_types/input_modes.py

def __len__(self) -> int:
    return len(self._modes)

LiveConstruct

Bases: Protocol

A renderable hosted in a surface's live zone (the Live capability).

The construct owns only its state and how to render it; the surface owns the cursor/mount and repaints. The contract is a single Rich renderable via __rich__ (a Table / Tree / Progress / Group / Text). The "raw" fallback needs no second renderer — it is Rich's own degradation: a non-terminal Console prints plain text, and rich.live.Live off-TTY prints the final state only.

Interactivity is a separate, optional capability. A construct that also implements the PanelHost action contract (get_available_actions(focused) plus Textual action_* methods, made focusable) can be mounted via Live.panel(...) where an event loop exists (PANEL / EXCLUSIVE); it degrades to passive render in STDOUT and to event-emission in MCP.

Return type is Any because _types is stdlib-only (no Rich import); the value is any Rich renderable.

__rich__()

Return a Rich renderable for the construct's current state.

Source code in src/functualize/_types/interactivity.py

def __rich__(self) -> Any:
    """Return a Rich renderable for the construct's current state."""
    ...

PromptChoice(value, label=None, description=None, disabled=False, group=None) dataclass

A single selectable choice within a prompt.

Attributes:

Name	Type	Description
value	str	The programmatic value returned when this choice is selected.
label	str | None	Display label (falls back to value if None).
description	str | None	Optional longer description for the choice.
disabled	bool	If True, shown but not selectable.
group	str | None	Optional group name for visual grouping of choices.

value instance-attribute

label = None class-attribute instance-attribute

description = None class-attribute instance-attribute

disabled = False class-attribute instance-attribute

group = None class-attribute instance-attribute

PromptCollector

Bases: Protocol

Something that can ask the user a question and return the answer.

Exactly one collector is active at a time — whichever owns the terminal (or the modal) right now. Implementations include the stdin fallback, a TUI's input bar, and a job-owned app's modal.

collect(request)

Ask the user, blocking until answered, timed out, or cancelled.

Source code in src/functualize/_types/interactivity.py

def collect(self, request: PromptRequest) -> PromptResponse:
    """Ask the user, blocking until answered, timed out, or cancelled."""
    ...

PromptIntent

Bases: Enum

Semantic intent of a prompt, guiding surface presentation.

CONFIRM_DESTRUCTIVE = 'confirm_destructive' class-attribute instance-attribute

CONFIRM_NEUTRAL = 'confirm_neutral' class-attribute instance-attribute

CONFIRM_PROCEED = 'confirm_proceed' class-attribute instance-attribute

SELECT = 'select' class-attribute instance-attribute

MULTI_SELECT = 'multi_select' class-attribute instance-attribute

TEXT_INPUT = 'text_input' class-attribute instance-attribute

SECRET_INPUT = 'secret_input' class-attribute instance-attribute

ACKNOWLEDGE = 'acknowledge' class-attribute instance-attribute

PromptRequest(question, intent=PromptIntent.TEXT_INPUT, choices=None, default=None, severity=PromptSeverity.INFO, context_message=None, context_data=None, placeholder=None, help_text=None, timeout=None, required=True, validator=None, validation_message=None, source_job=None, source_step=None) dataclass

A structured request for user input during job execution.

The wire format between a job and whatever is rendering it. Carries intent, severity, choices, context, and validation rules so that one request can be presented as a terminal question, a Textual modal, or an MCP gate checkpoint without the job knowing which.

Attributes:

Name	Type	Description
question	str	The prompt question text displayed to the user.
intent	PromptIntent	Semantic intent guiding surface presentation.
choices	list[PromptChoice] | None	Available choices for SELECT/MULTI_SELECT intents.
default	Any	Default value used on timeout or when no input provided.
severity	PromptSeverity	Visual severity level for styling.
context_message	str | None	Optional context message displayed alongside the prompt.
context_data	dict[str, Any] | None	Optional structured data displayed in a context panel.
placeholder	str | None	Placeholder text for text input fields.
help_text	str | None	Additional help text displayed below the prompt.
timeout	float | None	Timeout in seconds; None means wait indefinitely.
required	bool	If True and no Surface is available with no default, raises InputNotAvailable.
validator	str | Any | None	Regex pattern string or object with .validate_python() method.
validation_message	str | None	Custom message shown on validation failure.
source_job	str | None	Name of the job that initiated this prompt (auto-filled by rc.prompts.ask).
source_step	str | None	Name of the workflow step that initiated this prompt.

question instance-attribute

intent = PromptIntent.TEXT_INPUT class-attribute instance-attribute

choices = None class-attribute instance-attribute

default = None class-attribute instance-attribute

severity = PromptSeverity.INFO class-attribute instance-attribute

context_message = None class-attribute instance-attribute

context_data = None class-attribute instance-attribute

placeholder = None class-attribute instance-attribute

help_text = None class-attribute instance-attribute

timeout = None class-attribute instance-attribute

required = True class-attribute instance-attribute

validator = None class-attribute instance-attribute

validation_message = None class-attribute instance-attribute

source_job = None class-attribute instance-attribute

source_step = None class-attribute instance-attribute

PromptResponse(value, source='user') dataclass

Response from a user prompt interaction.

Attributes:

Name	Type	Description
value	Any	The response value (user input, default, or None if cancelled).
source	str	How the response was obtained. Constrained to: "user", "default", "timeout", "cancelled".

value instance-attribute

source = 'user' class-attribute instance-attribute

was_cancelled property

True if the user cancelled the prompt.

was_timeout property

True if the prompt timed out without user response.

is_user_input property

True if the response came from direct user input.

PromptSeverity

Bases: Enum

Visual severity level for prompt presentation.

Largely derivable from :class:PromptIntent — a destructive confirmation is a danger prompt, everything else is informational — so prefer letting :func:severity_for_intent supply it rather than passing it by hand. It remains settable for the cases where a caller genuinely wants to override the styling (a warning on a non-destructive action).

INFO = 'info' class-attribute instance-attribute

WARNING = 'warning' class-attribute instance-attribute

DANGER = 'danger' class-attribute instance-attribute

SUCCESS = 'success' class-attribute instance-attribute

Surface

Bases: Protocol

Something that renders a job's events.

Implementations include the TUI's output panel, flow-viz's plain-stdout tree, a job-owned Textual app, a log-file writer, and test recorders. Registered on the app; the engine fans every non-framework event out to all of them.

An implementation that can also answer questions additionally satisfies :class:PromptCollector.

Threading contract — the part that bites. handle_event is called from whatever thread the job runs on, which is a worker thread whenever a host owns the terminal (see _cli/tui/job_execution.py). An implementation that touches a UI must marshal onto its own loop (Textual: post_message / call_from_thread). Writing to a widget directly from handle_event freezes the app with no exception and no stack trace — the failure is silent, so it must be designed out here rather than debugged later.

handle_event(event)

Render or record one structured event. Called on worker threads.

Source code in src/functualize/_types/interactivity.py

def handle_event(self, event: StructuredEvent) -> None:
    """Render or record one structured event. Called on worker threads."""
    ...

AdapterPlugin

Bases: Protocol

Protocol for delivery surface adapters.

Adapters decouple delivery surfaces (CLI, HTTP, Lambda, MCP) from the application kernel. Each adapter implements a setup/run/shutdown lifecycle.

name instance-attribute

version instance-attribute

description instance-attribute

adapter_type instance-attribute

__call__(app)

Setup phase — called during boot to wire the adapter.

app: PluginHost since plugin-host-protocol/T9, replacing Any. This is the framework's own front door: an adapter is the thing the kernel hands itself to, so if any signature in the repository should name what it is being handed, it is this one.

Widening, not narrowing. An adapter that declares app: FunctualizeApp no longer satisfies this protocol — a parameter type is contravariant, so an implementation must accept at least what the protocol promises to pass, and FunctualizeApp is one PluginHost rather than any. The four concrete adapters that did were widened by T10.

Source code in src/functualize/_types/protocols.py

def __call__(self, app: PluginHost) -> None:
    """Setup phase — called during boot to wire the adapter.

    ``app: PluginHost`` since `plugin-host-protocol`/T9, replacing ``Any``.
    This is the framework's own front door: an adapter is the thing the
    kernel hands itself to, so if any signature in the repository should
    name what it is being handed, it is this one.

    **Widening, not narrowing.** An adapter that declares
    ``app: FunctualizeApp`` no longer satisfies this protocol — a parameter
    type is contravariant, so an implementation must accept *at least* what
    the protocol promises to pass, and ``FunctualizeApp`` is one
    ``PluginHost`` rather than any. The four concrete adapters that did
    were widened by T10.
    """
    ...

run(*args, **kwargs)

Universal entrypoint with platform-specific signatures.

Source code in src/functualize/_types/protocols.py

def run(self, *args: Any, **kwargs: Any) -> Any:
    """Universal entrypoint with platform-specific signatures."""
    ...

shutdown()

Graceful shutdown. No-op if not needed.

Source code in src/functualize/_types/protocols.py

def shutdown(self) -> None:
    """Graceful shutdown. No-op if not needed."""
    ...

AgentCapability

Bases: StrEnum

A constraint an executor promises it can enforce on a step's behalf.

Declared by the executor, required by the step, and compared before the walk starts. A step whose requirement the executor cannot honour is refused rather than run, because running it would leave the constraint silently unenforced — a workflow that appears to have restricted tools it left wide open.

A StrEnum rather than (str, Enum) for the same reason :class:~functualize._types.outcome.Family is one: the contracts spell it (str, Enum), and on this interpreter that is the same type with a UP042 warning attached.

ENFORCES_TOOL_ALLOWLIST = 'enforces_tool_allowlist' class-attribute instance-attribute

PRESERVES_ACTIVE_TIME_BUDGET = 'preserves_active_time_budget' class-attribute instance-attribute

SUPPORTS_VISIBLE_OUTPUT = 'supports_visible_output' class-attribute instance-attribute

AgentStepContext(request, step_name, instructions, tools, inputs, time_budget_s) dataclass

Everything an executor is given to perform one agent step.

It carries the run's :class:~functualize._types.run_request.RunRequest rather than restating its fields: the request already holds where the run came from and what it was asked for, and a second shape for those would be a second answer to where a run came from.

Attributes:

Name	Type	Description
request	RunRequest	The request the run reaching this step was built from.
step_name	str	The declaring node's name — the step being executed.
instructions	str	What the step asks the agent to do.
tools	tuple[str, ...]	The tool allowlist the step declared, normalized to a tuple. An empty tuple means the step declared no constraint, which is not the same statement as "this step may use no tools".
inputs	Mapping[str, Any]	The values the step binds into the agent's work. TRANSITIONAL(workflow-graph-semantics) — always empty today. The port's single construction site passes an empty mapping, because binding an upstream node's output into a downstream step is the typed-outcome plumbing that feature builds; there is no other source for it. An executor may read it and will get nothing (asp M-3). Declared now rather than added later so the payload shape a plugin compiles against does not change under it.
time_budget_s	float | None	The step's active-time budget in seconds, when it declared one.

request instance-attribute

step_name instance-attribute

instructions instance-attribute

tools instance-attribute

inputs instance-attribute

time_budget_s instance-attribute

AgentStepExecutor

Bases: Protocol

Protocol for running a workflow step by delegating it to an agent.

Registered by an app method — app.extensions.register_agent_step_executor — and never auto-discovered: auto-discovery is how a surface acquires behaviour nobody declared. GateResolver is the template, down to the registration door.

The engine asks an executor only for steps that named it, or for every agent step when exactly one executor is registered. Nothing falls back to a different executor, and nothing falls back to a human.

capabilities is a promise, and a missing flag is a refusal, not a default: a step requiring :attr:AgentCapability.ENFORCES_TOOL_ALLOWLIST from an executor that does not declare it fails validation, because running anyway would grant every tool the step meant to leave out.

Implementations are checked with isinstance; issubclass raises TypeError on this Protocol, because name and capabilities are data members — a fact no type checker will point out at the call site.

name instance-attribute

capabilities instance-attribute

execute(ctx)

Perform one agent step.

Parameters:

Name	Type	Description	Default
ctx	AgentStepContext	The step, its inputs, and the request that reached it.	required

Returns:

Type	Description
AgentStepResult	The step's result.

Raises:

Type	Description
Exception	Any exception fails the step. How a failure is routed around a step is not this port's business, and no exception here is answered by asking a human instead.

Source code in src/functualize/_types/protocols.py

def execute(self, ctx: AgentStepContext) -> AgentStepResult:
    """Perform one agent step.

    Args:
        ctx: The step, its inputs, and the request that reached it.

    Returns:
        The step's result.

    Raises:
        Exception: Any exception fails the step. How a failure is routed
            around a step is not this port's business, and no exception
            here is answered by asking a human instead.
    """
    ...

AgentStepResult(value, tool_calls=()) dataclass

What an executor returns for one agent step.

Attributes:

Name	Type	Description
value	Any	The step's result, recorded as the step's outcome.
tool_calls	tuple[Mapping[str, Any], ...]	The tool invocations the agent reported, in order. Empty for an executor that does not surface them — an audit trail, not a contract, so nothing may require a non-empty tuple. TRANSITIONAL(durable-run-layer) — the walker takes result.value and drops this, so an executor that fills it is writing the audit trail into nowhere. It lands when there is a run event stream to write it to; recording it in the step record first would put an unbounded, agent-controlled payload in the scope store (asp M-3).

value instance-attribute

tool_calls = () class-attribute instance-attribute

FormatProvider

Bases: Protocol

Protocol for configuration file format plugins.

Implementations parse configuration files into normalized dictionaries and serialize dictionaries back to formatted strings.

extensions()

Return file extensions this provider handles (e.g., ['.toml']).

Each extension MUST include the leading dot.

Source code in src/functualize/_types/protocols.py

def extensions(self) -> list[str]:
    """Return file extensions this provider handles (e.g., ['.toml']).

    Each extension MUST include the leading dot.
    """
    ...

parse(path)

Parse a configuration file and return a normalized dictionary.

Parameters:

Name	Type	Description	Default
path	str	Absolute path to the configuration file.	required

Returns:

Type	Description
dict[str, Any]	Normalized dict with primitive values, lists, or nested dicts.

Source code in src/functualize/_types/protocols.py

def parse(self, path: str) -> dict[str, Any]:
    """Parse a configuration file and return a normalized dictionary.

    Args:
        path: Absolute path to the configuration file.

    Returns:
        Normalized dict with primitive values, lists, or nested dicts.
    """
    ...

serialize(data)

Serialize a configuration dictionary to the provider's format.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Configuration dictionary to serialize.	required

Returns:

Type	Description
str	Formatted string representation.

Source code in src/functualize/_types/protocols.py

def serialize(self, data: dict[str, Any]) -> str:
    """Serialize a configuration dictionary to the provider's format.

    Args:
        data: Configuration dictionary to serialize.

    Returns:
        Formatted string representation.
    """
    ...

JobProvider

Bases: Protocol

Protocol for job descriptor sources.

Implementations provide job descriptors from various sources (filesystem scan, entry points, static definitions, etc.).

Note on workflows: the cached JobDescriptor.workflow shape is populated only by directory discovery, which projects it via the internal workflow_shape_of. A provider building descriptors by hand leaves it None and has no public way to set it — deliberately, to keep the cache projection one-sided. Consumers that need a provider-declared workflow's topology read it live from descriptor.function.__functualize_workflow__ when the cached shape is absent (e.g. the MCP WorkflowToolProvider); providers do not populate the field.

list_jobs()

Return all job descriptors from this source.

Source code in src/functualize/_types/protocols.py

def list_jobs(self) -> Sequence[JobDescriptor]:
    """Return all job descriptors from this source."""
    ...

get_job(name)

Retrieve a specific job by name. None if not found.

Source code in src/functualize/_types/protocols.py

def get_job(self, name: str) -> JobDescriptor | None:
    """Retrieve a specific job by name. None if not found."""
    ...

JobTransform

Bases: Protocol

Protocol for intercepting and modifying job descriptors.

Implementations transform job descriptors as they flow from providers to the registry.

transform_list(jobs)

Transform a list of job descriptors.

Source code in src/functualize/_types/protocols.py

def transform_list(self, jobs: Sequence[JobDescriptor]) -> Sequence[JobDescriptor]:
    """Transform a list of job descriptors."""
    ...

transform_get(name, descriptor)

Transform a single job descriptor lookup.

Source code in src/functualize/_types/protocols.py

def transform_get(
    self, name: str, descriptor: JobDescriptor | None
) -> JobDescriptor | None:
    """Transform a single job descriptor lookup."""
    ...

ModulePreFilter

Bases: Protocol

Decide whether a module is worth importing, without importing it.

Discovery reads a candidate file's AST before executing it, and a filter answers from that alone. The built-in filters (_primitives/pre_filter.py) express the require_* settings; a host whose jobs are, say, methods on classes cannot express itself in any of those and supplies its own.

Implementations satisfy this structurally -- there is nothing to inherit, and _primitives does not import this module's package upward.

fingerprint() is not decorative. The discovery cache persists negative pre-filter decisions and replays them, trusting them only while the discovery fingerprint matches. A caller-supplied predicate cannot join that hash by identity: _normalize_discovery_value renders an unknown value with str(), and str() of a function carries its address, so the digest would differ on every boot and invalidate the cache on every run. Omitting it instead reproduces the X1-X4 replay defect that CACHE_VERSION 15->16->17 and ADR-010/ADR-011 exist to close. A stable, caller-declared string is the only option that keeps the cache both warm and correct.

should_import(source_file)

Return whether this module should be imported for discovery.

Source code in src/functualize/_types/protocols.py

def should_import(self, source_file: Path) -> bool:
    """Return whether this module should be imported for discovery."""
    ...

fingerprint()

Stable identity of this filter's logic, for cache invalidation.

Must be identical across processes for identical behaviour, and must change when the predicate's behaviour changes. A host that forgets to bump it gets a stale cache -- the same contract as any cache key, and the same failure the require_* fields already have when a config is edited without invalidation.

Source code in src/functualize/_types/protocols.py

def fingerprint(self) -> str:
    """Stable identity of this filter's logic, for cache invalidation.

    Must be identical across processes for identical behaviour, and must
    change when the predicate's behaviour changes. A host that forgets to
    bump it gets a stale cache -- the same contract as any cache key, and
    the same failure the ``require_*`` fields already have when a config
    is edited without invalidation.
    """
    ...

PluginWithShutdown

Bases: Protocol

Protocol for plugins requiring cleanup on application shutdown.

Plugins satisfying this protocol will have on_shutdown called in reverse loading order when the application completes execution.

on_shutdown(app)

Called during application shutdown for resource cleanup.

Parameters:

Name	Type	Description	Default
app	PluginHost	The host being shut down, as the plugin port rather than Any (plugin-host-protocol/T9). A shutdown handler that reaches past these eleven members is reaching into an application that is already tearing itself down.	required

Source code in src/functualize/_types/protocols.py

def on_shutdown(self, app: PluginHost) -> None:
    """Called during application shutdown for resource cleanup.

    Args:
        app: The host being shut down, as the plugin port rather than
            ``Any`` (`plugin-host-protocol`/T9). A shutdown handler that
            reaches past these eleven members is reaching into an
            application that is already tearing itself down.
    """
    ...

Source

Bases: Protocol

Protocol for configuration value sources in the Resolution Chain.

Each source represents one origin of configuration values (CLI args, environment variables, remote providers, file-based config, defaults).

source_type property

Source type identifier (e.g., 'cli', 'env', 'remote', 'file', 'default').

source_id property

Source identifier (e.g., file path, provider name, 'environ').

get(key, section=None)

Retrieve a value for the given key.

Parameters:

Name	Type	Description	Default
key	str	The configuration key name.	required
section	str | None	Optional section/namespace.	None

Returns:

Type	Description
Any | None	The value if found, None if not present in this source.

Source code in src/functualize/_types/protocols.py

def get(self, key: str, section: str | None = None) -> Any | None:
    """Retrieve a value for the given key.

    Args:
        key: The configuration key name.
        section: Optional section/namespace.

    Returns:
        The value if found, None if not present in this source.
    """
    ...

has(key, section=None)

Check if this source can provide a value for the key.

Source code in src/functualize/_types/protocols.py

def has(self, key: str, section: str | None = None) -> bool:
    """Check if this source can provide a value for the key."""
    ...

keys(section)

Return all keys available for the given section.

Parameters:

Name	Type	Description	Default
section	str	The section/namespace to query.	required

Returns:

Type	Description
set[str]	Set of key names this source can provide for the section.

Source code in src/functualize/_types/protocols.py

def keys(self, section: str) -> set[str]:
    """Return all keys available for the given section.

    Args:
        section: The section/namespace to query.

    Returns:
        Set of key names this source can provide for the section.
    """
    ...

VaultKeyInitializer

Bases: VaultKeyProvider, Protocol

A key provider that can also create the key, not only read it.

Separate from :class:VaultKeyProvider rather than a method added to it, because widening that protocol would retroactively invalidate every structural implementation that satisfies it today — read-only providers are valid providers and must stay so. A provider opts in by having the method; nothing registers, and nothing inherits.

func builtin vault init looks for this capability. A provider that lacks it is not an error: the environment provider cannot create anything, because only the operator can set an environment variable, and init refuses with instructions rather than pretending otherwise.

Key scope is this provider's choice, and both shipped providers choose user-scope. project_id is carried here for symmetry with :meth:VaultKeyProvider.get_key and so a third-party KMS or hosted provider can hold one key per project — but the environment provider ignores it (one variable, many projects) and the keychain provider matches it deliberately. Isolation between projects comes from the separate vault files, not from separate keys. A seam whose two implementations disagreed about scope meant that which scope applied depended on whether an environment variable happened to be exported.

initialize_key(project_id)

Return the existing key, or create, persist, and return one.

Parameters:

Name	Type	Description	Default
project_id	str	Carried for providers that scope per project. Both shipped implementations ignore it; see the class docstring.	required

Returns:

Type	Description
bytes	Exactly KEY_BYTES bytes. Idempotent: a second call returns what
bytes	the first one persisted, never a fresh key — a provider that
bytes	generated a new key each time would silently strand every value
bytes	already written under the old one.

The key is never logged, printed or returned through any report. init reports which provider holds it, never the key itself; keygen is the one command that puts a key on screen, and it is explicitly the operator's to place.

Source code in src/functualize/_types/protocols.py

def initialize_key(self, project_id: str) -> bytes:
    """Return the existing key, or create, persist, and return one.

    Args:
        project_id: Carried for providers that scope per project. Both
            shipped implementations ignore it; see the class docstring.

    Returns:
        Exactly ``KEY_BYTES`` bytes. Idempotent: a second call returns what
        the first one persisted, never a fresh key — a provider that
        generated a new key each time would silently strand every value
        already written under the old one.

    The key is never logged, printed or returned through any report. `init`
    reports *which provider* holds it, never the key itself; `keygen` is the
    one command that puts a key on screen, and it is explicitly the
    operator's to place.
    """
    ...

VaultKeyProvider

Bases: Protocol

Protocol for supplying the key that opens the local secrets vault.

The vault caches values synced from remote providers (AWS Secrets Manager, Bitwarden, …) so that jobs resolve configuration without touching the network. It is encrypted at rest; this protocol is where the key comes from, and it is a seam rather than a fixed source so that an OS keychain, a cloud KMS, a password manager or a hosted control plane are all the same shape (ADR-016).

Two implementations ship: an environment-variable provider (non-interactive) and an OS keychain provider (interactive).

Resolution order is part of the contract. Non-interactive providers are consulted first, and interactive ones only when no key was found and a TTY is present. Reversed, an unattended run — CI, Lambda, a container — would block forever on a prompt nobody can answer.

identifier()

Return the short provider name, e.g. 'env' or 'keychain'.

Source code in src/functualize/_types/protocols.py

def identifier(self) -> str:
    """Return the short provider name, e.g. 'env' or 'keychain'."""
    ...

interactive()

Whether obtaining the key may prompt, block, or require a TTY.

A provider returning True is never consulted on an unattended run.

Source code in src/functualize/_types/protocols.py

def interactive(self) -> bool:
    """Whether obtaining the key may prompt, block, or require a TTY.

    A provider returning True is never consulted on an unattended run.
    """
    ...

is_available()

Whether this provider can supply a key in this environment.

Reports capability, not success: a keychain provider returns False where no keyring exists, rather than raising when asked for a key.

Source code in src/functualize/_types/protocols.py

def is_available(self) -> bool:
    """Whether this provider can supply a key in this environment.

    Reports capability, not success: a keychain provider returns False
    where no keyring exists, rather than raising when asked for a key.
    """
    ...

get_key(project_id)

Return the 32-byte key for a project's vault, or None.

Parameters:

Name	Type	Description	Default
project_id	str	The project identity the vault is scoped to. Vaults are per-project, so a provider may hold a distinct key per project.	required

Returns:

Type	Description
bytes | None	Exactly 32 bytes, or None when this provider has no key to offer.
bytes | None	Returning None is normal and lets resolution continue; it is not an
bytes | None	error.

Source code in src/functualize/_types/protocols.py

def get_key(self, project_id: str) -> bytes | None:
    """Return the 32-byte key for a project's vault, or None.

    Args:
        project_id: The project identity the vault is scoped to. Vaults are
            per-project, so a provider may hold a distinct key per project.

    Returns:
        Exactly 32 bytes, or None when this provider has no key to offer.
        Returning None is normal and lets resolution continue; it is not an
        error.
    """
    ...

AppSettingsSchema(settings, env_prefix='FUNCTUALIZE', sources=SettingsSources(), file_section_prefixes=_default_file_section_prefixes()) dataclass

The full settings declaration one app hands to the store.

Attributes:

Name	Type	Description
settings	tuple[Setting, ...]	The catalog, in display order.
env_prefix	str	Environment-variable prefix without the trailing underscore ("FUNCTUALIZE" → FUNCTUALIZE_TUI_THEME). Replaces the hardcoded literal in env_var_for.
sources	SettingsSources	File discovery declaration.
file_section_prefixes	Mapping[str, str]	{filename: section prefix}. A shared file such as pyproject.toml nests an app's settings under its own table (tool.functualize); a dedicated file uses the bare section. Declared as data — this replaces the "tool.functualize" if path.name == "pyproject.toml" else "" branch, which hardcodes both the filename and func's own table name and so cannot answer the question for a second app.

settings instance-attribute

env_prefix = 'FUNCTUALIZE' class-attribute instance-attribute

sources = field(default_factory=SettingsSources) class-attribute instance-attribute

file_section_prefixes = field(default_factory=_default_file_section_prefixes) class-attribute instance-attribute

env_var_for(setting)

The environment variable that overrides setting.

Byte-identical to the shipped env_var_for when env_prefix == "FUNCTUALIZE".

Source code in src/functualize/_types/settings.py

def env_var_for(self, setting: Setting) -> str:
    """The environment variable that overrides ``setting``.

    Byte-identical to the shipped ``env_var_for`` when
    ``env_prefix == "FUNCTUALIZE"``.
    """
    if setting.section:
        return f"{self.env_prefix}_{setting.section.upper()}_{setting.key.upper()}"
    return f"{self.env_prefix}_{setting.key.upper()}"

section_prefix_for(file_name)

The TOML table an app's settings nest under inside file_name.

Source code in src/functualize/_types/settings.py

def section_prefix_for(self, file_name: str) -> str:
    """The TOML table an app's settings nest under inside ``file_name``."""
    return self.file_section_prefixes.get(file_name, "")

section_in_file(setting, file_name)

Full dotted TOML section for setting inside file_name.

Mirrors the shipped section_for_file: prefix and section joined, empty parts dropped.

Source code in src/functualize/_types/settings.py

def section_in_file(self, setting: Setting, file_name: str) -> str:
    """Full dotted TOML section for ``setting`` inside ``file_name``.

    Mirrors the shipped ``section_for_file``: prefix and section joined,
    empty parts dropped.
    """
    parts = [p for p in (self.section_prefix_for(file_name), setting.section) if p]
    return ".".join(parts)

Setting(name, type, description, default=None, choices=None, min_value=None, max_value=None, max_items=None, cli_flag=None, phase=None) dataclass

One declared setting.

Attributes:

Name	Type	Description
name	str	Dotted canonical identity — "tui.theme", or a bare "dotenv" for a top-level key. This is the row key everywhere.
type	str	"enum" | "int" | "bool" | "list" | "str".
description	str	Human-readable help.
default	str | None	Built-in default as display text, or None when the setting genuinely has no default (most [discovery] filters).
choices	/ min_value / max_value / max_items	Validation bounds, carried over from the shipped SettingSchema unchanged.
cli_flag	str | None	Generated root CLI flag ("--log-level"), or None for a settings-only knob. Consumed by C3.1; nothing reads it yet.
phase	str | None	"early" marks a setting whose flag must be honoured by the pre-boot argv scan, before app construction. Consumed by C3.2.

name instance-attribute

type instance-attribute

description instance-attribute

default = None class-attribute instance-attribute

choices = None class-attribute instance-attribute

min_value = None class-attribute instance-attribute

max_value = None class-attribute instance-attribute

max_items = None class-attribute instance-attribute

cli_flag = None class-attribute instance-attribute

phase = None class-attribute instance-attribute

section property

TOML section this setting lives in; "" for a top-level key.

Derived from the dotted name rather than stored, so a setting cannot have a name and a section that disagree.

key property

The bare key inside :attr:section.

SettingsSources(global_file_name='config.toml', project_file_names=('pyproject.toml', '.functualize.toml', '.functualize/.functualize.toml'), env=True) dataclass

Where an app's settings are read from, in precedence order.

Precedence is fixed (default < global < project < env); what varies per app is the file names. Declaring them makes the store app-agnostic.

Attributes:

Name	Type	Description
global_file_name	str	File inside the user config dir (XDG-resolved).
project_file_names	tuple[str, ...]	Candidates for the upward project walk, nearest wins, in the order they are probed at each level.
env	bool	Whether environment variables participate at all.

global_file_name = 'config.toml' class-attribute instance-attribute

project_file_names = ('pyproject.toml', '.functualize.toml', '.functualize/.functualize.toml') class-attribute instance-attribute

env = True class-attribute instance-attribute

BarRenderer

Bases: Protocol

Overrides default header or status bar rendering.

When registered, replaces the default "join with double-space" rendering for the specified bar type. Last registered wins.

bar_type instance-attribute

render(items, context)

Render the bar from collected (id, text) pairs and context.

Source code in src/functualize/plugin/protocols.py

def render(self, items: list[tuple[str, str]], context: dict[str, object]) -> str:
    """Render the bar from collected (id, text) pairs and context."""
    ...

DisplayProvider

Bases: Protocol

Provides an above-header display panel with CWD-contextual visibility.

Display panels show ambient situational awareness (Docker services, Git status, etc.) and support auto-refresh and job-linking.

The widgets yielded by :meth:compose_display may satisfy :class:InteractiveContent (plus can_focus/action_*) to become interactive when the DISPLAY zone is focused — same contract as PanelHost panels.

display_id instance-attribute

display_title instance-attribute

display_priority instance-attribute

refresh_interval instance-attribute

linked_jobs instance-attribute

linked_groups instance-attribute

should_show(cwd, app)

Return True if this display should be visible for the given CWD.

Source code in src/functualize/plugin/protocols.py

def should_show(self, cwd: Path, app: FunctualizeApp) -> bool:
    """Return True if this display should be visible for the given CWD."""
    ...

compose_display()

Compose the display widget tree.

Source code in src/functualize/plugin/protocols.py

def compose_display(self) -> ComposeResult:
    """Compose the display widget tree."""
    ...

refresh()

Called at refresh_interval to update display content.

Source code in src/functualize/plugin/protocols.py

def refresh(self) -> None:
    """Called at refresh_interval to update display content."""
    ...

get_available_actions(focused)

Return (key, label) tuples for the dynamic footer.

Source code in src/functualize/plugin/protocols.py

def get_available_actions(self, focused: bool) -> list[tuple[str, str]]:
    """Return (key, label) tuples for the dynamic footer."""
    ...

HeaderItemProvider

Bases: Protocol

Provides an item rendered in the header bar.

Items are collected, filtered (None skipped), sorted by priority, and joined with double-space separator.

item_id instance-attribute

item_priority instance-attribute

render_item(app)

Render header item text, or None to skip.

Source code in src/functualize/plugin/protocols.py

def render_item(self, app: FunctualizeApp) -> str | None:
    """Render header item text, or None to skip."""
    ...

InteractiveContent

Bases: Protocol

The one interaction contract shared by every key-receiving widget.

Whether a widget lives in the PanelHost ring, a drill-down sub-view, or the display slot, it becomes interactive the same way (the PanelHost idiom — there is no second key-routing mechanism):

• can_focus = True so zone focus can land on it,
• action_* methods (action_cursor_down, action_drill_down, …) reached via KEYMAPS[mode] → KeyDispatcher._resolve_target,
• :meth:get_available_actions for the dynamic footer,
• drill-down via a namespaced Message the app routes to the host's push_view.

Implementing this protocol is opt-in: hosts fall back gracefully (footer default, keys inert) when a widget omits it.

get_available_actions(focused)

Return (key, label) tuples for the dynamic footer.

Source code in src/functualize/plugin/protocols.py

def get_available_actions(self, focused: bool) -> list[tuple[str, str]]:
    """Return (key, label) tuples for the dynamic footer."""
    ...

PanelProvider

Bases: Protocol

Provides a panel for the pre-flight or general ring.

Panels are shown in the Panel Slot below the SmartBar, navigable via Ctrl+H/J/K/L within their ring.

panel_id instance-attribute

panel_title instance-attribute

panel_priority instance-attribute

panel_category instance-attribute

should_show(pending)

Return True if this panel should appear in the ring.

Source code in src/functualize/plugin/protocols.py

def should_show(self, pending: PendingExecution | None) -> bool:
    """Return True if this panel should appear in the ring."""
    ...

compose_panel()

Compose the panel widget tree.

Source code in src/functualize/plugin/protocols.py

def compose_panel(self) -> ComposeResult:
    """Compose the panel widget tree."""
    ...

on_activate(pending)

Called when this panel becomes the active panel in the ring.

Source code in src/functualize/plugin/protocols.py

def on_activate(self, pending: PendingExecution | None) -> None:
    """Called when this panel becomes the active panel in the ring."""
    ...

get_available_actions(focused)

Return (key, label) tuples for the dynamic footer.

Source code in src/functualize/plugin/protocols.py

def get_available_actions(self, focused: bool) -> list[tuple[str, str]]:
    """Return (key, label) tuples for the dynamic footer."""
    ...

PostRunStampProvider

Bases: Protocol

Provides output printed to stdout on TUI exit.

Stamps are rendered after the TUI unmounts, giving plugins a chance to print summary information to the terminal.

render_stamp(session)

Render stamp text, or None to skip.

Source code in src/functualize/plugin/protocols.py

def render_stamp(self, session: SessionState) -> str | None:
    """Render stamp text, or None to skip."""
    ...

SessionState

Placeholder for runtime session state (to be implemented).

SignatureProvider

Bases: Protocol

Provides content for the signature slot above display panels.

Implementations render a single-line signature string shown at the top of the TUI. Multiple providers are stacked by priority (lower first).

priority instance-attribute

render_signature(app)

Render signature text, or None to skip this provider.

Source code in src/functualize/plugin/protocols.py

def render_signature(self, app: FunctualizeApp) -> str | None:
    """Render signature text, or None to skip this provider."""
    ...

StatusBarItemProvider

Bases: Protocol

Provides an item rendered in the status bar.

Items are collected, filtered (None skipped), sorted by priority, and joined with double-space separator.

item_id instance-attribute

item_priority instance-attribute

render_item(app, state)

Render status bar item text, or None to skip.

Source code in src/functualize/plugin/protocols.py

def render_item(self, app: FunctualizeApp, state: SessionState) -> str | None:
    """Render status bar item text, or None to skip."""
    ...

ThemeProvider

Bases: Protocol

Provides a CSS-based color theme for the TUI.

Themes are registered by theme_id. The active theme's CSS is loaded at startup and can be hot-switched via settings.

theme_id instance-attribute

theme_name instance-attribute

get_css()

Return the CSS string for this theme.

Source code in src/functualize/plugin/protocols.py

def get_css(self) -> str:
    """Return the CSS string for this theme."""
    ...

discover_domains()

Discover all installed domain SDKs by scanning entry points.

Scans the functualize.domains entry point group and loads each entry point, expecting a DomainMetadata-compatible instance (any frozen dataclass with the required fields).

Returns:

Type	Description
list[DomainMetadata]	List of successfully loaded DomainMetadata instances.

Source code in src/functualize/_plugins/domain_registry.py

def discover_domains() -> list[DomainMetadata]:
    """Discover all installed domain SDKs by scanning entry points.

    Scans the ``functualize.domains`` entry point group and loads each
    entry point, expecting a ``DomainMetadata``-compatible instance
    (any frozen dataclass with the required fields).

    Returns:
        List of successfully loaded DomainMetadata instances.
    """
    discovered: list[DomainMetadata] = []
    eps = entry_points(group=DOMAINS_ENTRY_POINT_GROUP)

    for ep in eps:
        try:
            metadata = ep.load()
        except Exception as exc:
            logger.warning(f"Failed to load domain entry point '{ep.name}': {exc}")
            continue

        # Validate structurally — the loaded object may be from a domain SDK's
        # local DomainMetadata copy. Convert to our canonical DomainMetadata.
        canonical = _to_canonical_metadata(metadata, ep.name)
        if canonical is None:
            continue

        discovered.append(canonical)
        logger.debug(f"Discovered domain '{canonical.name}' ({canonical.display_name})")

    return discovered

scan_domain_providers(metadata)

Scan a domain's entry point group for available implementation plugins.

Parameters:

Name	Type	Description	Default
metadata	DomainMetadata	The DomainMetadata instance whose entry_point_group to scan.	required

Returns:

Type	Description
dict[str, EntryPoint]	Dictionary mapping provider names to their entry points.

Source code in src/functualize/_plugins/domain_registry.py

def scan_domain_providers(
    metadata: DomainMetadata,
) -> dict[str, importlib.metadata.EntryPoint]:
    """Scan a domain's entry point group for available implementation plugins.

    Args:
        metadata: The DomainMetadata instance whose entry_point_group to scan.

    Returns:
        Dictionary mapping provider names to their entry points.
    """
    eps = entry_points(group=metadata.entry_point_group)
    return {ep.name: ep for ep in eps}

validate_extension_id(extension_id)

Validate an extension ID string.

Valid IDs are: - Non-empty - Lowercase alphanumeric, hyphens, and underscores only - Maximum 64 characters

Parameters:

Name	Type	Description	Default
extension_id	str	The ID string to validate.	required

Returns:

Type	Description
bool	True if valid, False otherwise.

Source code in src/functualize/plugin/protocols.py

def validate_extension_id(extension_id: str) -> bool:
    """Validate an extension ID string.

    Valid IDs are:
    - Non-empty
    - Lowercase alphanumeric, hyphens, and underscores only
    - Maximum 64 characters

    Args:
        extension_id: The ID string to validate.

    Returns:
        True if valid, False otherwise.
    """
    if not extension_id:
        return False
    if len(extension_id) > _EXTENSION_ID_MAX_LENGTH:
        return False
    return _EXTENSION_ID_PATTERN.match(extension_id) is not None

---

Overview

The functualize.plugin module is the entry point for plugin authors. It exports all protocols and types needed to build functualize plugins.

Module location: src/functualize/plugin/

from functualize.plugin import (
    EventBus,
    HookEvent,
    StructuredEvent,
    JobProvider,
    JobTransform,
    Job,
    AdapterPlugin,
    Surface,
    PromptCollector,
    LiveConstruct,
    PromptRequest,
    PromptSeverity,
    PluginMetadata,
    PluginWithShutdown,
    Source,
    FormatProvider,
    ModulePreFilter,
    DEFAULT_SIGIL,
    SettingsSources,
)

---

PluginMetadata

A typing.Protocol with runtime_checkable. Every plugin object must satisfy this protocol.

from functualize.plugin import PluginMetadata

@runtime_checkable
class PluginMetadata(Protocol):
    name: str
    version: str
    description: str

Attribute	Type	Constraint
name	str	Maximum 64 characters.
version	str	Must be a valid PEP 440 version string.
description	str	Maximum 256 characters.

---

Surface

A protocol for objects that render a job's events. The engine's single engine→UI channel: every non-framework event is fanned out to every registered surface.

from functualize.plugin import Surface

@runtime_checkable
class Surface(Protocol):
    def handle_event(self, event: StructuredEvent) -> None: ...

Register with app.extensions.register_surface(obj). handle_event is called on worker threads — a UI implementation must marshal onto its own loop (see functualize.ui.TextualApp, which does this for you). A surface may set needs_terminal = False to keep receiving events while a job owns the screen (log files, MCP progress, test recorders).

Exception safety

Exceptions raised inside handle_event are caught and logged at ERROR level by the event bus; dispatch continues, so a failing surface never interrupts job execution or starves other surfaces.

---

PromptCollector

A protocol for objects that answer a job's prompts. Exactly one collector is active at a time — the one that owns the terminal (or the modal) right now.

from functualize.plugin import PromptCollector

@runtime_checkable
class PromptCollector(Protocol):
    def collect(self, request: PromptRequest) -> PromptResponse: ...

An object may satisfy both Surface and PromptCollector (a full-screen TextualApp does). Also registered with app.extensions.register_surface(obj).

---

LiveConstruct

A protocol for a renderable hosted in a surface's live zone — a job's live: Live capability mounts these via live.add(construct).

from functualize.plugin import LiveConstruct

@runtime_checkable
class LiveConstruct(Protocol):
    def __rich__(self) -> Any: ...   # any Rich renderable

The surface owns the cursor and repaint; the construct just returns its current state as a Rich renderable. The "raw" fallback is Rich's own off-TTY degradation.

---

PromptRequest

A frozen dataclass carrying the complete specification for a user prompt.

from functualize.plugin import PromptRequest

request = PromptRequest(
    question="Select environment",
    intent=PromptIntent.SELECT,
    choices=[PromptChoice(value="staging"), PromptChoice(value="prod")],
)

---

PromptSeverity

Visual severity level for prompt presentation.

from functualize.plugin import PromptSeverity

class PromptSeverity(Enum):
    INFO = "info"
    WARNING = "warning"
    DANGER = "danger"
    SUCCESS = "success"

Largely derivable from PromptIntent — a destructive confirmation is a danger prompt, everything else is informational — so a PromptRequest normally lets the engine's own intent→severity mapping supply it rather than passing it by hand. It remains settable on PromptRequest for the cases where a caller genuinely wants to override the styling (e.g. a warning on a non-destructive action).

---

PluginWithShutdown

A protocol for plugins that need graceful shutdown:

from functualize.plugin import PluginWithShutdown

class MyPlugin:
    def on_shutdown(self, app) -> None:
        """Called during application shutdown."""
        ...

Shutdown methods are called in reverse loading order with a 5-second per-plugin timeout.

---

ModulePreFilter

Decides whether a module is worth importing, before discovery imports it. The built-in filters express the require_* settings; a host whose jobs no setting can describe supplies its own.

from pathlib import Path

from functualize.plugin import ModulePreFilter


class HasJobSuffix:
    def should_import(self, source_file: Path) -> bool:
        return source_file.stem.endswith(("_tasks", "_ops"))

    def fingerprint(self) -> str:
        return "has-job-suffix:v1"

Satisfied structurally — there is nothing to inherit. Supply it through DiscoveryConfig(pre_filter=...), where it is ANDed onto the built-in stack rather than replacing it, and runs last because its cost is unknown.

fingerprint() is not optional

The discovery cache persists negative pre-filter decisions and replays them while the fingerprint matches. Identity cannot stand in for it: str() of an object carries its memory address, so a digest built from the object would differ on every boot — invalidating the cache on every run while appearing to work. A filter with no fingerprint() is refused with a TypeError rather than cached wrongly.

Return the same string across processes for the same behaviour, and a new one when the predicate changes. Forgetting to bump it gives a stale cache — the same contract as any cache key.

See Jobs and Auto-Discovery for the full treatment and Hosting Functualize for the other host seams.

---

DEFAULT_SIGIL

The sigil the shell's default (command) input mode is registered under — the empty string. A mode is selected by the first character of the input text; the default mode's sigil is empty so it is the fallback for any input that starts with no other registered sigil, and cannot collide with a real one.

from functualize.plugin import DEFAULT_SIGIL, InputMode, InputModeRegistry

registry = InputModeRegistry()
registry.register(InputMode(
    sigil=DEFAULT_SIGIL,
    name="command",
    candidate_source=my_candidates,
    is_ready=lambda text: True,
    submit=run_command,
    history_namespace="command",
))

InputModeRegistry.resolve(text) falls back to whatever mode is registered under DEFAULT_SIGIL when text's first character matches no other registered sigil.

---

SettingsSources

Where a host app's settings are read from, in precedence order. Precedence itself is fixed (default < global < project < env); what varies per app is the file names — declaring them is what makes the settings store app-agnostic rather than hardcoded to func's own filenames.

from functualize.plugin import SettingsSources

@dataclass(frozen=True)
class SettingsSources:
    global_file_name: str = "config.toml"
    project_file_names: tuple[str, ...] = (
        "pyproject.toml",
        ".functualize.toml",
        ".functualize/.functualize.toml",
    )
    env: bool = True

Attribute	Type	Description
global_file_name	str	File inside the user config dir (XDG-resolved).
project_file_names	tuple[str, ...]	Candidates for the upward project walk, nearest wins, probed in this order at each level.
env	bool	Whether environment variables participate in resolution at all.

It is the sources field of AppSettingsSchema (also exported from functualize.plugin) — the full declaration a second app hands to the settings store to get the same machinery func uses under its own name.

---

Internal Location

Plugin loading machinery lives in functualize._plugins/:

• _plugins/loader.py — Discovery + dependency sort + loading (also defines PluginMetadata protocol)
• _plugins/config.py — PluginConfigRegistry

Internal API

Modules under functualize._plugins are implementation details. Import from functualize.plugin instead.