httk.workflow.compat.pwd

Importing one Python Workflow Definition document as one httk₂ job.

The Python Workflow Definition (PWD) is a small JSON exchange format: a list of nodes and a list of edges. A node is a Python function named module.function, a literal input, or a named output; an edge connects one node’s output port to another node’s input port. The format is deliberately machine-facing — several workflow engines read and write it — and this module is httk₂ reading it.

from httk.workflow import Workspace
from httk.workflow.compat.pwd import import_pwd

workspace = Workspace.initialize("workflow-workspace")
job = import_pwd(workspace, "workflow.json", modules=["workflow.py"], tag="arithmetic")

The import is one way and produces exactly one job. The whole graph runs inside that job, sequentially, in topological order, by the packaged pwd_runner.py — no runner file is written per workflow, and no per-node job is created: a PWD node is one Python call, which is not worth a claim, a lease and a process of its own.

The document travels in the job’s inputs when it fits within maximum_embedded_bytes, and in files/pwd.json with an inputs pointer when it does not, because inputs is bounded by MAXIMUM_INPUTS_BYTES and a generated document can be much larger than that.

Warning

Running a PWD document executes the Python functions it names. There is no sandbox and there cannot be one: the format’s whole content is module.function references. Import a document exactly as carefully as you would run the module it names. Passing allowed_modules records an allowlist of module prefixes in the job, which the runner refuses to import outside of.

Attributes

MAXIMUM_INPUTS_BYTES
DEFAULT_PLACEMENT
FILES_DIRECTORY
WORKFLOW
INITIAL_STEP
PACKAGE
RUNNER
KNOWN_VERSIONS
DOCUMENT_FILE
DEFAULT_MAXIMUM_EMBEDDED_BYTES

Exceptions

PwdFormatError

A document that is not a Python Workflow Definition this importer accepts.

Classes

Workspace	One self-contained httk workflow filesystem workspace.
ScaffoldedJob	One job this module submitted, and everything needed to look at it again.
PwdDocument	One validated PWD document, and the order its function nodes run in.

Functions

submit_integration_job(, ...)	Build one payload for a packaged integration runner and submit it.
load_pwd_document(→ PwdDocument)	Read and validate one PWD document from path.
validate_pwd_document(→ PwdDocument)	Validate the shape of one PWD document and order its nodes.
import_pwd(, module_path, workflow_inputs, ...)	Import one PWD document as one submitted job, and describe it.

Package Contents

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

httk.workflow.compat.pwd.MAXIMUM_INPUTS_BYTES = 262144

httk.workflow.compat.pwd.DEFAULT_PLACEMENT = 'jobs'

httk.workflow.compat.pwd.FILES_DIRECTORY = 'files'

class httk.workflow.compat.pwd.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.compat.pwd.submit_integration_job(workspace: httk.workflow.Workspace, *, runner_package: str, runner: str, workflow: str, initial_step: str, name: str, inputs: collections.abc.Mapping[str, object], documents: collections.abc.Mapping[str, str] | 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, data_mode: Literal['none', 'transactional'] = 'none', workdir_mode: Literal['persistent', 'isolated'] = 'persistent', required_capabilities: tuple[str, Ellipsis] = (), maximum_attempts_per_activation: int | None = None, template: str = 'import') → httk.workflow.scaffold.ScaffoldedJob

Build one payload for a packaged integration runner and submit it.

documents are written verbatim into the payload and files are copied into it, both under the naming rule of payload_relative(): a bare name lands in files/ and a name carrying a directory is used as written. The payload is built inside the workspace’s own scratch directory, so submitting it is a rename rather than a copy however large the staged inputs are.

httk.workflow.compat.pwd.WORKFLOW = 'pwd.workflow'

httk.workflow.compat.pwd.INITIAL_STEP = 'execute'

httk.workflow.compat.pwd.PACKAGE = 'httk.workflow.compat.pwd'

httk.workflow.compat.pwd.RUNNER = 'pwd_runner.py'

httk.workflow.compat.pwd.KNOWN_VERSIONS = ('0.1.0',)

httk.workflow.compat.pwd.DOCUMENT_FILE = 'files/pwd.json'

httk.workflow.compat.pwd.DEFAULT_MAXIMUM_EMBEDDED_BYTES = 131072

exception httk.workflow.compat.pwd.PwdFormatError

Bases: ValueError

A document that is not a Python Workflow Definition this importer accepts.

class httk.workflow.compat.pwd.PwdDocument

One validated PWD document, and the order its function nodes run in.

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

nodes: collections.abc.Mapping[int, collections.abc.Mapping[str, object]]

edges: tuple[collections.abc.Mapping[str, object], Ellipsis]

order: tuple[int, Ellipsis]

property version: str | None

property functions: tuple[str, Ellipsis]

Every module.function this document would import, in node order.

property input_names: tuple[str, Ellipsis]

The names of every input node, in node order.

property output_names: tuple[str, Ellipsis]

The names of every output node, in node order.

httk.workflow.compat.pwd.load_pwd_document(path: str | os.PathLike[str], *, allow_unknown_version: bool = False) → PwdDocument

Read and validate one PWD document from path.

httk.workflow.compat.pwd.validate_pwd_document(raw: object, *, source: str = 'the document', allow_unknown_version: bool = False) → PwdDocument

Validate the shape of one PWD document and order its nodes.

Every member the format defines is checked; every member it does not define is preserved untouched, so a document carrying an engine’s own annotations survives the round trip into the job payload. When the python-workflow-definition package happens to be installed it is asked for a second opinion — it is never a dependency of httk-workflow, only a stricter validator when it is there.

httk.workflow.compat.pwd.import_pwd(workspace: httk.workflow.Workspace, document_path: str | os.PathLike[str], *, placement: str | pathlib.PurePosixPath = DEFAULT_PLACEMENT, tag: str | None = None, name: str | None = None, priority: int | None = None, modules: collections.abc.Sequence[str | os.PathLike[str]] = (), module_path: collections.abc.Sequence[str] = (), workflow_inputs: collections.abc.Mapping[str, object] | None = None, allowed_modules: collections.abc.Sequence[str] = (), data_mode: Literal['none', 'transactional'] = 'none', maximum_attempts: int | None = 3, maximum_embedded_bytes: int = DEFAULT_MAXIMUM_EMBEDDED_BYTES, allow_unknown_version: bool = False) → httk.workflow.compat._integration.ScaffoldedJob

Import one PWD document as one submitted job, and describe it.

modules are Python files staged into the payload’s files/ directory, which the runner puts first on sys.path: a document naming workflow.get_sum is self-contained once workflow.py is staged that way. module_path adds further import roots by absolute path, for functions that live in an installed package on the machine that will run the job.

workflow_inputs overrides the value of input nodes by name, so one imported document runs with values the document itself does not carry. allowed_modules records a prefix allowlist the runner refuses to import outside of; the default is no allowlist, which means the document may import anything the interpreter can, exactly like running its module by hand.

The graph is validated, ordered and refused here rather than at run time: a document with a cycle, a dangling edge or a node that is not a callable reference never reaches a queue.