topmark.pipeline.steps.renderer ¶

def may_proceed(
    self,
    ctx: ProcessingContext,
) -> bool:
    """Determine if processing can proceed to the render step.

    Processing can proceed if:
    - Field generation completed with ``GENERATED`` or ``NO_FIELDS``.

    Args:
        ctx: The processing context for the current file.

    Returns:
        True if processing can proceed to the render step, False otherwise.
    """
    if ctx.is_halted:
        return False

    outcome: bool = ctx.status.generation in {
        GenerationStatus.GENERATED,
        GenerationStatus.NO_FIELDS,
    }
    return outcome

def run(
    self,
    ctx: ProcessingContext,
) -> None:
    """Render the expected header text from ``ctx.views.build.selected``.

    Args:
        ctx: Mutable context with:
            * ``header_processor`` - strategy providing ``render_header_lines()``;
            * ``build.selected`` - expected fields (for ``GENERATED``);
            * ``image`` - file image view (for indentation preservation and newline style).

    Raises:
        RuntimeError: If header processor is not defined.

    Mutations:
        ProcessingContext: The same context with ``ctx.views.render`` populated depending on
        the generation status:
            * ``NO_FIELDS`` - renders a markers-only header when allowed by policy; otherwise
              sets ``ctx.views.render = RenderView(lines=None, block=None)`` and skips.
            * ``GENERATED`` - sets ``ctx.views.render.lines`` and ``ctx.views.render.block``.
              If the selected mapping is empty, produces a completed empty render defensively.
            * any other status - returns unchanged.

    Notes:
        This step mutates ``ctx`` in place and performs no I/O.
    """
    logger.debug("ctx: %s", ctx)

    if ctx.header_processor is None:
        raise RuntimeError("Header processor not defined")

    # Nothing to render when no fields were generated; short-circuit safely.
    if ctx.status.generation == GenerationStatus.NO_FIELDS:
        if allow_empty_header(ctx):
            # Render markers-only (empty header) to enable deterministic compare/update.
            rendered_lines: list[str] = ctx.header_processor.render_header_lines(
                header_values={},  # no fields
                config=ctx.config,
                newline_style=ctx.header_newline_style or ctx.newline_style,
                header_indent_override=None,
            )
            ctx.views.render = RenderView(lines=rendered_lines, block="".join(rendered_lines))
            ctx.status.render = RenderStatus.RENDERED
        else:
            # Make it explicit that there is no "expected header" to compare against.
            ctx.views.render = RenderView(lines=None, block=None)
            ctx.status.render = RenderStatus.SKIPPED
            ctx.request_halt(reason=f"{self.__class__.__name__} skipped.", at_step=self)
        return

    # Now ctx.status.generation == GenerationStatus.GENERATED

    # Use builder output as the source of fields
    header_view: HeaderView | None = ctx.views.header
    builder_view: BuilderView | None = ctx.views.build
    selected_fields: Mapping[str, str] = (
        builder_view.selected if builder_view and builder_view.selected else {}
    )
    fields: dict[str, str] = {
        key: normalize_semantic_newlines(value) for key, value in selected_fields.items()
    }

    # Compute header_indent_override using the header view and file lines
    # Preserve pre-prefix indentation when replacing an existing header
    # (spaces/tabs before the prefix, e.g., "    //"). This keeps nested/indented
    # headers (like JSONC inside an object) visually stable after replacement.
    header_indent_override: str | None = None
    if header_view and header_view.range is not None:
        start_idx: int
        _end_idx: int
        start_idx, _end_idx = header_view.range
        # Fetch the first header line via iterator without materializing the file
        first_line_iter: islice[str] = islice(ctx.iter_image_lines(), start_idx, start_idx + 1)
        first_line: str | None = next(first_line_iter, None)
        if first_line is not None:
            leading_ws: str = first_line[: len(first_line) - len(first_line.lstrip())]
            if leading_ws and first_line.lstrip().startswith(ctx.header_processor.line_prefix):
                header_indent_override = leading_ws

    validation: HeaderFieldValidationResult = ctx.header_processor.validate_header_fields(
        field_names=ctx.config.header_fields,
        header_values=fields,
        config=ctx.config,
        header_indent_override=header_indent_override,
    )
    if not validation.is_valid:
        for issue in validation.issues:
            ctx.diagnostics.add_error(
                f"{_diagnostic_field_label(issue)} {issue.target} violates "
                f"{_diagnostic_rule_label(issue.rule)}."
            )
        ctx.status.render = RenderStatus.FAILED
        ctx.hint(
            axis=Axis.RENDER,
            code=KnownCode.RENDER_INVALID_FIELDS,
            cluster=Cluster.ERROR,
            message=(
                "header field validation failed "
                f"({len(validation.issues)} violation"
                f"{'' if len(validation.issues) == 1 else 's'})"
            ),
            terminal=True,
            reason=RenderStatus.FAILED.value,
        )
        ctx.request_halt(reason="unsafe header field content", at_step=self)
        return

    # Defensive: if mapping is empty, produce an empty render
    if not fields:
        ctx.status.render = RenderStatus.RENDERED
        ctx.views.render = RenderView(lines=[], block="")
        return

    soft_overflow_fields: set[str] = set()
    rendered_lines = ctx.header_processor.render_header_lines(
        header_values=fields,
        config=ctx.config,
        newline_style=ctx.header_newline_style or ctx.newline_style,
        # keep any other overrides you already pass (block_prefix/suffix,
        # line_prefix/suffix, etc.)
        header_indent_override=header_indent_override,  # preserve pre-prefix indent
        # line_indent_override stays as default so fields still use processor's
        # after-prefix spacing
        soft_overflow_fields=soft_overflow_fields,
    )
    for field_name in sorted(soft_overflow_fields):
        ctx.hint(
            axis=Axis.RENDER,
            code="render:soft-width-exceeded",
            cluster=Cluster.UNCHANGED,
            message=(
                f"header field {field_name!r} exceeds the configured soft width "
                f"of {ctx.config.max_header_line_length} code points"
            ),
        )

    # Generate the expected (updated) header block, preserving the file newline style
    ctx.status.render = RenderStatus.RENDERED
    ctx.views.render = RenderView(lines=rendered_lines, block="".join(rendered_lines))

    logger.debug("Rendered header block for %s:\n%s", ctx.path, ctx.views.render.block or "")

    return

def hint(
    self,
    ctx: ProcessingContext,
) -> None:
    """Attach render hints (non-binding).

    Args:
        ctx: The processing context.

    Raises:
        RuntimeError: If the context contains an unexpected render status value.
    """
    st: RenderStatus = ctx.status.render

    match st:
        # May proceed to next step:
        case RenderStatus.RENDERED:
            pass  # normal; no hint

        # Stop processing:
        case RenderStatus.SKIPPED:
            pass  # already halted by run()

        case RenderStatus.FAILED:
            pass  # validation failure already diagnosed and halted by run()

        # States owned outside this step:
        case RenderStatus.PENDING:  # pragma: no cover - BaseStep owns pending handling.
            # BaseStep.__call__() handles PENDING state (step did not complete)
            pass

        case _:  # pragma: no cover - exhaustive enum guard for untyped callers.
            raise RuntimeError(f"Unexpected RenderStatus found: {st!r}")