httk.workflow

Expose filesystem-native workflow execution for httk₂.

The package presents three layers, each with its own import home:

  • Filesystem protocol — the language-neutral on-disk contract lives in httk.workflow.protocol. Independent tools read and verify a workspace through it and the specification alone.

  • Execution / authoring — the surface a runner author uses. Runner, Attempt, and the small set of job and result types below are exported here; the lower-level runtime helpers live in httk.workflow.runtime and httk.workflow.runtime_utils, and job scaffolding in httk.workflow.scaffold.

  • Orchestration and management — Workspace, TaskManager, and job_records() drive and inspect a running workspace. The management operations that surround them (transfers, manifests, hygiene, configuration, adapters, supervision, and the VASP and v1 compatibility surfaces) live in their own named submodules rather than in this root.

The normal lifecycle is instantiate a job, run it, then collect its outputs. Only the deliberate top-level surface is re-exported here; everything else is reached through its submodule.

Submodules

Attributes

InstantiateHandler

Exceptions

FormatError

A workspace, job, journal frame, outcome, or request is malformed.

RunnerResolutionError

A shared runner cannot be resolved, staged, or verified.

TransactionError

A transactional-data manifest cannot be safely replayed.

TransitionLostError

Another actor committed a transition from the expected marker.

UnsupportedExtensionError

A workspace requires an extension this implementation does not support.

WorkflowError

Base class for workflow protocol failures.

WorkspaceCorruptionError

The authoritative filesystem state is internally inconsistent.

WorkspaceUnavailableError

The workspace cannot currently provide a coherent protocol view.

NotIdleError

A manager did not become idle within its timeout.

Classes

CollectedJob

Represent one job after workflow collecting and provenance assembly.

JobRecord

Everything a data layer needs about one job that stopped.

TaskManager

Execute and recover jobs in one workflow workspace.

WorkCensus

What one manager's scan found, tagged by why each job is or is not its work.

JobState

Atomic JSON application state that belongs to one job.

ScaffoldedJob

Describe one job this module submitted.

Attempt

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

ChildrenView

The children observed by the join that started this activation.

ChildResult

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

ChildSpec

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

Runner

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

RunnerRef

Which runner executes a child job synthesized by ChildSpec.

Workspace

Attach to one self-contained httk workflow filesystem workspace.

Functions

collect(workspace, *[, states, placement, ...])

Collect records through registered or explicitly allowed job collectors.

job_records(workspace, *[, states, placement, on_skipped])

Yield one JobRecord per finished job of workspace.

new_job(workspace, workflow, *[, inputs, files, ...])

Scaffold, submit, and describe one job of workflow.

new_jobs(workspace, workflow, items, *[, inputs, ...])

Scaffold and submit one job per member of items, lazily.

Package Contents

class httk.workflow.CollectedJob[source]

Represent one job after workflow collecting and provenance assembly.

Parameters:

  • workflow_id – Identify the workflow that produced the job.

  • outputs – Map declared output roles to collector results.

  • unfulfilled – Name declared output roles that carry no output. After a collector ran this is the roles it omitted; a degraded job (see missing_collector) lists every declared output role, because none of them was produced.

  • run – Carry the framework-assembled run provenance.

  • products – Carry the framework-assembled product links.

  • record – Preserve the mechanical job readout behind the collection.

  • missing_collector – Explain why collecting was unavailable, or leave it unset when collection completed.

  • products_unlinked – Name the declared product_of links skipped because the observed provenance held no matching input or output edge.

  • collector_exit_status – Report a nonzero executable-collector exit status observed after complete responses, or leave it unset.

  • identity_stable – Report whether a v1-harvested job’s identity is manifest-backed, or leave it unset for live collection.

workflow_id: str
outputs: collections.abc.Mapping[str, object]
unfulfilled: tuple[str, Ellipsis]
run: httk.core.Run
products: tuple[httk.core.ProductLink, Ellipsis]
record: JobRecord
missing_collector: str | None = None
products_unlinked: tuple[str, Ellipsis] = ()
collector_exit_status: int | None = None
identity_stable: bool | None = None
class httk.workflow.JobRecord[source]

Everything a data layer needs about one job that stopped.

Paths appear twice on purpose. The members payload_path, workdir_path, and data_path are workspace relative, which is what a stored record must hold so it survives moving the workspace; the properties payload, workdir, and data resolve them against the workspace this record was collected from, which is what code reading result files wants.

Parameters:

  • workspace_root – Identify the absolute workspace root.

  • workspace_id – Identify the workspace.

  • job_id – Identify the job.

  • job_key – Preserve the complete job key.

  • job – Preserve the validated immutable job definition.

  • runner_provenance – Preserve installed package provenance, when known.

  • state – Record the terminal state in which the job stopped.

  • failure – Record the terminal failure, when one exists.

  • placement – Locate the job within the workspace hierarchy.

  • payload_path – Locate the workspace-relative job payload.

  • workdir_path – Locate the last workspace-relative workdir, when known.

  • data_path – Locate transactional data, when the job has it.

  • data_generation – Record the committed data generation, when present.

  • provenance – Preserve the journal-derived timeline and damage flag.

  • runner_steps – Preserve the runner steps, when recorded.

  • children – Preserve labeled child references.

  • declarations – Preserve declared and observed workflow documents.

  • runner_description – Preserve the reserved runner description, when available.

