Diff Coverage

# Copyright 2026 Huawei Technologies Co., Ltd
# Licensed under the Apache License, Version 2.0
# ============================================================================
"""``python -m hyper_parallel.codegen`` entry point."""
from __future__ import annotations

import sys

from hyper_parallel.codegen.cli import main

if __name__ == "__main__":
    sys.exit(main())

        ``artifact_dir`` is ``<yaml_dir>/generated/``. Example directories
        contain one Codegen YAML, so another per-YAML directory is redundant.
        """
        yaml_path = os.path.abspath(yaml_path)
        artifact_dir = os.path.join(os.path.dirname(yaml_path), "generated")
        modeling_name = _default_modeling_name(model_name)
        return cls(
            yaml_path=yaml_path,
            artifact_dir=artifact_dir,
            modeling_path=os.path.join(artifact_dir, modeling_name),
            diff_path=os.path.join(artifact_dir, modeling_name + ".diff"),

        is kept as an empty string for backward-compatible metadata shape; the
        signature is computed from the resolved config projection, not from a
        YAML file path.
        """
        resolved_dir = os.path.abspath(artifact_dir or os.path.join(os.getcwd(), "generated"))
        modeling_name = _default_modeling_name(model_name)
        return cls(
            yaml_path="",
            artifact_dir=resolved_dir,
            modeling_path=os.path.join(resolved_dir, modeling_name),
            diff_path=os.path.join(resolved_dir, modeling_name + ".diff"),

        )


def _default_modeling_name(model_name: str | None) -> str:
    base = model_name or "model"
    # '' / '-' are not valid Python identifiers in module names.
    safe = "".join(c if c.isalnum() or c == "_" else "_" for c in base)
    return f"modeling_{safe}_gen_npu.py"


