httk.workflow.protocol

The language-neutral filesystem protocol surface of httk₂ workflows.

This module is the one deliberate public home of the on-disk protocol: the shapes, validators, and primitives an implementation in any language reads and writes to interoperate through a workspace. The normative specification is the filesystem protocol reference in the httk-workflow documentation; everything named here is what that document describes, and an independent inspection or verification tool should be able to work from this namespace and that document alone.

Nothing here is manager bookkeeping, a subprocess wrapper, a CLI handler, or a scheduling pass — those live in their own modules and are not part of the protocol. The implementations are owned by the modules re-exported below (models, journal, transactions, and the runtime builders), which are internal detail from the protocol’s point of view; import the names from here.

Attributes

CARRIED_STATE_MEMBERS
CORE_PROFILE
CORE_STATE_KINDS
QUIESCENT_KINDS
READABLE_CORE_PROFILES
RUNNER_SOURCES
STATE_KINDS
SUPPORTED_EXTENSIONS
TERMINAL_KINDS
WITHDRAWN_EXTENSIONS
JoinCondition
OutcomeAction

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

JournalFrame	One intact frame found by walking a segment from its header.
RecordVerification	The outcome of reading one referenced frame without raising.
Failure	One canonical structured failure record.
JobDefinition
Marker
RetentionPolicy	How long a workspace keeps the history it is allowed to collect.
RetryPolicy	The attempt budgets of one job and the failures it retries within them.
StateFrame	The members of one state frame, typed for the manager that uses them.
WorkspacePolicy	The tunables every implementation attaching to one workspace shares.
AttemptContext	The immutable identity and restart evidence for one running attempt.
ChildReference	The stable identity used in a native join.
JobSpec	Values needed to create an immutable native job definition.
OutcomeDraft	One unpublished outcome bundle below an attempt control directory.
ReplayableWorkdirBatch	A sealed, idempotently replayable set of workdir changes.
RunLog	Append-only structured application evidence in a workdir.
TransactionBuilder	Build a validated replayable transaction manifest.
MarkerFault	One state entry shaped like a marker that cannot be interpreted.

Functions