workspace_root: pathlib.Path
workspace_id: str
job_id: str
job_key: str
job: collections.abc.Mapping[str, object]
runner_provenance: collections.abc.Mapping[str, object] | None
state: str
failure: httk.workflow.models.Failure | None
placement: pathlib.PurePosixPath
payload_path: pathlib.PurePosixPath
workdir_path: pathlib.PurePosixPath | None
data_path: pathlib.PurePosixPath | None
data_generation: int | None
provenance: collections.abc.Mapping[str, object]
runner_steps: tuple[str, Ellipsis] | None
children: collections.abc.Mapping[str, collections.abc.Mapping[str, object]]
declarations: collections.abc.Mapping[str, collections.abc.Mapping[str, collections.abc.Mapping[str, object] | None]]
runner_description: collections.abc.Mapping[str, object] | None = None
property payload: pathlib.Path

The absolute payload directory of this job.

property workdir: pathlib.Path | None

The absolute workdir of this job’s last attempt, when one is known.

property data: pathlib.Path | None

The absolute transactional data directory, for a job that has one.

property gaps: bool

Whether part of this job’s recorded history could not be read.

as_mapping()[source]

Return the JSON representation of this record.

classmethod from_mapping(value)[source]

Rebuild one record from a serialized record mapping.

Parameters:

value (collections.abc.Mapping[str, object]) – Supply the mapping produced by as_mapping().

Returns:

The reconstructed job record.

Raises:

httk.workflow.errors.FormatError – If the mapping has the wrong format or invalid record members.

Return type:

JobRecord

httk.workflow.collect(workspace, *, states=DEFAULT_COLLECT_STATES, placement=None, allow_job_collector=False, on_skipped=None)[source]

Collect records through registered or explicitly allowed job collectors.

A fallback reads and verifies the package manifest from the pinned runner tree itself. A changed pinned tree raises _PinnedTreeError, which degrades that job and does not stop the rest of the sweep; other hook-loading errors propagate and stop iteration. An unusable observed provenance document degrades only its own job, exactly like every other per-job failure.

Parameters:

  • workspace (httk.workflow.workspace.Workspace) – Read jobs from this workspace.

  • states (collections.abc.Iterable[str]) – Select the stopped state kinds to report.

  • placement (str | pathlib.PurePosixPath | None) – Restrict results to this placement and its descendants.

  • allow_job_collector (bool) – Permit digest-verified collectors from job-pinned workspace package trees.

  • on_skipped (collections.abc.Callable[[str], None] | None) – Receive the job key of every selected job dropped for an unreadable job.json, forwarded to job_records().

Yields:

Framework-assembled collected jobs, including degraded jobs.

Raises:

ValueError – If a registered collector fails to resolve or returns invalid output roles.

httk.workflow.job_records(workspace, *, states=DEFAULT_COLLECT_STATES, placement=None, on_skipped=None)[source]

Yield one JobRecord per finished job of workspace.

states selects which stopped jobs are reported and defaults to the successful ones; every requested kind is validated against COLLECTABLE_KINDS before anything is read. placement restricts the job_records to the jobs at or below one placement, exactly as httk workflow job list --placement does.

The result is a lazy iterator over one scan of the requested state directories. Nothing is materialized, and building a record reads only that job’s own job.json and journal chain, so collecting is a single pass over a workspace of any size. Attach read-only — Workspace(root, mutable=False) — when nothing else in the process needs to write.

Parameters:

  • workspace (httk.workflow.workspace.Workspace) – Read jobs from this workspace.

  • states (collections.abc.Iterable[str]) – Select the stopped state kinds to report.

  • placement (str | pathlib.PurePosixPath | None) – Restrict results to this placement and its descendants.

  • on_skipped (collections.abc.Callable[[str], None] | None) – Receive the job key of every selected job dropped for an unreadable job.json, so a caller can count skips it never sees.

Yields:

Mechanical job records, one for each readable selected job.

Raises:

ValueError – If states contains no collectable state.

exception httk.workflow.FormatError[source]

Bases: WorkflowError, ValueError

A workspace, job, journal frame, outcome, or request is malformed.

exception httk.workflow.RunnerResolutionError(code, message)[source]

Bases: WorkflowError

A shared runner cannot be resolved, staged, or verified.

The failure carries the exact protocol failure code the manager records, so an unresolvable runner (runner_unavailable), a runner whose staged bytes disagree with the digest the job pinned (runner_mismatch), a missing registration (runner_not_built), and a failed foreground build (runner_build_failed) stay distinguishable to an operator.

Parameters:

  • code (str) – Protocol failure code recorded by the manager.

  • message (str) – Human-readable failure description.

code
exception httk.workflow.TransactionError[source]

Bases: WorkflowError

A transactional-data manifest cannot be safely replayed.

exception httk.workflow.TransitionLostError[source]

Bases: WorkflowError

Another actor committed a transition from the expected marker.

exception httk.workflow.UnsupportedExtensionError[source]

Bases: WorkflowError

A workspace requires an extension this implementation does not support.

exception httk.workflow.WorkflowError[source]

Bases: Exception

Base class for workflow protocol failures.

exception httk.workflow.WorkspaceCorruptionError[source]

Bases: WorkflowError

The authoritative filesystem state is internally inconsistent.

exception httk.workflow.WorkspaceUnavailableError[source]

Bases: WorkflowError

The workspace cannot currently provide a coherent protocol view.

exception httk.workflow.NotIdleError(census)[source]

Bases: TimeoutError

A manager did not become idle within its timeout.

