httk.workflow.compat.pwd ======================== .. py:module:: httk.workflow.compat.pwd .. autoapi-nested-parse:: 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. .. code-block:: python 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 :data:`~httk.workflow.models.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 ---------- .. autoapisummary:: httk.workflow.compat.pwd.MAXIMUM_INPUTS_BYTES httk.workflow.compat.pwd.DEFAULT_PLACEMENT httk.workflow.compat.pwd.FILES_DIRECTORY httk.workflow.compat.pwd.WORKFLOW httk.workflow.compat.pwd.INITIAL_STEP httk.workflow.compat.pwd.PACKAGE httk.workflow.compat.pwd.RUNNER httk.workflow.compat.pwd.KNOWN_VERSIONS httk.workflow.compat.pwd.DOCUMENT_FILE httk.workflow.compat.pwd.DEFAULT_MAXIMUM_EMBEDDED_BYTES Exceptions ---------- .. autoapisummary:: httk.workflow.compat.pwd.PwdFormatError Classes ------- .. autoapisummary:: httk.workflow.compat.pwd.Workspace httk.workflow.compat.pwd.ScaffoldedJob httk.workflow.compat.pwd.PwdDocument Functions --------- .. autoapisummary:: httk.workflow.compat.pwd.submit_integration_job httk.workflow.compat.pwd.load_pwd_document httk.workflow.compat.pwd.validate_pwd_document httk.workflow.compat.pwd.import_pwd Package Contents ---------------- .. py:class:: 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. .. py:attribute:: root .. py:attribute:: control .. py:attribute:: runners .. py:attribute:: durable :value: True .. py:attribute:: format .. py:attribute:: core_profile .. py:attribute:: extensions .. py:attribute:: workspace_id :value: '' .. py:method:: initialize(root: str | os.PathLike[str], *, extensions: collections.abc.Iterable[str] = (), durable: bool = True, policy: collections.abc.Mapping[str, object] | None = None) -> Workspace :classmethod: Create and return a new workspace. .. py:property:: policy :type: httk.workflow.models.WorkspacePolicy Return the shared tunables this workspace publishes to every attacher. .. py:property:: visibility_deadline :type: float Return how long a metadata visibility retry may keep probing. .. py:method:: 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. .. py:property:: settings :type: dict[str, object] Return this workspace's application settings, a flat dotted map. Application settings are distinct from :attr:`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 :meth:`~httk.workflow.sdk.Attempt.setting`. A workspace written before the section existed reads as an empty map. .. py:method:: 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 :meth:`set_policy` uses: an exclusively created temporary and a rename, so a reader never sees a torn object, and last writer wins. .. py:method:: unset_setting(key: str) -> dict[str, object] Remove one application setting, refusing one that is not set. .. py:method:: 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. .. py:method:: open_journal_writer(*, writer_id: str | None = None) -> httk.workflow.journal.JournalWriter Open one exclusive journal writer configured by workspace policy. .. py:method:: check(*, repair: bool = False, quarantine_unrepairable: bool = False) -> httk.workflow.fsck.FsckReport Verify that every marker resolves to its journal frame. .. py:method:: collect_garbage(*, dry_run: bool = False, now: float | None = None) -> httk.workflow.gc.GcReport Collect the disk this workspace's retention policy permits freeing. .. py:method:: upgrade(extensions: collections.abc.Iterable[str]) -> frozenset[str] Enable extensions that have an implemented in-place migration. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: import_bundle(bundle: str | os.PathLike[str]) -> dict[str, object] Import a validated detached transfer bundle. .. py:method:: acknowledge_transfer(acknowledgement: collections.abc.Mapping[str, object]) -> pathlib.Path Retire a source bundle after destination acknowledgement. .. py:method:: recover_transfers() -> list[dict[str, object]] Recover or report interrupted detached-transfer publications. .. py:method:: state_directory(kind: str, placement: pathlib.PurePosixPath) -> pathlib.Path .. py:method:: marker_path(kind: str, placement: pathlib.PurePosixPath, job_key: str, priority: int, generation: int, record_ref: str) -> pathlib.Path .. py:method:: payload_path(placement: pathlib.PurePosixPath, job_key: str) -> pathlib.Path .. py:method:: 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. .. py:method:: 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 :class:`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 :class:`~httk.workflow.workspace.MarkerStream` instead. .. py:method:: scan_markers(kinds: collections.abc.Iterable[str] | None = None) -> collections.abc.Iterable[httk.workflow.models.Marker] .. py:method:: 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. .. py:method:: invalidate_marker_index() -> None Drop the cached job-id index, so the next lookup rebuilds it. .. py:method:: find_markers(job_key: str, kinds: collections.abc.Iterable[str] | None = None) -> list[httk.workflow.models.Marker] .. py:method:: 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. .. py:method:: 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. .. py:method:: load_job(marker: httk.workflow.models.Marker) -> httk.workflow.models.JobDefinition .. py:method:: read_state(marker: httk.workflow.models.Marker) -> dict[str, Any] .. py:method:: 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. .. py:method:: 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 :meth:`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. .. py:method:: 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. .. py:method:: validate_job_payload(marker: httk.workflow.models.Marker) -> httk.workflow.models.JobDefinition Perform manager-side immutable submission validation. .. py:method:: quarantine(path: pathlib.Path, *, reason: str) -> pathlib.Path Move a malformed protocol entry into the canonical quarantine. .. py:method:: payload_digest(marker: httk.workflow.models.Marker) -> str Return the digest of one payload, ignoring runner-private entries. .. py:method:: publish_request(request: collections.abc.Mapping[str, object]) -> pathlib.Path Atomically publish an operator request. .. py:data:: MAXIMUM_INPUTS_BYTES :value: 262144 .. py:data:: DEFAULT_PLACEMENT :value: 'jobs' .. py:data:: FILES_DIRECTORY :value: 'files' .. py:class:: ScaffoldedJob One job this module submitted, and everything needed to look at it again. .. py:attribute:: job_id :type: str .. py:attribute:: job_key :type: str .. py:attribute:: tag :type: str | None .. py:attribute:: placement :type: pathlib.PurePosixPath .. py:attribute:: payload :type: pathlib.Path .. py:attribute:: marker :type: pathlib.Path .. py:attribute:: workflow :type: str .. py:attribute:: initial_step :type: str .. py:attribute:: template :type: str .. py:attribute:: runner :type: collections.abc.Mapping[str, object] .. py:method:: as_mapping() -> dict[str, object] Return the machine-readable report of this job. .. py:function:: 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 :func:`~httk.workflow.scaffold.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. .. py:data:: WORKFLOW :value: 'pwd.workflow' .. py:data:: INITIAL_STEP :value: 'execute' .. py:data:: PACKAGE :value: 'httk.workflow.compat.pwd' .. py:data:: RUNNER :value: 'pwd_runner.py' .. py:data:: KNOWN_VERSIONS :value: ('0.1.0',) .. py:data:: DOCUMENT_FILE :value: 'files/pwd.json' .. py:data:: DEFAULT_MAXIMUM_EMBEDDED_BYTES :value: 131072 .. py:exception:: PwdFormatError Bases: :py:obj:`ValueError` A document that is not a Python Workflow Definition this importer accepts. .. py:class:: PwdDocument One validated PWD document, and the order its function nodes run in. .. py:attribute:: raw :type: collections.abc.Mapping[str, object] .. py:attribute:: nodes :type: collections.abc.Mapping[int, collections.abc.Mapping[str, object]] .. py:attribute:: edges :type: tuple[collections.abc.Mapping[str, object], Ellipsis] .. py:attribute:: order :type: tuple[int, Ellipsis] .. py:property:: version :type: str | None .. py:property:: functions :type: tuple[str, Ellipsis] Every ``module.function`` this document would import, in node order. .. py:property:: input_names :type: tuple[str, Ellipsis] The names of every input node, in node order. .. py:property:: output_names :type: tuple[str, Ellipsis] The names of every output node, in node order. .. py:function:: load_pwd_document(path: str | os.PathLike[str], *, allow_unknown_version: bool = False) -> PwdDocument Read and validate one PWD document from *path*. .. py:function:: 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. .. py:function:: 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.