encode_record_ref(→ str)	Encode one canonical hwref-v1 reference.
iter_journal_frames(...)	Yield every intact frame of every segment of every writer.
iter_segment_frames(...)	Yield every intact frame of one segment.
parse_record_ref(→ tuple[str, int, int, int, str])	Parse one canonical hwref-v1 reference.
read_record(→ dict[str, Any])	Read and verify a journal record, retrying visibility-short reads.
segment_path(→ pathlib.Path)	Return the segment file one record reference names.
verify_record(→ RecordVerification)	Read one referenced frame, reporting damage rather than raising.
canonical_uuid(→ str)
is_payload_private(→ bool)	Report whether one payload entry name is runner-private scratch.
job_digest(→ str)	Return the normative immutable job digest of stored job.json bytes.
make_job_key(→ str)
marker_basename(→ str)
normalize_placement(→ pathlib.PurePosixPath)
parse_job_key(→ tuple[str | None, str])
parse_package_runner(→ tuple[str, ...)	Split the reserved pkg:<module>/<resource> installed runner form.
to_base36(→ str)
validate_attempt_control(→ str)	Validate one attempt-control directory name read from a state frame.
validate_declaration_name(→ str)	Validate one declaration name of a job.
validate_declarations(→ dict[str, dict[str, object]])	Validate the optional declarations object of a job.
validate_failure(→ Failure)	Validate one published failure object.
validate_inputs(→ dict[str, object])	Validate the optional application-defined inputs object of a job.
validate_label(→ str)
validate_runner_path(→ pathlib.PurePosixPath)	Validate runner.path against the root implied by runner.source.
validate_sha256(→ str)	Validate one lowercase hexadecimal SHA-256 digest string.
validate_step(→ str)
join_mapping(→ dict[str, object])	Return the validated join member of one waiting outcome.
prepare_job_payload(→ httk.workflow.models.JobDefinition)	Create and validate job.json in an existing prepared payload.
replay_transaction(→ bool)	Idempotently apply one published transaction.

Module Contents

exception httk.workflow.protocol.FormatError

Bases: WorkflowError, ValueError

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

exception httk.workflow.protocol.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.protocol.TransactionError

Bases: WorkflowError

A transactional-data manifest cannot be safely replayed.

exception httk.workflow.protocol.TransitionLostError

Bases: WorkflowError

Another actor committed a transition from the expected marker.

exception httk.workflow.protocol.UnsupportedExtensionError

Bases: WorkflowError

A workspace requires an extension this implementation does not support.

exception httk.workflow.protocol.WorkflowError

Bases: Exception

Base class for workflow protocol failures.

exception httk.workflow.protocol.WorkspaceCorruptionError

Bases: WorkflowError

The authoritative filesystem state is internally inconsistent.

exception httk.workflow.protocol.WorkspaceUnavailableError

Bases: WorkflowError

The workspace cannot currently provide a coherent protocol view.

class httk.workflow.protocol.JournalFrame

One intact frame found by walking a segment from its header.

record_ref: str

writer_id: str

segment: int

offset: int

frame: dict[str, Any]

class httk.workflow.protocol.RecordVerification

The outcome of reading one referenced frame without raising.

record_ref: str

frame: dict[str, Any] | None

problem: str | None

detail: str

property ok: bool

Report whether the referenced frame was read and verified.

httk.workflow.protocol.encode_record_ref(writer_id: str, segment: int, offset: int, length: int, checksum: bytes) → str

Encode one canonical hwref-v1 reference.

httk.workflow.protocol.iter_journal_frames(control_dir: pathlib.Path) → collections.abc.Iterator[JournalFrame]

Yield every intact frame of every segment of every writer.

httk.workflow.protocol.iter_segment_frames(path: pathlib.Path, writer_id: str, segment: int) → collections.abc.Iterator[JournalFrame]

Yield every intact frame of one segment.

The walk is deliberately forgiving. A damaged frame whose framing is still intact is skipped, because the frames behind it remain locatable and are exactly what a repair is looking for; a torn or partially visible tail is the normal state of a segment a live writer is appending to and simply ends the walk.

httk.workflow.protocol.parse_record_ref(record_ref: str) → tuple[str, int, int, int, str]

Parse one canonical hwref-v1 reference.

httk.workflow.protocol.read_record(control_dir: pathlib.Path, record_ref: str, *, deadline_seconds: float | None = None) → dict[str, Any]

Read and verify a journal record, retrying visibility-short reads.

A frame that is absent, short, or undecodable may be an extension of a segment that has not reached this client yet, so it is retried with bounded backoff until deadline_seconds — the workspace’s configured visibility deadline — expires. Damage that no amount of waiting can repair is reported at once.

httk.workflow.protocol.segment_path(control_dir: pathlib.Path, writer_id: str, segment: int) → pathlib.Path

Return the segment file one record reference names.

httk.workflow.protocol.verify_record(control_dir: pathlib.Path, record_ref: str, *, deadline_seconds: float | None = None) → RecordVerification

Read one referenced frame, reporting damage rather than raising.

This is the reading half of a workspace check: it distinguishes a segment that is gone from one that is truncated, corrupt, or simply not holding the frame the reference names, which is what a repair decision needs.

httk.workflow.protocol.CARRIED_STATE_MEMBERS = ('step', 'activation_id', 'activation_ordinal', 'attempt_id', 'attempt_ordinal',...

httk.workflow.protocol.CORE_PROFILE = 'core-v2'

httk.workflow.protocol.CORE_STATE_KINDS

httk.workflow.protocol.QUIESCENT_KINDS

httk.workflow.protocol.READABLE_CORE_PROFILES

httk.workflow.protocol.RUNNER_SOURCES

httk.workflow.protocol.STATE_KINDS = ('submitted', 'ready', 'claimed', 'running', 'committing', 'cancelling', 'relocating',...

httk.workflow.protocol.SUPPORTED_EXTENSIONS

httk.workflow.protocol.TERMINAL_KINDS

httk.workflow.protocol.WITHDRAWN_EXTENSIONS

class httk.workflow.protocol.Failure

One canonical structured failure record.

Every failure published by a runner, a bridge, or the manager itself uses exactly this shape: a stable machine code, one human message, optional structured details, and the advisory retryable flag. Retry policy is keyed on code alone; retryable is recorded evidence and never a manager decision.

code: str

message: str

details: collections.abc.Mapping[str, object] | None = None

retryable: bool = False

as_mapping() → dict[str, object]

Return the canonical JSON representation of this failure.

class httk.workflow.protocol.JobDefinition

id: str

tag: str | None

name: str

workflow: str

runner_backend: str

runner_source: str

runner_path: pathlib.PurePosixPath

runner_sha256: str | None

runner_arguments: tuple[str, Ellipsis]

workdir_mode: str

workdir_path: pathlib.PurePosixPath

data_mode: str

initial_step: str

priority: int

claim_pool: str

required_capabilities: frozenset[str]

retry_policy: RetryPolicy

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

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

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

parent: collections.abc.Mapping[str, object] | None

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

stored_digest: str | None = None

property job_key: str

property digest: str

Return the immutable job digest.

Normatively the digest is job_digest() over the stored job.json file bytes exactly as submitted, which is what every definition read through from_bytes() carries. A definition composed in memory has no stored bytes yet, so its canonical serialization is hashed instead; the two agree as soon as that serialization is what gets written.

classmethod from_bytes(data: bytes, *, name: str = 'job.json') → JobDefinition

Parse stored job.json bytes, pinning the normative job digest.

classmethod from_path(path: pathlib.Path) → JobDefinition

Read one stored job.json, pinning the normative job digest.

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

class httk.workflow.protocol.Marker

kind: str

placement: pathlib.PurePosixPath

job_key: str

priority: int

generation: int

record_ref: str

path: pathlib.Path

property job_id: str

classmethod from_path(state_root: pathlib.Path, path: pathlib.Path) → Marker

class httk.workflow.protocol.RetentionPolicy

How long a workspace keeps the history it is allowed to collect.

Every member is optional and means “no configured limit” when absent. The collector that acts on these numbers is a separate concern; the workspace only carries them so that every implementation attaching to it agrees on what may be removed and when.

attempt_control_days: float | None = None

journal_days: float | None = None

trash_days: float | None = None

classmethod from_mapping(value: object, name: str = 'policy.retention') → RetentionPolicy

as_mapping() → dict[str, object]

Return the JSON representation, omitting unconfigured limits.

class httk.workflow.protocol.RetryPolicy

The attempt budgets of one job and the failures it retries within them.

Two independent rules make a failure retry-eligible, and both are bounded by exactly the same budgets:

• retry_on lists failure codes. A manager-detected failure — a lost lease, a process failure, an unusable outcome — is retried when its code appears in this set.

• A runner-declared failure published with retryable: true is retried whether or not its code appears in retry_on, because the runner that produced the failure is the authority on whether repeating the attempt can help.

Neither rule can exceed maximum_attempts_per_activation or maximum_total_attempts: an exhausted budget always ends the job.

maximum_attempts_per_activation: int | None

maximum_total_attempts: int | None

maximum_activations: int | None

retry_on: frozenset[str]

classmethod from_mapping(value: object) → RetryPolicy

class httk.workflow.protocol.StateFrame

The members of one state frame, typed for the manager that uses them.

The frame is held exactly as it is on disk, so every member round-trips verbatim — including one written by a newer implementation, by an enabled extension, or by a workspace older than this code. What this implementation reads and writes goes through the typed accessors and through of(), so a mistyped member name is a type error at the call site rather than a silently defaulted value at runtime.

The envelope members format, workspace_id, job_id, kind, state_generation, and their siblings belong to the transition that publishes a frame and are supplied by the workspace, never here.

members: collections.abc.Mapping[str, Any]

classmethod from_mapping(value: object, name: str = 'state frame') → StateFrame

Read one stored state frame, keeping every member verbatim.

as_mapping() → dict[str, object]

Return the JSON representation, member for member.

has(name: str) → bool

Report whether the frame carries name at all, null included.

classmethod of(base: StateFrame | None = None, *, step: str = _UNSET, activation_id: str = _UNSET, activation_ordinal: int = _UNSET, attempt_id: str = _UNSET, attempt_ordinal: int = _UNSET, total_attempts: int = _UNSET, data_generation: int | None = _UNSET, join_summary: collections.abc.Sequence[object] | None = _UNSET, runner_steps: collections.abc.Sequence[str] = _UNSET, manager_id: str = _UNSET, writer_id: str = _UNSET, claim_id: str = _UNSET, attempt_control: str = _UNSET, lease_seconds: float = _UNSET, matched_pool: str = _UNSET, matched_capabilities: collections.abc.Sequence[str] = _UNSET, started_at: str = _UNSET, workdir: str = _UNSET, outcome_action: str = _UNSET, child_digests: collections.abc.Mapping[str, str] = _UNSET, child_labels: collections.abc.Mapping[str, str] = _UNSET, next_step: str = _UNSET, join: collections.abc.Mapping[str, object] = _UNSET, pause: object = _UNSET, failure: collections.abc.Mapping[str, object] = _UNSET, job_digest: str = _UNSET, unclean_restart: bool = _UNSET, unsafe_persistent_takeover: bool = _UNSET, takeover_evidence: collections.abc.Mapping[str, object] = _UNSET, cancellation: collections.abc.Mapping[str, object] = _UNSET, previous_attempt_id: str | None = _UNSET, operator: object = _UNSET, operator_key: object = _UNSET, operator_reason: object = _UNSET, request_id: object = _UNSET, revival_hazard: collections.abc.Mapping[str, object] = _UNSET, reason: str = _UNSET) → StateFrame

Return base with the named members set, absent ones untouched.

Every member a manager writes is one declared keyword, so the complete vocabulary of a state frame is visible in one signature and no call site can invent a member by misspelling one. Passing None writes the JSON null the protocol distinguishes from an absent member.

carried() → StateFrame

Return only the members every transition of this activation repeats.

select(names: collections.abc.Sequence[str]) → StateFrame

Return only the named members this frame actually carries.

property step: str | None

property activation_id: str | None

property activation_ordinal: int | None

property attempt_id: str | None

property attempt_ordinal: int | None

property total_attempts: int | None

property data_generation: int | None

property join_summary: object

property manager_id: str | None

Return the owning manager, refusing anything that is not one.

The value is joined below managers/ to reach a heartbeat, so a frame that does not name a canonical manager UUID is a protocol violation rather than a path to try.

property attempt_control: str | None

Return the validated attempt-control component of this frame.

property lease_seconds: float | None

property started_at: str | None

property workdir: str | None

property next_step: str | None

property join: collections.abc.Mapping[str, object] | None

property failure: collections.abc.Mapping[str, object] | None

property pause: object

property cancellation: collections.abc.Mapping[str, object] | None

property child_digests: collections.abc.Mapping[str, object] | None

property previous_attempt_id: str | None

property unclean_restart: bool

property unsafe_persistent_takeover: bool

property reason: str | None

class httk.workflow.protocol.WorkspacePolicy

The tunables every implementation attaching to one workspace shares.

These are workspace properties rather than per-process options: two managers on different hosts must agree on how long a marker may take to become visible and on how long an unheartbeaten lease means anything. They live in format.json beside the format and profile declarations, and a workspace written before this section existed simply reads as the defaults.

visibility_deadline_seconds: float = 5.0

lease_seconds: float = 900.0

journal_segment_bytes: int = 67108864

retention: RetentionPolicy

classmethod from_mapping(value: object, name: str = 'policy') → WorkspacePolicy

Validate one complete policy object, filling in absent members.

as_mapping() → dict[str, object]

Return the complete JSON representation stored in format.json.

updated(changes: collections.abc.Mapping[str, object], name: str = 'policy') → WorkspacePolicy

Return this policy with changes applied and revalidated.

httk.workflow.protocol.canonical_uuid(value: object, name: str = 'id') → str

httk.workflow.protocol.is_payload_private(name: str) → bool

Report whether one payload entry name is runner-private scratch.

A runner-private entry is excluded from every payload digest, so publishing an outcome or writing job state can never change the digest of a payload that a manager, a transfer, or a registration check must still recognize.

httk.workflow.protocol.job_digest(data: bytes) → str

Return the normative immutable job digest of stored job.json bytes.

The digest of a job is the SHA-256 over the job.json file bytes exactly as submitted. Nothing rewrites or renormalizes those bytes, so the digest is reproducible by any implementation with only a hash utility.

httk.workflow.protocol.make_job_key(job_id: str, tag: str | None) → str

httk.workflow.protocol.marker_basename(job_key: str, priority: int, generation: int, record_ref: str) → str

httk.workflow.protocol.normalize_placement(value: str | pathlib.PurePosixPath) → pathlib.PurePosixPath

httk.workflow.protocol.parse_job_key(value: str) → tuple[str | None, str]

httk.workflow.protocol.parse_package_runner(value: str) → tuple[str, pathlib.PurePosixPath] | None

Split the reserved pkg:<module>/<resource> installed runner form.

Return None when value is an ordinary relative runner path, so callers can treat the reserved form as one alternative spelling of runner.path rather than as a separate protocol member.

httk.workflow.protocol.to_base36(value: int) → str

httk.workflow.protocol.validate_attempt_control(value: object, name: str = 'attempt_control') → str

Validate one attempt-control directory name read from a state frame.

The name is joined below a job payload to reach the control directory of an attempt, so it is exactly one relative component of the canonical .httk-attempt.<attempt-id> shape. Validating it before it is joined is what keeps a hostile or damaged frame from naming a path outside the job.

httk.workflow.protocol.validate_declaration_name(value: object, name: str = 'declaration name') → str

Validate one declaration name of a job.

The name keys the declarations object of job.json and is also the basename of the runtime-refined document below .httk-job/declarations/, so it must be a safe single path component and nothing else.

httk.workflow.protocol.validate_declarations(value: object, name: str = 'declarations') → dict[str, dict[str, object]]

Validate the optional declarations object of a job.

Each member is one workflow-declaration document carried verbatim: the protocol checks that a declaration is a JSON object and never looks inside it, because what the members mean is owned by the vocabulary the document names itself — the OPTIMADE workflow-declaration work is standardizing exactly that, and an engine that reinterpreted it would only be able to disagree with it. The bytes live in job.json and are therefore covered by the immutable job digest like every other member.

httk.workflow.protocol.validate_failure(value: object, name: str = 'failure') → Failure

Validate one published failure object.

httk.workflow.protocol.validate_inputs(value: object, name: str = 'inputs') → dict[str, object]

Validate the optional application-defined inputs object of a job.

The member is opaque to the protocol: only its shape, its key syntax, and its serialized size are checked. Its bytes are part of job.json and are therefore covered by the immutable job digest like every other member.

httk.workflow.protocol.validate_label(value: object, name: str) → str

httk.workflow.protocol.validate_runner_path(value: object, source: str) → pathlib.PurePosixPath

Validate runner.path against the root implied by runner.source.

Every source resolves the same relative path below a different root: the job payload, the workspace runner store, or one configured installed-runner search path. The path must therefore stay below its root under every source, and only an installed runner may use the reserved pkg: form.

httk.workflow.protocol.validate_sha256(value: object, name: str) → str

Validate one lowercase hexadecimal SHA-256 digest string.

httk.workflow.protocol.validate_step(value: object, name: str = 'step') → str

class httk.workflow.protocol.AttemptContext

The immutable identity and restart evidence for one running attempt.

workspace_id: str

job_id: str

job_key: str

placement: str

step: str

activation_id: str

attempt_id: str

activation_ordinal: int | None

attempt_ordinal: int | None

total_attempts: int | None

is_restart: bool

is_unclean_restart: bool

attempt_reason: str | None

previous_attempt_id: str | None

activation_reason: str | None

workdir_mode: str | None

workdir_reused: bool

unsafe_persistent_takeover: bool

data_generation: int | None

durable: bool

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

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

join: object

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

classmethod read(path: str | os.PathLike[str]) → Self

Read and validate a manager-written attempt context.

class httk.workflow.protocol.ChildReference

The stable identity used in a native join.

workspace_id: str

job_id: str

job_key: str

placement_hint: str

as_mapping() → dict[str, str]

class httk.workflow.protocol.JobSpec

Values needed to create an immutable native job definition.

A payload runner is a file inside the job payload. A workspace or installed runner lives outside the payload and must therefore pin its own runner_sha256, which is how one published runner serves a whole campaign of jobs without being copied per job.

name: str

workflow: str

runner_path: str

initial_step: str = 'start'

tag: str | None = None

job_id: str | None = None

runner_backend: str = 'path'

runner_source: Literal['payload', 'workspace', 'installed'] = 'payload'

runner_sha256: str | None = None

runner_arguments: tuple[str, Ellipsis] = ()

workdir_mode: Literal['persistent', 'isolated'] = 'persistent'

workdir_path: str = 'run'

data_mode: Literal['none', 'transactional'] = 'none'

priority: int = 500

claim_pool: str = 'default'

required_capabilities: tuple[str, Ellipsis] = ()

maximum_attempts_per_activation: int | None = None

maximum_total_attempts: int | None = None

maximum_activations: int | None = None

retry_on: tuple[str, Ellipsis] = ()

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

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

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

as_mapping(*, parent: collections.abc.Mapping[str, object] | None = None) → dict[str, object]

type httk.workflow.protocol.JoinCondition = Literal['all_succeeded', 'all_terminal', 'any_succeeded', 'at_least']

type httk.workflow.protocol.OutcomeAction = Literal['advance', 'retry', 'wait', 'succeed', 'fail', 'pause']

class httk.workflow.protocol.OutcomeDraft(context: httk.workflow.runtime.AttemptContext, control: pathlib.Path, root: pathlib.Path | None = None, *, durable: bool = False)

One unpublished outcome bundle below an attempt control directory.

The draft is the single place that writes the protocol shapes of an outcome: its transaction, its spawn set, and the atomic rename that publishes it. It is bound to nothing but the attempt identity and the control directory, so the authoring SDK and the Bash bridge publish through exactly one implementation.

context

control

durable = False

root

transaction() → TransactionBuilder

add_child(payload: str | os.PathLike[str], placement: str | pathlib.PurePosixPath, *, label: str | None = None) → ChildReference

Register one prepared payload directory as a child of this outcome.

add_child_job(job: collections.abc.Mapping[str, object], placement: str | pathlib.PurePosixPath, *, label: str) → ChildReference

Register one synthesized child that needs no prepared payload.

A child whose runner lives outside the payload — a workspace or installed runner — is completely described by its job.json, so a partitioned campaign can spawn children without copying a payload tree per child.

property children: tuple[ChildReference, Ellipsis]

publish(action: OutcomeAction, *, next_step: str | None = None, priority: int | None = None, failure: collections.abc.Mapping[str, object] | None = None, retry: collections.abc.Mapping[str, object] | None = None, join: collections.abc.Mapping[str, object] | None = None, pause: collections.abc.Mapping[str, object] | None = None, message: str | None = None, expected_data_generation: int | None = None, runner_steps: collections.abc.Sequence[str] | None = None) → pathlib.Path

class httk.workflow.protocol.ReplayableWorkdirBatch(workdir: pathlib.Path, root: pathlib.Path, *, durable: bool = False)

A sealed, idempotently replayable set of workdir changes.

workdir

root

durable = False

transaction

classmethod create(workdir: str | os.PathLike[str], *, durable: bool = False) → ReplayableWorkdirBatch

seal() → pathlib.Path

commit() → pathlib.Path

static recover(workdir: str | os.PathLike[str], *, durable: bool = False) → tuple[pathlib.Path, Ellipsis]

class httk.workflow.protocol.RunLog(workdir: str | os.PathLike[str])

Append-only structured application evidence in a workdir.

path

append(kind: str, message: str, *, files: collections.abc.Sequence[str | os.PathLike[str]] = ()) → None

class httk.workflow.protocol.TransactionBuilder(root: pathlib.Path, *, expected_generation: int, durable: bool = False)

Build a validated replayable transaction manifest.

root

expected_generation

durable = False

classmethod resume(root: str | os.PathLike[str], *, expected_generation: int, durable: bool = False) → Self

Reattach to a transaction an earlier process of this attempt sealed.

A Bash runner publishes one outcome through many short-lived processes, so the staged manifest on disk — not any in-memory counter — is what carries the operations of a draft from one call to the next. Resuming reads it back, so appending an operation continues the same sequence and the same overlap checks as the process that staged the first one.

make_dir(operation_id: str, path: str | os.PathLike[str]) → None

put_file(operation_id: str, source: str | os.PathLike[str], path: str | os.PathLike[str]) → None

put_tree(operation_id: str, source: str | os.PathLike[str], path: str | os.PathLike[str], *, replace: bool = False) → None

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

seal() → pathlib.Path

httk.workflow.protocol.join_mapping(children: collections.abc.Sequence[ChildReference], condition: JoinCondition = 'all_succeeded', count: int | None = None, on_impossible_step: str | None = None) → dict[str, object]

Return the validated join member of one waiting outcome.

httk.workflow.protocol.prepare_job_payload(destination: str | os.PathLike[str], spec: JobSpec, *, parent: collections.abc.Mapping[str, object] | None = None, durable: bool = False) → httk.workflow.models.JobDefinition

Create and validate job.json in an existing prepared payload.

durable synchronizes the written job.json for a caller preparing a payload directly on durable storage; it defaults to False because a payload prepared here is not yet a workspace artifact, and its submission is what makes it authoritative and durable.

httk.workflow.protocol.replay_transaction(transaction_dir: pathlib.Path, data_dir: pathlib.Path, *, expected_generation: int, durable: bool = False) → bool

Idempotently apply one published transaction.

Returns whether the manifest contains operations and therefore advances the data generation.

When durable is set, every destination this replay installs and every directory whose entries it changes — including the parents that gained or lost a name, and the trash a removal moved into — is synchronized before this call returns. The manager relies on that ordering: it appends the destination state frame and renames the marker out of committing only after replay returns, so a committed transaction is on storage before the marker that claims it is.

class httk.workflow.protocol.MarkerFault

One state entry shaped like a marker that cannot be interpreted.

path: pathlib.Path

reason: str