It carries the WorkCensus of the final scan so a caller can turn the failure into advice that names the actual pool, capability, or executor mismatches rather than a generic hint. It subclasses TimeoutError, so existing except TimeoutError callers keep working.

Parameters:

census (WorkCensus) – The work census of the manager’s last scan.

census
class httk.workflow.TaskManager(workspace, *, pools=('default',), capabilities=(), maximum_workers=1, lease_seconds=None, heartbeat_interval=30.0, unsafe_persistent_takeover=False, unsafe_isolated_takeover=False, takeover_grace_factor=DEFAULT_TAKEOVER_GRACE_FACTOR, executors=(), allowed_executors=None, accept_any_pool=False, join_grace_seconds=3600.0, cancel_grace_seconds=DEFAULT_CANCEL_GRACE_SECONDS, maximum_pass_markers=DEFAULT_MAXIMUM_PASS_MARKERS, discovery_budget=DEFAULT_DISCOVERY_BUDGET, placement_prefixes=(), runner_search_paths=(), runner_modules=DEFAULT_RUNNER_MODULES, gc_interval=None)[source]

Execute and recover jobs in one workflow workspace.

Parameters:

  • workspace (httk.workflow.workspace.Workspace) – Attach the manager to this workspace.

  • pools (collections.abc.Sequence[str]) – Accept jobs assigned to these pools.

  • capabilities (collections.abc.Sequence[str]) – Advertise these execution capabilities.

  • maximum_workers (int) – Limit the number of local attempts.

  • lease_seconds (float | None) – Override the workspace claim lease.

  • heartbeat_interval (float) – Set the requested manager heartbeat interval.

  • unsafe_persistent_takeover (bool) – Permit takeover based on persistent evidence.

  • unsafe_isolated_takeover (bool) – Permit takeover based on isolated evidence.

  • takeover_grace_factor (float) – Multiply the lease to determine takeover grace.

  • executors (collections.abc.Sequence[httk.workflow.executors.RunnerExecutor]) – Add runner executors to the built-in executor.

  • allowed_executors (collections.abc.Sequence[str] | None) – Restrict jobs to these installed executors.

  • accept_any_pool (bool) – Accept jobs without requiring a configured pool match.

  • join_grace_seconds (float) – Wait this long for unresolved join children.

  • cancel_grace_seconds (float) – Wait this long after cancellation before killing.

  • maximum_pass_markers (int) – Bound markers processed in one scheduling pass.

  • discovery_budget (int) – Bound entries visited in one scheduling pass.

  • placement_prefixes (collections.abc.Sequence[str]) – Restrict scheduling to these placement subtrees.

  • runner_search_paths (collections.abc.Iterable[str | os.PathLike[str]]) – Search these locations for installed runners.

  • runner_modules (collections.abc.Iterable[str]) – Search these module prefixes for packaged runners.

  • gc_interval (float | None) – Run background collection at this interval when supplied.

Raises:
workspace
uid
runner_search_paths: tuple[pathlib.Path, Ellipsis]
runner_modules: tuple[str, Ellipsis] = ('httk.workflow',)
pools
capabilities
maximum_workers = 1
lease_seconds
heartbeat_interval = 30.0
unsafe_persistent_takeover = False
unsafe_isolated_takeover = False
takeover_grace_factor = 2.0
join_grace_seconds = 3600.0
cancel_grace_seconds = 10.0
maximum_pass_markers = 256
discovery_budget = 4096
placement_prefixes: tuple[pathlib.PurePosixPath, Ellipsis]
gc_interval = None
executors
allowed_executors
accept_any_pool = False
manager_id = ''
hostname
writer
close()[source]

Close local attempt logs and the manager’s journal writer.

property manager_directory: pathlib.Path

Return this manager’s own directory below managers/.

Returns:

The manager directory path.

Return type:

pathlib.Path

property heartbeat_period: float

Return how long this manager may actually go without heartbeating.

A configured interval longer than the lease it claims work under would let a manager expire its own claims, so the interval is capped at a fraction of the lease however it was configured.

Returns:

The effective heartbeat interval.

Return type:

float

heartbeat(*, force=False)[source]

Publish a manager heartbeat when the effective interval has elapsed.

Parameters:

force (bool) – Publish immediately instead of honoring the interval.

tick()[source]

Perform one nonblocking scheduling and recovery pass.

Returns:

Whether the pass changed or launched workflow state.

Return type:

bool

serve(*, poll_interval=1.0, drain_timeout=30.0, drain_grace_seconds=10.0)[source]

Run until interrupted, draining running attempts on a stop signal.

A first SIGTERM or SIGINT — what a batch system sends at walltime — stops claiming new work, terminates the local attempts, and keeps ticking so their outcomes are committed. A second signal exits at once. The drain is process-local: everything an interrupted attempt needs is already recorded by the transitions it produces, and any attempt left behind is recovered from its expired lease.

Parameters:

  • poll_interval (float) – Wait this long between scheduling passes.

  • drain_timeout (float) – Stop draining after this much time.

  • drain_grace_seconds (float) – Kill attempts after this much drain grace.

run_until_idle(*, timeout=60.0, poll_interval=0.02)[source]

Run until no local process or claimable marker remains, and report it.

A job this manager cannot progress — one whose pool, capability, or executor it does not serve, or one waiting on children or paused for an operator — does not keep it awake: it is counted in the returned census instead. The census is what the caller prints as the idle summary.

