topmark.pipeline.steps.builder

topmark / pipeline / steps / builder

Builder step for TopMark pipeline.

This step computes the field dictionaries that will later be rendered into a TopMark header. It derives built-in fields from the filesystem (e.g., file, file_relpath) and merges them with FrozenConfig.field_values, then selects only those keys listed in FrozenConfig.header_fields.

Outputs

• ctx.views.build.builtins: the derived built-in field mapping.
• ctx.views.build.selected: the filtered/merged mapping to be rendered by the renderer.
• ctx.status.generation: set to GENERATED (or NO_FIELDS when no fields are configured).

Notes

This step does not render text. The renderer consumes ctx.views.build.selected and produces the final lines/block in ctx.views.render.

BuilderStep

BuilderStep()

Bases: BaseStep

Compute field dictionaries for rendering a TopMark header.

Derives built-in fields from the filesystem, merges with configured values, and selects only configured header fields. Produces a BuilderView.

Axes written

• generation

Sets

• GenerationStatus: {PENDING, GENERATED, NO_FIELDS}

Source code in src/topmark/pipeline/steps/builder.py

def __init__(self) -> None:
    super().__init__(
        name=self.__class__.__name__,
        primary_axis=Axis.GENERATION,
        axes_written=(Axis.GENERATION,),
    )

may_proceed

may_proceed(ctx)

Determine if processing can proceed to the build step.

Return whether the builder has the prerequisites to compute field values.

The builder should run whenever content is available, even for empty-like images. Empty-file insertion policy is a mutation concern and is handled later by planner/updater logic via allow_insert_into_empty_like(), not by may_proceed().

Parameters:

Name	Type	Description	Default
ctx	ProcessingContext	The processing context for the current file.	required

Returns:

Type	Description
bool	True if the builder can compute header field values for this context.

