httk.workflow.sdk

The Python authoring SDK for native httk₂ workflow runners.

One runner is one program that implements the steps of one workflow. Steps are registered on a Runner, and Runner.main() dispatches the step the manager asked for to the handler that implements it, giving it one Attempt object:

from httk.workflow import ChildSpec, Runner

run = Runner("defects")


@run.step
def characterize(a):
    for site in range(a.parameter("sites")):
        a.spawn(ChildSpec(step="relax", parameters={"site": site}), label=f"site-{site}")
    a.gather("aggregate", on_impossible="triage")


if __name__ == "__main__":
    raise SystemExit(run.main())

Nothing declares the shape of the workflow up front: a step decides at run time which children to spawn and which step runs next, so the graph of a job is whatever its steps published. Exactly one outcome is published per attempt, and the handler that returns without publishing one, or raises, is reported as such instead of leaving the attempt ambiguous.

Attributes

RUNNER_DESCRIPTION_FORMAT
RUNNER_ERROR_FORMAT
InstantiateHandler

Classes

RunnerRef	Which runner executes a child job synthesized by ChildSpec.
ChildSpec	A complete child job described by the step and parameters it starts with.
ChildResult	What one gathering step may know about one child it spawned.
ChildrenView	The children observed by the join that started this activation.
Attempt	Everything one attempt of one step may read, do, and publish.
Runner	The registered steps of one workflow and the dispatch into them.

Module Contents

httk.workflow.sdk.RUNNER_DESCRIPTION_FORMAT = 'httk-workflow-runner-description'

httk.workflow.sdk.RUNNER_ERROR_FORMAT = 'httk-workflow-runner-error'

type httk.workflow.sdk.InstantiateHandler = Callable[[Any], object]

class httk.workflow.sdk.RunnerRef

Which runner executes a child job synthesized by ChildSpec.

A synthesized child has no payload of its own, so its runner must be one that lives outside a payload: an entry of the workspace runner store, or an installed runner on the machine that runs it. inherit() copies the reference of the spawning job itself, which is what a campaign whose steps all live in one published runner wants.

Parameters:

• source – The location from which the child runner is loaded.

• path – The workspace or installed runner path when one is selected.

• sha256 – The digest pin for a workspace or installed runner.

source: Literal['inherit', 'workspace', 'installed'] = 'inherit'

path: str | None = None

sha256: str | None = None

classmethod inherit()

Reference exactly the runner of the spawning job.

Returns:

The inherited runner reference.

Return type:

RunnerRef

classmethod workspace(path, sha256)

Reference one runner published in the workspace runner store.

Parameters:

• path (str | pathlib.PurePosixPath) – The path within the workspace runner store.

• sha256 (str) – The runner digest.

Returns:

The workspace runner reference.

Return type:

RunnerRef

classmethod installed(path, sha256)

Reference one runner installed on the machine that runs the child.

Parameters:

• path (str | pathlib.PurePosixPath) – The installed runner path.

• sha256 (str) – The runner digest.

Returns:

The installed runner reference.

Return type:

RunnerRef

class httk.workflow.sdk.ChildSpec

A complete child job described by the step and parameters it starts with.

Everything not given follows the spawning job: its workflow, its claim pool, its priority, its resources, and its runner. The child therefore differs from its parent in exactly what the campaign varies, which is normally only step and parameters.

Parameters:

• step – The first step the child runs.

• parameters – The opaque implementation knobs given to the child.

• declarations – The workflow declarations carried by the child.

• runner – The runner reference used to execute the child.

• name – The child job name, or a generated name when omitted.

• workflow – The child’s workflow identifier, or the parent’s when omitted.

• tag – The child’s optional job tag, or the spawn label when omitted.

• workdir_mode – Whether the child’s workdir persists or is isolated.

• workdir_path – The child’s relative workdir path.

• data_mode – Whether the child has transactional data.

• priority – The child’s priority, or the parent’s when omitted.

• claim_pool – The child’s claim pool, or the parent’s when omitted.

• required_capabilities – Capabilities required by the child.

• resources – Resources requested by the child, or the parent’s when omitted.

• maximum_attempts_per_activation – The child’s per-activation attempt budget.

• maximum_total_attempts – The child’s total attempt budget.

• maximum_activations – The child’s activation budget.

• retry_on – Failure codes eligible for manager-detected retry.

step: str

parameters: collections.abc.Mapping[str, object]

declarations: collections.abc.Mapping[str, collections.abc.Mapping[str, object]]

runner: RunnerRef

name: str | None = None

workflow: str | None = None

tag: str | None = None

workdir_mode: Literal['persistent', 'isolated'] = 'persistent'