Parameters:

  • timeout (float) – Stop waiting after this many seconds.

  • poll_interval (float) – Wait this long between scheduling passes.

Returns:

The work census of the settled workspace.

Raises:

httk.workflow.manager.NotIdleError – If the manager does not become idle before the timeout.

Return type:

WorkCensus

class httk.workflow.WorkCensus[source]

What one manager’s scan found, tagged by why each job is or is not its work.

ready_blocked groups the ready and unregisterable-submitted jobs this manager cannot progress by the requirement it lacks — executor, pool, or capability — mapping each requirement to the count of jobs it would turn away. Every such job is attributed to exactly one requirement, so the grouped counts sum to ready_blocked_total.

Parameters:

  • succeeded – Terminal jobs that succeeded.

  • failed – Terminal jobs that failed.

  • ready_claimable – Ready jobs this manager could claim right now.

  • ready_blocked – Requirement kind to requirement to blocked job count.

  • waiting – Jobs waiting on their join children.

  • paused – Jobs paused for an operator.

  • actionable_count – Jobs this manager can still make progress on.

  • unreadable – Committing or cancelling jobs whose definition cannot be read.

succeeded: int
failed: int
ready_claimable: int
ready_blocked: collections.abc.Mapping[str, collections.abc.Mapping[str, int]]
waiting: int
paused: int
actionable_count: int
unreadable: int = 0
property actionable: bool

Whether this manager still has work it can make progress on.

Returns:

Whether any counted job is this manager’s to progress.

Return type:

bool

property ready_blocked_total: int

The number of jobs no requirement of this manager can claim here.

Returns:

The total blocked job count.

Return type:

int

summary_line()[source]

Render the one-line idle summary an operator reads on exit.

Returns:

The idle summary line.

Return type:

str

mismatch_advice()[source]

Name the requirements blocked jobs need that this manager lacks.

Returns:

The mismatch advice, or None when nothing is blocked.

Return type:

str | None

timeout_message(seconds)[source]

Render the not-idle advice, naming mismatches when there are any.

Parameters:

seconds (float) – The idle timeout that elapsed.

Returns:

The not-idle advice line.

Return type:

str

class httk.workflow.JobState(payload, *, durable=False)[source]

Bases: collections.abc.MutableMapping[str, object]

Atomic JSON application state that belongs to one job.

The state lives at .httk-job/state.json inside the job payload, so it survives every step advance, every retry, and every isolated workdir of the job, and it travels with the payload when the job is transferred. It is runner-private: the directory is excluded from every payload digest, so writing state never disturbs the immutability checks of the payload.

Keys are nonempty strings and values must be JSON. Each mutation rewrites the whole document through an atomic replace, so a crash leaves either the previous state or the new one.

Parameters:

  • payload (str | os.PathLike[str]) – Locate the job payload containing the state directory.

  • durable (bool) – Synchronize each atomic state replacement.

path
durable = False
read()[source]

Return the whole state document.

Returns:

The current JSON state mapping.

Return type:

dict[str, object]

merge(values)[source]

Write several keys in one atomic replace.

Parameters:

values (collections.abc.Mapping[str, object]) – Supply the keys and values to merge.

Raises:

ValueError – If a key or value is not valid JSON state.

set(name, value)[source]

Store one value, an alias of state[name] = value.

Parameters:

  • name (str) – Name the state key.

  • value (object) – Supply the JSON-compatible value.

Raises:

ValueError – If the key or value is invalid JSON state.

delete(name)[source]

Remove one key, reporting whether it was present.

Parameters:

name (str) – Name the state key.

Returns:

Whether the key was present.

Return type:

bool

class httk.workflow.ScaffoldedJob[source]

Describe one job this module submitted.

Parameters:

  • job_id – Identify the submitted job.

  • job_key – Identify the job payload and state markers.

  • tag – Preserve the optional job tag.

  • placement – Locate the job within the workspace.

  • payload – Locate the submitted payload.

  • marker – Locate the submitted state marker.

  • workflow – Name the workflow the job runs.

  • initial_step – Name the step the job starts at.

  • runner – Describe the pinned runner.

  • warnings – Preserve the preparation warnings raised for this workflow.

job_id: str
job_key: str
tag: str | None
placement: pathlib.PurePosixPath
payload: pathlib.Path
marker: pathlib.Path
workflow: str
initial_step: str
runner: collections.abc.Mapping[str, object]
warnings: tuple[str, Ellipsis] = ()
as_mapping()[source]

Return the machine-readable report of this job.

Returns:

The serialized job report.

Return type:

dict[str, object]

httk.workflow.new_job(workspace, workflow, *, inputs=None, files=None, parameters=None, environment=None, tag=None, placement=DEFAULT_PLACEMENT, priority=None, workdir_mode='persistent', data_mode=None, publish='workspace', step=None, format=None, workflow_id=None, name=None)[source]

Scaffold, submit, and describe one job of workflow.

workflow is a registered workflow name — see registered_workflows() — or the path of a runner file. files maps payload names to the files to stage there: a bare name lands in the payload’s FILES_DIRECTORY, which is where a packaged runner reads its inputs, and a name with a directory in it is used verbatim. inputs stages the workflow’s declared objects into the payload; parameters is the job’s opaque implementation mapping.

data_mode defaults to what the workflow needs — transactional for a workflow whose runner publishes collected results, and none for a runner that said nothing. publish workspace publishes the runner file into the workspace runner store and pins its digest; installed references a packaged runner through the reserved pkg: form instead and copies nothing. It is ignored for language workflows, whose realization chooses the runner itself.

