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.

StatusSnapshot `dataclass` ¶

StatusSnapshot(
    *,
    resolve,
    fs,
    content,
    header,
    generation,
    render,
    strip,
    comparison,
    plan,
    patch,
    write,
)

Immutable snapshot of per-axis pipeline status.

A StatusSnapshot captures the durable result-facing status values from a mutable ProcessingStatus instance without retaining the mutable source object. It is intended for post-run result models that should remain stable after the pipeline context continues to mutate or is released.

Attributes:

Name	Type	Description
`resolve`	`ResolveStatus`	Snapshot of file-type resolution status.
`fs`	`FsStatus`	Snapshot of file system status.
`content`	`ContentStatus`	Snapshot of content-level status.
`header`	`HeaderStatus`	Snapshot of header detection and parsing status.
`generation`	`GenerationStatus`	Snapshot of header generation status.
`render`	`RenderStatus`	Snapshot of header rendering status.
`strip`	`StripStatus`	Snapshot of header stripping status.
`comparison`	`ComparisonStatus`	Snapshot of comparison status.
`plan`	`PlanStatus`	Snapshot of update-planning status.
`patch`	`PatchStatus`	Snapshot of patch generation status.
`write`	`WriteStatus`	Snapshot of write status.

from_status `classmethod` ¶

from_status(status)

Create an immutable status snapshot from mutable pipeline status.

Parameters:

Name	Type	Description	Default
`status`	`ProcessingStatus`	Mutable status object to snapshot.	required

Returns:

Type	Description
`StatusSnapshot`	Detached immutable status snapshot containing the current per-axis values.

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

@classmethod
def from_status(cls, status: ProcessingStatus) -> StatusSnapshot:
    """Create an immutable status snapshot from mutable pipeline status.

    Args:
        status: Mutable status object to snapshot.

    Returns:
        Detached immutable status snapshot containing the current per-axis values.
    """
    return cls(
        resolve=status.resolve,
        fs=status.fs,
        content=status.content,
        header=status.header,
        generation=status.generation,
        render=status.render,
        strip=status.strip,
        comparison=status.comparison,
        plan=status.plan,
        patch=status.patch,
        write=status.write,
    )

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.

Raises:

Type	Description
`RuntimeError`	if an invalid `Axis` value has been given (defensive behavior).

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.

    Raises:
        RuntimeError: if an invalid `Axis` value has been given (defensive behavior).
    """
    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
        case _:  # pragma: no cover - defensive against invalid untyped callers
            raise RuntimeError(f"Invalid axis: {axis!r}")

to_dict ¶

to_dict()

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

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.

    Returns:
        Mapping from axis name to its status payload.
    """
    data: dict[str, AxisStatusPayload] = {}
    for item in fields(self):
        axis_name: str = item.name
        status: BaseStatus = getattr(self, axis_name)
        data[axis_name] = {
            "axis": axis_name,
            "name": status.name,
            "label": status.value,
        }
    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

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.

Raises:

Type	Description
`RuntimeError`	if an invalid `Axis` value has been given (defensive behavior).

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.

    Raises:
        RuntimeError: if an invalid `Axis` value has been given (defensive behavior).
    """
    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
        case _:  # pragma: no cover - defensive against invalid untyped callers
            raise RuntimeError(f"Invalid axis: {axis!r}")

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.

topmark.pipeline.context.status ¶

AxisStatusPayload ¶

StatusSnapshot dataclass ¶

from_status classmethod ¶

get ¶

to_dict ¶

has_write_outcome ¶

ProcessingStatus dataclass ¶

get ¶

reset ¶

to_dict ¶

has_write_outcome ¶

StepStatus dataclass ¶

StatusSnapshot `dataclass` ¶

from_status `classmethod` ¶

ProcessingStatus `dataclass` ¶

StepStatus `dataclass` ¶