Source code in src/topmark/pipeline/steps/builder.py

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

    Return whether the builder has the prerequisites to compute field values.

    The builder should run whenever content is available, even for empty-like
    images. Empty-file insertion policy is a *mutation* concern and is handled
    later by planner/updater logic via
    [`allow_insert_into_empty_like()`][topmark.pipeline.context.policy.allow_insert_into_empty_like],
    not by `may_proceed()`.

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

    Returns:
        True if the builder can compute header field values for this context.
    """
    if ctx.is_halted:
        return False

    # Normal path:
    # - non-empty files require ContentStatus.OK
    # - empty files can proceed only when allowed by policy (also ContentStatus.OK)
    return ctx.status.content == ContentStatus.OK

run

run(ctx)

Build the field dictionaries used to render a TopMark header.

This step analyzes the processing context and configuration to compute: * derived built-in fields based on the file path; and * the final field mapping to be rendered (subset/overrides as per config).

Parameters:

Name	Type	Description	Default
ctx	ProcessingContext	The current processing context containing file info, configuration, and status.	required

Mutations

• ctx.views.build.builtins: built-in field mapping.
• ctx.views.build.selected: selected/merged field mapping.
• ctx.status.generation: updated to GENERATED or NO_FIELDS.

Notes

Diagnostic messages are added if unknown header fields are referenced in FrozenConfig.header_fields or when built-ins are overridden by FrozenConfig.field_values.

Path semantics

• file_path is ctx.path as supplied when the processing context was bootstrapped. In normal file-list processing this is TopMark's canonical processing path, not necessarily the original CLI/config spelling. Lower-level callers may still pass a symlink spelling directly; the builder canonicalizes filesystem metadata again.
• For regular filesystem input, file, file_relpath, and file_abspath are derived from the canonical filesystem spelling of the resolved processing target. Symlink spelling is not preserved in generated header metadata. This keeps repeated runs idempotent when a symlink and its target are both selected, and avoids casing-only metadata drift on case-insensitive filesystems.
• In stdin mode, file, file_relpath, and file_abspath are derived from the logical stdin_filename, not the materialized temporary file path.
• file_relpath and relpath are computed by calling compute_relpath(header_path, relative_to) and serialized with format_header_metadata_path().
• relative_to defaults to Path.cwd() when no explicit root is configured. If FrozenConfig.relative_to is set and exists on disk, it is canonicalized before relative paths are computed.
• relpath becomes "." at repo root.

Header metadata path fields are serialized with POSIX separators on all platforms. This behavior intentionally keeps file_relpath stable for common cases like running TopMark from the repository root (so pyproject.toml yields file_relpath = "pyproject.toml"). If you need file_relpath computed relative to a specific root, set FrozenConfig.relative_to (via config or CLI/API overrides). User-supplied path spelling is not treated as header metadata once a regular filesystem path has been successfully resolved.

Source code in src/topmark/pipeline/steps/builder.py

def run(self, ctx: ProcessingContext) -> None:
    """Build the field dictionaries used to render a TopMark header.

    This step analyzes the processing context and configuration to compute:
    * derived built-in fields based on the file path; and
    * the final field mapping to be rendered (subset/overrides as per config).

    Args:
        ctx: The current processing context containing file info,
            configuration, and status.

    Mutations:
        - `ctx.views.build.builtins`: built-in field mapping.
        - `ctx.views.build.selected`: selected/merged field mapping.
        - `ctx.status.generation`: updated to `GENERATED` or `NO_FIELDS`.

    Notes:
        Diagnostic messages are added if unknown header fields are referenced in
        `FrozenConfig.header_fields` or when built-ins are overridden by
        `FrozenConfig.field_values`.

    Path semantics:
        - `file_path` is `ctx.path` as supplied when the processing context was bootstrapped.
          In normal file-list processing this is TopMark's canonical processing path, not
          necessarily the original CLI/config spelling. Lower-level callers may still pass a
          symlink spelling directly; the builder canonicalizes filesystem metadata again.
        - For regular filesystem input, `file`, `file_relpath`, and `file_abspath`
          are derived from the canonical filesystem spelling of the resolved
          processing target. Symlink spelling is not preserved in generated
          header metadata. This keeps repeated runs idempotent when a symlink
          and its target are both selected, and avoids casing-only metadata drift
          on case-insensitive filesystems.
        - In stdin mode, `file`, `file_relpath`, and `file_abspath` are derived from
          the logical `stdin_filename`, not the materialized temporary file path.
        - `file_relpath` and `relpath` are computed by calling
          `compute_relpath(header_path, relative_to)` and serialized with
          `format_header_metadata_path()`.
        - `relative_to` defaults to `Path.cwd()` when no explicit root is configured.
          If `FrozenConfig.relative_to` is set and exists on disk, it is canonicalized
          before relative paths are computed.
        - `relpath` becomes "." at repo root.

        Header metadata path fields are serialized with POSIX separators on all
        platforms. This behavior intentionally keeps `file_relpath` stable for common
        cases like running TopMark from the repository root (so `pyproject.toml` yields
        `file_relpath = "pyproject.toml"`). If you need `file_relpath` computed relative
        to a specific root, set `FrozenConfig.relative_to` (via config or CLI/API
        overrides). User-supplied path spelling is not treated as header metadata once a
        regular filesystem path has been successfully resolved.
    """
    logger.debug("ctx: %s", ctx)

    config: FrozenConfig = ctx.config

    if check_permitted_by_policy(ctx) is False:
        ctx.status.generation = GenerationStatus.SKIPPED
        reason = "header field generation skipped by policy"
        ctx.diagnostics.add_info(reason)
        ctx.request_halt(reason=reason, at_step=self)
        return

    if not config.header_fields:
        # No header fields specified in the configuration
        ctx.status.generation = GenerationStatus.NO_FIELDS
        ctx.diagnostics.add_info("No header fields specified.")
        return

    file_path: Path = ctx.path

    # Prepare built-in fields related to the file system.
    #
    # STDIN content mode needs special care:
    #   - `ctx.path` is the *materialized* temp file that TopMark can read/write.
    #   - `run_options.stdin_filename` is a user-supplied *logical* filename that may not exist
    #     on disk (it is used only for metadata and discovery anchoring).
    #
    # Therefore:
    #   - `file` / `file_relpath` / `file_abspath` / `relpath` are derived from the logical
    #     header path.
    #   - `abspath` is still derived from the on-disk content path, since it represents the
    #     containing directory of the actual processed content.

    content_path: Path = file_path

    # In stdin mode, use `stdin_filename` (if provided) for logical header metadata.
    # Otherwise, use the canonical filesystem spelling of the existing content path.
    # This keeps generated metadata tied to the resolved processing target instead of
    # invocation spelling. In particular, symlinked filesystem inputs generate path
    # metadata for the target that TopMark will actually read/write. A separate warning
    # about original symlink spelling would require retaining that spelling before file-list
    # resolution canonicalizes `ctx.path`, so this step deliberately documents rather than
    # diagnoses that policy.
    header_path: Path = (
        Path(ctx.run_options.stdin_filename)
        if (ctx.run_options.stdin_mode and ctx.run_options.stdin_filename)
        else canonical_processing_path(content_path)
    )

    # File absolute path is derived from the logical header path so stdin mode reports the
    # user-facing logical filename rather than the ephemeral materialized temp file.
    absolute_path: Path = header_path.resolve()
    content_absolute_path: Path = content_path.resolve(strict=True)

    # Relative paths are computed against `relative_to` using the logical header path.
    # Note: `header_path` may not exist in stdin mode, so `compute_relpath()` must not rely
    # on strict filesystem resolution.
    relative_to: Path = (
        canonicalize_existing_path(Path(config.relative_to))
        if config.relative_to and Path(config.relative_to).exists()
        else Path.cwd()
    )
    # Determine relative path from the file to the root path
    # Default to the current working directory if 'relative_to' is not configured
    relative_path: Path = compute_relpath(header_path, relative_to)

    builtin_fields: dict[str, str] = {
        # Base file name (without any path)
        "file": header_path.name,
        # File name with its relative path
        "file_relpath": format_header_metadata_path(relative_path),
        # File name with its absolute path
        "file_abspath": format_header_metadata_path(absolute_path),
        # Parent directory path (relative)
        "relpath": format_header_metadata_path(relative_path.parent) if relative_path else "",
        # Parent directory path (absolute, of actual content)
        "abspath": format_header_metadata_path(content_absolute_path.parent),
    }

    # Merge in any additional fields from the configuration (may override built-ins).
    if config.field_values:
        # Warn if configuration fields override built-in fields (potentially accidental)
        builtin_overlap: list[str] = [
            key for key in config.field_values if key in builtin_fields
        ]
        if len(builtin_overlap) > 0:
            builtin_overlap_repr: str = ", ".join(
                key for key in config.field_values if key in builtin_fields
            )
            ctx.diagnostics.add_warning(f"Redefined built-in fields: {builtin_overlap_repr}")

    # Merge built-ins with configuration-defined values; allow overrides; restrict
    # to header_fields.
    all_fields: dict[str, str] = {
        **builtin_fields,
        **config.field_values,
    }

    result: dict[str, str] = {}

    for key in config.header_fields:
        value: str | None = all_fields.get(key)
        if value is None:
            ctx.diagnostics.add_error(f"Unknown header field: {key}")
        else:
            result[key] = value

    # Populate BuilderView with builtins and selected field mappings
    ctx.views.build = BuilderView(builtins=builtin_fields, selected=result)
    # Populate RenderView with mapping only; lines/block are filled by renderer
    ctx.status.generation = GenerationStatus.GENERATED

    logger.debug(
        "Builder: %s - header status=%s, selected fields:\n%s",
        ctx.path,
        ctx.status.header.value,
        "\n".join(
            f"  {key:<20} : {value}" for key, value in (ctx.views.build.selected or {}).items()
        ),
    )
    logger.info(
        "Builder completed for %s: header status=%s, generation status=%s",
        ctx.path,
        ctx.status.header.value,
        ctx.status.generation.value,
    )

    return

hint

hint(ctx)

Attach generation hints (non-binding).

Parameters:

Name	Type	Description	Default
ctx	ProcessingContext	The processing context.	required

Source code in src/topmark/pipeline/steps/builder.py

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

    Args:
        ctx: The processing context.
    """
    st: GenerationStatus = ctx.status.generation

    # May proceed to next step (always):
    if st == GenerationStatus.GENERATED:
        pass  # expected path; silent
    # May proceed to next step (render empty header):
    elif st == GenerationStatus.NO_FIELDS:
        ctx.hint(
            axis=Axis.GENERATION,
            code=KnownCode.GENERATION_NO_FIELDS,
            cluster=Cluster.BLOCKED_POLICY,
            message="no header fields configured",
            terminal=False,
        )
    # Stop processing:
    elif st == GenerationStatus.SKIPPED:
        ctx.hint(
            axis=Axis.GENERATION,
            code=KnownCode.PLAN_SKIP,
            cluster=Cluster.BLOCKED_POLICY,
            message="header field generation skipped",
            terminal=True,
        )
    elif st == GenerationStatus.PENDING:
        # builder did not complete
        ctx.request_halt(reason=f"{self.__class__.__name__} did not set state.", at_step=self)