Parameters:

  • workspace (httk.workflow.workspace.Workspace) – Provide the workspace receiving the job.

  • workflow (str | os.PathLike[str]) – Select the workflow or runner file.

  • inputs (collections.abc.Mapping[str, object] | None) – Supply declared workflow inputs.

  • files (collections.abc.Mapping[str, str | os.PathLike[str]] | None) – Map payload names to files to stage.

  • parameters (collections.abc.Mapping[str, object] | None) – Supply opaque job parameters.

  • environment (collections.abc.Mapping[str, object] | None) – Supply overrides for declared workflow environment values.

  • tag (str | None) – Set the job tag.

  • placement (str | pathlib.PurePosixPath) – Place the job within the workspace.

  • priority (int | None) – Set the scheduling priority.

  • workdir_mode (WorkdirMode) – Select the job workdir mode.

  • data_mode (DataMode | None) – Override the workflow data mode.

  • publish (PublishMode) – Select workspace publication or installed reference.

  • step (str | None) – Override the workflow’s initial step.

  • format (str | None) – Force a language for a bare workflow document or directory.

  • workflow_id (str | None) – Override the workflow id in the job definition.

  • name (str | None) – Set the job’s display name.

Returns:

The submitted job description.

Raises:

ValueError – If workflow, inputs, placement, or job settings are invalid.

Return type:

ScaffoldedJob

httk.workflow.new_jobs(workspace, workflow, items, *, inputs=None, files=None, parameters=None, environment=None, tag=None, placement=DEFAULT_PLACEMENT, priority=None, workdir_mode='persistent', data_mode=None, publish='workspace', step=None, format=None, workflow_id=None, name=None)[source]

Scaffold and submit one job per member of items, lazily.

Every keyword is the shared value of the whole campaign, and every member of one JobItem is what that job varies: inputs and files are merged over the shared mappings, and tag, name, placement, and priority replace the shared value.

This is the pattern for a campaign of any size. The workflow is resolved once and its runner published once, however many jobs follow, so every job costs exactly one payload directory and one state marker; items is consumed as an iterator and the results are yielded as they are submitted, so a structure generator can be turned into jobs without either side of the loop ever being materialized.

Parameters:

  • workspace (httk.workflow.workspace.Workspace) – Provide the workspace receiving the jobs.

  • workflow (str | os.PathLike[str]) – Select the workflow or runner file.

  • items (collections.abc.Iterable[JobItem]) – Yield per-job overrides.

  • inputs (collections.abc.Mapping[str, object] | None) – Supply shared declared workflow inputs.

  • files (collections.abc.Mapping[str, str | os.PathLike[str]] | None) – Supply shared payload files.

  • parameters (collections.abc.Mapping[str, object] | None) – Supply shared opaque job parameters.

  • environment (collections.abc.Mapping[str, object] | None) – Supply shared declared environment overrides.

  • tag (str | None) – Set the shared job tag.

  • placement (str | pathlib.PurePosixPath) – Set the shared workspace placement.

  • priority (int | None) – Set the shared scheduling priority.

  • workdir_mode (WorkdirMode) – Select the shared workdir mode.

  • data_mode (DataMode | None) – Override the workflow data mode.

  • publish (PublishMode) – Select workspace publication or installed reference.

  • step (str | None) – Override the workflow’s initial step.

  • format (str | None) – Force a language for a bare workflow document or directory.

  • workflow_id (str | None) – Override the workflow id in each job definition.

  • name (str | None) – Set the shared display name.

Returns:

An iterator yielding each submitted job description.

Yield:

Each submitted job description.

Raises:

ValueError – If workflow, inputs, placement, or job settings are invalid.

Return type:

collections.abc.Iterator[ScaffoldedJob]

def structures():
    for path in sorted(Path("structures").glob("POSCAR.*")):
        yield {"files": {"POSCAR": path}, "tag": structure_tag(path)}

for job in new_jobs(workspace, "some-workflow", structures(), parameters={"kpoint_density": 30.0}):
    print(job.job_key)
class httk.workflow.Attempt(context, *, control, payload, workdir, workspace, data=None, step=None, runner=None)[source]

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)[source]

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:
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)[source]

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)[source]

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)[source]

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)[source]

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:
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)[source]

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)[source]

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()[source]

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)[source]

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:
Returns:

The generated transaction operation identifier.

Raises:

ValueError – If this job has no transactional data.

Return type:

str

remove(destination, *, missing_ok=False)[source]

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)[source]

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)[source]

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)[source]

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()[source]

Publish the successful completion of this job.

Returns:

The path of the published outcome.

Return type:

pathlib.Path

retry(reason)[source]

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)[source]

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)[source]

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.ChildrenView[source]

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)[source]

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.ChildResult[source]

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.ChildSpec[source]

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] = ()
type httk.workflow.InstantiateHandler = Callable[[Any], object][source]
class httk.workflow.Runner(workflow, *, inputs=None)[source]

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[source]
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)[source]

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()[source]

Return the machine-readable description of this runner.

Returns:

The runner description document.

Return type:

dict[str, object]

main(argv=None)[source]

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

class httk.workflow.RunnerRef[source]

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()[source]

Reference exactly the runner of the spawning job.

Returns:

The inherited runner reference.

Return type:

RunnerRef

classmethod workspace(path, sha256)[source]