workdir_path: str = 'run'

data_mode: Literal['none', 'transactional'] = 'none'

priority: int | None = None

claim_pool: str | None = None

required_capabilities: tuple[str, Ellipsis] = ()

resources: collections.abc.Mapping[str, object] | None = None

maximum_attempts_per_activation: int | None = None

maximum_total_attempts: int | None = None

maximum_activations: int | None = None

retry_on: tuple[str, Ellipsis] = ()

class httk.workflow.sdk.ChildResult

What one gathering step may know about one child it spawned.

Every member is derived from authoritative state by the manager before the gathering activation starts, so reading a child is a pure read of the attempt context and never a scan of the workspace. Paths are absolute.

label: str | None

job_id: str

job_key: str

kind: str

failure: httk.workflow.models.Failure | None

placement: pathlib.PurePosixPath

payload: pathlib.Path

workdir: pathlib.Path | None

data: pathlib.Path | None

data_generation: int | None

raw: collections.abc.Mapping[str, object]

property succeeded: bool

Report whether this child ended successfully.

property failed: bool

Report whether this child ended badly.

class httk.workflow.sdk.ChildrenView

The children observed by the join that started this activation.

The view is empty for an activation that follows no join, so a step can read it unconditionally.

all: tuple[ChildResult, Ellipsis] = ()

property succeeded: tuple[ChildResult, Ellipsis]

The children that ended successfully, in spawn order.

property failed: tuple[ChildResult, Ellipsis]

The children that ended badly, in spawn order.

property labels: tuple[str, Ellipsis]

The labels of every observed child, in spawn order.

get(label, default=None)

Return the child spawned under label, or default.

Parameters:

• label (str) – The unique spawn label to find.

• default (ChildResult | None) – The value to return when no child has that label.

Returns:

The matching child, or the default value.

Return type:

ChildResult | None

class httk.workflow.sdk.Attempt(context, *, control, payload, workdir, workspace, data=None, step=None, runner=None)

Everything one attempt of one step may read, do, and publish.

An attempt owns exactly one implicit outcome draft. The draft is created by the first spawn(), put(), or remove(), and it is published by exactly one of advance(), gather(), succeed(), retry(), pause(), or fail(). Publication is the single atomic rename the manager observes, so nothing a step did takes effect until the step says how it ended.

Parameters:

• context (httk.workflow.runtime.AttemptContext) – The manager-written identity and restart context.

• control (pathlib.Path) – The attempt control directory.

• payload (pathlib.Path) – The immutable job payload directory.

• workdir (pathlib.Path) – The directory in which the step works.

• workspace (pathlib.Path) – The workspace root containing the job.

• data (pathlib.Path | None) – The job’s transactional data directory, when enabled.

• step (str | None) – The step this attempt runs, or the context step when omitted.

• runner (Runner | None) – The runner dispatching this attempt, when available.

context

control

payload

workdir

workspace

data = None

step

state

log

classmethod initialize(environment=None, *, runner=None)

Bind this process to its attempt and recover an interrupted one.

Recovery replays every workdir batch an earlier attempt sealed but did not get to apply, so a handler always starts from a workdir whose sealed changes are complete. This is the only constructor a runner needs.

Parameters:

• environment (collections.abc.Mapping[str, str] | None) – The process environment, or the current environment when omitted.

• runner (Runner | None) – The runner dispatching this attempt, when called by Runner.main().

Returns:

The initialized attempt.

Raises:

ValueError – If the manager-written attempt environment or context is invalid.

Return type:

Self

property job: httk.workflow.models.JobDefinition

The immutable definition of the job this attempt belongs to.

property parameters: collections.abc.Mapping[str, object]

The application-defined parameters object of this job.

property children: ChildrenView

The children observed by the join that started this activation.

property published: bool

Report whether this attempt already published its outcome.

parameter(name, default=_MISSING)

Return one member of the job’s parameters object.

Without a default, a missing parameter is a KeyError: a step that needs a parameter cannot run without it, and saying so immediately is better than failing later on a value that was never there.

Parameters:

• name (str) – The parameter name to look up.

• default (object) – The value to return when the parameter is absent.

Returns:

The parameter value or the supplied default.

Raises:

KeyError – If the parameter is absent and no default was supplied.

Return type:

object

setting(name, default=None)

Resolve one application setting through its layers.

The layers are consulted most-specific first, and the first that has the name wins: this job’s parameters object, then the environment variable HTTK_ + the name upper-cased with dots as underscores (so vasp.command reads HTTK_VASP_COMMAND), then the workspace’s application settings, then default. This is how a step reads the VASP command a workspace was configured with without the operator exporting it for every job, while still letting one job or one shell override it.

