Job Module — RunContext

job

Public job authoring API for functualize.

This package is the primary import point for job authors:

from functualize.job import RunContext, Log, Invoke, Prompt, Perf, State
from functualize.job import JobContext, JobConfigView
from functualize.job import Arg, Option, Stdin

It also exports job decorator utilities for metadata and hooks.

__all__ = ['RunContext', 'RunStatus', 'FromJob', 'Log', 'Invoke', 'Prompt', 'Perf', 'Shell', 'ShellError', 'ShellResult', 'Responder', 'FailingResponder', 'Sources', 'Freshness', 'FreshnessVerdict', 'State', 'Stdout', 'JobContext', 'JobConfigView', 'TTY', 'Live', 'TerminalUnavailable', 'Arg', 'Option', 'Stdin', 'GroupOptions', 'job', 'JobDeclaration', 'Deps', 'Fingerprint', 'Guards', 'Exec', 'Retry', 'Precondition', 'Call', 'call', '_make_global_only_decorator', '_make_hook_decorator', '_make_middleware_decorator', 'suppress_live', 'surface_hint'] module-attribute

FromJob(job, /, *, run=True)

A reference to another job, used in a parameter annotation.

See the module docstring for the two accepted forms. Instances are the metadata carried inside Annotated; FromJob[...] builds that Annotated for you.