Reference one runner published in the workspace runner store.

Parameters:
Returns:

The workspace runner reference.

Return type:

RunnerRef

classmethod installed(path, sha256)[source]

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

Parameters:
Returns:

The installed runner reference.

Return type:

RunnerRef

class httk.workflow.Workspace(root, *, mutable=True, durable=True, marker_index_capacity=DEFAULT_MARKER_INDEX_CAPACITY)[source]

Attach to one self-contained httk workflow filesystem workspace.

Parameters:

  • root (str | os.PathLike[str]) – Locate the workspace root.

  • mutable (bool) – Preserve the attachment mutability option accepted by callers.

  • durable (bool) – Enable storage-crash durability for filesystem publications.

  • marker_index_capacity (int) – Bound the in-memory active-marker index.

Raises:
root
control
runners
runner_builds
durable = True
format
core_profile
extensions
workspace_id = ''
ensure_directory(path)[source]

Create a directory below the workspace root.

Parameters:

path (pathlib.Path) – Identify the directory to create.

Returns:

The created directory path.

Raises:

ValueError – If the path is outside the workspace root.

Return type:

pathlib.Path

classmethod initialize(root, *, extensions=(), durable=True, policy=None)[source]

Create and return a new workspace.

Parameters:
Returns:

The initialized workspace.

Raises:
Return type:

Workspace

classmethod default()[source]

Resolve the project or per-user default workspace, creating it if needed.

Returns:

The default workspace.

Return type:

Self

property policy: httk.workflow.models.WorkspacePolicy

Return the tunables this workspace publishes to every attacher.

Returns:

The current workspace policy.

Return type:

httk.workflow.models.WorkspacePolicy

property visibility_deadline: float

Return how long a metadata visibility retry may keep probing.

Returns:

The workspace metadata visibility deadline.

Return type:

float

set_policy(changes)[source]

Validate changes, merge them into the stored policy, and publish it.

The write is an ordinary read-modify-write of format.json through an exclusively created temporary file and a rename, so a reader never sees a torn object. It is deliberately not serialized against another writer: policy is administrative, changes are rare, and last writer wins.

Parameters:

changes (collections.abc.Mapping[str, object]) – Supply policy values to validate and merge.

Returns:

The resulting workspace policy.

Return type:

httk.workflow.models.WorkspacePolicy

property settings: dict[str, object]

Return this workspace’s application settings, a flat dotted map.

Application settings are distinct from policy, which tunes the engine. These are the values an application step resolves at run time — the VASP command, a pseudopotential library — one layer of the job-parameters → environment → workspace → default resolution a runner reads through setting(). A workspace written before the section existed reads as an empty map.

Returns:

The workspace’s application settings.

Return type:

dict[str, object]

read_settings()[source]

Read and validate the current settings from disk.

Returns:

The current application settings.

Raises:

httk.workflow.errors.FormatError – If the stored settings are not valid.

Return type:

dict[str, object]

set_setting(key, value)[source]

Store one application setting and return the resulting map.

The write is the same read-modify-write of format.json that set_policy() uses: an exclusively created temporary and a rename, so a reader never sees a torn object, and last writer wins.

Parameters:

  • key (str) – Name the application setting to store.

  • value (object) – Supply the setting value.

Returns:

The resulting application settings.

Raises:

ValueError – If the setting name or value is invalid, or its environment name collides.

Return type:

dict[str, object]

unset_setting(key)[source]

Remove one application setting, refusing one that is not set.

Parameters:

key (str) – Name the application setting to remove.

Returns:

The resulting application settings.

Raises:

ValueError – If the setting is not set.

Return type:

dict[str, object]

seed_settings(seeds)[source]

Merge seeds into the settings, keeping any value already set.

Seeding happens once, when a workspace bound to a remote is created: the remote definition’s whitelisted settings become the workspace’s starting application settings. An explicit setting already present is never overwritten, so a value the operator chose outlives a reseed.

Parameters:

seeds (collections.abc.Mapping[str, object]) – Supply initial application settings to merge.

Returns:

The resulting application settings.

Raises:

ValueError – If a supplied setting is invalid.

Return type:

dict[str, object]

read_workflow_preludes()[source]

Read and validate the workflow-in-workspace preludes from disk.

A workflow prelude is shell text run to initialize the environment before each launch of a runner for that workflow. A workspace written before the section existed reads as an empty map.

Returns:

The current map of workflow id to prelude text.

Raises:

httk.workflow.errors.FormatError – If the stored preludes are not valid.

Return type:

dict[str, str]

set_workflow_prelude(workflow_id, value)[source]

Store one workflow prelude and return the resulting map.

The write is the same read-modify-write of format.json that set_setting() uses: an exclusively created temporary and a rename, so a reader never sees a torn object, and last writer wins.

Parameters:

  • workflow_id (str) – Name the workflow whose prelude to store.

  • value (str) – Supply the shell prelude text.

Returns:

The resulting map of workflow id to prelude text.

Raises:

ValueError – If the workflow id or prelude value is invalid.

Return type:

dict[str, str]

unset_workflow_prelude(workflow_id)[source]

Remove one workflow prelude, refusing one that is not set.

Parameters:

workflow_id (str) – Name the workflow whose prelude to remove.

Returns:

The resulting map of workflow id to prelude text.

Raises:

ValueError – If the prelude is not set.

Return type:

dict[str, str]

open_journal_writer(*, writer_id=None)[source]

Open one exclusive journal writer configured by workspace policy.

