topmark.pipeline.context.status

topmark / pipeline / context / status

Status types and helpers for per-axis progress tracking in the TopMark pipeline.

This module defines the core dataclasses and typed payloads used to represent per-axis status within the TopMark processing pipeline. Each pipeline axis (resolve, fs, content, header, generation, render, strip, comparison, plan, patch, write) exposes its own status enum, and ProcessingStatus collects these into a single structure that serves as the authoritative source of truth for all status evaluation.

The supporting AxisStatusPayload provides a stable, JSON-serializable representation used in machine-readable output (--json / NDJSON), ensuring that external tools can reliably consume pipeline results without depending on internal enum details.

A small StepStatus value object is also provided for step-level diagnostics, enabling the runner and CLI to report per-step outcomes without exposing internal step implementation details.

Sections

AxisStatusPayload: TypedDict describing the serializable payload for a single axis.

ProcessingStatus: Aggregated structure holding the current status for all pipeline axes. Used by steps, the runner, and the CLI as the unified representation of pipeline progress and outcomes.

StepStatus: Lightweight pairing of a step name with its coarse status for diagnostic and summarization purposes.

This module contains no policy logic, no hint generation, and no coupling to file-type processors. It is intentionally pure and side-effect-free so that status evaluation remains predictable, testable, and import-safe.

AxisStatusPayload

Bases: TypedDict

Typed payload describing the status of a single pipeline axis.

Keys

axis: Machine-friendly axis name (for example, "fs" or "header"). name: Enum member name for the axis status (for example, "OK" or "MISSING"). label: Human-readable label for the status, derived from the enum's .value.

ProcessingStatus dataclass

ProcessingStatus(
    *,
    resolve=ResolveStatus.PENDING,
    fs=FsStatus.PENDING,
    content=ContentStatus.PENDING,
    header=HeaderStatus.PENDING,
    generation=GenerationStatus.PENDING,
    render=RenderStatus.PENDING,
    strip=StripStatus.PENDING,
    comparison=ComparisonStatus.PENDING,
    plan=PlanStatus.PENDING,
    patch=PatchStatus.PENDING,
    write=WriteStatus.PENDING,
)

Tracks the status of each processing phase for a single file.

Each attribute corresponds to a pipeline axis and is represented by a dedicated status enum. This dataclass is the single source of truth for per-axis status in the pipeline.

Attributes:

Name	Type	Description
resolve	ResolveStatus	Status of file-type resolution.
fs	FsStatus	File system status (existence, permissions, binary guard).
content	ContentStatus	Status of content-level checks (BOM, shebang, mixed newlines, readability).
header	HeaderStatus	Status of header detection and parsing.
generation	GenerationStatus	Status of header field/value generation.
render	RenderStatus	Status of header rendering for the active file type.
strip	StripStatus	Status of header stripping preparation and execution.
comparison	ComparisonStatus	Status of comparing original vs. updated content.
plan	PlanStatus	Status of planning updates prior to writing.
patch	PatchStatus	Status of patch generation.
write	WriteStatus	Status of writing changes back to the file system.

get

get(axis)

Get the status for a given Axis.

Parameters:

Name	Type	Description	Default
axis	Axis	The Axis we want to get the status for;	required

Returns:

Type	Description
BaseStatus	The status for the given Axis.

Source code in src/topmark/pipeline/context/status.py

def get(self, axis: Axis) -> BaseStatus:
    """Get the status for a given Axis.

    Args:
        axis: The Axis we want to get the status for;

    Returns:
        The status for the given Axis.
    """
    match axis:
        case Axis.RESOLVE:
            return self.resolve
        case Axis.FS:
            return self.fs
        case Axis.CONTENT:
            return self.content
        case Axis.HEADER:
            return self.header
        case Axis.GENERATION:
            return self.generation
        case Axis.RENDER:
            return self.render
        case Axis.STRIP:
            return self.strip
        case Axis.COMPARISON:
            return self.comparison
        case Axis.PLAN:
            return self.plan
        case Axis.PATCH:
            return self.patch
        case Axis.WRITE:
            return self.write

reset

reset()

Reset all status fields to PENDING.

This helper is mainly intended for tests or reuse of an existing ProcessingStatus instance.

Returns:

Name	Type	Description
None	None	The instance is mutated in place.

Source code in src/topmark/pipeline/context/status.py

def reset(self) -> None:
    """Reset all status fields to ``PENDING``.

    This helper is mainly intended for tests or reuse of an existing
    ``ProcessingStatus`` instance.

    Returns:
        None: The instance is mutated in place.
    """
    self.resolve = ResolveStatus.PENDING
    self.fs = FsStatus.PENDING
    self.content = ContentStatus.PENDING
    self.header = HeaderStatus.PENDING
    self.generation = GenerationStatus.PENDING
    self.render = RenderStatus.PENDING
    self.strip = StripStatus.PENDING
    self.comparison = ComparisonStatus.PENDING
    self.plan = PlanStatus.PENDING
    self.patch = PatchStatus.PENDING
    self.write = WriteStatus.PENDING

to_dict

to_dict()

Return axis → {axis, name, label} payload for all axes.

This mirrors AxisStatusPayload and is useful if you want a single, centralized representation of axis status data for machine-readable output.

Returns:

Type	Description
dict[str, AxisStatusPayload]	Mapping from axis name to its status payload.

Source code in src/topmark/pipeline/context/status.py

def to_dict(self) -> dict[str, AxisStatusPayload]:
    """Return axis → {axis, name, label} payload for all axes.

    This mirrors ``AxisStatusPayload`` and is useful if you want a single,
    centralized representation of axis status data for machine-readable output.

    Returns:
        Mapping from axis name to its status payload.
    """
    data: dict[str, AxisStatusPayload] = {}
    for axis in Axis:
        attr_name: str = axis.value  # e.g. "resolve"
        status: BaseStatus = getattr(self, attr_name)
        label: str = status.value  # ColoredStrEnum: (label, color)
        axis_name: str = axis.value
        data[axis_name] = {
            "axis": axis_name,
            "name": status.name,
            "label": label,
        }
    return data

has_write_outcome

has_write_outcome()

Return True if the write axis has reached a non-pending outcome.

Returns:

Type	Description
bool	True when the write status is anything other than PENDING.

Source code in src/topmark/pipeline/context/status.py

def has_write_outcome(self) -> bool:
    """Return True if the write axis has reached a non-pending outcome.

    Returns:
        `True` when the write status is anything other than `PENDING`.
    """
    return self.write is not WriteStatus.PENDING

StepStatus dataclass

StepStatus(step, status)

Lightweight pairing of a step name with its coarse status.

This structure is primarily used for diagnostics and summaries that need to report per-step outcomes.

Attributes:

Name	Type	Description
step	str	Name of the pipeline step.
status	BaseStatus	Coarse status for the step, typically derived from its primary axis.