httk.workflow

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 harvest() 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.

Only the deliberate top-level surface is re-exported here; everything else is reached through its submodule.

Submodules

• httk.workflow.adapter_protocol
• httk.workflow.adapters
• httk.workflow.backends
• httk.workflow.compat
• httk.workflow.configuration
• httk.workflow.errors
• httk.workflow.harvesting
• httk.workflow.hygiene
• httk.workflow.manifests
• httk.workflow.projects
• httk.workflow.protocol
• httk.workflow.runtime
• httk.workflow.runtime_utils
• httk.workflow.scaffold
• httk.workflow.sdk
• httk.workflow.shell_bridge
• httk.workflow.supervision
• httk.workflow.transfers
• httk.workflow.vasp

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.

Classes

HarvestRecord	Everything a data layer needs about one job that stopped.
TaskManager	Execute and recover jobs in one workflow workspace.
JobState	Atomic JSON application state that belongs to one job.
ScaffoldedJob	One job this module submitted, and everything needed to look at it again.
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 inputs 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	One self-contained httk workflow filesystem workspace.

Functions

harvest(→ collections.abc.Iterator[HarvestRecord])	Yield one HarvestRecord per finished job of workspace.
new_job(→ ScaffoldedJob)	Scaffold, submit, and describe one job of template.
new_jobs(→ collections.abc.Iterator[ScaffoldedJob])	Scaffold and submit one job per member of items, lazily.

Package Contents

exception httk.workflow.FormatError

Bases: WorkflowError, ValueError

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

exception httk.workflow.RunnerResolutionError(code: str, message: str)

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) and a runner whose staged bytes disagree with the digest the job pinned (runner_mismatch) stay distinguishable to an operator.

code

exception httk.workflow.TransactionError

Bases: WorkflowError

A transactional-data manifest cannot be safely replayed.

exception httk.workflow.TransitionLostError

Bases: WorkflowError

Another actor committed a transition from the expected marker.

exception httk.workflow.UnsupportedExtensionError

Bases: WorkflowError

A workspace requires an extension this implementation does not support.

exception httk.workflow.WorkflowError

Bases: Exception

Base class for workflow protocol failures.

exception httk.workflow.WorkspaceCorruptionError

Bases: WorkflowError

The authoritative filesystem state is internally inconsistent.

exception httk.workflow.WorkspaceUnavailableError

Bases: WorkflowError

The workspace cannot currently provide a coherent protocol view.

class httk.workflow.HarvestRecord

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 harvested from, which is what code reading result files wants.

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() → dict[str, object]

Return the JSON representation of this record.

classmethod from_mapping(value: collections.abc.Mapping[str, object]) → HarvestRecord

Rebuild one record from the mapping as_mapping() produced.

httk.workflow.harvest(workspace: httk.workflow.workspace.Workspace, *, states: collections.abc.Iterable[str] = DEFAULT_HARVEST_STATES, placement: str | pathlib.PurePosixPath | None = None) → collections.abc.Iterator[HarvestRecord]

Yield one HarvestRecord per finished job of workspace.

states selects which stopped jobs are reported and defaults to the successful ones; every requested kind is validated against HARVESTABLE_KINDS before anything is read. placement restricts the harvest 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 harvesting 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.

class httk.workflow.TaskManager(workspace: httk.workflow.workspace.Workspace, *, pools: collections.abc.Sequence[str] = ('default',), capabilities: collections.abc.Sequence[str] = (), maximum_workers: int = 1, lease_seconds: float | None = None, heartbeat_interval: float = 30.0, unsafe_persistent_takeover: bool = False, unsafe_isolated_takeover: bool = False, takeover_grace_factor: float = DEFAULT_TAKEOVER_GRACE_FACTOR, runner_backends: collections.abc.Sequence[httk.workflow.backends.RunnerBackend] = (), allowed_backends: collections.abc.Sequence[str] | None = None, accept_any_pool: bool = False, join_grace_seconds: float = 3600.0, cancel_grace_seconds: float = DEFAULT_CANCEL_GRACE_SECONDS, maximum_pass_markers: int = DEFAULT_MAXIMUM_PASS_MARKERS, discovery_budget: int = DEFAULT_DISCOVERY_BUDGET, placement_prefixes: collections.abc.Sequence[str] = (), runner_search_paths: collections.abc.Iterable[str | os.PathLike[str]] = (), runner_modules: collections.abc.Iterable[str] = DEFAULT_RUNNER_MODULES, gc_interval: float | None = None)