Parameters:

writer_id (str | None) – Reuse a canonical writer identity when one is supplied.

Returns:

The exclusive journal writer.

Raises:

ValueError – If the writer identity is not canonical.

Return type:

httk.workflow.journal.JournalWriter

check(*, repair=False, quarantine_unrepairable=False)[source]

Verify that every marker resolves to its journal frame.

Parameters:

  • repair (bool) – Repair recoverable marker or journal inconsistencies.

  • quarantine_unrepairable (bool) – Quarantine entries that cannot be repaired.

Returns:

The workspace check report.

Return type:

httk.workflow.fsck.FsckReport

collect_garbage(*, dry_run=False, now=None)[source]

Collect the disk this workspace’s retention policy permits freeing.

Parameters:

  • dry_run (bool) – Report eligible removals without changing the workspace.

  • now (float | None) – Use this timestamp when evaluating retention deadlines.

Returns:

The garbage-collection report.

Return type:

httk.workflow.gc.GcReport

runner_store_path(path)[source]

Return the store location of one workspace runner.

The store is flat and name-keyed below .httk-workflow/runners/. Relative subdirectories are permitted so a campaign can group runners, but a name can never escape the store.

Parameters:

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

Returns:

The runner’s store path.

Raises:

httk.workflow.errors.FormatError – If the runner path is invalid or escapes the store.

Return type:

pathlib.Path

publish_runner(source, *, name=None, replace=False)[source]

Install one runner in the workspace store and describe the reference.

Publication is content addressed: republishing identical bytes is an idempotent no-op, and replacing a name whose content differs requires replace so a live campaign referring to the old digest can never be changed underneath by accident.

Parameters:

  • source (str | os.PathLike[str]) – Locate the runner file or directory to publish.

  • name (str | pathlib.PurePosixPath | None) – Choose the store name, defaulting to the source name.

  • replace (bool) – Replace a different existing runner with the same name.

Returns:

The published runner reference.

Raises:
Return type:

dict[str, object]

detach(job_id, *, destination_workspace_id, destination_remote=None, destination_placement=None, transfer_id=None)[source]

Seal one quiescent job as a detached transfer bundle.

Parameters:

  • job_id (str) – Identify the job to detach.

  • destination_workspace_id (str) – Identify the destination workspace.

  • destination_remote (str | None) – Name the destination remote, when applicable.

  • destination_placement (str | pathlib.PurePosixPath | None) – Choose the destination placement.

  • transfer_id (str | None) – Reuse a transfer identity when resuming a publication.

Returns:

The sealed transfer bundle path.

Return type:

pathlib.Path

import_bundle(bundle)[source]

Import a validated detached transfer bundle.

Parameters:

bundle (str | os.PathLike[str]) – Locate the detached transfer bundle.

Returns:

The imported transfer description.

Return type:

dict[str, object]

acknowledge_transfer(acknowledgement)[source]

Retire a source bundle after destination acknowledgement.

Parameters:

acknowledgement (collections.abc.Mapping[str, object]) – Supply the destination acknowledgement.

Returns:

The retired source bundle path.

Return type:

pathlib.Path

recover_transfers()[source]

Recover or report interrupted detached-transfer publications.

Returns:

Descriptions of recovered or still-pending transfers.

Return type:

list[dict[str, object]]

state_directory(kind, placement)[source]

Return the state directory for a kind and placement.

Parameters:
Returns:

The state directory path.

Raises:

ValueError – If the state kind is unknown.

Return type:

pathlib.Path

marker_path(kind, placement, job_key, priority, generation, record_ref)[source]

Build the path of one state marker.

Parameters:

  • kind (str) – Select the state kind.

  • placement (pathlib.PurePosixPath) – Select the marker placement.

  • job_key (str) – Identify the job.

  • priority (int) – Set the marker priority.

  • generation (int) – Set the state generation.

  • record_ref (str) – Identify the journal record.

Returns:

The marker path.

Raises:

ValueError – If the state kind is unknown.

Return type:

pathlib.Path

payload_path(placement, job_key)[source]

Return the payload path for one job placement.

Parameters:
Returns:

The payload directory path.

Return type:

pathlib.Path

walk_markers(kinds=None, *, roots=(), heartbeat=None, heartbeat_every=DISCOVERY_HEARTBEAT_STRIDE)[source]

Stream every schedulable marker of kinds, exhaustively.

This is the streaming, cursorless counterpart of a bounded pass: it walks the same scandir tree with no discovery budget, reports every fault, and takes a heartbeat opportunity every heartbeat_every entries so a long exhaustive pass — polling running attempts, recovering claims — keeps its lease alive from inside the walk. A pass MAY restrict itself to placement roots; the debug workspace narrows what it surfaces through its private _scheduling_includes hook.

Parameters:
Yield:

Each schedulable marker found during the walk.

scan_marker_entries(kinds=None)[source]

Yield every marker below state/, reporting damage per entry.

One unusable entry must never hide the rest of the workspace, so a marker-shaped basename that fails validation is reported as a MarkerFault instead of aborting the scan. This is the exhaustive walk the workspace tools (fsck, collection, status, collect) use; the scheduling passes use the bounded MarkerStream instead.

Parameters:

kinds (collections.abc.Iterable[str] | None) – Restrict the scan to these state kinds.

Yield:

Each valid marker or reported marker fault.

scan_markers(kinds=None)[source]

Yield every valid marker below state/.

Parameters:

