httk.workflow.collecting

The results-collect contract: everything a data layer needs about one job.

Collecting is the read-only counterpart of running work. A manager decides what happens next; a job_records reports what already happened, once, per job that stopped, in a shape a data layer can store without knowing anything about markers, journals, or leases.

That shape — JobRecord — is the layering boundary of httk₂. httk-workflow has no database dependency and never will: it produces records, and something else consumes them. A consumer therefore reads results like this, and nothing in this module knows what store or load_vasp are:

for record in job_records(workspace):
    store.save(load_vasp(record))

Every member of a record is derived from exactly the authoritative state a manager reads — the marker below state/, the journal frames that marker’s chain names, and the immutable job.json — so a record never says anything the workspace does not. Two properties follow from that and are the reason this module exists at all:

• The executed code is pinned. A record carries the immutable job digest and the complete runner identity: executor, source, path, and the SHA-256 the job pinned for every runner that lives outside its payload. For a runner named by the reserved pkg: form the installed distribution and its version are reported as well, so a stored result names the software that produced it.

• Damage is reported, never guessed. A job whose journal chain is broken is still collected, with whatever remains readable and gaps set, because a result that exists must not become invisible just because part of its history did not survive.

job_records() is lazily evaluated over one scan of the workspace. By design it iterates jobs without materializing the workspace, and building one record reads only that job’s own payload and journal chain.

Attributes

COLLECT_FORMAT
COLLECT_FORMAT_VERSION
COLLECTABLE_KINDS
DEFAULT_COLLECT_STATES

Classes

CollectedJob	Represent one job after workflow collecting and provenance assembly.
JobRecord	Everything a data layer needs about one job that stopped.

Functions

module_distribution(module)	Return the (name, version) of the distribution installing module.
runner_provenance(job)	Return what installation provenance exists for one job's runner.
timeline(frames)	Return the activation and attempt timeline of one job, oldest first.
children_of(frames)	Return the labeled children one job registered, keyed by spawn label.
declarations_of(job, payload)	Return the declarations of one job and whether any observed one is lost.
record_of(workspace, marker)	Return the job_records record of the one job marker names.
collect_kinds(states)	Validate the requested state kinds against what collect may read.
job_records(workspace, *[, states, placement, on_skipped])	Yield one JobRecord per finished job of workspace.
collect(workspace, *[, states, placement, ...])	Collect records through registered or explicitly allowed job collectors.

Module Contents

httk.workflow.collecting.COLLECT_FORMAT = 'httk-workflow-collect'

httk.workflow.collecting.COLLECT_FORMAT_VERSION = 1

httk.workflow.collecting.COLLECTABLE_KINDS

httk.workflow.collecting.DEFAULT_COLLECT_STATES = ('succeeded',)

class httk.workflow.collecting.CollectedJob

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

httk.workflow.collecting.module_distribution(module)

Return the (name, version) of the distribution installing module.

The answer is read from installation metadata alone and never by importing anything: a module name comes out of an untrusted job.json, and importing it to ask which package it belongs to would execute code during a read-only job_records. A wheel installation is recognized by the module path recorded in its file list, and an editable installation by the source tree its direct_url.json names. Anything else — a module on PYTHONPATH that no installed distribution owns — is reported as unknown rather than guessed.

Parameters:

module (str) – Name the module whose installed distribution to locate.

Returns:

The distribution name and version, or None when ownership is unknown.

Return type:

tuple[str, str] | None

httk.workflow.collecting.runner_provenance(job)

Return what installation provenance exists for one job’s runner.

Only the reserved pkg:<module>/<resource> form of an installed runner resolves to a Python distribution, so every other runner reports None: a payload runner is pinned by the job digest, and a workspace or plain installed runner is pinned by runner.sha256 and nothing else is known about where it came from.

Parameters:

job (httk.workflow.models.JobDefinition) – Supply the validated job definition and runner identity.

Returns:

Installation metadata for a reserved package runner, or None.

Return type:

dict[str, object] | None

httk.workflow.collecting.timeline(frames)

Return the activation and attempt timeline of one job, oldest first.

The frames are exactly the ones the workspace’s job_frames reader walked, so this is a pure regrouping of recorded history: an activation is every consecutive frame sharing one activation_id, and an attempt is opened by the claimed frame that consumed a budget and closed by the first frame that reported how it ended. A frame the journal could not return sets gaps and is skipped, which keeps a job with a damaged history collectable instead of silent.

Parameters:

frames (collections.abc.Sequence[collections.abc.Mapping[str, Any]]) – Supply the journal frames read for one job.

Returns:

The oldest-first activation and attempt timeline with its damage flag.

Return type:

dict[str, object]

httk.workflow.collecting.children_of(frames)

Return the labeled children one job registered, keyed by spawn label.

A campaign therefore collects as a tree: every record names the children it spawned, and each of those is a job a consumer collects in its own right. A label is mandatory in core-v2, so an unlabeled child reference — only possible in a workspace written by an older profile — is left out rather than given an invented name. A label reused by a later activation names the child of the most recent spawn under it.

Parameters:

frames (collections.abc.Sequence[collections.abc.Mapping[str, Any]]) – Supply the state and journal frames for one job.

Returns:

Child records keyed by their spawn labels.

Return type:

dict[str, dict[str, object]]

class httk.workflow.collecting.JobRecord

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

Return the JSON representation of this record.

classmethod from_mapping(value)

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.collecting.declarations_of(job, payload)

Return the declarations of one job and whether any observed one is lost.

Every name either source knows appears exactly once. declared is the document job.json carried, and observed is the runtime-refined one the job wrote below .httk-job/declarations/; both are carried verbatim and reported side by side, because merging them would require understanding a vocabulary this module deliberately does not implement. An observed document that cannot be read is reported as None with the damage flag set, exactly like every other unreadable evidence a job_records still reports.

Parameters:

• job (httk.workflow.models.JobDefinition) – Supply the validated job whose declared documents are available.

• payload (pathlib.Path) – Locate the payload containing observed documents.

Returns:

The side-by-side declarations and whether reading observed data found damage.

Return type:

tuple[dict[str, dict[str, collections.abc.Mapping[str, object] | None]], bool]

httk.workflow.collecting.record_of(workspace, marker)

Return the job_records record of the one job marker names.

None means this job has no readable job.json and therefore no definition to report: the whole contract of a record is the validated job behind a result, so an unusable payload is reported through the module logger and left to a workspace tool instead of being described by guesswork.

Parameters:

• workspace (httk.workflow.workspace.Workspace) – Read the workspace containing the marked job.

• marker (httk.workflow.models.Marker) – Identify the stopped job to read.

Returns:

The validated job record, or None when its payload is unreadable.

Return type:

JobRecord | None

httk.workflow.collecting.collect_kinds(states)

Validate the requested state kinds against what collect may read.

Parameters:

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

Returns:

The distinct validated state kinds in request order.

Raises:

ValueError – If no state or an uncollectable state is requested.

Return type:

tuple[str, Ellipsis]

httk.workflow.collecting.job_records(workspace, *, states=DEFAULT_COLLECT_STATES, placement=None, on_skipped=None)

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.

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

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.