Execute and recover jobs in one workflow workspace.

workspace

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

runner_backends

allowed_backends

accept_any_pool = False

manager_id = ''

hostname

writer

close() → None

property manager_directory: pathlib.Path

Return this manager’s own directory below managers/.

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.

heartbeat(*, force: bool = False) → None

tick() → bool

Perform one nonblocking scheduling and recovery pass.

serve(*, poll_interval: float = 1.0, drain_timeout: float = 30.0, drain_grace_seconds: float = 10.0) → None

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.

run_until_idle(*, timeout: float = 60.0, poll_interval: float = 0.02) → None

Run until no local process or immediately actionable marker remains.

class httk.workflow.JobState(payload: str | os.PathLike[str], *, durable: bool = False)

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.

path

durable = False

read() → dict[str, object]

Return the whole state document.

merge(values: collections.abc.Mapping[str, object]) → None

Write several keys in one atomic replace.

set(name: str, value: object) → None

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

delete(name: str) → bool

Remove one key, reporting whether it was present.

class httk.workflow.ScaffoldedJob

One job this module submitted, and everything needed to look at it again.

job_id: str

job_key: str

tag: str | None

placement: pathlib.PurePosixPath

payload: pathlib.Path

marker: pathlib.Path

workflow: str

initial_step: str

template: str

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

as_mapping() → dict[str, object]

Return the machine-readable report of this job.

httk.workflow.new_job(workspace: httk.workflow.workspace.Workspace, template: str | os.PathLike[str], *, inputs: collections.abc.Mapping[str, object] | None = None, files: collections.abc.Mapping[str, str | os.PathLike[str]] | None = None, tag: str | None = None, placement: str | pathlib.PurePosixPath = DEFAULT_PLACEMENT, priority: int | None = None, workdir_mode: WorkdirMode = 'persistent', data_mode: DataMode | None = None, publish: PublishMode = 'workspace', step: str | None = None, workflow: str | None = None, name: str | None = None) → ScaffoldedJob

Scaffold, submit, and describe one job of template.

template is a registered template name — see registered_templates() — 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 becomes the job’s inputs object, whose members are documented per runner.

data_mode defaults to what the template needs — transactional for a template 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.

httk.workflow.new_jobs(workspace: httk.workflow.workspace.Workspace, template: str | os.PathLike[str], items: collections.abc.Iterable[JobItem], *, inputs: collections.abc.Mapping[str, object] | None = None, files: collections.abc.Mapping[str, str | os.PathLike[str]] | None = None, tag: str | None = None, placement: str | pathlib.PurePosixPath = DEFAULT_PLACEMENT, priority: int | None = None, workdir_mode: WorkdirMode = 'persistent', data_mode: DataMode | None = None, publish: PublishMode = 'workspace', step: str | None = None, workflow: str | None = None, name: str | None = None) → collections.abc.Iterator[ScaffoldedJob]

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 template 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.

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-template", structures(), inputs={"kpoint_density": 30.0}):
    print(job.job_key)

