httk.workflow.compat.cwl

Importing one CWL document as one httk₂ job.

The Common Workflow Language is supported here as a workflow language, not as an execution engine. A document is parsed and validated with cwl-utils, normalized into one self-contained JSON plan, and then executed entirely by httk₂’s own packaged cwl_runner.py on httk₂’s own manager: claims, leases, attempts, checkpoints, labeled children and journalled state frames are the same ones every other job of a workspace gets. cwltool is not used, not bundled and never invoked.

from httk.workflow import Workspace
from httk.workflow.compat.cwl import import_cwl

workspace = Workspace.initialize("workflow-workspace")
imported = import_cwl(workspace, "flow.cwl", "job.yml", tag="echo")
print(imported.job.job_key, imported.warnings)

Parsing needs the optional extra:

pip install httk-workflow[cwl]

which brings cwl-utils and cwl-upgrader. Nothing else in httk-workflow depends on either, and the packaged runner that executes the normalized plan needs neither: the plan is plain JSON, so the machine that runs an imported CWL job does not need a CWL library at all.

The supported subset

Supported: Workflow and CommandLineTool of CWL v1.0, v1.1 and v1.2 (older versions are upgraded when cwl-upgrader is installed); baseCommand and arguments; input bindings with position, prefix, separate and itemSeparator; the types File, Directory, string, int, long, float, double, boolean, Any, arrays of those, and optional (?) forms of all of them; stdout/stderr shortcuts and stdout/stderr redirection; output collection through outputBinding.glob with loadContents; EnvVarRequirement, ResourceRequirement (recorded), ToolTimeLimit (honoured), successCodes; workflow steps with source, default, linkMerge and plain-reference valueFrom; single-input scatter and dotproduct scatter; subworkflows; and when written as one plain parameter reference.

Rejected, always with the feature name and where it was found: any JavaScript (InlineJavascriptRequirement, ${...} bodies, and anything inside $(...) beyond a plain inputs.x/runtime.x reference), ExpressionTool and Operation, nested_crossproduct and flat_crossproduct, streamable inputs and outputs, secondaryFiles, outputEval, ShellCommandRequirement, SchemaDefRequirement, InitialWorkDirRequirement, stdin redirection, record and enum schemas, and CWL v1.2 loops.

DockerRequirement is neither rejected nor honoured: it is recorded as the required capability docker on the job — so only a manager declaring that capability will claim it — and reported as a warning, because httk₂ runs the command directly and does not pull, build or enter the image.

Attributes

DEFAULT_PLACEMENT
FILES_DIRECTORY
WORKFLOW
INITIAL_STEP
PACKAGE
RUNNER
DOCUMENT_FILE
INPUTS_FILE
STAGED_DIRECTORY
DOCKER_CAPABILITY

Exceptions

UnsupportedCwlError	One CWL feature outside the subset httk₂ executes.
CwlImportError	A CWL document that cannot be read, parsed or staged at all.

Classes

Workspace	One self-contained httk workflow filesystem workspace.
ScaffoldedJob	One job this module submitted, and everything needed to look at it again.
CwlNotes	What one normalization pass accumulates besides the plan itself.
ImportedCwl	One imported CWL document: the job, the plan, and what was noted.

Functions

submit_integration_job(, ...)	Build one payload for a packaged integration runner and submit it.
load_cwl_plan(→ tuple[dict[str, object], CwlNotes])	Parse, check and normalize one CWL document into the plan httk executes.
load_cwl_inputs(→ dict[str, object])	Read one CWL input object, as YAML when a YAML reader is installed.
stage_cwl_inputs(→ dict[str, object])	Return the input object with every File and Directory staged in the payload.
import_cwl(→ ImportedCwl)	Import one CWL document and its input object as one submitted job.

Package Contents

class httk.workflow.compat.cwl.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.cwl.DEFAULT_PLACEMENT = 'jobs'

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

class httk.workflow.compat.cwl.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.cwl.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.cwl.WORKFLOW = 'cwl.workflow'

httk.workflow.compat.cwl.INITIAL_STEP = 'start'

httk.workflow.compat.cwl.PACKAGE = 'httk.workflow.compat.cwl'

httk.workflow.compat.cwl.RUNNER = 'cwl_runner.py'

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

httk.workflow.compat.cwl.INPUTS_FILE = 'files/inputs.json'

httk.workflow.compat.cwl.STAGED_DIRECTORY = 'files/inputs'

httk.workflow.compat.cwl.DOCKER_CAPABILITY = 'docker'

exception httk.workflow.compat.cwl.UnsupportedCwlError

Bases: ValueError

One CWL feature outside the subset httk₂ executes.

exception httk.workflow.compat.cwl.CwlImportError

Bases: ValueError

A CWL document that cannot be read, parsed or staged at all.

class httk.workflow.compat.cwl.CwlNotes

What one normalization pass accumulates besides the plan itself.

A note is never a refusal: everything here was accepted, and every entry is something an operator should nevertheless be told — a container that will not be entered, a hint that was dropped — or something the job must carry, like the capabilities a document implies.

warnings: list[str] = []

capabilities: set[str]

class httk.workflow.compat.cwl.ImportedCwl

One imported CWL document: the job, the plan, and what was noted.

job: httk.workflow.compat._integration.ScaffoldedJob

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

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

warnings: tuple[str, Ellipsis]

httk.workflow.compat.cwl.load_cwl_plan(workflow_path: str | os.PathLike[str]) → tuple[dict[str, object], CwlNotes]

Parse, check and normalize one CWL document into the plan httk executes.

httk.workflow.compat.cwl.load_cwl_inputs(inputs_path: str | os.PathLike[str]) → dict[str, object]

Read one CWL input object, as YAML when a YAML reader is installed.

httk.workflow.compat.cwl.stage_cwl_inputs(values: collections.abc.Mapping[str, object], base: pathlib.Path, documents: dict[str, str], files: dict[str, str | os.PathLike[str]]) → dict[str, object]

Return the input object with every File and Directory staged in the payload.

A staged entry carries a payload-relative payload member instead of a path, so the job is self-contained: it travels to another machine with the files its first tool reads.

httk.workflow.compat.cwl.import_cwl(workspace: httk.workflow.Workspace, workflow_path: str | os.PathLike[str], inputs_path: str | os.PathLike[str], *, placement: str | pathlib.PurePosixPath = DEFAULT_PLACEMENT, tag: str | None = None, name: str | None = None, priority: int | None = None, data_mode: Literal['none', 'transactional'] = 'none', maximum_attempts: int | None = 3, timeout: float | None = None) → ImportedCwl

Import one CWL document and its input object as one submitted job.

The document is parsed with cwl-utils, checked against the supported subset, normalized into one self-contained JSON plan with every run: reference inlined, and staged in the payload together with the input object and every local file it names. The job runs the packaged cwl_runner.py through the reserved installed form, so no runner file is written per workflow.

Anything outside the subset raises UnsupportedCwlError naming the feature and where it was found; anything accepted with a caveat — a DockerRequirement, an ignored hint — is reported in warnings.