Parameters:

• name (str) – The dotted application setting name to resolve.

• default (object) – The value to return when no layer defines the setting.

Returns:

The first value found in the resolution layers, or the default.

Return type:

object

environment(name, default=_MISSING)

Resolve one declared workflow environment value through its layers.

Once the runner’s start gate has run, this attempt reads the immutable snapshot resolved there. Before that gate, overrides, the declared setting’s environment variable, workspace settings, the declaration default, and default are consulted in that order.

Parameters:

• name (str) – The declared environment name to look up.

• default (object) – The value to return when no declared value exists.

Returns:

The resolved environment value.

Raises:

KeyError – If the name is undeclared or unresolved without a default.

Return type:

object

declare(name, document)

Record the observed workflow declaration name of this job.

The static declarations of a job are the ones job.json carried at submission, and they cannot change. A dynamic campaign nevertheless only learns at run time what it actually consumed and produced, so a step writes the refined document here and it is stored beside the job state as .httk-job/declarations/<name>.json, atomically. The bytes are carried verbatim: nothing here interprets the document, whose own members say which vocabulary and version it follows.

The write is runner-private, so it never disturbs the payload digest, and repeating it overwrites: what a job observed is whatever its last word on the subject was. A collect reports the observed document beside the declared one and never merges the two.

Parameters:

• name (str) – The declaration name to record.

• document (collections.abc.Mapping[str, object]) – The declaration document to store verbatim.

Returns:

The path of the stored observed declaration.

Raises:

httk.workflow.errors.FormatError – If the declaration name or document is invalid.

Return type:

pathlib.Path

declaration(name)

Return the workflow declaration name, observed first.

The document this job observed is returned when one was written, otherwise the one job.json declared, otherwise None.

Parameters:

name (str) – The declaration name to read.

Returns:

The observed or declared document, or None when absent.

Raises:

httk.workflow.errors.FormatError – If the declaration name is invalid or its document is malformed.

Return type:

collections.abc.Mapping[str, object] | None

run(argv, *, timeout=None, cwd=None, environment=None, termination_grace=10.0)

Run an argv array in the workdir and reap its process group.

Parameters:

• argv (collections.abc.Sequence[str]) – The command and its arguments.

• timeout (float | None) – The maximum runtime before terminating the process group.

• cwd (str | os.PathLike[str] | None) – The working directory, or this attempt’s workdir when omitted.

• environment (collections.abc.Mapping[str, str] | None) – The complete child environment, or the process environment when omitted.

• termination_grace (float) – The grace period after a timeout before forceful termination.

Returns:

The completed command result.

Return type:

httk.workflow.runtime.CommandResult

workdir_batch()

Start a replayable group of workdir changes.

Returns:

A batch that seals changes for replay after interruption.

Return type:

httk.workflow.runtime_builders.ReplayableWorkdirBatch

put(source, destination)

Stage one file or directory for the job’s transactional data.

The operation is applied by the manager when the outcome is committed, exactly once, whatever happens to this process in between. Operation identifiers are generated in call order, so replaying the same step produces the same manifest.

Parameters:

• source (str | os.PathLike[str]) – The file or directory to stage.

• destination (str | os.PathLike[str]) – The destination path in transactional data.

Returns:

The generated transaction operation identifier.

Raises:

ValueError – If this job has no transactional data.

Return type:

str

remove(destination, *, missing_ok=False)

Remove one path from the job’s transactional data.

Parameters:

• destination (str | os.PathLike[str]) – The path to remove from transactional data.

• missing_ok (bool) – Whether an absent destination is acceptable.

Returns:

The generated transaction operation identifier.

Raises:

ValueError – If this job has no transactional data.

Return type:

str

spawn(child, *, label, placement=None)

Register one child job under label, to be created on publication.

child is either a ChildSpec, which needs no payload at all, or the path of a prepared payload directory. The label is mandatory and must be unique within one attempt: it is how gather() and children name this child later.

Parameters:

• child (ChildSpec | str | os.PathLike[str]) – The child specification or prepared payload directory.

• label (str) – The unique label used to observe the child later.

• placement (str | pathlib.PurePosixPath | None) – The workspace placement for the child, or this attempt’s placement when omitted.

Returns:

The reference to the registered child.

Raises:

ValueError – If the child or label is invalid, or the label is reused.

Return type:

httk.workflow.runtime_builders.ChildReference

advance(step, *, state=None, priority=None)

Publish a new activation of this job at step.

state is written to state before the outcome is published, so the step that runs next always finds the state that decided to run it.

Parameters:

• step (str) – The next registered step.

• state (collections.abc.Mapping[str, object] | None) – State members to merge before publication.

• priority (int | None) – The priority of the new activation, when changed.

Returns:

The path of the published outcome.

Raises:

RuntimeError – If this attempt already published an outcome.

Return type:

pathlib.Path

gather(step, *, when='all_succeeded', count=None, on_impossible=None, rejoin=(), priority=None)

Wait for this attempt’s and optionally earlier children, then run step.

The join names children spawn() registered on this attempt and labels from earlier join activations named by rejoin. when is one of all_succeeded, all_terminal, any_succeeded, any_terminal, or at_least with count. When the condition can no longer be met, the job advances to on_impossible if one is named and fails with dependency_failure otherwise.

Parameters:

• step (str) – The step to run when the join condition is met.

• when (httk.workflow.runtime_builders.JoinCondition) – The child completion condition.

• count (int | None) – The required count when when is at_least.

• on_impossible (str | None) – The step to run when the condition cannot be met.

• rejoin (collections.abc.Iterable[str]) – Labels of children observed by an earlier join activation.

• priority (int | None) – The priority of the join activation, when changed.

Returns:

The path of the published wait outcome.

Raises:

ValueError – If no children were spawned or rejoined, a rejoined label is unknown, or a named step is invalid.

Return type:

pathlib.Path

succeed()

Publish the successful completion of this job.

Returns:

The path of the published outcome.

Return type:

pathlib.Path

retry(reason)

Ask for another attempt of this same activation.

Parameters:

reason (str) – The reason recorded with the retry request.

Returns:

The path of the published outcome.

Return type:

pathlib.Path

pause(reason)

Pause this job until an operator resumes it.

Parameters:

reason (str) – The reason recorded with the pause request.

Returns:

The path of the published outcome.

Return type:

pathlib.Path

fail(code, message, *, details=None, retryable=False, priority=None)

Publish a structured terminal failure.

code is the token a job lists in retry_on. retryable declares that repeating this attempt could help, which the manager honours within the attempt budgets of the job.

Parameters:

• code (str) – The stable failure code.

• message (str) – The human-readable failure message.

• details (collections.abc.Mapping[str, object] | None) – Optional structured failure details.

• retryable (bool) – Whether repeating this attempt could help.

• priority (int | None) – The terminal priority, when changed.

Returns:

The path of the published outcome.

Return type:

pathlib.Path

class httk.workflow.sdk.Runner(workflow, *, inputs=None)

The registered steps of one workflow and the dispatch into them.

A runner is created once at module level, its steps are registered with step() before any work happens, and main() is what the manager invokes. Registration is therefore complete before the first step runs, which is what lets every step name in a published outcome be checked against the steps that really exist.

Parameters:

• workflow (str) – The workflow identifier and registry key implemented by this runner.

• inputs (collections.abc.Mapping[str, str | None] | None) – The immutable creation-time staged-input declarations.

workflow

property inputs: collections.abc.Mapping[str, str | None]

The immutable declared-input staging map.

property steps: frozenset[str]

The names of every registered step.

property has_instantiate: bool

Whether this runner has a creation-time instantiate hook.

step(function: StepHandler) → StepHandler

step(*, name: str | None = None) → collections.abc.Callable[[StepHandler], StepHandler]

Register one step handler, named after the function unless name is given.

Parameters:

• function – The step handler, or None when used as a decorator factory.

• name – The registered step name, or the handler name when omitted.

Returns:

The handler or a decorator that registers it.

Raises:

ValueError – If the step name is invalid or already registered.

instantiate(function)

Register the hook receiving httk.workflow.scaffold.InstantiateContext.

Parameters:

function (InstantiateHandler) – The creation-time instantiate handler.

Returns:

The handler, unchanged.

Raises:

ValueError – If an instantiate handler is already registered.

Return type:

InstantiateHandler

description()

Return the machine-readable description of this runner.

Returns:

The runner description document.

Return type:

dict[str, object]

main(argv=None)

Run the step this process was launched for and publish its outcome.

Asked to describe itself — through HTTK_WORKFLOW_DESCRIBE=1 or --describe — the runner prints its description and exits without touching anything, so a tool can enumerate the steps of a runner it is not running.

Every ending is an outcome: a step that publishes none is reported as no_outcome, an unimplemented step as unknown_step, and a step that raises leaves an error.json breadcrumb and lets the exception reach the manager, whose retry policy owns what happens next.

Parameters:

argv (collections.abc.Sequence[str] | None) – Command-line arguments, or the process arguments when omitted.

Returns:

Zero after dispatching or describing the runner.

Return type:

int