class httk.workflow.Attempt(context: httk.workflow.runtime.AttemptContext, *, control: pathlib.Path, payload: pathlib.Path, workdir: pathlib.Path, workspace: pathlib.Path, data: pathlib.Path | None = None, step: str | None = None, runner: Runner | None = 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.

context

control

payload

workdir

workspace

data = None

step

state

log

classmethod initialize(environment: collections.abc.Mapping[str, str] | None = None, *, runner: Runner | None = None) → Self

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.

property job: httk.workflow.models.JobDefinition

The immutable definition of the job this attempt belongs to.

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

The application-defined inputs 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.

input(name: str, default: object = _MISSING) → object

Return one member of the job’s inputs object.

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

setting(name: str, default: object = None) → object

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 inputs 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.

declare(name: str, document: collections.abc.Mapping[str, object]) → pathlib.Path

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 harvest reports the observed document beside the declared one and never merges the two.

declaration(name: str) → collections.abc.Mapping[str, object] | None

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.

run(argv: collections.abc.Sequence[str], *, timeout: float | None = None, cwd: str | os.PathLike[str] | None = None, environment: collections.abc.Mapping[str, str] | None = None, termination_grace: float = 10.0) → httk.workflow.runtime.CommandResult

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

workdir_batch() → httk.workflow.runtime_builders.ReplayableWorkdirBatch

Start a replayable group of workdir changes.

put(source: str | os.PathLike[str], destination: str | os.PathLike[str]) → str

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.

remove(destination: str | os.PathLike[str], *, missing_ok: bool = False) → str

Remove one path from the job’s transactional data.

spawn(child: ChildSpec | str | os.PathLike[str], *, label: str, placement: str | pathlib.PurePosixPath | None = None) → httk.workflow.runtime_builders.ChildReference

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.

advance(step: str, *, state: collections.abc.Mapping[str, object] | None = None, priority: int | None = None) → pathlib.Path

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.

gather(step: str, *, when: httk.workflow.runtime_builders.JoinCondition = 'all_succeeded', count: int | None = None, on_impossible: str | None = None) → pathlib.Path

Wait for the children spawned on this attempt, then run step.

The join names exactly the children spawn() registered on this attempt, which is also the bundle that creates them, so the manager can always resolve it. when is one of all_succeeded, all_terminal, any_succeeded, 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.

succeed() → pathlib.Path

Publish the successful completion of this job.

retry(reason: str) → pathlib.Path

Ask for another attempt of this same activation.

pause(reason: str) → pathlib.Path

Pause this job until an operator resumes it.

fail(code: str, message: str, *, details: collections.abc.Mapping[str, object] | None = None, retryable: bool = False) → pathlib.Path

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.

class httk.workflow.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: str, default: ChildResult | None = None) → ChildResult | None

Return the child spawned under label, or default.

class httk.workflow.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.ChildSpec

A complete child job described by the step and inputs 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 inputs.

step: str

inputs: 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.Runner(workflow: str)

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.

workflow

property steps: frozenset[str]

The names of every registered step.

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.

description() → dict[str, object]

Return the machine-readable description of this runner.

main(argv: collections.abc.Sequence[str] | None = None) → int

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.

class httk.workflow.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.

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

path: str | None = None

sha256: str | None = None

classmethod inherit() → RunnerRef

Reference exactly the runner of the spawning job.

classmethod workspace(path: str | pathlib.PurePosixPath, sha256: str) → RunnerRef

Reference one runner published in the workspace runner store.

classmethod installed(path: str | pathlib.PurePosixPath, sha256: str) → RunnerRef

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

class httk.workflow.Workspace(root: str | os.PathLike[str], *, mutable: bool = True, durable: bool = True, marker_index_capacity: int = DEFAULT_MARKER_INDEX_CAPACITY)

One self-contained httk workflow filesystem workspace.

root

control

runners

durable = True

format

core_profile

extensions

workspace_id = ''

classmethod initialize(root: str | os.PathLike[str], *, extensions: collections.abc.Iterable[str] = (), durable: bool = True, policy: collections.abc.Mapping[str, object] | None = None) → Workspace

Create and return a new workspace.

property policy: httk.workflow.models.WorkspacePolicy

Return the shared tunables this workspace publishes to every attacher.

property visibility_deadline: float

Return how long a metadata visibility retry may keep probing.

set_policy(changes: collections.abc.Mapping[str, object]) → httk.workflow.models.WorkspacePolicy

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.

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-inputs → environment → workspace → default resolution a runner reads through setting(). A workspace written before the section existed reads as an empty map.

set_setting(key: str, value: object) → dict[str, object]

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.

unset_setting(key: str) → dict[str, object]

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

seed_settings(seeds: collections.abc.Mapping[str, object]) → dict[str, object]

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 queue 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.