kinds (collections.abc.Iterable[str] | None) – Restrict the scan to these state kinds.

Yield:

Each valid marker.

report_marker_fault(fault)[source]

Report an uninterpretable state entry loudly once, then quietly.

A marker whose basename or placement cannot be parsed is workspace corruption rather than a job state: the core profile leaves its repair to an explicit workspace tool, so a manager only reports it and never schedules or relocates it.

Parameters:

fault (MarkerFault) – Describe the unusable state entry to report.

invalidate_marker_index()[source]

Drop the cached job-id index, so the next lookup rebuilds it.

find_markers(job_key, kinds=None)[source]

Find all current markers for a job key.

Parameters:
Returns:

The matching current markers.

Raises:

httk.workflow.errors.WorkspaceCorruptionError – If more than one current marker identifies the job.

Return type:

list[httk.workflow.models.Marker]

find_marker_by_id(job_id)[source]

Return the one current marker of job_id, or None if it has none.

Resolution follows the specified ladder: the in-memory index, then a targeted probe of the finite state set at the placement the index last saw, then one complete rescan. Absence is only ever reported after that rescan, so a job another actor has just created or moved is never mistaken for a job that does not exist.

Parameters:

job_id (str) – Identify the job to find.

Returns:

The current marker, or None when the job has no marker.

Raises:

httk.workflow.errors.WorkspaceCorruptionError – If more than one current marker identifies the job.

Return type:

httk.workflow.models.Marker | None

find_marker_at(job_key, placement)[source]

Find job_key by checking the finite state set at a placement.

This is the first rung of the resolution ladder: a join child carrying a placement hint is resolved here, without the index and without a scan. The index is used only as a shortcut when it already names this job at exactly this placement, which turns the bounded directory sweep below into one confirmed lookup.

Parameters:

  • job_key (str) – Identify the job to find.

  • placement (pathlib.PurePosixPath) – Restrict the lookup to this placement.

Returns:

The matching current marker, or None when none exists.

Raises:

httk.workflow.errors.WorkspaceCorruptionError – If more than one marker exists at the placement.

Return type:

httk.workflow.models.Marker | None

load_job(marker)[source]

Load and validate the job definition referenced by a marker.

Parameters:

marker (httk.workflow.models.Marker) – Identify the job payload to load.

Returns:

The validated job definition.

Raises:

httk.workflow.errors.FormatError – If the payload identity disagrees with the marker.

Return type:

httk.workflow.models.JobDefinition

read_state(marker)[source]

Read and validate the state frame referenced by a marker.

Parameters:

marker (httk.workflow.models.Marker) – Identify the state frame to read.

Returns:

The validated state frame.

Raises:
Return type:

dict[str, Any]

transition(writer, marker, kind, updates, *, priority=None)[source]

Append a state frame and atomically move marker to it.

Parameters:

  • writer (httk.workflow.journal.JournalWriter) – Append the new state frame through this journal writer.

  • marker (httk.workflow.models.Marker) – Identify the current marker to advance.

  • kind (str) – Select the next state kind.

  • updates (collections.abc.Mapping[str, object]) – Add state members to the new frame.

  • priority (int | None) – Override the marker priority when supplied.

Returns:

The marker after the transition.

Raises:
Return type:

httk.workflow.models.Marker

repoint_marker(writer, marker, frame)[source]

Publish a repair frame for marker and move the marker onto it.

This is the repair counterpart of transition(). The caller supplies the complete frame because what needs repairing is precisely the frame the marker references now, which cannot be read and therefore cannot be carried forward automatically. The frame must still name this marker’s job and kind at the next generation, so a repair can never disguise a state change as a repair.

Parameters:

  • writer (httk.workflow.journal.JournalWriter) – Append the repair frame through this journal writer.

  • marker (httk.workflow.models.Marker) – Identify the damaged marker to repair.

  • frame (collections.abc.Mapping[str, object]) – Supply the complete replacement state frame.

Returns:

The marker after the repair frame is published.

Raises:

httk.workflow.errors.FormatError – If the repair frame changes required marker identity.

Return type:

httk.workflow.models.Marker

submit(source, placement, *, move=False)[source]

Copy or move a complete payload into the workspace and publish it.

Parameters:
Returns:

The submitted job marker.

Raises:

  • FileExistsError – If the target payload already exists.

  • httk.workflow.workspace.WorkspaceOperationError – If a move crosses filesystems.

Return type:

httk.workflow.models.Marker

validate_job_payload(marker)[source]

Perform manager-side immutable submission validation.

Parameters:

marker (httk.workflow.models.Marker) – Identify the submitted payload to validate.

Returns:

The validated job definition.

Return type:

httk.workflow.models.JobDefinition

quarantine(path, *, reason)[source]

Move a malformed protocol entry into the canonical quarantine.

Parameters:

  • path (pathlib.Path) – Locate the malformed protocol entry.

  • reason (str) – Record why the entry was quarantined.

Returns:

The quarantine directory containing the entry.

Return type:

pathlib.Path

payload_digest(marker)[source]

Return the digest of one payload, ignoring runner-private entries.

Parameters:

marker (httk.workflow.models.Marker) – Identify the job payload to digest.

Returns:

The payload digest.

Return type:

str

publish_request(request)[source]

Atomically publish an operator request.

Parameters:

request (collections.abc.Mapping[str, object]) – Supply the operator request to publish.

Returns:

The ready request path.

Return type:

pathlib.Path