run=False reads without causing work. The default is that referencing a value declares the dependency — the idiom every comparable system follows (doit's getargs "creates an implicit setup-task"; Dagster infers upstreams "from the arguments to the decorated function"). run=False opts out: use the recorded value if there is one, never trigger execution, and contribute no dependency edge. That is what a reporting job wants — read the last build's result, do not cause a build.

With run=False and nothing recorded, the parameter falls back to its default; a parameter with no default raises, because silently injecting None would make "never ran" indistinguishable from "returned nothing".

Parameters:

Name	Type	Description	Default
job	str | Callable[..., Any]	The upstream job — its registered name, or the decorated function. Positional-only, matching :class:~functualize._types.workflow.Tool.	required
run	bool	Whether a missing or stale value may trigger the upstream.	True

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

def __init__(self, job: str | Callable[..., Any], /, *, run: bool = True) -> None:
    if isinstance(job, str):
        if not job.strip():
            raise ValueError("FromJob reference must not be empty")
    elif not callable(job):
        raise TypeError(
            f"FromJob must reference a registered job name or a callable, "
            f"got {type(job).__name__}"
        )
    object.__setattr__(self, "_job", job)
    object.__setattr__(self, "_run", bool(run))

__slots__ = ('_job', '_run') class-attribute instance-attribute

job property

The referenced job, as written.

run property

Whether resolving this reference may execute the upstream.

name property

The upstream job's name.

__setattr__(name, value)

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

def __setattr__(self, name: str, value: Any) -> None:
    raise AttributeError("FromJob is immutable")

__class_getitem__(job)

Reject the subscript form with the syntax that actually works.

X[...] is the conventional way to parameterize a Python generic, so it is the first thing a reader will try. Without this they would get either a bare "not subscriptable" or — worse — a working runtime object that their type checker rejects, with nothing pointing at the fix.

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

def __class_getitem__(cls, job: Any) -> Any:
    """Reject the subscript form with the syntax that actually works.

    ``X[...]`` is the conventional way to parameterize a Python generic, so
    it is the first thing a reader will try. Without this they would get
    either a bare "not subscriptable" or — worse — a working runtime object
    that their type checker rejects, with nothing pointing at the fix.
    """
    name = getattr(job, "__name__", job)
    raise TypeError(
        f"FromJob[...] is not a type. Write "
        f"Annotated[<return type>, FromJob({name!r})] instead — a function "
        f"object is not valid in a type position (PEP 484), so the "
        f"subscript form cannot type-check."
    )

__repr__()

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

def __repr__(self) -> str:
    suffix = "" if self._run else ", run=False"
    return f"FromJob({self.name!r}{suffix})"

__eq__(other)

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

def __eq__(self, other: object) -> bool:
    if not isinstance(other, FromJob):
        return NotImplemented
    return self.name == other.name and self._run == other._run

__hash__()

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

def __hash__(self) -> int:
    return hash(("FromJob", self.name, self._run))

Call(target, kwargs=dict()) dataclass

A parameterized dependency reference — a job plus bound keyword args.

Produced by the call() factory. Lets a dependency carry config overrides, which parameterized deps require: e.g. call(build, target="wheel") is a distinct dep from call(build, target="sdist") (proposal §A.4).

target instance-attribute

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

__post_init__()

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

def __post_init__(self) -> None:
    if not isinstance(self.target, str) and not callable(self.target):
        raise ValueError(
            "call() target must be a job-name string or a callable, "
            f"got {type(self.target).__name__}"
        )

to_dict()

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

def to_dict(self) -> dict[str, Any]:
    return {
        "ref": _ref_name(self.target),
        "opaque": not isinstance(self.target, str),
        "kwargs": dict(sorted(self.kwargs.items())),
    }

from_dict(data) classmethod

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Call:
    return cls(data["ref"], dict(data.get("kwargs", {})))

Deps(*refs, policy='fail-fast') dataclass

A job's dependency set and its failure policy (proposal §A.4).

Refs are positional and may each be a registered-job name string, a callable (resolved to a name at discovery — refactor-safe and IDE-navigable), or a call(...) for parameterized deps. Unknown names, unregistered callables, and cycles are boot errors (resolved at discovery).

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

def __init__(
    self,
    *refs: str | Callable[..., Any] | Call,
    policy: Literal["fail-fast", "keep-going"] = "fail-fast",
) -> None:
    object.__setattr__(self, "refs", tuple(refs))
    object.__setattr__(self, "policy", policy)
    self.__post_init__()

refs instance-attribute

policy instance-attribute

__post_init__()

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

def __post_init__(self) -> None:
    if self.policy not in ("fail-fast", "keep-going"):
        raise ValueError(
            f"Deps.policy must be 'fail-fast' or 'keep-going', got {self.policy!r}"
        )
    for ref in self.refs:
        if not isinstance(ref, (str, Call)) and not callable(ref):
            raise ValueError(
                "Deps refs must be job-name strings, callables, or call(...), "
                f"got {type(ref).__name__}"
            )

to_dict()

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

def to_dict(self) -> dict[str, Any]:
    refs: list[Any] = []
    for ref in self.refs:
        if isinstance(ref, Call):
            refs.append(ref.to_dict())
        else:
            refs.append({"ref": _ref_name(ref), "opaque": not isinstance(ref, str)})
    return {"refs": refs, "policy": self.policy}

from_dict(data) classmethod

Reconstruct from cache form. Opaque callable refs materialize as their recorded name string (the live callable is unavailable off-cache).

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Deps:
    """Reconstruct from cache form. Opaque callable refs materialize as
    their recorded name string (the live callable is unavailable off-cache)."""
    refs: list[str | Call] = []
    for r in data["refs"]:
        if "kwargs" in r:
            refs.append(Call(r["ref"], dict(r.get("kwargs", {}))))
        else:
            refs.append(r["ref"])
    return cls(*refs, policy=data["policy"])

Exec(retry=None, platforms=None, run='always', silent=False) dataclass

Execution policy for a job (proposal §A.5).

run selects dedup behavior: "always", "once" (per session, ignoring args), or "when_changed" (per session, keyed on identical resolved args).

There is no job-level timeout, deliberately. Python cannot preempt a running function, so any such field could only report an overrun while the work continued — and a caller believing the job stopped may release a lock or delete a file the still-live job is using. The two mature runners in this space reached the same conclusion: invoke has no task-level timeout (only run(cmd, timeout=), a threading.Timer that SIGKILLs a subprocess), and doit has none at all — its doit.tools.timeout is an uptodate checker, a freshness TTL, which is what Fingerprint does here. Bound the work where the OS can enforce it: sh(..., timeout=N) kills the process group (§B.4).

retry = None class-attribute instance-attribute

platforms = None class-attribute instance-attribute

run = 'always' class-attribute instance-attribute

silent = False class-attribute instance-attribute

__post_init__()

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

def __post_init__(self) -> None:
    if self.platforms is not None:
        object.__setattr__(self, "platforms", tuple(self.platforms))
    if self.retry is not None and not isinstance(self.retry, Retry):
        raise ValueError("Exec.retry must be a Retry or None")
    if self.platforms is not None:
        for plat in self.platforms:
            if not isinstance(plat, str):
                raise ValueError("Exec.platforms items must be strings")
    if self.run not in ("always", "once", "when_changed"):
        raise ValueError(
            f"Exec.run must be 'always', 'once', or 'when_changed', "
            f"got {self.run!r}"
        )
    if not isinstance(self.silent, bool):
        raise ValueError("Exec.silent must be a bool")

to_dict()

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

def to_dict(self) -> dict[str, Any]:
    return {
        "retry": self.retry.to_dict() if self.retry is not None else None,
        "platforms": list(self.platforms) if self.platforms is not None else None,
        "run": self.run,
        "silent": self.silent,
    }

from_dict(data) classmethod

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Exec:
    return cls(
        retry=Retry.from_dict(data["retry"]) if data["retry"] else None,
        platforms=tuple(data["platforms"])
        if data["platforms"] is not None
        else None,
        run=data["run"],
        silent=data["silent"],
    )

Fingerprint(sources=(), generates=(), method='checksum', decides=False) dataclass

Up-to-date-checking inputs and outputs for a job (proposal §A.3, §D.3).

sources and generates are both glob patterns — generates=["dist/*.whl"] is a declaration about a wheel whose version is not known in advance, not about a file literally named *.whl. method selects the staleness test.

Neither is required to live under the project (ADR-013). An absolute pattern, a ../ pattern, and a pattern reaching through a symlinked directory are all declarable; each path is recorded as written, so an absolute one does not match on another machine and that machine re-runs the job once.

A declared output that is not on disk forces a run under every method, and a pattern matching nothing counts as not on disk. Otherwise a job whose inputs were unchanged would report fresh with its promised artifact deleted.

decides moves the skip decision from the framework to the job:

============================ =============================================== Declaration A fresh verdict means ============================ =============================================== Fingerprint(sources=[…]) the job is skipped — the default, unchanged decides=True the body runs and decides what to do ============================ ===============================================

The default is the first row on purpose: opting out is a decision a job author makes, never one they inherit. A job that opts in reads :class:~functualize.job.Freshness and returns whatever it likes — its cached artifact, or real work. It does not get to report RunStatus.SKIPPED: "the framework skipped me" and "I ran and decided to do nothing" stay distinguishable in history.

sources = () class-attribute instance-attribute

generates = () class-attribute instance-attribute

method = 'checksum' class-attribute instance-attribute

decides = False class-attribute instance-attribute

__post_init__()

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

def __post_init__(self) -> None:
    object.__setattr__(self, "sources", tuple(self.sources))
    object.__setattr__(self, "generates", tuple(self.generates))
    if self.method not in ("checksum", "timestamp", "none"):
        raise ValueError(
            f"Fingerprint.method must be 'checksum', 'timestamp', or 'none', "
            f"got {self.method!r}"
        )
    for label, seq in (("sources", self.sources), ("generates", self.generates)):
        for item in seq:
            if not isinstance(item, str):
                raise ValueError(
                    f"Fingerprint.{label} items must be strings, "
                    f"got {type(item).__name__}"
                )
    if self.method == "timestamp" and not self.generates:
        raise ValueError(
            "Fingerprint(method='timestamp') requires 'generates' — "
            "timestamp comparison needs output targets to check against."
        )
    if not isinstance(self.decides, bool):
        raise ValueError(
            f"Fingerprint.decides must be a bool, got {type(self.decides).__name__}"
        )

to_dict()

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

def to_dict(self) -> dict[str, Any]:
    return {
        "sources": list(self.sources),
        "generates": list(self.generates),
        "method": self.method,
        # Round-tripped, not omitted. It is read off the live function on a
        # cold boot and off this dict on every warm one, so a field that
        # serializes nowhere silently reverts to its default on the second
        # invocation of a project — the cold/warm divergence class.
        "decides": self.decides,
    }

from_dict(data) classmethod

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Fingerprint:
    return cls(
        sources=tuple(data["sources"]),
        generates=tuple(data["generates"]),
        method=data["method"],
        # `.get` because a discovery cache written before this field existed
        # is discarded only when the *version* changes; the default here is
        # the field's own, so an old entry reads exactly as it always did.
        decides=data.get("decides", False),
    )

Guards(preconditions=(), status=()) dataclass

Pre-flight and up-to-date guards for a job (proposal §A.3, §D.2).

preconditions refuse the run when unmet; status checks report the job already up-to-date (skip). Precondition items may be shell strings, callables, or Precondition objects; status items are shell strings or callables.

preconditions = () class-attribute instance-attribute

status = () class-attribute instance-attribute

__post_init__()

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

def __post_init__(self) -> None:
    object.__setattr__(self, "preconditions", tuple(self.preconditions))
    object.__setattr__(self, "status", tuple(self.status))
    for item in self.preconditions:
        if not isinstance(item, (str, Precondition)) and not callable(item):
            raise ValueError(
                "Guards.preconditions items must be shell strings, callables, "
                f"or Precondition objects, got {type(item).__name__}"
            )
    for item in self.status:
        if not isinstance(item, str) and not callable(item):
            raise ValueError(
                "Guards.status items must be shell strings or callables, "
                f"got {type(item).__name__}"
            )

to_dict()

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

def to_dict(self) -> dict[str, Any]:
    preconditions: list[Any] = []
    for item in self.preconditions:
        if isinstance(item, Precondition):
            preconditions.append(item.to_dict())
        else:
            preconditions.append(
                {
                    "check": _ref_name(item),
                    "opaque": not isinstance(item, str),
                    "msg": None,
                }
            )
    status = [
        {"check": _ref_name(item), "opaque": not isinstance(item, str)}
        for item in self.status
    ]
    return {"preconditions": preconditions, "status": status}

from_dict(data) classmethod

Reconstruct from cache form. Precondition items materialize as Precondition objects; status items as their recorded check strings.

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Guards:
    """Reconstruct from cache form. Precondition items materialize as
    ``Precondition`` objects; status items as their recorded check strings."""
    preconditions = tuple(
        Precondition(p["check"], p.get("msg")) for p in data["preconditions"]
    )
    status = tuple(s["check"] for s in data["status"])
    return cls(preconditions=preconditions, status=status)

JobDeclaration(group=None, extra_description=None, category=None, examples=(), tags=(), visibility='external', config_section=None, deps=None, cache=None, guards=None, exec=None) dataclass

The frozen, aggregate declaration attached by @job (proposal §A.3).

Identity/description fields stay flat (they are what @job is about); operational concerns are the grouped value objects. name/group are stored as declared (None means "fall back to convention"): discovery resolves group against the module-level JOB_GROUP and name against __name__ (proposal §A.3). Serializes to/from the discovery cache via to_dict/from_dict; opaque callables/exception-types do not round-trip (see module docstring).

group = None class-attribute instance-attribute

extra_description = None class-attribute instance-attribute

category = None class-attribute instance-attribute

examples = () class-attribute instance-attribute

tags = () class-attribute instance-attribute

visibility = 'external' class-attribute instance-attribute

config_section = None class-attribute instance-attribute

deps = None class-attribute instance-attribute

cache = None class-attribute instance-attribute

guards = None class-attribute instance-attribute

exec = None class-attribute instance-attribute

__post_init__()

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

def __post_init__(self) -> None:
    # None is accepted as "none of these" and normalized to an empty tuple.
    object.__setattr__(self, "examples", tuple(self.examples or ()))
    object.__setattr__(self, "tags", tuple(self.tags or ()))
    if self.group is not None and not isinstance(self.group, str):
        raise ValueError("@job group must be a string or None")
    for label, seq in (("examples", self.examples), ("tags", self.tags)):
        for item in seq:
            if not isinstance(item, str):
                raise ValueError(f"@job {label} items must be strings")
    if self.visibility not in ("external", "internal"):
        raise ValueError(
            f"@job visibility must be 'external' or 'internal', "
            f"got {self.visibility!r}"
        )
    for label, obj, typ in (
        ("deps", self.deps, Deps),
        ("cache", self.cache, Fingerprint),
        ("guards", self.guards, Guards),
        ("exec", self.exec, Exec),
    ):
        if obj is not None and not isinstance(obj, typ):
            raise ValueError(
                f"@job {label} must be a {typ.__name__} or None, "
                f"got {type(obj).__name__}"
            )

to_dict()

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

def to_dict(self) -> dict[str, Any]:
    return {
        "group": self.group,
        "extra_description": self.extra_description,
        "category": self.category,
        "examples": list(self.examples),
        "tags": list(self.tags),
        "visibility": self.visibility,
        "config_section": self.config_section,
        "deps": self.deps.to_dict() if self.deps is not None else None,
        "cache": self.cache.to_dict() if self.cache is not None else None,
        "guards": self.guards.to_dict() if self.guards is not None else None,
        "exec": self.exec.to_dict() if self.exec is not None else None,
    }

from_dict(data) classmethod

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> JobDeclaration:
    return cls(
        group=data["group"],
        extra_description=data["extra_description"],
        category=data["category"],
        examples=tuple(data["examples"]),
        tags=tuple(data["tags"]),
        visibility=data["visibility"],
        config_section=data["config_section"],
        deps=Deps.from_dict(data["deps"]) if data["deps"] else None,
        cache=Fingerprint.from_dict(data["cache"]) if data["cache"] else None,
        guards=Guards.from_dict(data["guards"]) if data["guards"] else None,
        exec=Exec.from_dict(data["exec"]) if data["exec"] else None,
    )

Precondition(cmd_or_callable, msg=None) dataclass

A pre-flight check plus an optional human-facing failure message.

The check is a shell-command string (run, non-zero = refuse) or a callable (falsy return = refuse) — proposal §A.3, §D.2.

cmd_or_callable instance-attribute

msg = None class-attribute instance-attribute

__post_init__()

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

def __post_init__(self) -> None:
    if not isinstance(self.cmd_or_callable, str) and not callable(
        self.cmd_or_callable
    ):
        raise ValueError(
            "Precondition check must be a shell-command string or a callable, "
            f"got {type(self.cmd_or_callable).__name__}"
        )
    if self.msg is not None and not isinstance(self.msg, str):
        raise ValueError("Precondition.msg must be a string or None")

to_dict()

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

def to_dict(self) -> dict[str, Any]:
    is_str = isinstance(self.cmd_or_callable, str)
    return {
        "check": _ref_name(self.cmd_or_callable),
        "opaque": not is_str,
        "msg": self.msg,
    }

from_dict(data) classmethod

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Precondition:
    return cls(data["check"], data.get("msg"))

Retry(attempts, backoff='exponential', on=(), on_exit_codes=()) dataclass

Retry policy for a job (proposal §A.5).

on narrows retries to specific exception types; on_exit_codes narrows to specific ShellError exit codes. Empty means "retry on any failure".

attempts instance-attribute

backoff = 'exponential' class-attribute instance-attribute

on = () class-attribute instance-attribute

on_exit_codes = () class-attribute instance-attribute

__post_init__()

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

def __post_init__(self) -> None:
    object.__setattr__(self, "on", tuple(self.on))
    object.__setattr__(self, "on_exit_codes", tuple(self.on_exit_codes))
    if not isinstance(self.attempts, int) or isinstance(self.attempts, bool):
        raise ValueError("Retry.attempts must be an int")
    if self.attempts < 1:
        raise ValueError(f"Retry.attempts must be >= 1, got {self.attempts}")
    if self.backoff not in ("exponential", "linear", "constant"):
        raise ValueError(
            f"Retry.backoff must be 'exponential', 'linear', or 'constant', "
            f"got {self.backoff!r}"
        )
    for exc in self.on:
        if not (isinstance(exc, type) and issubclass(exc, BaseException)):
            raise ValueError(f"Retry.on items must be exception types, got {exc!r}")
    for code in self.on_exit_codes:
        if not isinstance(code, int) or isinstance(code, bool):
            raise ValueError(
                f"Retry.on_exit_codes items must be ints, got {code!r}"
            )

to_dict()

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

def to_dict(self) -> dict[str, Any]:
    return {
        "attempts": self.attempts,
        "backoff": self.backoff,
        "on": [exc.__name__ for exc in self.on],
        "on_exit_codes": list(self.on_exit_codes),
    }

from_dict(data) classmethod

Reconstruct from cache form. on exception types cannot round-trip (recorded by name only) and materialize empty off-cache.

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Retry:
    """Reconstruct from cache form. ``on`` exception types cannot round-trip
    (recorded by name only) and materialize empty off-cache."""
    return cls(
        attempts=data["attempts"],
        backoff=data["backoff"],
        on=(),
        on_exit_codes=tuple(data["on_exit_codes"]),
    )

TTY(*, caps, available, funcapp=None)

Terminal-ownership capability delivered by DI to a job that owns its UI.

You cannot obtain the handle without declaring the parameter, and you cannot declare the parameter without the router seeing it — one artifact serves both the static routing decision and the runtime handle.

Parameters:

Name	Type	Description	Default
caps	dict[type, Any]	The per-invocation capability map; ctx resolves the RunContext from it lazily (so declaration order does not matter).	required
available	bool	Whether terminal ownership can be granted in this context.	required

Source code in src/functualize/_engine/capabilities/tty.py

def __init__(
    self, *, caps: dict[type, Any], available: bool, funcapp: Any = None
) -> None:
    self._caps = caps
    self._available = available
    self._funcapp = funcapp

ctx property

The RunContext for this execution — the app's API handle.

Resolved lazily from the per-invocation capability map, so it is populated by the time the job body runs regardless of parameter order. Returns None only if the job declared no rc: RunContext parameter (a fuller guarantee arrives with the Phase 5 orchestrator handoff).

run(app)

Run a job-owned app while it owns the terminal.

Raises:

Type	Description
TerminalUnavailable	If terminal ownership cannot be granted here (MCP / CI / piped / background).
TypeError	If app is not runnable (no .run() method).

Source code in src/functualize/_engine/capabilities/tty.py

def run(self, app: Any) -> Any:
    """Run a job-owned app while it owns the terminal.

    Raises:
        TerminalUnavailable: If terminal ownership cannot be granted here
            (MCP / CI / piped / background).
        TypeError: If ``app`` is not runnable (no ``.run()`` method).
    """
    if not self._available:
        raise TerminalUnavailable(
            "This job needs an interactive terminal (it declares "
            "`tty: TTY`). Run it from `func` at a real TTY — it cannot run "
            "over MCP, in CI, or with piped I/O."
        )
    run = getattr(app, "run", None)
    if not callable(run):
        raise TypeError(
            "tty.run(app) expects a runnable app with a .run() method "
            "(e.g. a functualize.ui.TextualApp); got "
            f"{type(app).__name__}."
        )

    # A Surface-conforming app becomes the active surface for its window:
    # child rc.invoke() events fan out to it and nested prompts route to
    # its collect(). Popped in finally so a crash still unwinds the stack.
    from functualize._types.interactivity import Surface

    pushed = False
    if self._funcapp is not None and isinstance(app, Surface):
        self._funcapp.push_surface(app)
        pushed = True
    try:
        return run()
    finally:
        if pushed:
            self._funcapp.pop_surface(app)

FailingResponder(pattern, response, sentinel) dataclass

Bases: Responder

A :class:Responder that also aborts on a failure sentinel (§B.4).

Responds like :class:Responder, but if sentinel appears in the output the command is killed and a :class:ShellError is raised. This is how a sudo password responder aborts on Sorry, try again instead of answering the re-prompt forever.

Subclasses :class:Responder so it is accepted anywhere watchers are.

Attributes:

Name	Type	Description
pattern	str | Pattern[str]	Regex whose match triggers response (inherited).
response	str	Text written to stdin on each new pattern match (inherited).
sentinel	str | Pattern[str]	Regex whose appearance aborts the command.

sentinel instance-attribute

Freshness(verdict=None)

The verdict this job's own Fingerprint produced, for the job to act on.

Declared as a job parameter, like every other capability::

@job(cache=Fingerprint(sources=["src/**/*.py"], decides=True))
def build(fresh: Freshness) -> str:
    verdict = fresh.verdict()
    if verdict is not None and verdict.is_fresh:
        return "artifact already current"
    return rebuild()

decides=True is what makes the branch above reachable, and it is not decoration. Without it the engine skips the job for you and returns before the body runs, so is_fresh is the one value that branch can never see — which is the misconception this capability exists to remove, reproduced in its own documentation until a review ran the snippet (jof S1). The rule: ask for the verdict only where you have taken the decision back.

:meth:verdict returns None when this job declares no Fingerprint — there was no decision, and a fabricated one would be a lie. A job that declares one gets the verdict whether or not the framework used it to skip.

Source code in src/functualize/_engine/capabilities/freshness.py

def __init__(self, verdict: FreshnessVerdict | None = None) -> None:
    self._verdict = verdict

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

verdict()

The decision this run's pre-flight made, or None if none was made.

Source code in src/functualize/_engine/capabilities/freshness.py

def verdict(self) -> FreshnessVerdict | None:
    """The decision this run's pre-flight made, or None if none was made."""
    return self._verdict

FreshnessVerdict(state, key, recorded_value, declared_sources, declared_generates, source_map) dataclass

What the pre-flight decided about this run, and the inputs it decided on.

It is the decision itself, not a summary of it: source_map is the very mapping Sources exposes, and the declared patterns are the ones the job's own Fingerprint named. A job asking "why am I fresh?" therefore gets one answer from one capability rather than two answers free to drift.

Attributes:

Name	Type	Description
state	GuardState	The pipeline outcome — SKIP_FRESH for a job whose declared inputs are current, RUN when they are not.
key	str	The fingerprint key the decision was computed under, so a job can see which recorded run it is being compared against.
recorded_value	Any | None	What the previous run recorded, when the pre-flight read a record and the verdict was a skip.
declared_sources	tuple[str, ...]	The input patterns as the job declared them.
declared_generates	tuple[str, ...]	The output patterns as the job declared them.
source_map	Mapping[str, Mapping[str, Any]]	{path: {mtime, size, sha256}} for every match of the declared inputs — the decision's own mapping, bound by reference, so asking for it costs no second copy.

state instance-attribute

key instance-attribute

recorded_value instance-attribute

declared_sources instance-attribute

declared_generates instance-attribute

source_map instance-attribute

is_fresh property

True when the engine decided this job's declared inputs are current.

Only SKIP_FRESH answers True. A satisfied status guard is a different claim about a different thing ("already done"), and it is deliberately not folded in here — the two states are distinct for the same reason they are distinct in the guard pipeline.

Invoke

Job-to-job invocation capability.

Allows a job to invoke other jobs by name or function reference, run jobs in parallel, or introspect job schemas. The actual execution is delegated to the engine — this class raises NotImplementedError until wired.

The engine replaces this stub with a fully-wired instance at runtime.

__call__(job_or_fn, *, config=None, awaits_input=None, available_tools=None, force_gate=False, gate_strategy=None, timeout=None, group_option_values=None, **kwargs)

Invoke a job by name or function reference.

Parameters:

Name	Type	Description	Default
job_or_fn	str | Callable[..., Any]	Job name string or registered callable.	required
config	BaseModel | None	Typed config (mutually exclusive with **kwargs).	None
awaits_input	type[BaseModel] | None	BaseModel subclass describing input the job needs; its JSON schema is attached to the result metadata for the caller (and drives gate resolution when gate params are set).	None
available_tools	list[str] | None	Restrict tool visibility at the gate (max 64).	None
force_gate	bool	Dispatch gate strategy even when fully resolved.	False
gate_strategy	GateStrategy | str | list[GateStrategy | str] | None	Override configured gate strategy for this invocation.	None
timeout	float | None	Optional timeout in seconds.	None
**kwargs	Any	Arguments to pass to the job.	{}

Returns:

Type	Description
JobResult	A JobResult with status, return_value, exception, metadata.

Raises:

Type	Description
JobNotFoundError	If callable is not registered.
ValueError	If both config and kwargs are provided.
ValueError	If available_tools contains unregistered tool names.
ValueError	If available_tools has more than 64 entries.

Source code in src/functualize/_engine/capabilities/invoke.py

def __call__(
    self,
    job_or_fn: str | Callable[..., Any],
    *,
    config: BaseModel | None = None,
    awaits_input: type[BaseModel] | None = None,
    available_tools: list[str] | None = None,
    force_gate: bool = False,
    gate_strategy: GateStrategy | str | list[GateStrategy | str] | None = None,
    timeout: float | None = None,
    group_option_values: Mapping[str, Any] | None = None,
    **kwargs: Any,
) -> JobResult:
    """Invoke a job by name or function reference.

    Args:
        job_or_fn: Job name string or registered callable.
        config: Typed config (mutually exclusive with **kwargs).
        awaits_input: BaseModel subclass describing input the job needs;
            its JSON schema is attached to the result metadata for the
            caller (and drives gate resolution when gate params are set).
        available_tools: Restrict tool visibility at the gate (max 64).
        force_gate: Dispatch gate strategy even when fully resolved.
        gate_strategy: Override configured gate strategy for this invocation.
        timeout: Optional timeout in seconds.
        **kwargs: Arguments to pass to the job.

    Returns:
        A JobResult with status, return_value, exception, metadata.

    Raises:
        JobNotFoundError: If callable is not registered.
        ValueError: If both config and kwargs are provided.
        ValueError: If available_tools contains unregistered tool names.
        ValueError: If available_tools has more than 64 entries.
    """

    # --- Resolve job_or_fn ---
    self._resolve_job_name(job_or_fn)

    # --- Config-kwargs mutual exclusivity check ---
    if config is not None and kwargs:
        raise ValueError(
            "Cannot pass both 'config' and keyword arguments to invoke(). "
            "Use one or the other."
        )

    # --- Extract config fields as kwargs if config provided ---
    if config is not None:
        kwargs = config.model_dump()

    # --- Validate available_tools ---
    if available_tools is not None:
        if len(available_tools) > 64:
            raise ValueError(
                f"available_tools must have at most 64 entries, "
                f"got {len(available_tools)}"
            )
        self._validate_tool_names(available_tools)

    raise NotImplementedError(
        "Invoke capability is not wired. "
        "This instance must be replaced by the engine at runtime."
    )

parallel(jobs, *, timeout=None)

Invoke 1-32 jobs concurrently, returning results in input order.

Parameters:

Name	Type	Description	Default
jobs	Sequence[tuple[str | Callable[..., Any], dict[str, Any]]]	List of (job_or_fn, kwargs) tuples to execute concurrently.	required
timeout	float | None	Seconds the batch may run before unfinished jobs are reported as timed out. None uses :data:DEFAULT_PARALLEL_TIMEOUT; <= 0 waits indefinitely.	None

Returns:

Type	Description
list[JobResult]	List of JobResult objects in the same order as input.

Raises:

Type	Description
ValueError	If more than 32 jobs are specified.

Source code in src/functualize/_engine/capabilities/invoke.py

def parallel(
    self,
    jobs: Sequence[tuple[str | Callable[..., Any], dict[str, Any]]],
    *,
    timeout: float | None = None,
) -> list[JobResult]:
    """Invoke 1-32 jobs concurrently, returning results in input order.

    Args:
        jobs: List of (job_or_fn, kwargs) tuples to execute concurrently.
        timeout: Seconds the batch may run before unfinished jobs are
            reported as timed out. ``None`` uses
            :data:`DEFAULT_PARALLEL_TIMEOUT`; ``<= 0`` waits indefinitely.

    Returns:
        List of JobResult objects in the same order as input.

    Raises:
        ValueError: If more than 32 jobs are specified.
    """
    if len(jobs) > 32:
        raise ValueError(
            f"Invoke.parallel accepts at most 32 jobs, got {len(jobs)}"
        )

    raise NotImplementedError(
        "Invoke.parallel is not wired. "
        "This instance must be replaced by the engine at runtime."
    )

schema(job_or_fn)

Retrieve the JobDescriptor for a job by name or function reference.

Parameters:

Name	Type	Description	Default
job_or_fn	str | Callable[..., Any]	Job name string or registered callable.	required

Returns:

Type	Description
JobDescriptor	A JobDescriptor for the referenced job.

Raises:

Type	Description
JobNotFoundError	If the callable/name is not registered.

Source code in src/functualize/_engine/capabilities/invoke.py

def schema(self, job_or_fn: str | Callable[..., Any]) -> JobDescriptor:
    """Retrieve the JobDescriptor for a job by name or function reference.

    Args:
        job_or_fn: Job name string or registered callable.

    Returns:
        A JobDescriptor for the referenced job.

    Raises:
        JobNotFoundError: If the callable/name is not registered.
    """
    raise NotImplementedError(
        "Invoke.schema is not wired. "
        "This instance must be replaced by the engine at runtime."
    )

JobConfigView(resolution_chain, default_section_prefix='general')

Scoped, read-write config access for job execution.

Wraps ResolutionChain (parse-once, shared) with: - A section prefix (scoped to job name) - An in-memory override layer for programmatic set() - Same get/set/get_model API that RunContext.config exposed

Initialize with shared resolution chain and optional prefix.

Parameters:

Name	Type	Description	Default
resolution_chain	ResolutionChain	The app's ResolutionChain (shared, read-only).	required
default_section_prefix	str	Initial section prefix for key lookups.	'general'

Source code in src/functualize/_config/job_config.py

def __init__(
    self,
    resolution_chain: ResolutionChain,
    default_section_prefix: str = "general",
) -> None:
    """Initialize with shared resolution chain and optional prefix.

    Args:
        resolution_chain: The app's ResolutionChain (shared, read-only).
        default_section_prefix: Initial section prefix for key lookups.
    """
    self._chain = resolution_chain
    self._default_section_prefix = default_section_prefix
    self._overrides: dict[str, Any] = {}

get(key, default=None, section=None)

Retrieve a configuration value.

Resolution priority: 1. In-memory overrides (from set()) 2. ResolutionChain (env vars → file sources → defaults) 3. Caller-provided default parameter

This method never raises — it always returns a value or None.

Parameters:

Name	Type	Description	Default
key	str	The configuration key.	required
default	Any	Fallback value if key not found anywhere.	None
section	str | None	Section override. None uses default_section_prefix.	None

Returns:

Type	Description
Any	Resolved value or default.

Source code in src/functualize/_config/job_config.py

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

    Resolution priority:
    1. In-memory overrides (from set())
    2. ResolutionChain (env vars → file sources → defaults)
    3. Caller-provided default parameter

    This method never raises — it always returns a value or None.

    Args:
        key: The configuration key.
        default: Fallback value if key not found anywhere.
        section: Section override. None uses default_section_prefix.

    Returns:
        Resolved value or default.
    """
    effective_section = (
        section if section is not None else self._default_section_prefix
    )
    combined_key = _env_token(effective_section, key)

    # Step 1: Check in-memory overrides
    if combined_key in self._overrides:
        return self._overrides[combined_key]

    # Step 2-3: Delegate to ResolutionChain
    try:
        resolved = self._chain.resolve(key, effective_section)
        return resolved.value
    except MissingKeyError:
        return default

resolve_with_source(key, section=None)

(source_type, source_id, value) for key, or None if unset.

The provenance-carrying sibling of :meth:get, for surfaces that must report where a value came from as well as what it is. It consults the same layers in the same order — overrides first, then the chain — which is the whole reason it exists here rather than in the caller: the display seam previously reached past this view to config_view._chain and so could not see the override layer at all. A value set through :meth:set was what the run used and not what any surface showed.

Never raises: an unresolved key is an answer, not an error.

Source code in src/functualize/_config/job_config.py

def resolve_with_source(
    self,
    key: str,
    section: str | None = None,
) -> tuple[str, str, Any] | None:
    """``(source_type, source_id, value)`` for ``key``, or None if unset.

    The provenance-carrying sibling of :meth:`get`, for surfaces that must
    report *where* a value came from as well as what it is. It consults the
    same layers in the same order — overrides first, then the chain — which
    is the whole reason it exists here rather than in the caller: the
    display seam previously reached past this view to
    ``config_view._chain`` and so could not see the override layer at all.
    A value set through :meth:`set` was what the run used and not what any
    surface showed.

    Never raises: an unresolved key is an answer, not an error.
    """
    effective_section = (
        section if section is not None else self._default_section_prefix
    )
    combined_key = _env_token(effective_section, key)

    if combined_key in self._overrides:
        return ("override", "set() at runtime", self._overrides[combined_key])

    try:
        resolved = self._chain.resolve(key, effective_section)
    except MissingKeyError:
        return None
    except Exception:
        # Introspection must never mask the real failure a run would report.
        return None
    if resolved is None or resolved.value is None:
        return None
    return (resolved.source_type, resolved.source_id, resolved.value)

set(key, value, section=None)

Store a value in the in-memory override layer.

Does NOT write to disk. Override persists for this instance's lifetime.

Parameters:

Name	Type	Description	Default
key	str	The configuration key.	required
value	Any	The value to store.	required
section	str | None	Section override. None uses default_section_prefix.	None

Source code in src/functualize/_config/job_config.py

def set(
    self,
    key: str,
    value: Any,
    section: str | None = None,
) -> None:
    """Store a value in the in-memory override layer.

    Does NOT write to disk. Override persists for this instance's lifetime.

    Args:
        key: The configuration key.
        value: The value to store.
        section: Section override. None uses default_section_prefix.
    """
    effective_section = (
        section if section is not None else self._default_section_prefix
    )
    combined_key = _env_token(effective_section, key)
    self._overrides[combined_key] = value

get_model(model_class, section=None)

Resolve a Pydantic model from configuration values.

For each field in the model, resolves the value using the same priority as get(). Passes collected values to Pydantic for validation.

Parameters:

Name	Type	Description	Default
model_class	type[T]	Pydantic BaseModel subclass.	required
section	str | None	Section override. None uses default_section_prefix.	None

Returns:

Type	Description
T	Validated model instance.

Raises:

Type	Description
ValidationError	If values don't satisfy model schema.

Source code in src/functualize/_config/job_config.py

def get_model(
    self,
    model_class: type[T],
    section: str | None = None,
) -> T:
    """Resolve a Pydantic model from configuration values.

    For each field in the model, resolves the value using the same
    priority as get(). Passes collected values to Pydantic for validation.

    Args:
        model_class: Pydantic BaseModel subclass.
        section: Section override. None uses default_section_prefix.

    Returns:
        Validated model instance.

    Raises:
        pydantic.ValidationError: If values don't satisfy model schema.
    """
    effective_section = (
        section if section is not None else self._default_section_prefix
    )

    data: dict[str, Any] = {}

    # Use model_fields if available (Pydantic v2), fallback for duck typing
    fields = getattr(model_class, "model_fields", {})
    for field_name in fields:
        value = self.get(field_name, default=None, section=effective_section)
        if value is not None:
            data[field_name] = value

    return model_class(**data)

set_prefix(prefix)

Update the default section prefix.

Called by RunContext.init to scope lookups to the job name.

Parameters:

Name	Type	Description	Default
prefix	str	New default section prefix.	required

Source code in src/functualize/_config/job_config.py

def set_prefix(self, prefix: str) -> None:
    """Update the default section prefix.

    Called by RunContext.__init__ to scope lookups to the job name.

    Args:
        prefix: New default section prefix.
    """
    self._default_section_prefix = prefix

JobContext(name, trace_id=None, span_id=None, cwd=None, job_directory=None, invoke_depth=0, scope_id=None, metadata=(lambda: MappingProxyType({}))()) dataclass

Immutable execution context for the current job invocation.

Attributes:

Name	Type	Description
name	str	The job name being executed.
trace_id	str | None	Optional distributed trace identifier.
span_id	str | None	Optional 16 hex-character span identifier from PropagationContext.
cwd	Path | None	Optional working directory for the job execution.
job_directory	Path | None	Optional filesystem directory containing the job's source module.
invoke_depth	int	Nesting depth of invocations (0 at top-level, increments per nest).
scope_id	str | None	Optional active WorkflowScope ID.
metadata	MappingProxyType[str, Any]	Read-only mapping of arbitrary key-value metadata.

name instance-attribute

trace_id = None class-attribute instance-attribute

span_id = None class-attribute instance-attribute

cwd = None class-attribute instance-attribute

job_directory = None class-attribute instance-attribute

invoke_depth = 0 class-attribute instance-attribute

scope_id = None class-attribute instance-attribute

metadata = field(default_factory=(lambda: MappingProxyType({}))) class-attribute instance-attribute

__post_init__()

Source code in src/functualize/_engine/capabilities/job_context.py

def __post_init__(self) -> None:
    if self.invoke_depth < 0:
        raise ValueError(f"invoke_depth must be >= 0, got {self.invoke_depth}")

Live(*, _zone=None)

Per-invocation live-display channel, bound to the active rendering surface.

Parameters:

Name	Type	Description	Default
_zone	Any | None	The surface-side live zone that actually mounts constructs, or None for the degraded/kernel case (handles become no-ops).	None

Source code in src/functualize/_engine/capabilities/live.py

def __init__(self, *, _zone: Any | None = None) -> None:
    self._zone = _zone

suppress(name)

Hide an ambient construct for this invocation.

Ambient constructs are the ones a plugin registered to render by default (a flow-viz tree, say). A job that wants a quiet output for one run drops it here::

def simple(live: Live) -> None:
    live.suppress("flow-viz")

The declarative equivalent is @job(suppress_live=["flow-viz"]); project-wide, it is [live] suppress. No-op in a degraded context.

Source code in src/functualize/_engine/capabilities/live.py

def suppress(self, name: str) -> None:
    """Hide an ambient construct for this invocation.

    Ambient constructs are the ones a plugin registered to render by
    default (a flow-viz tree, say). A job that wants a quiet output for
    one run drops it here::

        def simple(live: Live) -> None:
            live.suppress("flow-viz")

    The declarative equivalent is ``@job(suppress_live=["flow-viz"])``;
    project-wide, it is ``[live] suppress``. No-op in a degraded context.
    """
    if self._zone is not None:
        suppressor = getattr(self._zone, "suppress_ambient", None)
        if callable(suppressor):
            suppressor(name)

suppress_all()

Hide every ambient construct for this invocation.

Source code in src/functualize/_engine/capabilities/live.py

def suppress_all(self) -> None:
    """Hide every ambient construct for this invocation."""
    if self._zone is not None:
        suppressor = getattr(self._zone, "suppress_all_ambient", None)
        if callable(suppressor):
            suppressor()

add(construct)

Mount a passive construct (a Rich renderable) in the live zone.

Source code in src/functualize/_engine/capabilities/live.py

def add(self, construct: LiveConstruct) -> LiveHandle:
    """Mount a passive construct (a Rich renderable) in the live zone."""
    if self._zone is not None:
        return self._zone.add(construct)  # type: ignore[no-any-return]
    return LiveHandle(construct)

panel(construct)

Mount an interactive construct as a PanelHost panel (j/k/Enter).

Requires an event loop; where none exists (STDOUT) it degrades to a passive render, and in MCP to event-emission — the same handle, honest degradation. In the kernel it is a no-op.

Source code in src/functualize/_engine/capabilities/live.py

def panel(self, construct: LiveConstruct) -> LiveHandle:
    """Mount an interactive construct as a PanelHost panel (j/k/Enter).

    Requires an event loop; where none exists (STDOUT) it degrades to a
    passive render, and in MCP to event-emission — the same handle, honest
    degradation. In the kernel it is a no-op.
    """
    if self._zone is not None:
        return self._zone.panel(construct)  # type: ignore[no-any-return]
    return LiveHandle(construct)

Log(job_name=None)

Structured logging capability for job functions.

Supports both direct call syntax and named level methods

log("message") # defaults to info log("message", level="warning") log.error("something broke")

When constructed with a job_name, log records are emitted to the per-job logger functualize.job.<job_name>, which allows the TUI and other handlers to capture output for the specific job. Without a job name the fallback functualize.job logger is used (backward-compatible).

Source code in src/functualize/_engine/capabilities/log.py

def __init__(self, job_name: str | None = None) -> None:
    if job_name:
        self._logger = logging.getLogger(f"functualize.job.{job_name}")
    else:
        self._logger = _DEFAULT_LOGGER

__call__(message, level='info')

Log a message at the specified level.

Parameters:

Name	Type	Description	Default
message	object	The message to log (will be converted to str).	required
level	str	One of "debug", "info", "warning", "error", "critical".	'info'

Raises:

Type	Description
ValueError	If level is not a valid log level.

Source code in src/functualize/_engine/capabilities/log.py

def __call__(self, message: object, level: str = "info") -> None:
    """Log a message at the specified level.

    Args:
        message: The message to log (will be converted to str).
        level: One of "debug", "info", "warning", "error", "critical".

    Raises:
        ValueError: If level is not a valid log level.
    """
    validate_log_level(level)
    numeric_level = getattr(logging, level.upper())
    self._logger.log(numeric_level, str(message))

info(msg)

Log at INFO level.

Source code in src/functualize/_engine/capabilities/log.py

def info(self, msg: object) -> None:
    """Log at INFO level."""
    self(msg, level="info")

warning(msg)

Log at WARNING level.

Source code in src/functualize/_engine/capabilities/log.py

def warning(self, msg: object) -> None:
    """Log at WARNING level."""
    self(msg, level="warning")

error(msg)

Log at ERROR level.

Source code in src/functualize/_engine/capabilities/log.py

def error(self, msg: object) -> None:
    """Log at ERROR level."""
    self(msg, level="error")

debug(msg)

Log at DEBUG level.

Source code in src/functualize/_engine/capabilities/log.py

def debug(self, msg: object) -> None:
    """Log at DEBUG level."""
    self(msg, level="debug")

Perf

Performance measurement capability.

Provides methods for marking instants and measuring durations during job execution. The actual implementation is backed by the observability layer and wired at runtime.

This class raises NotImplementedError until wired.

mark(name)

Record an instant performance mark.

Parameters:

Name	Type	Description	Default
name	str	The mark name (non-empty, max 256 characters).	required

Raises:

Type	Description
NotImplementedError	Until wired by the observability layer.

Source code in src/functualize/_engine/capabilities/perf.py

def mark(self, name: str) -> None:
    """Record an instant performance mark.

    Args:
        name: The mark name (non-empty, max 256 characters).

    Raises:
        NotImplementedError: Until wired by the observability layer.
    """
    raise NotImplementedError(
        "Perf.mark is not wired. "
        "This instance must be replaced by the observability layer at runtime."
    )

mark_start(name)

Start a named timing phase.

Parameters:

Name	Type	Description	Default
name	str	The phase name (non-empty, max 256 characters).	required

Raises:

Type	Description
NotImplementedError	Until wired by the observability layer.

Source code in src/functualize/_engine/capabilities/perf.py

def mark_start(self, name: str) -> None:
    """Start a named timing phase.

    Args:
        name: The phase name (non-empty, max 256 characters).

    Raises:
        NotImplementedError: Until wired by the observability layer.
    """
    raise NotImplementedError(
        "Perf.mark_start is not wired. "
        "This instance must be replaced by the observability layer at runtime."
    )

mark_end(name)

End a named timing phase.

Parameters:

Name	Type	Description	Default
name	str	The phase name (must match a previous mark_start call).	required

Raises:

Type	Description
NotImplementedError	Until wired by the observability layer.

Source code in src/functualize/_engine/capabilities/perf.py

def mark_end(self, name: str) -> None:
    """End a named timing phase.

    Args:
        name: The phase name (must match a previous mark_start call).

    Raises:
        NotImplementedError: Until wired by the observability layer.
    """
    raise NotImplementedError(
        "Perf.mark_end is not wired. "
        "This instance must be replaced by the observability layer at runtime."
    )

phases(include=None, exclude=None)

Retrieve recorded phases with optional filtering.

Parameters:

Name	Type	Description	Default
include	str | None	Optional regex pattern — only return phases whose names match.	None
exclude	str | None	Optional regex pattern — exclude phases whose names match.	None

Returns:

Type	Description
list[Phase]	List of Phase objects matching the filters.

Raises:

Type	Description
NotImplementedError	Until wired by the observability layer.

Source code in src/functualize/_engine/capabilities/perf.py

def phases(
    self, include: str | None = None, exclude: str | None = None
) -> list[Phase]:
    """Retrieve recorded phases with optional filtering.

    Args:
        include: Optional regex pattern — only return phases whose names match.
        exclude: Optional regex pattern — exclude phases whose names match.

    Returns:
        List of Phase objects matching the filters.

    Raises:
        NotImplementedError: Until wired by the observability layer.
    """
    raise NotImplementedError(
        "Perf.phases is not wired. "
        "This instance must be replaced by the observability layer at runtime."
    )

Prompt(*, _provider=None, _rc=None, _caps=None)

Ask the person on the other end, or be told there is none.

Reached two ways, and it is the same object either way (ADR-021): rc.prompts.confirm(...) and a prompt: Prompt parameter.

Parameters:

Name	Type	Description	Default
_provider	PromptCollector | None	A collector bound at construction. The kernel's own callers (Shell.sudo, missing-value prompting) resolve a collector first and pass it here, so those paths do not depend on a RunContext.	None
_caps	dict[type, Any] | None	The run's capability map, from which the RunContext — and through it the live surface stack — is found at call time. Lazy on purpose: DI resolution runs before the RunContext exists, so capturing it eagerly captures None (the defect capability-duality/T1 fixed for Invoke).	None

Source code in src/functualize/_engine/capabilities/prompt.py

def __init__(
    self,
    *,
    _provider: PromptCollector | None = None,
    _rc: Any | None = None,
    _caps: dict[type, Any] | None = None,
) -> None:
    self._provider = _provider
    self._explicit_rc = _rc
    self._caps = _caps

__slots__ = ('_caps', '_explicit_rc', '_provider') class-attribute instance-attribute

ask(request)

Send a fully-constructed request to the surface that can answer it.

The low-level form every convenience method routes through. Use it when you need control over the PromptRequest fields.

Raises:

Type	Description
InputNotAvailable	Nothing can collect and the request is required with no default — the case where returning a default would fabricate an answer nobody gave.

Source code in src/functualize/_engine/capabilities/prompt.py

def ask(self, request: PromptRequest) -> PromptResponse:
    """Send a fully-constructed request to the surface that can answer it.

    The low-level form every convenience method routes through. Use it when
    you need control over the `PromptRequest` fields.

    Raises:
        InputNotAvailable: Nothing can collect and the request is required
            with no default — the case where returning a default would
            fabricate an answer nobody gave.
    """
    filled = dataclasses.replace(request, source_job=self._job_name)
    provider = self._get_input_provider()
    if provider is None:
        if filled.required and filled.default is None:
            raise InputNotAvailable(
                f"No InputProvider registered and prompt requires input "
                f"(job={self._job_name!r}, question={filled.question!r}). "
                f"Prompts need either an interactive terminal or a "
                f"registered surface (see docs/guides/interactivity.md)."
            )
        return PromptResponse(value=filled.default, source="default")
    return provider.collect(filled)

confirm(question, *, destructive=False, default=None, context_message=None, context_data=None)

Ask a yes/no question.

With no default the question is required: off a terminal it raises rather than assuming an answer. Pass default to make the non-interactive answer explicit.

Source code in src/functualize/_engine/capabilities/prompt.py

def confirm(
    self,
    question: str,
    *,
    destructive: bool = False,
    default: bool | None = None,
    context_message: str | None = None,
    context_data: dict[str, Any] | None = None,
) -> bool:
    """Ask a yes/no question.

    With no ``default`` the question is *required*: off a terminal it
    raises rather than assuming an answer. Pass ``default`` to make the
    non-interactive answer explicit.
    """
    intent = (
        PromptIntent.CONFIRM_DESTRUCTIVE
        if destructive
        else PromptIntent.CONFIRM_NEUTRAL
    )
    # Derived, not hand-mapped — one source of truth for the styling.
    response = self.ask(
        PromptRequest(
            question=question,
            intent=intent,
            severity=severity_for_intent(intent),
            default=default,
            context_message=context_message,
            context_data=context_data,
            required=default is None,
        )
    )
    if response.was_cancelled:
        return False
    value = response.value
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        return value.lower() in ("yes", "y", "true", "1")
    return bool(value) if value is not None else False

choice(question, choices, *, default=None, context_message=None)

Present options and return the selected value.

Source code in src/functualize/_engine/capabilities/prompt.py

def choice(
    self,
    question: str,
    choices: list[str] | list[PromptChoice],
    *,
    default: str | None = None,
    context_message: str | None = None,
) -> str:
    """Present options and return the selected value."""
    normalized = [
        PromptChoice(value=c) if isinstance(c, str) else c for c in choices
    ]
    response = self.ask(
        PromptRequest(
            question=question,
            intent=PromptIntent.SELECT,
            choices=normalized,
            default=default,
            context_message=context_message,
            required=default is None,
        )
    )
    return str(response.value) if response.value is not None else ""

text(question, *, default=None, secret=False, placeholder=None, validator=None, context_message=None)

Ask for free-form text. secret=True asks without echoing.

Source code in src/functualize/_engine/capabilities/prompt.py

def text(
    self,
    question: str,
    *,
    default: str | None = None,
    secret: bool = False,
    placeholder: str | None = None,
    validator: str | Any | None = None,
    context_message: str | None = None,
) -> str:
    """Ask for free-form text. ``secret=True`` asks without echoing."""
    intent = PromptIntent.SECRET_INPUT if secret else PromptIntent.TEXT_INPUT
    response = self.ask(
        PromptRequest(
            question=question,
            intent=intent,
            default=default,
            placeholder=placeholder,
            validator=validator,
            context_message=context_message,
            required=default is None,
        )
    )
    return str(response.value) if response.value is not None else ""

Responder(pattern, response) dataclass

Answers an interactive prompt in a command's live output (§B.4).

When pattern (a regex) appears in new output, response is written to the child's stdin. Use for scripted interactions — a tool asking Continue? [y/N], a password prompt, etc. response should include its own trailing newline if the program expects Enter.

Attributes:

Name	Type	Description
pattern	str | Pattern[str]	Regex searched against live output (str or compiled pattern).
response	str	Text written to stdin on each new match.

pattern instance-attribute

response instance-attribute

Shell

Bases: Protocol

DI-injectable shell-command capability (proposal Part B).

Three command forms (§B.1), in order of preference:

• list — sh(["docker", "build", "-t", tag, "."]) — no shell, no quoting problem. The documented default idiom.
• template — sh("docker build -t {tag} .", tag=version) — each substituted value is shlex.quote-d before interpolation, then run without a shell.
• raw — sh("a | b", shell=True) — explicit shell interpretation; the caller owns quoting.

A raw string with neither shell=True nor template params is an error (never a silent shlex.split) — that ambiguity is where injection bugs live.

__call__(command, *, capture=True, stream=None, check=True, cwd=None, env=None, replace_env=False, in_stream=None, timeout=None, retry=None, shell=False, pty=False, watchers=None, background=False, label=None, silent=False, **template_params)

Run command and return a :class:ShellResult.

stream=True routes live output to the surface's channel — stdout when piped, the panel in the TUI (§C.1); a callable is an explicit sink. silent=True suppresses the command echo, which is otherwise written to the surface's diagnostic channel (stderr when piped) so it can never corrupt piped data.

pty allocates a pseudo-terminal (POSIX; degrades to pipes on Windows) so programs that check isatty() behave interactively. watchers are :class:Responder/:class:FailingResponder objects that answer prompts in the live output by writing to stdin.

Raises:

Type	Description
ShellError	If the command exits non-zero and check is True, on timeout (always), or when a :class:FailingResponder sentinel appears.
ValueError	If a raw string is passed without shell=True or template params (ambiguous — see class docstring).

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

def __call__(
    self,
    command: list[str] | str,
    *,
    capture: bool = True,
    stream: Callable[[str], None] | bool | None = None,
    check: bool = True,
    cwd: str | None = None,
    env: Mapping[str, str] | None = None,
    replace_env: bool = False,
    in_stream: str | None = None,
    timeout: float | None = None,
    retry: Retry | None = None,
    shell: bool = False,
    pty: bool = False,
    watchers: Sequence[Responder] | None = None,
    background: bool = False,
    label: str | None = None,
    silent: bool = False,
    **template_params: Any,
) -> ShellResult:
    """Run ``command`` and return a :class:`ShellResult`.

    ``stream=True`` routes live output to the **surface's** channel —
    stdout when piped, the panel in the TUI (§C.1); a callable is an
    explicit sink. ``silent=True`` suppresses the command echo, which is
    otherwise written to the surface's *diagnostic* channel (stderr when
    piped) so it can never corrupt piped data.

    ``pty`` allocates a pseudo-terminal (POSIX; degrades to pipes on
    Windows) so programs that check ``isatty()`` behave interactively.
    ``watchers`` are :class:`Responder`/:class:`FailingResponder` objects
    that answer prompts in the live output by writing to stdin.

    Raises:
        ShellError: If the command exits non-zero and ``check`` is True, on
            timeout (always), or when a :class:`FailingResponder` sentinel
            appears.
        ValueError: If a raw string is passed without ``shell=True`` or
            template params (ambiguous — see class docstring).
    """
    ...

cd(path)

Run commands in path for the duration of the block (§B.3).

Nestable — a nested cd resolves relative to the enclosing one. Overridden by an explicit per-call cwd=.

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

def cd(self, path: str) -> AbstractContextManager[None]:
    """Run commands in ``path`` for the duration of the block (§B.3).

    Nestable — a nested ``cd`` resolves relative to the enclosing one.
    Overridden by an explicit per-call ``cwd=``.
    """
    ...

prefix(command)

Prepend command to every command in the block (§B.3).

with sh.prefix(["poetry", "run"]): sh(["pytest"]) runs poetry run pytest. Nestable; outer prefixes apply before inner.

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

def prefix(self, command: list[str] | str) -> AbstractContextManager[None]:
    """Prepend ``command`` to every command in the block (§B.3).

    ``with sh.prefix(["poetry", "run"]): sh(["pytest"])`` runs
    ``poetry run pytest``. Nestable; outer prefixes apply before inner.
    """
    ...

defer(command, **kwargs)

Register a cleanup command to run when the job exits (§B.5).

Deferred commands run LIFO on the engine's job-exit unwind — on success, on failure, on Ctrl+C, and after a timeout — not through a user try/finally, which a killed subprocess tree or a hard timeout skips. kwargs are the options :meth:__call__ takes.

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

def defer(self, command: list[str] | str, **kwargs: Any) -> None:
    """Register a cleanup command to run when the job exits (§B.5).

    Deferred commands run **LIFO** on the engine's job-exit unwind — on
    success, on failure, on Ctrl+C, and after a timeout — not through a
    user ``try``/``finally``, which a killed subprocess tree or a hard
    timeout skips. ``kwargs`` are the options :meth:`__call__` takes.
    """
    ...

run_deferred()

Run and clear all deferred commands, LIFO (engine-owned, §B.5).

Best-effort: a cleanup that fails must not mask the job's own outcome and must not stop the remaining cleanups.

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

def run_deferred(self) -> None:
    """Run and clear all deferred commands, LIFO (engine-owned, §B.5).

    Best-effort: a cleanup that fails must not mask the job's own outcome
    and must not stop the remaining cleanups.
    """
    ...

sudo(command, *, preserve_env=False, password=None, watchers=None, **kwargs)

Run command under sudo -S with an auto password responder (§B.4).

The password comes from password= or the injected [shell] sudo_password secret, and is fed to sudo's stdin rather than placed in the echoed command. Requires the list command form.

Raises:

Type	Description
ValueError	No password is available, or command is not a list.
ShellError	The command itself failed.

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

def sudo(
    self,
    command: list[str],
    *,
    preserve_env: bool = False,
    password: Any = None,
    watchers: Sequence[Responder] | None = None,
    **kwargs: Any,
) -> ShellResult:
    """Run ``command`` under ``sudo -S`` with an auto password responder (§B.4).

    The password comes from ``password=`` or the injected
    ``[shell] sudo_password`` secret, and is fed to sudo's stdin rather
    than placed in the echoed command. Requires the list command form.

    Raises:
        ValueError: No password is available, or ``command`` is not a list.
        ShellError: The command itself failed.
    """
    ...

ShellError(result)

Bases: Exception

Raised when a command exits non-zero under check=True (§B.2).

Carries the full :class:ShellResult so callers can inspect output.

Attributes:

Name	Type	Description
result		The ShellResult for the failed command.

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

def __init__(self, result: ShellResult) -> None:
    self.result = result
    super().__init__(f"Command failed (exit {result.returncode}): {result.command}")

result = result instance-attribute

ShellResult(returncode, stdout, stderr, command, duration_ms, pid=None) dataclass

The outcome of a single shell command (proposal §B.2).

Attributes:

Name	Type	Description
returncode	int	Process exit code.
stdout	str	Captured standard output (empty string when not captured).
stderr	str	Captured standard error (empty string when not captured).
command	str	Display form of the command (secrets masked — §B.6).
duration_ms	float	Wall-clock duration in milliseconds.
pid	int | None	The child process id (or None if it never started).

returncode instance-attribute

stdout instance-attribute

stderr instance-attribute

command instance-attribute

duration_ms instance-attribute

pid = None class-attribute instance-attribute

ok property

True when the command exited zero.

Sources(source_map=None, *, declared=False, generates=())

The files this job's Fingerprint(sources=...) resolved to.

Reads as a mapping of project-relative POSIX path → {"mtime", "size", "sha256"}::

@job(cache=Fingerprint(sources=["src/**/*.yaml"]))
def parse(sources: Sources) -> Parsed:
    files = {path: Path(path).read_text() for path in sources.keys()}

declared is not the same question as emptiness, and conflating them is the bug this exists next to:

============================================ ========== ========= Declaration declared items() ============================================ ========== ========= sources=["src/*.yaml"], files present True populated sources=["absent/*.yaml"], no match True empty no Fingerprint, or no sources False empty ============================================ ========== =========

The middle row is the R3 refusal's trigger, read here through the same mechanism rather than a second one.

Source code in src/functualize/_engine/capabilities/sources.py

def __init__(
    self,
    source_map: Mapping[str, Any] | None = None,
    *,
    declared: bool = False,
    generates: Sequence[str] = (),
) -> None:
    self._map: Mapping[str, Any] = dict(source_map or {})
    self._declared = declared
    self._generates: tuple[str, ...] = tuple(generates)

__slots__ = ('_declared', '_generates', '_map') class-attribute instance-attribute

declared property

True when the job declares Fingerprint(sources=...) at all.

Tells "declared, nothing matched" (True, empty) apart from "declared no sources" (False, empty) — which an empty mapping alone cannot.

generates property

Declared outputs, as project-relative POSIX paths.

items()

Source code in src/functualize/_engine/capabilities/sources.py

def items(self) -> ItemsView[str, Any]:
    return self._map.items()

keys()

Source code in src/functualize/_engine/capabilities/sources.py

def keys(self) -> KeysView[str]:
    return self._map.keys()

values()

Source code in src/functualize/_engine/capabilities/sources.py

def values(self) -> Any:
    return self._map.values()

get(path, default=None)

Source code in src/functualize/_engine/capabilities/sources.py

def get(self, path: str, default: Any = None) -> Any:
    return self._map.get(path, default)

__len__()

Source code in src/functualize/_engine/capabilities/sources.py

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

__iter__()

Source code in src/functualize/_engine/capabilities/sources.py

def __iter__(self) -> Any:
    return iter(self._map)

__contains__(path)

Source code in src/functualize/_engine/capabilities/sources.py

def __contains__(self, path: str) -> bool:
    return path in self._map

__getitem__(path)

Source code in src/functualize/_engine/capabilities/sources.py

def __getitem__(self, path: str) -> Any:
    return self._map[path]

__bool__()

Truthy when at least one input resolved.

Note this is emptiness, not :attr:declared — a job that declared sources which matched nothing is falsy here and declared. That pair is the whole point; see the class docstring.

Source code in src/functualize/_engine/capabilities/sources.py

def __bool__(self) -> bool:
    """Truthy when at least one input resolved.

    Note this is emptiness, not :attr:`declared` — a job that declared
    sources which matched nothing is falsy here *and* ``declared``. That
    pair is the whole point; see the class docstring.
    """
    return bool(self._map)

__repr__()

Source code in src/functualize/_engine/capabilities/sources.py

def __repr__(self) -> str:
    return (
        f"Sources(declared={self._declared}, resolved={len(self._map)}, "
        f"generates={list(self._generates)!r})"
    )

State(backend)

A run's key-value store, shared by every job in the run.

Reached as rc.state, or by declaring state: State. Both are the same object (ADR-021).

Keys are flat and the whole run shares one namespace, so two jobs can pick the same name. The answer is a naming convention rather than a framework namespace — write state.set("fetch.rows", n) and read the namespace back with state.keys("fetch.*"). One concept (a string) instead of two.

Values must be JSON-serializable; that is checked at write time, where the offending call is still on the stack, rather than at the write of the file.

Every write is one lock-read-write cycle against scopes.json. A job writing a handful of keys will not notice; one writing hundreds in a loop should hold :meth:batch.

Source code in src/functualize/_engine/capabilities/state.py

def __init__(self, backend: ScopeBackedStateStore | None) -> None:
    self._backend = backend

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

get(key, default=None)

The value stored under key, or default.

Source code in src/functualize/_engine/capabilities/state.py

def get(self, key: str, default: Any = None) -> Any:
    """The value stored under ``key``, or ``default``."""
    return self._bound().get(key, default)

set(key, value)

Store value under key.

Raises:

Type	Description
TypeError	value is not JSON-serializable. Raised by the backing store, which is the thing that has to write it.

Source code in src/functualize/_engine/capabilities/state.py

def set(self, key: str, value: Any) -> None:
    """Store ``value`` under ``key``.

    Raises:
        TypeError: ``value`` is not JSON-serializable. Raised by the
            backing store, which is the thing that has to write it.
    """
    self._bound().set(key, value)

delete(key)

Remove key. A no-op when it is not there.

Source code in src/functualize/_engine/capabilities/state.py

def delete(self, key: str) -> None:
    """Remove ``key``. A no-op when it is not there."""
    self._bound().delete(key)

keys(pattern='')

Stored key names, optionally filtered by a glob pattern.

* stops at a . and ** crosses it::

state.keys("fetch.*")    # one level under `fetch`
state.keys("fetch.**")   # every depth under `fetch`
state.keys("*.rows")     # the same leaf in any namespace
state.keys()             # everything

So "fetch.*" cannot match "fetchmeta.x": the literal dot has to match a real dot. A bare prefix is a startswith and does reach the neighbouring namespace, which is why the pattern form is documented. A pattern containing no * keeps the prefix meaning.

The matcher is the one the rest of the codebase uses — rc.events.on_event("job.*") and perf-phase filtering call the same function. A second glob implementation that agreed with this one today is the divergence ADR-021 exists to prevent.

Raises:

Type	Description
TypeError	pattern is not a string.

Source code in src/functualize/_engine/capabilities/state.py

def keys(self, pattern: str = "") -> list[str]:
    """Stored key names, optionally filtered by a glob ``pattern``.

    ``*`` stops at a ``.`` and ``**`` crosses it::

        state.keys("fetch.*")    # one level under `fetch`
        state.keys("fetch.**")   # every depth under `fetch`
        state.keys("*.rows")     # the same leaf in any namespace
        state.keys()             # everything

    So ``"fetch.*"`` cannot match ``"fetchmeta.x"``: the literal dot has to
    match a real dot. A bare prefix is a ``startswith`` and does reach the
    neighbouring namespace, which is why the pattern form is documented. A
    pattern containing no ``*`` keeps the prefix meaning.

    The matcher is the one the rest of the codebase uses —
    ``rc.events.on_event("job.*")`` and perf-phase filtering call the same
    function. A second glob implementation that agreed with this one today
    is the divergence ADR-021 exists to prevent.

    Raises:
        TypeError: ``pattern`` is not a string.
    """
    if not isinstance(pattern, str):
        raise TypeError(f"pattern must be a str, got {type(pattern).__name__}")
    names = list(self._bound().keys())
    if not pattern:
        return names
    from functualize._events._pattern_matcher import matches_pattern

    return [name for name in names if matches_pattern(name, pattern)]

to_dict()

Every key this run holds, as a plain dict.

Source code in src/functualize/_engine/capabilities/state.py

def to_dict(self) -> dict[str, Any]:
    """Every key this run holds, as a plain dict."""
    return self._bound().to_dict()

clear()

Drop every key this run holds.

Source code in src/functualize/_engine/capabilities/state.py

def clear(self) -> None:
    """Drop every key this run holds."""
    self._bound().clear()

batch()

Hold the file lock across many writes, writing once at the end.

Every :meth:set is otherwise its own lock-read-write cycle against scopes.json::

with state.batch():
    for name, n in counts.items():
        state.set(f"fetch.{name}", n)

An exception inside the block discards the block's writes rather than persisting some of them. A backend that cannot batch yields a plain null context, so this is always safe to write.

Source code in src/functualize/_engine/capabilities/state.py

def batch(self) -> Any:
    """Hold the file lock across many writes, writing once at the end.

    Every :meth:`set` is otherwise its own lock-read-write cycle against
    ``scopes.json``::

        with state.batch():
            for name, n in counts.items():
                state.set(f"fetch.{name}", n)

    An exception inside the block discards the block's writes rather than
    persisting some of them. A backend that cannot batch yields a plain
    null context, so this is always safe to write.
    """
    backend = self._bound()
    batch = getattr(backend, "batch", None)
    if batch is None:
        from contextlib import nullcontext

        return nullcontext()
    return batch()

Stdout

Bases: Protocol

DI-injectable explicit stdout data channel (proposal Part C, revised).

Two methods, two intents:

• emit(value) — serialize value to stdout per the resolved --emit-format format, one logical document per call, flushed per call. value may be str/bytes, dict/list, a pydantic model, a dataclass, or an iterable of those. --emit-format decides list handling: emit([a, b, c]) is one JSON array under json and one line per item under ndjson. To stream rows explicitly, loop for r in rows: out.emit(r).
• write(data) — raw verbatim passthrough (str or bytes), no serialization. For cat-like filters and binary output.

Unlike the old implicit return-value emission, an explicit emit/write is never surface-suppressed: it always writes (on a TTY too), routed so it coordinates with an active TUI, with secrets masked.

emit(value)

Serialize value to stdout per the resolved --emit-format format.

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

def emit(self, value: Any) -> None:
    """Serialize ``value`` to stdout per the resolved ``--emit-format`` format."""
    ...

write(data)

Write data to stdout verbatim (no serialization).

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

def write(self, data: str | bytes) -> None:
    """Write ``data`` to stdout verbatim (no serialization)."""
    ...

TerminalUnavailable(message=None, *, job_name=None)

Bases: Exception

Raised when a job needs an interactive terminal but none is available.

A job that declares tty: TTY (a hard requirement) owns the terminal for the duration of tty.run(app). In contexts that cannot grant terminal ownership — MCP, CI, piped/redirected I/O, background execution — the job is refused with this error (pre-flight where the router can see the requires_tty marker; at tty.run time otherwise), naming the fix.

Attributes:

Name	Type	Description
job_name		The job that required a terminal, if known.

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

def __init__(
    self, message: str | None = None, *, job_name: str | None = None
) -> None:
    self.job_name = job_name
    super().__init__(
        message
        or "This job needs an interactive terminal (it declares `tty: TTY`)."
    )

job_name = job_name instance-attribute

RunContext(name, config, logger, metadata=None, *, plugin_configs=None, resources=None, perf_timeline=None, _workflow_scope=None, _invoke_depth=0, _parent_request=None, _run_id=None, _max_invoke_depth=10, _execution_engine=None, cwd=None, job_directory=None, _di_registry=None, _caps=None)

Execution context injected into each job — thin facade delegating to capabilities.

Source code in src/functualize/_engine/capabilities/runcontext.py

def __init__(
    self,
    name: str,
    config: JobConfigView,
    logger: Logger,
    metadata: dict[str, Any] | None = None,
    *,
    plugin_configs: dict[str, BaseModel] | None = None,
    resources: dict[str, Any] | None = None,
    perf_timeline: PerfTimeline | None = None,
    _workflow_scope: WorkflowScope | None = None,
    _invoke_depth: int = 0,
    _parent_request: Any = None,
    _run_id: str | None = None,
    _max_invoke_depth: int = 10,
    _execution_engine: Any = None,
    cwd: Path | None = None,
    job_directory: Path | None = None,
    _di_registry: DIRegistry | None = None,
    _caps: dict[type, Any] | None = None,
):
    self._name = name
    self._config = config
    self._logger = logger
    self._metadata: dict[str, Any] = metadata.copy() if metadata else {}
    self._metadata.setdefault("run_type", RunType.JOB)
    self._metadata.setdefault("run_status", RunStatus.RUNNING)
    self._metadata.setdefault("start_time", datetime.now(UTC))
    self._metadata.setdefault("end_time", None)
    self._metadata.setdefault("duration", None)
    self._job_config: Any = None
    self._plugin_configs: dict[str, BaseModel] | None = plugin_configs
    self._state: State | None = None
    #: Deliberately absent. `rc.state` resolves through the capability
    #: map and the scope; a per-context store parameter was a third way to
    #: obtain one, used only by tests, and three doors onto one fact is
    #: what ADR-021 exists to remove.
    self._resources: dict[str, Any] | None = resources
    self._perf_timeline: PerfTimeline | None = perf_timeline
    self._workflow_scope: WorkflowScope | None = _workflow_scope
    self._invoke_depth: int = _invoke_depth
    self._parent_request: Any = _parent_request
    #: The run-log id of *this* run, so `rc.invoke` children can name
    #: their parent. Carried rather than looked up: a batch item runs on
    #: a worker thread, where a `ContextVar` would be empty.
    self._run_id: str | None = _run_id
    self._max_invoke_depth: int = _max_invoke_depth
    self._execution_engine: Any = _execution_engine
    self._cwd: Path | None = cwd
    self._job_directory: Path | None = job_directory
    self._result_metadata: dict[str, Any] = {}
    self._di_registry: DIRegistry | None = _di_registry
    # The live per-invocation capability map (the same dict the engine
    # fills as it resolves bindings) — log() reads the job's own Log out
    # of it, so rc.log() and a `log: Log` parameter share one sink.
    self._caps: dict[type, Any] | None = _caps
    self._config.set_prefix(name)
    # Capability instances (lazily created)
    self._invoke_capability: Invoke | None = None
    self._workflow_tracker: WorkflowTracker | None = None
    # Callback registrations (for backward compat)
    self._status_callbacks: list[Any] = []
    self._phase_callbacks: list[Any] = []
    self._log_callbacks: list[Any] = []
    #: Facades. `RunContext` reached 800 lines by being the one object a
    #: job holds, so everything a job might ever want was a method on it
    #: (T8). These group the rarer capabilities behind a name that says
    #: which subject they belong to; the core a job actually reaches for —
    #: `config`, `log`, `invoke`, `state`, `cwd` — stays flat.
    self._discovery: DiscoveryFacade | None = None
    self._wiring: WiringFacade | None = None
    self._events: ObservabilityFacade | None = None
    self._prompts: Prompt | None = None

MAX_INVOKE_DEPTH = 10 class-attribute

prompts property

rc.prompts — ask the person on the other end, if there is one.

The same object a prompt: Prompt parameter receives (ADR-021). Resolved through the capability map so the two doors cannot drift; the fallback below builds one only for a context with no map at all.

events property

rc.events — events, phases, run status and the perf timeline.

wiring property

rc.wiring — the plugin configs and resources this app provides.

discovery property

rc.discovery — read-only questions about the registered jobs.

config property

name property

metadata property

result_metadata property

job_config property writable

workflow_scope property

cwd property

This run's working directory: the one it named, or the project's.

The run's own cwd wins when the request carried one. Otherwise the answer is the project root the engine's host knows, not the process's working directory — which is what this used to return, and which is a different directory whenever the two disagree.

job_directory property

state property

rc.state — the run's shared, durable key-value store.

The same object a state: State parameter receives (ADR-021). Resolved through the capability map so the two doors cannot drift; the scope is consulted only to build one when the job declared no state: parameter and nothing has therefore materialised it yet.

on_log(callback)

Register a callback invoked on log emissions.

Source code in src/functualize/_engine/capabilities/runcontext.py

def on_log(self, callback: Any) -> None:
    """Register a callback invoked on log emissions."""
    self._log_callbacks.append(callback)

set_result_metadata(key, value)

Source code in src/functualize/_engine/capabilities/runcontext.py

def set_result_metadata(self, key: str, value: Any) -> None:
    if (
        key in self._result_metadata
        or len(self._result_metadata) < self._MAX_RESULT_METADATA_KEYS
    ):
        self._result_metadata[key] = value

__getitem__(key)

__getitem__(key: type) -> Any

__getitem__(key: tuple[type, str]) -> Any

__getitem__(key: str) -> Any

rc[T] — the run's instance of T.

The capability map first, the DI registry second (ADR-021). A per-invocation capability lives in the map and is never in the registry, so consulting only the registry reported a capability the job was holding in its own hand as missing: rc[Log] raised MissingProviderError and Log in rc was False while a log: Log parameter had the object.

A qualified or named lookup skips the map deliberately. There is no "the" Conn when two are registered under different qualifiers, so those forms are exactly the ones that must reach the registry and let it answer — see ADR-021 for the classes that cannot share one object.

Source code in src/functualize/_engine/capabilities/runcontext.py

def __getitem__(self, key: type | str | tuple[type, str]) -> Any:
    """`rc[T]` — the run's instance of ``T``.

    **The capability map first, the DI registry second** (ADR-021). A
    per-invocation capability lives in the map and is never in the
    registry, so consulting only the registry reported a capability the job
    was *holding in its own hand* as missing: `rc[Log]` raised
    `MissingProviderError` and `Log in rc` was False while a `log: Log`
    parameter had the object.

    A qualified or named lookup skips the map deliberately. There is no
    "the" `Conn` when two are registered under different qualifiers, so
    those forms are exactly the ones that must reach the registry and let
    it answer — see ADR-021 for the classes that cannot share one object.
    """
    from functualize._primitives.di import (
        AmbiguousProviderError,
        MissingProviderError,
    )

    if isinstance(key, type):
        found = self._cap_or_none(key)
        if found is not None:
            return found

    if self._di_registry is None:
        raise RuntimeError(
            "Cannot use subscript access: RunContext has no DI registry attached"
        )
    if isinstance(key, tuple):
        type_, qualifier = key
        try:
            return self._di_registry.resolve(type_, qualifier=qualifier)
        except AmbiguousProviderError:
            raise MissingProviderError(
                type_=type_,
                job_name=self._name,
                available=self._di_registry.available_types(),
            ) from None
    elif isinstance(key, str):
        try:
            return self._di_registry.resolve_named(key)
        except MissingProviderError:
            raise MissingProviderError(
                type_=str,
                job_name=self._name,
                available=self._di_registry.available_types(),
            ) from None
    else:
        return self._di_registry.resolve(key)

__contains__(key)

T in rc — is T reachable from this run?

Same order as :meth:__getitem__, and for the same reason: a capability in the map is reachable even though the registry has never heard of it.

Source code in src/functualize/_engine/capabilities/runcontext.py

def __contains__(self, key: type | str) -> bool:
    """`T in rc` — is ``T`` reachable from this run?

    Same order as :meth:`__getitem__`, and for the same reason: a
    capability in the map is reachable even though the registry has never
    heard of it.
    """
    if isinstance(key, type) and self._cap_or_none(key) is not None:
        return True
    if self._di_registry is None:
        return False
    if isinstance(key, str):
        return self._di_registry.has_named(key)
    return self._di_registry.has(key)

invoke(job_name, *, _propagate_scope=True, timeout=None, **kwargs)

Invoke another registered job. Delegates to Invoke capability.

Source code in src/functualize/_engine/capabilities/runcontext.py

def invoke(
    self,
    job_name: str,
    *,
    _propagate_scope: bool = True,
    timeout: float | None = None,
    **kwargs: Any,
) -> JobResult:
    """Invoke another registered job. Delegates to Invoke capability."""
    return self._get_invoke()(job_name, timeout=timeout, **kwargs)

invoke_parallel(jobs)

Invoke multiple jobs concurrently. Delegates to Invoke.parallel().

Source code in src/functualize/_engine/capabilities/runcontext.py

def invoke_parallel(
    self, jobs: list[tuple[str, dict[str, Any]]]
) -> list[JobResult]:
    """Invoke multiple jobs concurrently. Delegates to Invoke.parallel()."""
    return self._get_invoke().parallel(jobs)

log(message, level='info')

Source code in src/functualize/_engine/capabilities/runcontext.py

def log(self, message: object, level: str = "info") -> None:
    # Validate before the callbacks so an invalid level fails the same way
    # whichever sink is behind it — the Log capability, or the fallback
    # logger whose getattr would otherwise raise AttributeError instead.
    validate_log_level(level)
    msg = str(message)
    # Invoke log callbacks BEFORE emitting to logger
    for cb in self._log_callbacks:
        try:
            result = cb(level, msg)
            # If callback returns None, suppress the message
            if result is None:
                return
            # If callback returns a string, use it as the new message
            if isinstance(result, str):
                msg = result
        except Exception:
            self._logger.warning(
                "Log callback %r raised an exception", cb, exc_info=True
            )
    sink = self._log_sink()
    if sink is not None:
        sink(msg, level=level)
    else:
        getattr(self._logger, level)(msg)

RunStatus

Bases: Enum

Status of a run context execution.

SUCCESS = 'Success' class-attribute instance-attribute

FAILURE = 'Failure' class-attribute instance-attribute

BLOCKED = 'Blocked' class-attribute instance-attribute

SKIPPED = 'Skipped' class-attribute instance-attribute

RUNNING = 'Running' class-attribute instance-attribute

CANCELLED = 'Cancelled' class-attribute instance-attribute

TIMEOUT = 'Timeout' class-attribute instance-attribute

UNKNOWN = 'Unknown' class-attribute instance-attribute

REFUSED = 'Refused' class-attribute instance-attribute

resumable property

True when the run stopped at a declared pause point, not an error.

A workflow that reaches a Gate with no input is BLOCKED, not FAILURE: it did everything it was asked to, and re-invoking it with the input deposited continues from where it stopped. Callers that treat "not SUCCESS" as "broken" — CI gates, Deps satisfaction, TUI colouring — need to tell the two apart.

terminal property

True when the phase is over and cannot be transitioned from.

One definition, because there were two and they disagreed. _engine/capabilities/workflow.py and _engine/capabilities/runcontext.py each carried a _TERMINAL_STATES frozenset; one included REFUSED and one did not. The first one's own comment explains why the omission matters — "a refused step simply never gets end_time or duration, so it reads as still running in every consumer of this record" — and its sibling had exactly that defect, one module away. The runcontext copy's test mirrored the omission and asserted agreement with it, so the two could never converge by failing.

This is the state machine's question — may a phase transition out of here — and not the delivery question functualize.types.is_failure answers. The membership is the one workflow.py reasoned about, kept exactly: BLOCKED is a declared pause and resumable is the property that says so; SKIPPED is a phase a run can still move on from; RUNNING is the non-terminal case by definition; and UNKNOWN is deliberately left transitionable, because a status nobody could determine should not also be a state nobody can leave.

ran property

True when the job's body actually executed.

SKIPPED is not a failure — a guard or an up-to-date fingerprint answered "no work to do", which is the point of declaring them. But it is not SUCCESS either: a caller that treats them alike cannot tell a build that ran from one that was already current, and func why exists precisely to answer that.

Arg(help=None, metavar=None, show_default=True) dataclass

Mark a parameter as a positional CLI argument.

Non-CLI adapters ignore this marker — they see only the base type.

Usage::

from typing import Annotated
from functualize.job import Arg

def deploy(target: Annotated[str, Arg(help="Deploy target")]):
    ...

help = None class-attribute instance-attribute

metavar = None class-attribute instance-attribute

show_default = True class-attribute instance-attribute

Option(*args, help=None, hidden=False, envvar=None) dataclass

Mark a parameter as a named CLI option with optional short flag.

Non-CLI adapters ignore this marker — they see only the base type.

Usage::

from typing import Annotated
from functualize.job import Option

def deploy(target: Annotated[str, Option("-t", "--target", help="Deploy target")]):
    ...

# Short flag only — long flag derived from param name:
def deploy(verbose: Annotated[bool, Option("-v")]):
    ...

Accept positional strings as flag names: Option("-t", "--target").

Positional arg detection: - Single dash + one char (e.g. -t) → short flag - Double dash + name (e.g. --target) → long flag - Two positional args: detect by dash count, assign accordingly

Parameters:

Name	Type	Description	Default
*args	str	Flag name strings (short like -t, long like --target).	()
help	str | None	Help text for the option.	None
hidden	bool	Whether to hide this option from --help output.	False
envvar	str | None	Environment variable name to read as fallback.	None

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

def __init__(
    self,
    *args: str,
    help: str | None = None,
    hidden: bool = False,
    envvar: str | None = None,
) -> None:
    """Accept positional strings as flag names: Option("-t", "--target").

    Positional arg detection:
    - Single dash + one char (e.g. ``-t``) → short flag
    - Double dash + name (e.g. ``--target``) → long flag
    - Two positional args: detect by dash count, assign accordingly

    Args:
        *args: Flag name strings (short like ``-t``, long like ``--target``).
        help: Help text for the option.
        hidden: Whether to hide this option from --help output.
        envvar: Environment variable name to read as fallback.
    """
    short: str | None = None
    long: str | None = None

    for arg in args:
        if arg.startswith("--"):
            long = arg
        elif arg.startswith("-") and len(arg) == 2:
            short = arg
        else:
            # Not a recognized short flag format — treat as long
            long = arg

    object.__setattr__(self, "short", short)
    object.__setattr__(self, "long", long)
    object.__setattr__(self, "help", help)
    object.__setattr__(self, "hidden", hidden)
    object.__setattr__(self, "envvar", envvar)

short instance-attribute

long instance-attribute

help instance-attribute

hidden instance-attribute

envvar instance-attribute

Stdin(flag=None, help=None, encoding='utf-8') dataclass

Mark a parameter as stdin-aware.

When stdin is piped and no explicit CLI value provided, reads from stdin. Flag value always wins over stdin (explicit > implicit).

Usage::

from typing import Annotated
from functualize.job import Stdin

def transform(data: Annotated[str, Stdin(flag="--data", help="Input data")]):
    ...

# Pipe usage: echo "hello" | func transform

flag = None class-attribute instance-attribute

help = None class-attribute instance-attribute

encoding = 'utf-8' class-attribute instance-attribute

GroupOptions

Bases: BaseModel

Base class for per-group option declarations.

Subclass with a group= class keyword to bind the declaration to a dotted group path. Fields carry the same Option markers used on job parameters and are introspectable via model_fields.

__group_path__ = '' class-attribute

__init_subclass__(*, group=None, **kwargs)

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

def __init_subclass__(cls, *, group: str | None = None, **kwargs: Any) -> None:
    super().__init_subclass__(**kwargs)
    if group is not None:
        cls.__group_path__ = group

call(fn_or_name, **kwargs)

Build a parameterized dependency reference (proposal §A.4).

Example::

deps=Deps(call(build, target="wheel"))

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

def call(fn_or_name: str | Callable[..., Any], **kwargs: Any) -> Call:
    """Build a parameterized dependency reference (proposal §A.4).

    Example::

        deps=Deps(call(build, target="wheel"))
    """
    return Call(fn_or_name, kwargs)

job(_func=None, *, group=None, extra_description=None, category=None, examples=(), tags=(), visibility='external', config_section=None, deps=None, cache=None, guards=None, exec=None)

Declare a job's identity and operational contract (proposal §A.3–A.6).

Identity and description are flat kwargs (that is what @job is about); operational concerns are grouped, self-validating value objects (Deps/Fingerprint/Guards/Exec). Usable bare (@job) for plain opt-in or with any subset of kwargs::

@job
def build(sh: Shell): ...

@job(
    group="infra",
    deps=Deps("lint", "test"),
    cache=Fingerprint(sources=["src/**/*.py"], generates=["dist/*.whl"]),
    exec=Exec(retry=Retry(attempts=2)),
)
def deploy(sh: Shell, config: DeployConfig): ...

The frozen :class:JobDeclaration is stored on the function as __functualize_job__; discovery reads it (falling back to convention). group overrides the module-level JOB_GROUP; the addressable name is always derived from __name__ by normalization, so there is exactly one spelling of a job. Identity-preserving: decorated is original always holds, so @job composes with other job decorators in any order.

Raises:

Type	Description
ValueError	If any field or value object violates its invariants (validated eagerly at decoration time).

Source code in src/functualize/job/decorators.py

def job(
    _func: F | None = None,
    *,
    group: str | None = None,
    extra_description: str | None = None,
    category: str | None = None,
    examples: tuple[str, ...] | list[str] = (),
    tags: tuple[str, ...] | list[str] = (),
    visibility: Literal["external", "internal"] = "external",
    config_section: str | None = None,
    deps: Deps | None = None,
    cache: Fingerprint | None = None,
    guards: Guards | None = None,
    exec: Exec | None = None,
) -> F | Callable[[F], F]:
    """Declare a job's identity and operational contract (proposal §A.3–A.6).

    Identity and description are flat kwargs (that is what ``@job`` is about);
    operational concerns are grouped, self-validating value objects
    (``Deps``/``Fingerprint``/``Guards``/``Exec``). Usable bare (``@job``) for
    plain opt-in or with any subset of kwargs::

        @job
        def build(sh: Shell): ...

        @job(
            group="infra",
            deps=Deps("lint", "test"),
            cache=Fingerprint(sources=["src/**/*.py"], generates=["dist/*.whl"]),
            exec=Exec(retry=Retry(attempts=2)),
        )
        def deploy(sh: Shell, config: DeployConfig): ...

    The frozen :class:`JobDeclaration` is stored on the function as
    ``__functualize_job__``; discovery reads it (falling back to convention).
    ``group`` overrides the module-level ``JOB_GROUP``; the addressable name
    is always derived from ``__name__`` by normalization, so there is exactly
    one spelling of a job. Identity-preserving: ``decorated is original`` always holds,
    so ``@job`` composes with other job decorators in any order.

    Raises:
        ValueError: If any field or value object violates its invariants
            (validated eagerly at decoration time).
    """
    declaration = JobDeclaration(
        group=group,
        extra_description=extra_description,
        category=category,
        examples=tuple(examples),
        tags=tuple(tags),
        visibility=visibility,
        config_section=config_section,
        deps=deps,
        cache=cache,
        guards=guards,
        exec=exec,
    )

    def apply(func: F) -> F:
        func.__functualize_job__ = declaration  # type: ignore[attr-defined]
        return func

    # Bare @job (function passed directly) vs @job(...)/@job() (returns decorator).
    if _func is not None:
        return apply(_func)
    return apply

suppress_live(*names)

Opt a job out of one or more ambient live constructs.

Ambient constructs are the ones a plugin registered to render by default (a flow-viz execution tree, say). A job whose output reads better without one says so declaratively::

@suppress_live("flow-viz")
def simple_task(log: Log) -> None:
    log("No tree needed here")

The imperative equivalent is live.suppress("flow-viz") inside the body; project-wide, it is [live] suppress in config.

Identity-preserving, like every decorator here: decorated is original.

Parameters:

Name	Type	Description	Default
*names	str	Ambient construct names to suppress. Passing none is a no-op.	()

Returns:

Type	Description
Callable[[F], F]	A decorator attaching the declaration to the function.

Source code in src/functualize/job/decorators.py

def suppress_live(*names: str) -> Callable[[F], F]:
    """Opt a job out of one or more ambient live constructs.

    Ambient constructs are the ones a plugin registered to render by default
    (a flow-viz execution tree, say). A job whose output reads better without
    one says so declaratively::

        @suppress_live("flow-viz")
        def simple_task(log: Log) -> None:
            log("No tree needed here")

    The imperative equivalent is ``live.suppress("flow-viz")`` inside the body;
    project-wide, it is ``[live] suppress`` in config.

    Identity-preserving, like every decorator here: ``decorated is original``.

    Args:
        *names: Ambient construct names to suppress. Passing none is a no-op.

    Returns:
        A decorator attaching the declaration to the function.
    """

    def decorator(func: F) -> F:
        existing = getattr(func, "__functualize_suppress_live__", ())
        func.__functualize_suppress_live__ = (  # type: ignore[attr-defined]
            tuple(existing) + names
        )
        return func

    return decorator

surface_hint(surface)

Declare a job's preferred render surface.

The surface-resolution ladder consults this before the tui.default_surface setting and the framework default::

@surface_hint("stdout")
def report(log: Log) -> None:
    log("Renders on the released terminal, even from the TUI")

A hint is a preference, not a requirement — a HARD constraint like a bare tty: TTY capability still wins, and a "panel" hint is ignored on a direct CLI run (there is no TUI to render a panel).

Identity-preserving, like every decorator here: decorated is original.

Parameters:

Name	Type	Description	Default
surface	str	"stdout" or "panel".	required

Returns:

Type	Description
Callable[[F], F]	A decorator attaching the declaration to the function.

Raises:

Type	Description
ValueError	If surface is not a recognized surface name.

Source code in src/functualize/job/decorators.py

def surface_hint(surface: str) -> Callable[[F], F]:
    """Declare a job's preferred render surface.

    The surface-resolution ladder consults this before the
    ``tui.default_surface`` setting and the framework default::

        @surface_hint("stdout")
        def report(log: Log) -> None:
            log("Renders on the released terminal, even from the TUI")

    A hint is a preference, not a requirement — a HARD constraint like a
    bare ``tty: TTY`` capability still wins, and a "panel" hint is ignored
    on a direct CLI run (there is no TUI to render a panel).

    Identity-preserving, like every decorator here: ``decorated is original``.

    Args:
        surface: "stdout" or "panel".

    Returns:
        A decorator attaching the declaration to the function.

    Raises:
        ValueError: If ``surface`` is not a recognized surface name.
    """
    if surface not in _VALID_SURFACE_HINTS:
        raise ValueError(
            f"surface_hint must be one of {_VALID_SURFACE_HINTS}, got {surface!r}"
        )

    def decorator(func: F) -> F:
        func.__functualize_surface_hint__ = surface  # type: ignore[attr-defined]
        return func

    return decorator

__getattr__(name)

Lazily materialize GroupOptions on first access (PEP 562).

Defining a Pydantic BaseModel subclass runs pydantic's plugin loader, which imports every installed pydantic plugin — logfire pulls in rich, for instance. functualize/__init__.py reaches this package on every functualize import, so defining the class eagerly would drag those dependencies into every process and break the CLI-dependency isolation guarantee (tests/test_typer_isolation.py, now a CLI-dependency absence test).

Deferring costs nothing: only a project that actually declares group options ever touches the class, and from functualize.job import GroupOptions still works — Python falls back to this hook.

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

def __getattr__(name: str) -> Any:
    """Lazily materialize ``GroupOptions`` on first access (PEP 562).

    Defining a Pydantic ``BaseModel`` subclass runs pydantic's plugin loader,
    which imports every installed pydantic plugin — ``logfire`` pulls in
    ``rich``, for instance. ``functualize/__init__.py`` reaches this package on
    *every* functualize import, so defining the class eagerly would drag those
    dependencies into every process and break the CLI-dependency isolation
    guarantee (``tests/test_typer_isolation.py``, now a CLI-dependency absence test).

    Deferring costs nothing: only a project that actually declares group
    options ever touches the class, and ``from functualize.job import
    GroupOptions`` still works — Python falls back to this hook.
    """
    if name == "GroupOptions":
        from functualize._types.group_options import GroupOptions

        return GroupOptions
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

---

RunContext API Reference

functualize.job.RunContext is a thin facade (a 32-line re-export module) over the ~780-LOC capability class in functualize._engine.capabilities.runcontext, which delegates work to per-capability classes (Log, Invoke, Perf, WorkflowTracker, ...). It provides configuration access, logging, metadata tracking, phase tracking, job invocation, event emission, and prompting.

Module location: src/functualize/job/context.py

from functualize.job import (
    RunContext, Log, Invoke, Prompt, Perf, State, Sources, JobContext, JobConfigView,
)

Core Properties

Property	Type	Description
name	str	The job name for this execution.
metadata	dict[str, Any]	Execution metadata (run_type, run_status, start_time, end_time, duration).
result_metadata	dict[str, Any]	Mutable metadata dict carried to JobResult (64-key max).
config	JobConfigView	The resolved configuration view for this job.
job_config	Any	The Pydantic config model instance (settable).
phases	list[JobPhase]	All tracked job phases in this execution.
cwd	Path	The working directory for this execution.
job_directory	Path \| None	The directory containing the job module.
run_status	RunStatus	Current execution status.
run_duration	float	Elapsed time in seconds since execution started.

---

Logging

rc.log(message, level="info")

Emit a log message through the job's own Log capability when the job declares one, so both routes share a sink; otherwise it writes to the job's functualize.job.<name> logger, which is where Log would have written too.

level must be one of debug, info, warning, error, critical — anything else raises ValueError.

from functualize.job import Log, RunContext

def my_job(rc: RunContext) -> None:
    rc.log("Starting processing")                # info (default)
    rc.log("Connecting to DB", level="debug")
    rc.log("Retrying request", level="warning")


def either_way(rc: RunContext, log: Log) -> None:
    log("same sink")                             # the injected capability
    rc.log("same sink")                          # routed to that same instance

---

Status Tracking

rc.events.set_run_status(status, message="")

Transition the execution status. Terminal states cannot be transitioned from.

from functualize.types import RunStatus

rc.events.set_run_status(RunStatus.SUCCESS, "All records processed")

---

Job Phases

rc.events.track_phase(phase_name, message, status=None)

Create or update a named job phase. Delegates to the WorkflowTracker capability class.

from functualize.types import RunStatus

rc.events.track_phase("extract", "Fetching from API")
rc.events.track_phase("extract", "Got 1000 records", RunStatus.SUCCESS)

---

Job Invocation

rc.invoke(job_name, *, timeout=None, **kwargs)

Invoke a sibling job as a child execution. Delegates to the Invoke capability class.

result = rc.invoke("validate-data", source="api")
result = rc.invoke("slow-job", timeout=30.0, batch_size=100)

Returns: JobResult with status, duration, return_value, and exception.

rc.invoke_parallel(jobs)

Invoke multiple jobs concurrently. Delegates to Invoke.parallel().

jobs = [
    ("process-shard", {"shard_id": 0}),
    ("process-shard", {"shard_id": 1}),
    ("process-shard", {"shard_id": 2}),
]
results = rc.invoke_parallel(jobs)

Returns: list[JobResult] in the same positional order as the input.

---

Event Emission

rc.events.emit(event_name, resource="", **payload)

Emit a custom structured event. Delegates directly to EventBus.emit().

rc.events.emit("etl.extract.complete", resource="customers", record_count=1500)

---

Prompting

rc.prompts.ask(request)

Present a structured prompt to the user via the active PromptCollector.

from functualize.plugin import PromptRequest

response = rc.prompts.ask(PromptRequest(
    question="Select environment",
    choices=[...],
))

Convenience Methods

• rc.prompts.confirm(question, *, destructive=False, default=None) — Yes/no confirmation
• rc.prompts.choice(question, choices, *, default=None) — Single selection
• rc.prompts.text(question, *, default=None, secret=False) — Text input

---

Performance Instrumentation

rc.events.perf_mark(name) / rc.events.perf_mark_start(name) / rc.events.perf_mark_end(name)

Record performance marks. Delegates to the Perf capability class.

---

DI Access

rc[SomeType] / SomeType in rc

Access dependency-injected services via subscript notation.

db = rc[DatabaseConnection]

---

Shell Capability — Errors and Responders

The Shell capability (sh: Shell) and its ShellResult/Responder types are covered in the Shell Capability guide. Two of its vocabulary types are documented here.

ShellError

Raised when a command exits non-zero under check=True — also on timeout, and when a FailingResponder sentinel appears in the live output.

class ShellError(Exception):
    def __init__(self, result: ShellResult) -> None: ...

Attribute	Type	Description
result	ShellResult	The full result of the failed command, so callers can inspect stdout/stderr/returncode.

from functualize.job import Shell, ShellError

def deploy(sh: Shell) -> None:
    try:
        sh(["docker", "build", "-t", "app", "."])
    except ShellError as e:
        print(e.result.stderr)

FailingResponder

A Responder that also aborts on a failure sentinel. Responds like Responder — writing response to the child's stdin whenever pattern matches new output — but if sentinel appears the command is killed and a ShellError is raised. This is how a sudo password responder aborts on Sorry, try again instead of answering the re-prompt forever.

from functualize.job import FailingResponder

FailingResponder(
    pattern=r"\[sudo\] password.*:",
    response="s3cr3t\n",
    sentinel="Sorry, try again",
)

Attribute	Type	Description
pattern	str \| re.Pattern[str]	Regex whose match in live output triggers response (inherited from Responder).
response	str	Text written to the child's stdin on each match (inherited from Responder).
sentinel	str \| re.Pattern[str]	Regex whose appearance aborts the command with ShellError.

Subclasses Responder, so it is accepted anywhere a watchers= sequence is.

---

Freshness Capability

FreshnessVerdict

The verdict functualize.job.Freshness exposes to a job whose @job(cache=Fingerprint(..., decides=True)) opted in to seeing its own freshness decision, instead of the engine silently returning before the body ran.

@dataclass(frozen=True)
class FreshnessVerdict:
    state: GuardState
    key: str
    recorded_value: Any | None
    declared_sources: tuple[str, ...]
    declared_generates: tuple[str, ...]
    source_map: Mapping[str, Mapping[str, Any]]

Attribute	Type	Description
state	GuardState	The pipeline outcome — SKIP_FRESH when the job's declared inputs are current, RUN otherwise.
key	str	The fingerprint key the decision was computed under.
recorded_value	Any \| None	What the previous run recorded, when the pre-flight read a record and the verdict was a skip.
declared_sources	tuple[str, ...]	The input patterns as the job's own Fingerprint declared them.
declared_generates	tuple[str, ...]	The output patterns as the job's own Fingerprint declared them.
source_map	Mapping[str, Mapping[str, Any]]	{path: {mtime, size, sha256}} for every match of the declared inputs.
is_fresh (property)	bool	True only when state is GuardState.SKIP_FRESH. Deliberately does not fold in a satisfied status guard — "already done" is a different claim from "declared inputs are current".

from functualize.job import Freshness
from functualize.job.decorators import job, Fingerprint

@job(cache=Fingerprint(sources=["src/**/*.py"], decides=True))
def build(fresh: Freshness) -> str:
    verdict = fresh.verdict()
    if verdict is not None and verdict.is_fresh:
        return "artifact already current"
    return rebuild()

decides=True is what makes the branch above reachable — without it the engine skips the job for you and returns before the body runs, so is_fresh is the one value that branch could never see.