open_journal_writer(*, writer_id: str | None = None) → httk.workflow.journal.JournalWriter

Open one exclusive journal writer configured by workspace policy.

check(*, repair: bool = False, quarantine_unrepairable: bool = False) → httk.workflow.fsck.FsckReport

Verify that every marker resolves to its journal frame.

collect_garbage(*, dry_run: bool = False, now: float | None = None) → httk.workflow.gc.GcReport

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

upgrade(extensions: collections.abc.Iterable[str]) → frozenset[str]

Enable extensions that have an implemented in-place migration.

runner_store_path(path: str | pathlib.PurePosixPath) → pathlib.Path

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.

publish_runner(source: str | os.PathLike[str], *, name: str | pathlib.PurePosixPath | None = None, replace: bool = False) → dict[str, object]

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.

detach(job_id: str, *, destination_workspace_id: str, destination_placement: str | pathlib.PurePosixPath | None = None, transfer_id: str | None = None) → pathlib.Path

Seal one quiescent job as a detached transfer bundle.

import_bundle(bundle: str | os.PathLike[str]) → dict[str, object]

Import a validated detached transfer bundle.

acknowledge_transfer(acknowledgement: collections.abc.Mapping[str, object]) → pathlib.Path

Retire a source bundle after destination acknowledgement.

recover_transfers() → list[dict[str, object]]

Recover or report interrupted detached-transfer publications.

state_directory(kind: str, placement: pathlib.PurePosixPath) → pathlib.Path

marker_path(kind: str, placement: pathlib.PurePosixPath, job_key: str, priority: int, generation: int, record_ref: str) → pathlib.Path

payload_path(placement: pathlib.PurePosixPath, job_key: str) → pathlib.Path

walk_markers(kinds: collections.abc.Iterable[str] | None = None, *, roots: collections.abc.Sequence[pathlib.PurePosixPath] = (), heartbeat: collections.abc.Callable[[], None] | None = None, heartbeat_every: int = DISCOVERY_HEARTBEAT_STRIDE) → collections.abc.Iterator[httk.workflow.models.Marker]

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.

scan_marker_entries(kinds: collections.abc.Iterable[str] | None = None) → collections.abc.Iterator[httk.workflow.models.Marker | MarkerFault]

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, harvest) use; the scheduling passes use the bounded MarkerStream instead.

scan_markers(kinds: collections.abc.Iterable[str] | None = None) → collections.abc.Iterable[httk.workflow.models.Marker]

report_marker_fault(fault: MarkerFault) → None

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.

invalidate_marker_index() → None

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

find_markers(job_key: str, kinds: collections.abc.Iterable[str] | None = None) → list[httk.workflow.models.Marker]

find_marker_by_id(job_id: str) → httk.workflow.models.Marker | None

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.

find_marker_at(job_key: str, placement: pathlib.PurePosixPath) → httk.workflow.models.Marker | None

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.

load_job(marker: httk.workflow.models.Marker) → httk.workflow.models.JobDefinition

read_state(marker: httk.workflow.models.Marker) → dict[str, Any]

transition(writer: httk.workflow.journal.JournalWriter, marker: httk.workflow.models.Marker, kind: str, updates: collections.abc.Mapping[str, object], *, priority: int | None = None) → httk.workflow.models.Marker

Append a state frame and atomically move marker to it.

repoint_marker(writer: httk.workflow.journal.JournalWriter, marker: httk.workflow.models.Marker, frame: collections.abc.Mapping[str, object]) → httk.workflow.models.Marker

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.

submit(source: str | os.PathLike[str], placement: str | pathlib.PurePosixPath, *, move: bool = False) → httk.workflow.models.Marker

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

validate_job_payload(marker: httk.workflow.models.Marker) → httk.workflow.models.JobDefinition

Perform manager-side immutable submission validation.

quarantine(path: pathlib.Path, *, reason: str) → pathlib.Path

Move a malformed protocol entry into the canonical quarantine.

payload_digest(marker: httk.workflow.models.Marker) → str

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

publish_request(request: collections.abc.Mapping[str, object]) → pathlib.Path

Atomically publish an operator request.