def resolve_artifact_layout(
    yaml_path: str, model_name: str | None = None

def resolve_artifact_layout(
    yaml_path: str, model_name: str | None = None
) -> ArtifactLayout:
    """Resolve the artifact layout for ``yaml_path``."""
    return ArtifactLayout.build(yaml_path, model_name)


def resolve_default_artifact_layout(
    artifact_dir: str | None = None,

    artifact_dir: str | None = None,
    model_name: str | None = None,
) -> ArtifactLayout:
    """Resolve the default model-construction artifact layout."""
    return ArtifactLayout.build_default(artifact_dir, model_name)


def ensure_artifact_dir(layout: ArtifactLayout) -> None:
    """Create the artifact directory (recursive, no-op if present)."""
    Path(layout.artifact_dir).mkdir(parents=True, exist_ok=True)


def write_bundle_atomic(
    layout: ArtifactLayout, files: dict[str, str], meta: object = None

    three steps: move the old bundle aside, rename the staged dir into place,
    then delete the backup.  A reader on ``artifact_dir`` therefore sees either
    the old complete bundle or the new complete bundle — never a mix.
    """
    ensure_artifact_dir(layout)
    parent = os.path.dirname(layout.artifact_dir) or "."

    tmpdir = tempfile.mkdtemp(prefix=".codegen_tmp_", dir=parent)
    old_dir = None
    try:
        _stage_bundle(tmpdir, layout, files, has_meta=meta is not None)
        if meta is not None:
            _stage_meta(tmpdir, layout, meta)

        if os.path.exists(layout.artifact_dir):
            old_dir = tempfile.mkdtemp(prefix=".codegen_old_", dir=parent)
            os.rmdir(old_dir)  # reserve a name; must be absent for os.replace
            os.replace(layout.artifact_dir, old_dir)
        os.replace(tmpdir, layout.artifact_dir)
        if old_dir is not None:
            shutil.rmtree(old_dir, ignore_errors=True)
            old_dir = None
    finally:
        # If the staged dir never made it into place, put the old bundle back.
        if old_dir is not None and os.path.isdir(old_dir):
            if not os.path.exists(layout.artifact_dir):
                os.replace(old_dir, layout.artifact_dir)
            else:
                shutil.rmtree(old_dir, ignore_errors=True)
        if os.path.isdir(tmpdir):
            shutil.rmtree(tmpdir, ignore_errors=True)


def _stage_bundle(
    tmpdir: str,

    ``has_meta`` is False the prior ``codegen_meta.json`` is also preserved —
    otherwise the caller is going to write a fresh one into the staging dir.
    """

    def _is_tmp_name(name: str) -> bool:
        return name.startswith(".codegen_tmp_") or name.startswith(".codegen_old_")

    normalized = dict(files)
    for name in normalized:
        _validate_bundle_filename(name)

    init_present = "__init__.py" in normalized
    if not has_meta and os.path.isdir(layout.artifact_dir):
        for name in os.listdir(layout.artifact_dir):
            if _is_tmp_name(name):
                continue
            if name == "codegen_meta.json":
                continue  # preserved below, after the loop
            if name in normalized:
                continue
            src = os.path.join(layout.artifact_dir, name)
            if not os.path.isfile(src):
                continue
            if name == "__init__.py":
                init_present = True
            shutil.copy2(src, os.path.join(tmpdir, name))

    if not init_present:
        normalized["__init__.py"] = ""

    for name, content in normalized.items():
        # newline="" disables the platform's default CRLF translation so the
        # staged bytes equal the emitted text bytes exactly.  On Windows the
        # default (newline=None) would turn every LF into CRLF, inflating each
        # file by one byte per line and breaking the drift/hash checks that

        # staged bytes equal the emitted text bytes exactly.  On Windows the
        # default (newline=None) would turn every LF into CRLF, inflating each
        # file by one byte per line and breaking the drift/hash checks that
        # compare on-disk bytes against the re-emitted LF text.
        with open(
            os.path.join(tmpdir, name), "w", encoding="utf-8", newline=""
        ) as handle:
            handle.write(content)

    if not has_meta and os.path.isfile(layout.meta_path):
        shutil.copy2(
            layout.meta_path, os.path.join(tmpdir, os.path.basename(layout.meta_path))
        )

def _validate_bundle_filename(name: str) -> None:
    """Reject absolute, nested, and parent-relative bundle paths."""
    if (
        not name
        or os.path.isabs(name)
        or os.path.basename(name) != name
        or "/" in name

        or "/" in name
        or "\\" in name
        or name in {".", ".."}
    ):
        raise ValueError(f"bundle file keys must be plain names, got {name!r}")


def _stage_meta(tmpdir: str, layout: ArtifactLayout, meta: object) -> None:
    """Hash the staged content files then write ``codegen_meta.json`` last."""
    from hyper_parallel.codegen.meta import (
        record_output_hashes,
        write_codegen_meta,
    )

    record_output_hashes(meta, layout, base_dir=tmpdir)
    meta_path = os.path.join(tmpdir, os.path.basename(layout.meta_path))
    write_codegen_meta(meta, meta_path)


def clean_temp_artifacts(layout: ArtifactLayout) -> None:
    """Remove stale ``.codegen_tmp_*`` / ``.codegen_old_*`` dirs.

    Staging now happens in the artifact dir's *parent*, so these leftovers are
    scanned there rather than inside the bundle.
    """
    parent = os.path.dirname(layout.artifact_dir) or "."
    if not os.path.isdir(parent):
        return
    for name in os.listdir(parent):
        if name.startswith(".codegen_tmp_") or name.startswith(".codegen_old_"):
            shutil.rmtree(os.path.join(parent, name), ignore_errors=True)


__all__ = [
    "ArtifactLayout",

    end: int
    replacement: str

    def __post_init__(self) -> None:
        if self.start < 0 or self.end < self.start:
            raise ValueError(
                f"TextEdit has an invalid span: start={self.start} end={self.end}"
            )

    every later offset by the edit's length delta, so each edit is re-anchored
    by the cumulative drift of the edits before it.  Edits must be
    non-overlapping and stay within the original text — checked up front.
    """
    ordered = sorted(edits, key=lambda e: (e.start, e.end))
    if ordered:
        prev_end = ordered[0].start
        for edit in ordered:
            if edit.start < prev_end:
                raise ValueError(
                    f"apply_edits: overlapping edits — [{edit.start}, {edit.end}) "
                    f"overlaps the previous edit ending at {prev_end}"
                )
            if edit.end > len(text):
                raise ValueError(
                    f"apply_edits: edit [{edit.start}, {edit.end}) exceeds text "
                    f"length {len(text)}"
                )
            prev_end = edit.end

    out = list(text)
    shift = 0
    for edit in ordered:
        start = edit.start + shift
        end = edit.end + shift
        replacement = edit.replacement
        out[start:end] = replacement
        shift += len(replacement) - (edit.end - edit.start)
    return "".join(out)


def replace_function_body(
    source_text: str,

    Returns an empty tuple when there is nothing to replace (a function with
    no body span and no way to anchor).
    """
    if func.body_start is None or func.body_end is None:
        return ()
    # Re-indent per line, preserving blank lines and inner indentation.  The
    # caller renders ``new_body`` at column-0-relative depth (blank lines carry
    # no indentation), so a statement's relative nesting survives; only the
    # method's own ``indent`` is applied once.
    body_lines = [
        (indent + line) if line.strip() else ""
        for line in new_body.splitlines(keepends=True)
    ]
    replacement = "".join(body_lines)
    return (TextEdit(func.body_start, func.body_end, replacement),)


def insert_after_imports(
    source_text: str,

    ``None`` ``import_end`` means no import block: fall back to
    ``default_offset`` (the docstring end or 0).  Returns ``None`` when there is
    no sensible anchor — caller treats that as "insert at the top".
    """
    anchor = import_end if import_end is not None else default_offset
    if anchor < 0 or anchor > len(source_text):
        return None
    # Keep the inserted text on its own lines, bracketed by newlines, so it
    # never glues onto a trailing comment or the first statement.
    return TextEdit(anchor, anchor, "\n" + insert_text + "\n")


__all__ = [
    "TextEdit",

    docstring_end: Optional[int]

    def find_class(self, name: str) -> Optional[ClassInfo]:
        """Return the ``ClassInfo`` for a top-level class, or ``None``."""
        return self.classes.get(name)

    def find_forward(self, class_name: str) -> Optional[FunctionInfo]:
        """Return the ``FunctionInfo`` for ``class_name.forward``, or ``None``.

        A class without a ``forward`` (not a module) returns ``None``; the
        caller decides whether that is an error.
        """
        info = self.classes.get(class_name)
        if info is None:
            return None
        return info.methods.get("forward")


def build_source_index(source_text: str) -> SourceIndex:
    """Parse ``source_text`` and answer the three structural questions.

    The index is built against the *original* text.  Caller must apply edits to
    the same unmodified string (edits.py ``apply_edits`` splices by offset).
    """
    tree = ast.parse(source_text)

    line_offsets = _line_offsets(source_text)

    def offset_for(line_no: int) -> int:
        """Byte offset of the start of ``line_no`` (1-based)."""
        if 1 <= line_no <= len(line_offsets):
            return line_offsets[line_no - 1]
        return len(source_text)

    classes: dict[str, ClassInfo] = {}
    import_end: Optional[int] = None
    docstring_end: Optional[int] = None

    body = tree.body
    # Module docstring: the first statement is a constant string (PEP 257).  Its
    # closing quote is the first insertion point for appended constants, so
    # Frozen constants belong after the docstring and before imports.
    if body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant):
        if isinstance(body[0].value.value, str) and body[0].end_lineno is not None:
            docstring_end = offset_for(body[0].end_lineno)

    # Import block: the leading (or maximal) run of Import/ImportFrom at module
    # level.  In a well-formed module imports are first, so we take the span
    # from the first import to the last import before the first non-import

    # level.  In a well-formed module imports are first, so we take the span
    # from the first import to the last import before the first non-import
    # statement.  Anything after the docstring but before the first import (a
    # comment, ``__future__`` in a string, a module-level flag) is skipped.
    for node in body:
        if isinstance(node, (ast.Import, ast.ImportFrom)):
            if node.end_lineno is not None:
                import_end = offset_for(node.end_lineno)

    for node in body:
        if isinstance(node, ast.ClassDef):
            class_info = _build_class_info(node, source_text, offset_for)
            classes[class_info.name] = class_info

    return SourceIndex(
        source_text=source_text,
        classes=classes,
        import_end=import_end,
        docstring_end=docstring_end,

    )


def _build_class_info(node: ast.ClassDef, text: str, offset_for) -> ClassInfo:
    methods: dict[str, FunctionInfo] = {}
    for child in node.body:
        if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
            methods[child.name] = _build_function_info(child, text, offset_for)
    end_offset = (
        offset_for(node.end_lineno) if node.end_lineno is not None else len(text)
    )
    return ClassInfo(
        name=node.name,
        class_line=node.lineno,
        class_offset=offset_for(node.lineno),
        end_offset=end_offset,

def _build_function_info(node: ast.AST, text: str, offset_for) -> FunctionInfo:
    """Collect offsets, parameters and ``*args``/``**kwargs`` names for a def node."""
    decorators = getattr(node, "decorator_list", [])
    first_def_line = node.lineno
    if decorators:
        first_def_line = min(decorators[0].lineno, first_def_line)
    def_offset = offset_for(first_def_line)

    # Positional parameters (``self`` excluded) — the names the boundary's
    # ``arg_name`` binds against.  Only POSITIONAL_ONLY / POSITIONAL_OR_KEYWORD
    # are part of a positional call; ``*args`` / ``**kwargs`` are not.
    params = getattr(node, "args", None)
    positional: list[str] = []
    kwonly: list[str] = []
    vararg_name: Optional[str] = None
    kwarg_name: Optional[str] = None
    if params is not None:
        for arg in params.posonlyargs + params.args:
            if arg.arg != "self":
                positional.append(arg.arg)
        kwonly = [arg.arg for arg in params.kwonlyargs]
        vararg_name = params.vararg.arg if params.vararg is not None else None
        kwarg_name = params.kwarg.arg if params.kwarg is not None else None

    body = node.body
    body_start: Optional[int] = None
    body_end: Optional[int] = None
    # A function whose body is a single `pass` / `...` / single constant
    # statement on the same line has no replaceable body span.
    has_body = any(
        not (
            isinstance(stmt, ast.Pass)
            or (isinstance(stmt, ast.Expr) and isinstance(stmt.value, ast.Constant) and stmt.value.value is Ellipsis)
        )

            or (isinstance(stmt, ast.Expr) and isinstance(stmt.value, ast.Constant) and stmt.value.value is Ellipsis)
        )
        for stmt in body
    )
    if has_body:
        first_stmt = body[0]
        body_start = offset_for(first_stmt.lineno)
        last_stmt = body[-1]
        # End just past the last statement's final character AND its trailing
        # newline (when present).  ``offset_for(end_lineno)`` is only the START
        # of that line — using it here would leave the last statement (e.g.
        # ``return x``) outside the replaceable span, so a body swap would

        # start-of-line offset plus the statement's column + 1 lands just past
        # the line's newline; when the statement is the last line and the file
        # has no trailing newline, that would overshoot EOF, so clamp to
        # ``len(text)``.
        body_end = (
            min(offset_for(last_stmt.end_lineno) + last_stmt.end_col_offset + 1, len(text))
            if last_stmt.end_lineno is not None
            else len(text)
        )
    return FunctionInfo(
        name=node.name,
        body_start=body_start,
        body_end=body_end,
        def_offset=def_offset,

    strips that common prefix so the isolated def parses as a top-level
    statement (an indented first line would raise ``IndentationError`` and
    fail every method closed).
    """
    if func.body_start is None or func.body_end is None:
        return False
    def_line_start = _def_keyword_offset(source_text, func)
    span = textwrap.dedent(source_text[def_line_start:func.body_end])
    try:
        tree = ast.parse(span)
    except SyntaxError:
        return False
    for node in ast.walk(tree):
        if isinstance(node, ast.Return):
            value = node.value
            if value is None:
                continue
            if isinstance(value, (ast.Tuple, ast.List, ast.Starred)):
                return False
    return True


def returns_any_value(source_text: str, func: FunctionInfo) -> bool:
    """Whether every ``return`` exits with a single value or a tuple value.

    ``outputs[0]`` and rebuild the original object, matching the compiled plan's
    index-0 behavior. A bare ``return`` has no value to transform; a list return
    would need list-specific rebuilding, so it fails closed for now.
    """
    if func.body_start is None or func.body_end is None:
        return False
    def_line_start = _def_keyword_offset(source_text, func)
    span = textwrap.dedent(source_text[def_line_start:func.body_end])
    try:
        tree = ast.parse(span)
    except SyntaxError:
        return False
    seen_return = False
    for node in ast.walk(tree):
        if isinstance(node, ast.Return):
            seen_return = True
            if node.value is None:
                return False
            if isinstance(node.value, (ast.List, ast.Starred)):
                return False
    return seen_return


def _def_keyword_offset(source_text: str, func: FunctionInfo) -> int:
    """Byte offset of the ``def`` keyword line, skipping decorator lines.

    one is present); the span fed to ``ast.parse`` must start at the ``def``
    keyword itself.  Shared logic with ``_build_forward_impl_edit`` in
    emit/parallel (kept local — three lines, not worth a cross-module helper).
    """
    def_line_start = func.def_offset
    for line in source_text[def_line_start:func.body_start].splitlines(keepends=True):
        if line.lstrip().startswith("def "):
            break
        def_line_start += len(line)
    return def_line_start


def _line_offsets(text: str) -> list[int]:
    """Byte offset of each line's start (``offsets[line_no - 1]`` is 1-based line ``line_no``).

    Single implementation, shared with ``codegen.source.compat`` (which imports
    it from here — the existing layering direction is ``source`` -> ``astkit``).
    """
    offsets: list[int] = [0]
    for line in text.splitlines(keepends=True):
        offsets.append(offsets[-1] + len(line))
    return offsets

against the *hashes recorded in meta*): this check re-derives the expected
bytes from the meta itself, so it catches an edit that was also re-touched in
meta, or an emit-layer regression.
"""
from __future__ import annotations

import logging
import os
from typing import Any, Optional

from hyper_parallel.codegen.artifact import resolve_artifact_layout

logger = logging.getLogger(__name__)


def verify_drift_intact(meta: Any, yaml_path: str) -> list[str]:
    """Regenerate the bundle from ``meta`` and compare with the on-disk files.

    Returns a list of drift descriptions (empty = intact).  Raises
    ``FileNotFoundError`` when meta or the bundle directory is missing.  When

    the original modeling source file can no longer be read (e.g. a remote-code
    cache was evicted), drift is *reported*, not silently skipped: the artifact
    cannot be re-validated against its origin.
    """
    layout = resolve_artifact_layout(yaml_path, model_name=_model_name(meta))
    if not os.path.isdir(layout.artifact_dir):
        raise FileNotFoundError(f"codegen drift: bundle missing at {layout.artifact_dir}")

    from hyper_parallel.codegen.emit import emit_bundle

    files = emit_bundle(layout, meta)

    drifts: list[str] = []
    for name, expected in files.items():
        path = os.path.join(layout.artifact_dir, name)
        if not os.path.isfile(path):
            drifts.append(f"{name}: missing on disk")
            continue
        with open(path, "r", encoding="utf-8", errors="replace") as handle:
            actual = handle.read()
        if actual != expected:
            drifts.append(f"{name}: bytes differ from re-emitted bundle")
    return drifts


def _model_name(meta: Any) -> Optional[str]:
    """The model name recorded in meta (the generated filename stems from it)."""
    return getattr(meta, "model_name", None)


__all__ = [
    "verify_drift_intact",
]

of an environment mismatch, it just cannot be validated by the pin — so the
caller exits 0 either way, and reserves exit 1 for "the pin itself cannot be
parsed" / "the installed version cannot be read".
"""
from __future__ import annotations

import logging
from typing import Optional, Tuple

logger = logging.getLogger(__name__)

# The public status the CLI maps to an exit code.
OK = "OK"
SKIP = "SKIP"


def verify_transformers_pin(spec: str) -> Tuple[str, str, str]:
    """Compare the installed transformers to ``spec``.

    Returns ``(status, installed, spec)`` where ``status`` is ``OK`` when the
    installed version satisfies ``spec`` and ``SKIP`` otherwise.  Raises

    ``ValueError`` when ``spec`` cannot be parsed and ``ImportError`` when the
    installed transformers version cannot be read (both are exit-1 conditions
    for the CLI).
    """
    installed = _installed_version()
    if installed is None:
        raise ImportError("codegen check: cannot read installed transformers version")

    from packaging.specifiers import SpecifierSet

    try:
        pin = SpecifierSet(spec)
    except Exception as exc:
        raise ValueError(f"codegen check: invalid transformers pin {spec!r}: {exc}") from exc

    if installed in pin:
        return OK, installed, spec
    logger.warning(
        "codegen check: installed transformers %s does not satisfy pin %s — "
        "SKIP (artifact is not stale, pin is not satisfiable here)",
        installed, spec,
    )
    return SKIP, installed, spec


def _installed_version() -> Optional[str]:
    """Installed transformers version, or ``None`` when not importable."""
    try:
        return _version_from_importlib_metadata()
    except ImportError:
        return None


def _version_from_importlib_metadata() -> str:
    import importlib.metadata

    return importlib.metadata.version("transformers")


__all__ = [
    "OK",
    "SKIP",
    "verify_transformers_pin",
]

    A mismatch means the YAML / source / env changed since generation — the
    bundle is stale and must be regenerated before training.
    """
    if meta.signature != current_signature:
        raise RuntimeError(
            f"codegen preflight: signature mismatch for {meta.yaml_path} "
            f"(stored {meta.signature}, current {current_signature}); regenerate"
        )

    The meta file itself is not in that set: a file cannot carry its own
    sha256 (writing the value changes the bytes it hashes), so meta
    integrity is covered by the signature and schema checks instead.
    """
    core_paths = {
        "modeling": layout.modeling_path,
        "diff": layout.diff_path,
        "init": layout.init_path,
    }
    for key in core_paths:
        if key not in meta.outputs:
            raise RuntimeError(f"codegen preflight: missing {key} output hash in meta")

    for sibling in getattr(meta, "remote_siblings", None) or []:
        if sibling not in meta.outputs:
            raise RuntimeError(
                f"codegen preflight: missing output hash for remote sibling {sibling!r}"
            )

    for key, recorded in meta.outputs.items():
        path = core_paths.get(key)
        if path is None:
            if (
                not key
                or os.path.basename(key) != key
                or "/" in key
                or "\\" in key

                or "/" in key
                or "\\" in key
                or key in {".", ".."}
            ):
                raise RuntimeError(
                    f"codegen preflight: invalid output filename {key!r}"
                )
            path = os.path.join(layout.artifact_dir, key)
        if not os.path.isfile(path):
            raise RuntimeError(
                f"codegen preflight: {key} file missing at {path} "
                f"(recorded sha256 {recorded.get('sha256')})"
            )
        actual = sha256_file(path)
        if recorded.get("sha256") != actual:
            raise RuntimeError(
                f"codegen preflight: {key} hash drift at {path} "
                f"(recorded {recorded.get('sha256')}, actual {actual})"
            )

    ``sys.modules`` by the loader, so preflight and the later build reuse the
    same import. A minimal placeholder also imports because preflight checks
    bundle structure, while the emit layer owns lowering completeness.
    """
    from hyper_parallel.codegen.loader import import_generated_module

    import_generated_module(layout.artifact_dir)


def verify_meta_required_fields(meta: Any) -> None:
    """Fail fast when the meta record is missing required schema fields.

    Delegates to the meta module's schema validation; a stale or hand-edited
    meta file that lost a required field must not pass preflight.
    """
    from hyper_parallel.codegen.meta import validate_meta_schema

    validate_meta_schema(meta_to_dict(meta))


def warn_skipped_overrides(meta: Any) -> None:
    """Warn (not error) when generation skipped any plan_overrides entries.

    An entry skipped during generation (or a ``when``-gated entry that never
    applied) must be surfaced
    so the user knows the artifact does not contain what the YAML declared.
    """
    skipped = getattr(meta, "skipped_overrides", None) or []
    if not skipped:
        return
    lines = [f"codegen: {len(skipped)} plan_overrides entries were skipped:"]
    for item in skipped:
        match = item.get("match") if isinstance(item, dict) else None
        reason = item.get("reason") if isinstance(item, dict) else item
        lines.append(f"  - match={match!r}: {reason}")
    logger.warning("\n".join(lines))


def verify_boundary_forms(meta: Any, layout: Any) -> None:
    """Fail fast when persisted forward forms disagree with the frozen plan.

    A drift means the modeling file was edited after generation (or the
    emitter and this check disagree); the install-time validation would
    silently fall back, masking a broken artifact, so fail fast instead.
    """
    import re

    from hyper_parallel.codegen.astkit.index import build_source_index
    from hyper_parallel.codegen.emit.parallel import (
        TOGGLE_FORM_MARKER,
        TP_FORM_ATTRIBUTE,
        TP_FORM_MARKER,
        iter_emitted_forms,

        TP_FORM_ATTRIBUTE,
        TP_FORM_MARKER,
        iter_emitted_forms,
    )
    from hyper_parallel.codegen.plan.boundary_forms import (
        FORM_IDENTITY,
        FORM_TP_COLLECTIVE,
    )

    with open(layout.modeling_path, "r", encoding="utf-8") as handle:
        source_text = handle.read()
    index = build_source_index(source_text)
    boundary_classes = getattr(meta, "boundary_classes", None) or {}
    source = getattr(meta, "source", None)
    module_name = (
        source.get("module_name")
        if isinstance(source, dict)
        else getattr(source, "module_name", "")
    )

        if isinstance(source, dict)
        else getattr(source, "module_name", "")
    )

    for class_name, _, emitted, func, _, deferred_bias in iter_emitted_forms(
        source_text,
        meta,
        boundary_classes=boundary_classes,
        module_name=module_name,

        meta,
        boundary_classes=boundary_classes,
        module_name=module_name,
    ):
        info = index.find_class(class_name)
        has_impl = info is not None and "_forward_impl" in info.methods
        class_body = source_text[info.class_offset:info.end_offset] if info else ""
        has_marker = f'{TP_FORM_ATTRIBUTE} = "{TP_FORM_MARKER}"' in class_body
        # The opt-in toggle template marks its classes with TOGGLE_FORM_MARKER:
        # a boundary the classifier calls tp_collective carries either marker
        # depending on which generation mode emitted it.
        has_toggle_marker = (
            f'{TP_FORM_ATTRIBUTE} = "{TOGGLE_FORM_MARKER}"' in class_body
        )
        has_inline_marker = "_hp_inline_boundary = True" in class_body
        body = source_text[func.body_start:func.body_end]
        # Every generated communication path — inlined or lowered — reads the
        # instance-attribute channel (``self._hp_tp`` / ``self._hp_cp_mesh``
        # / ``self._hp_ep_group`` / ``self.<axis>_enable``) the runtime binds
        # at install time.  The retired module-level accessor must never come

        # / ``self._hp_ep_group`` / ``self.<axis>_enable``) the runtime binds
        # at install time.  The retired module-level accessor must never come
        # back: a body still reading it was emitted by the accessor era, and the
        # class would raise NameError at train time.
        stray_accessor = re.search(r"get_parallel_state|get_inline_parallel_state", body)
        if stray_accessor is not None:
            raise RuntimeError(
                f"codegen preflight: class {class_name} still reads the removed "
                f"module-level accessor {stray_accessor.group(0)!r}; the parallel "
                "channel is bound on the instance at install time — regenerate "
                "the artifact"

                f"module-level accessor {stray_accessor.group(0)!r}; the parallel "
                "channel is bound on the instance at install time — regenerate "
                "the artifact"
            )
        has_inline_channel = any(
            token in body
            for token in (
                "self._hp_tp",
                "self._hp_cp_mesh",

                "self.ep_enable",
                "ep_all_to_all",
            )
        )
        if has_inline_marker or (
            has_inline_channel and not (has_impl or has_marker or has_toggle_marker)
        ):
            # A fully inlined component body (the inline-marker shape): it
            # keeps no extracted ``_forward_impl`` and no form marker, and

        ):
            # A fully inlined component body (the inline-marker shape): it
            # keeps no extracted ``_forward_impl`` and no form marker, and
            # must expose the generated communication path through the channel.
            if not has_inline_channel:
                raise RuntimeError(
                    f"codegen preflight: inline boundary class {class_name} "
                    "does not expose a generated communication path; "
                    "regenerate the artifact"
                )
            if has_marker:
                raise RuntimeError(
                    f"codegen preflight: clean inline boundary class "
                    f"{class_name} also carries the {TP_FORM_MARKER} marker; "
                    "regenerate the artifact"
                )
            continue
        if emitted == FORM_IDENTITY:
            if has_impl or has_marker:
                raise RuntimeError(
                    f"codegen preflight: class {class_name} was pruned as "
                    f"identity but the source carries a rewritten forward "
                    f"(marker={has_marker}, _forward_impl={has_impl}); "
                    "regenerate the artifact"

                    f"identity but the source carries a rewritten forward "
                    f"(marker={has_marker}, _forward_impl={has_impl}); "
                    "regenerate the artifact"
                )
            continue
        if not has_impl:
            raise RuntimeError(
                f"codegen preflight: class {class_name} should carry an "
                f"extracted _forward_impl for form {emitted!r} but does not; "
                "regenerate the artifact"
            )
        if emitted == FORM_TP_COLLECTIVE:
            if not (has_marker or has_toggle_marker):
                raise RuntimeError(
                    f"codegen preflight: class {class_name} should carry the "
                    f"{TP_FORM_MARKER} marker but does not; regenerate the "
                    "artifact"
                )
            body = source_text[func.body_start:func.body_end]
            # The static template's own channel is ``self._hp_tp``; a
            # class whose frozen entry declares D-22 deferred bias params
            # additionally calls ``self._hp_deferred_bias`` at the exit
            # (bound at install time).  Anything else is drift.
            allowed = "tp|deferred_bias" if deferred_bias else "tp"
            stray = re.search(rf"self\._hp_(?!({allowed})\b)", body)
            if stray is not None:
                raise RuntimeError(
                    f"codegen preflight: static tp_collective forward on "
                    f"class {class_name} references {stray.group(0)}* "
                    "attributes outside the static template; regenerate the "
                    "artifact"

                    f"class {class_name} references {stray.group(0)}* "
                    "attributes outside the static template; regenerate the "
                    "artifact"
                )
        elif has_marker:
            raise RuntimeError(
                f"codegen preflight: class {class_name} carries the "
                f"{TP_FORM_MARKER} marker but its emitted form is "
                f"{emitted!r}; regenerate the artifact"
            )

    variant) fails fast instead of silently running with a stale sharding
    contract.  Degrades to a warning when ``model`` is not given (preflight
    without a model handle can still check the plan's internal consistency).
    """
    param_plan = getattr(meta, "param_plan", None) or {}
    if not param_plan:
        raise RuntimeError(
            "codegen preflight: param plan is empty — generation did not "
            "freeze a sharding plan (derivation failed or was skipped); "
            "check generation logs and regenerate the artifact"
        )

            "freeze a sharding plan (derivation failed or was skipped); "
            "check generation logs and regenerate the artifact"
        )

    if model is None:
        logger.warning(
            "codegen preflight: no model handle — param-plan coverage "
            "skipped (only internal consistency checked)"
        )
        return

    # Every planned parameter key must resolve to a model parameter.
    planned: set[str] = set()
    missing: list[str] = []
    for boundary, entry in param_plan.items():
        params = entry.get("params", {}) if isinstance(entry, dict) else {}
        for name in params:
            fqn = f"{boundary}.{name}" if name else boundary
            planned.add(fqn)
            if _get_parameter(model, fqn) is None:
                missing.append(fqn)
    if missing:
        raise RuntimeError(
            f"codegen preflight: param plan references parameters missing "
            f"from the model: {missing}"
        )

        )

    # Every trainable model parameter must be covered by an explicit plan
    # entry (exact FQN or a leaf under a grouped key).
    uncovered = [
        fqn
        for fqn in _iter_trainable_params(model)
        if not any(fqn == key or fqn.startswith(f"{key}.") for key in planned)
    ]
    if uncovered:
        raise RuntimeError(
            f"codegen preflight: {len(uncovered)} trainable parameters are "
            f"not covered by the frozen plan: {uncovered[:20]}"
        )

def meta_to_dict(meta: Any) -> dict[str, Any]:
    """Serialize meta to dict, or pass through a dict already."""
    if isinstance(meta, dict):
        return meta
    from hyper_parallel.codegen.meta import meta_to_dict as _meta_to_dict

    return _meta_to_dict(meta)


def _iter_trainable_params(model: Any):
    """Yield ``fqn`` for every trainable parameter of ``model`` (or its config)."""
    try:
        import torch  # pylint: disable=unused-import
    except ImportError:
        return
    seen: set[int] = set()
    for module_name, module in model.named_modules():
        for param_name, param in module.named_parameters(recurse=False):
            if not param.requires_grad:
                continue
            fqn = f"{module_name}.{param_name}" if module_name else param_name
            if id(param) in seen:
                continue
            seen.add(id(param))
            yield fqn


def _get_parameter(model: Any, fqn: str) -> Any:
    """Look up a parameter by dotted FQN, tolerating an absent attribute."""
    obj = model
    for part in fqn.split("."):
        try:
            obj = getattr(obj, part)
        except AttributeError:
            return None
    return obj if hasattr(obj, "requires_grad") else None


__all__ = [
    "verify_boundary_forms",

Exit codes are 0 for success or a skipped pin check, 1 for a check/clean
failure, and 2 for a usage error.
"""
from __future__ import annotations

import argparse
import logging
import os
import sys
from typing import Iterator, Optional

logger = logging.getLogger(__name__)


def main(argv: Optional[list[str]] = None) -> int:
    """argparse entry; returns the process exit code."""
    parser = _build_parser()
    args, config_overrides = parser.parse_known_args(argv)
    setattr(args, "_config_overrides", config_overrides)
    if not getattr(args, "command", None):
        parser.print_usage(sys.stderr)
        return 2
    _configure_logging(getattr(args, "verbose", False))
    try:
        return args.func(args)
    except Exception as exc:  # noqa: BLE001 - a CLI reports, does not traceback
        logger.error("codegen: %s", exc)
        return 1


# ---------------------------------------------------------------------------
# commands

# ---------------------------------------------------------------------------
# commands
# ---------------------------------------------------------------------------

def cmd_generate(args: argparse.Namespace) -> int:
    """``generate --config <yaml>`` — build/reuse the bundle for one YAML."""
    config = _load_config(args.config, _cli_overrides(args))
    from hyper_parallel.codegen.manager import ensure_codegen_artifact

    if not getattr(config, "codegen", False):
        logger.info("codegen: generate skipped — codegen is false")
        return 0
    layout = ensure_codegen_artifact(config, args.config)
    if layout is None:
        return 0
    from hyper_parallel.codegen.meta import load_codegen_meta

    meta = load_codegen_meta(layout.meta_path)
    signature = meta.signature if meta is not None else "?"
    logger.info("codegen: artifact %s (signature %s)", layout.artifact_dir, signature)
    return 0


def cmd_check(args: argparse.Namespace) -> int:
    """``check --config <yaml>`` — full preflight for one YAML."""
    config = _load_config(args.config, _cli_overrides(args))
    if not getattr(config, "codegen", False):
        logger.info("codegen: SKIP (codegen disabled)")
        return 0
    from hyper_parallel.codegen.manager import preflight_integrity_check

    preflight_integrity_check(config, args.config)
    logger.info("codegen: PASS %s", args.config)
    return 0


def cmd_check_all(args: argparse.Namespace) -> int:
    """Run preflight for every discovered YAML and check cached artifact drift."""
    if _cli_overrides(args):
        raise ValueError("check --all does not accept per-config CLI overrides")
    failed = 0
    for yaml_path in discover_yaml_files(args.root):
        try:
            config = _load_config(yaml_path)
            if not getattr(config, "codegen", False):
                logger.info("codegen: SKIP (codegen disabled) %s", yaml_path)
                continue
            from hyper_parallel.codegen.manager import preflight_integrity_check

            preflight_integrity_check(config, yaml_path)
            logger.info("codegen: PASS %s", yaml_path)
            # Drift can only be checked when an artifact already exists.
            from hyper_parallel.codegen.check.drift import verify_drift_intact
            from hyper_parallel.codegen.meta import load_codegen_meta
            from hyper_parallel.codegen.artifact import resolve_artifact_layout

            layout = resolve_artifact_layout(yaml_path)
            meta = load_codegen_meta(layout.meta_path)
            if meta is not None:
                drifts = verify_drift_intact(meta, yaml_path)
                if drifts:
                    failed = 1
                    logger.error("codegen: drift in %s: %s", yaml_path, "; ".join(drifts))
                else:
                    logger.info("codegen: drift intact %s", yaml_path)
        except Exception as exc:  # noqa: BLE001 - aggregate, keep scanning
            failed = 1
            logger.error("codegen: FAIL %s: %s", yaml_path, exc)
    return failed


def cmd_check_pin(args: argparse.Namespace) -> int:
    """Validate a Transformers version constraint without regenerating artifacts."""
    from hyper_parallel.codegen.check.pin import verify_transformers_pin

    status, installed, spec = verify_transformers_pin(args.transformers_pin)
    logger.info("codegen: transformers pin %s — installed %s (%s)", spec, installed, status)
    return 0  # OK and SKIP both exit 0


# ---------------------------------------------------------------------------
# discovery & helpers