Diff Coverage

    no aliasing between model types.  When ``models/<model_type>/`` does not
    exist yet, ``ArtifactLayout.build_model_anchored`` warns and the directory is
    created on demand by :func:`ensure_artifact_dir`.
    """
    return ArtifactLayout.build_model_anchored(model_type, model_name=model_name)


def ensure_artifact_dir(layout: ArtifactLayout) -> None:
    """Create the artifact directory (recursive, no-op if present).

    tmpdir = tempfile.mkdtemp(prefix=".codegen_tmp_", dir=parent)
    old_dir = None
    try:
        _stage_bundle(tmpdir, files)

        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

    *,
    timeout_s: float = 600.0,
) -> Iterator[None]:
    """Serialize bundle publication and recover locks left by dead owners."""
    parent = os.path.dirname(layout.artifact_dir) or "."
    bundle_name = os.path.basename(layout.artifact_dir.rstrip(os.sep)) or "generated"
    lock_path = os.path.join(parent, f".{bundle_name}.codegen.lock")
    deadline = time.monotonic() + timeout_s
    descriptor = None
    owner_identity = None
    while descriptor is None:
        try:
            descriptor = os.open(lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
            try:
                os.write(descriptor, str(os.getpid()).encode("ascii"))
            except BaseException:
                os.close(descriptor)
                descriptor = None
                try:
                    os.unlink(lock_path)
                except FileNotFoundError:
                    pass
                raise
            owner_identity = _file_identity(lock_path)
        except FileExistsError:
            if _remove_stale_lock(lock_path):
                continue
            if time.monotonic() >= deadline:
                raise TimeoutError(
                    f"codegen: timed out waiting for artifact lock {lock_path}"
                ) from None
            time.sleep(0.1)
    try:
        yield
    finally:
        os.close(descriptor)
        if owner_identity is not None and _file_identity(lock_path) == owner_identity:
            try:
                os.unlink(lock_path)
            except FileNotFoundError:
                pass


def _remove_stale_lock(lock_path: str) -> bool:
    """Remove ``lock_path`` when its recorded owner process is dead."""
    try:
        identity = _file_identity(lock_path)
        stat_result = os.stat(lock_path)
        with open(lock_path, "r", encoding="ascii") as handle:
            value = handle.read().strip()
    except (FileNotFoundError, OSError, UnicodeError):
        return False

    try:
        pid = int(value)
    except ValueError:
        # There is a tiny interval between O_EXCL creation and writing the PID.
        # Only reclaim malformed locks after that initialization window.
        age_s = max(0.0, time.time() - stat_result.st_mtime)
        stale = age_s >= _LOCK_INITIALIZATION_GRACE_S
    else:
        stale = not _pid_is_alive(pid)
    if not stale or _file_identity(lock_path) != identity:
        return False
    try:
        os.unlink(lock_path)
    except FileNotFoundError:
        return True
    except OSError:
        return False
    _LOGGER.warning("codegen: removed stale artifact lock %s", lock_path)
    return True


def _pid_is_alive(pid: int) -> bool:
    """Return whether ``pid`` still identifies a live process."""
    if pid <= 0:
        return False
    if os.name == "nt":
        return _windows_pid_is_alive(pid)
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    except OSError as exc:
        if exc.errno == errno.ESRCH:
            return False
        if exc.errno == errno.EPERM:
            return True
        return True
    return True


def _windows_pid_is_alive(pid: int) -> bool:
    """Query a Windows process without using ``os.kill(pid, 0)``.

    Windows does not implement the POSIX signal-zero probe; using it may
    terminate the target process. ``OpenProcess`` plus ``GetExitCodeProcess``
    is read-only and reports ``STILL_ACTIVE`` for a live owner.
    """
    kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
    open_process = kernel32.OpenProcess
    open_process.argtypes = (ctypes.c_ulong, ctypes.c_int, ctypes.c_ulong)
    open_process.restype = ctypes.c_void_p
    get_exit_code = kernel32.GetExitCodeProcess
    get_exit_code.argtypes = (ctypes.c_void_p, ctypes.POINTER(ctypes.c_ulong))
    get_exit_code.restype = ctypes.c_int
    close_handle = kernel32.CloseHandle
    close_handle.argtypes = (ctypes.c_void_p,)
    close_handle.restype = ctypes.c_int
    process = open_process(0x1000, False, pid)
    if not process:
        # Access denied proves that the process exists. Other errors are
        # treated conservatively except ERROR_INVALID_PARAMETER (no such PID).
        return ctypes.get_last_error() != 87
    try:
        exit_code = ctypes.c_ulong()
        if not get_exit_code(process, ctypes.byref(exit_code)):
            return True
        return exit_code.value == 259  # STILL_ACTIVE
    finally:
        close_handle(process)


def _file_identity(path: str) -> tuple[int, int, int] | None:
    """Return enough stat identity to avoid unlinking a replaced lock file."""
    try:
        result = os.stat(path)
    except FileNotFoundError:
        return None
    return result.st_ino, result.st_mtime_ns, result.st_size


__all__ = [
    "ArtifactLayout",

    Raises:
        RuntimeError: If the native model does not publish a complete artifact.
    """
    return _build_and_publish(config, artifact_dir)


def _build_and_publish(config: Any, artifact_dir: Optional[str]) -> ArtifactLayout:
    """Build and publish from a process-group-free planning topology."""
    distributed_setup = create_offline_distributed_setup(config)
    normalize_distributed_setup_overrides(distributed_setup, config)

    session = CodegenBuildSession()
    with use_codegen_model_build_overlay(session, artifact_dir) as result, use_codegen_planner_overlay():
        config.model.build(
            distributed_setup=distributed_setup,
            peft_config=config.peft,
            activation_checkpoint=config.activation_checkpoint.mode,
            activation_checkpoint_selection=config.activation_checkpoint.selection,

            compile_config=config.compile,
            model_init_dtype=config.model_init_dtype,
        )

    if result.artifact_layout is None:
        failure = getattr(result.snapshot, "failure", None)
        if failure is not None:
            raise RuntimeError(
                f"codegen {failure.stage} failed: {failure.reason}"
            )
        raise RuntimeError("native model build completed without publishing a codegen artifact")
    return result.artifact_layout


def create_offline_distributed_setup(config: Any) -> DistributedSetup:
    """Project YAML parallel settings into a process-group-free setup."""
    topology_spec = ParallelTopologySpec.from_config(config)
    layout = resolve_parallel_layout(
        topology_spec,
        infer_minimum_world_size(topology_spec),
    )
    topology = LogicalDeviceMesh(
        mesh_dim_names=("dp", "cp", "tp"),
        mesh_shape=(layout.dp_size, topology_spec.cp_size, topology_spec.tp_size),
    )
    mesh_context = MeshContext.from_layout(layout)
    mesh_context.device_mesh = topology
    return DistributedSetup.from_config(
        config,
        mesh_context=mesh_context,
        strategy_enabled=True,
    )

        return 2
    _configure_logging(getattr(args, "verbose", False))
    try:
        return args.func(args)
    except Exception as exc:  # A CLI reports failures without a traceback by default.
        if getattr(args, "verbose", False):
            logger.exception("codegen: %s", exc)
        else:
            logger.error("codegen: %s", exc)
        return 1


def cmd_generate(args: argparse.Namespace) -> int:

def cmd_generate(args: argparse.Namespace) -> int:
    """Observe a meta model and generate a fresh bundle."""
    if int(os.environ.get("WORLD_SIZE", "1")) > 1:
        raise RuntimeError(
            "codegen generate is a single-process CPU/meta command; "
            "run it with python instead of torchrun"
        )
    config = _load_config(args.config, _cli_overrides(args))
    layout = generate_from_native_build(config, artifact_dir=args.output)
    logger.info("codegen: artifact %s", layout.artifact_dir)
    return 0


def _load_config(yaml_path: str, overrides: Optional[list[str]] = None):

def _load_config(yaml_path: str, overrides: Optional[list[str]] = None):
    """Load a resolved TrainerConfig without building trainer runtime components."""
    return config_parser.load_training_config(yaml_path, overrides or [])


def _cli_overrides(args: argparse.Namespace) -> list[str]:
    """Return typed config overrides left after argparse processing."""

        prog="hyper_parallel.codegen",
        description="Generate a model bundle from a training YAML.",
    )
    parser.add_argument("--verbose", action="store_true", help="verbose logging")
    subparsers = parser.add_subparsers(dest="command")

    generate = subparsers.add_parser(
        "generate",
        help="observe the model on meta and generate its bundle",
    )
    generate.add_argument("--config", required=True, help="path to the training YAML")
    generate.add_argument("--output", help="explicit artifact output directory")
    generate.set_defaults(func=cmd_generate)

    return parser

    source, and ``__init__.py``; the caller (manager) stages and atomically
    swaps the complete bundle directory.

    """
    if getattr(meta, "status", None) == "not_required":
        logger.info("codegen: rendered audit-only bundle %s", layout.artifact_dir)
        modeling_name = os.path.basename(layout.modeling_path)
        return {
            modeling_name: _prepare_generated_python(
                copy_original_modeling(meta.source)
            ),
            os.path.basename(layout.diff_path): _NO_PATCH_DIFF,

            ),
            os.path.basename(layout.diff_path): _NO_PATCH_DIFF,
            "__init__.py": _strip_trailing_whitespace(emit_init_file(layout)),
        }
    if patch_plan is None:
        patch_plan = emission_patches_from_meta(meta)
    original_text = copy_original_modeling(meta.source)
    generated_text = _prepare_generated_python(
        emit_modeling_file(meta, patch_plan=patch_plan)
    )

    modeling_name = os.path.basename(layout.modeling_path)
    diff_text = _strip_trailing_whitespace(
        emit_diff(original_text, generated_text, layout)
    )

    logger.info(

    # package ``hyper_parallel_generated.<digest>``, so those siblings must be
    # IN the bundle and registered under that package.  Copy the transitive
    # closure of the source's relative imports here using their original
    # basenames, so the generated bundle remains self-contained.
    resolved = resolve_remote_siblings(meta.source)
    if resolved:
        siblings, sibling_diffs = _emit_remote_siblings(resolved, meta, patch_plan)
        import_map = {
            os.path.basename(path): name
            for path, name in zip(resolved.values(), siblings)
        }
        files[modeling_name] = _rewrite_sibling_imports(
            files[modeling_name], import_map
        )
        files[os.path.basename(layout.diff_path)] = _strip_trailing_whitespace(
            emit_diff(original_text, files[modeling_name], layout)
        )
        files.update(siblings)
        files.update(sibling_diffs)
    verify_generated_source_structure(files)
    verify_recorded_patch_bundle(files, patch_plan=patch_plan)
    return files


def _emit_remote_siblings(

    meta: Any,
    patch_plan: Any,
) -> tuple[dict[str, str], dict[str, str]]:
    """Materialize remote-code siblings and their visible diffs."""
    siblings, originals, import_map = {}, {}, {}
    for path in resolved.values():
        basename = os.path.basename(path)
        with open(path, "r", encoding="utf-8", errors="replace") as handle:
            original = handle.read()
        materialized = _strip_trailing_whitespace(
            _materialize_sibling(original, meta, patch_plan, source_path=path)
        )
        rendered = _prepare_generated_python(materialized)
        output_name = (
            _generated_sibling_name(basename)
            if materialized != _strip_trailing_whitespace(original)
            else basename
        )
        import_map[basename] = output_name
        siblings[output_name] = rendered
        originals[output_name] = (basename, original)
    siblings = {
        name: _rewrite_sibling_imports(text, import_map)
        for name, text in siblings.items()
    }
    diffs = _remote_sibling_diffs(siblings, originals)
    return siblings, diffs


def _remote_sibling_diffs(
    siblings: dict[str, str],

    siblings: dict[str, str],
    originals: dict[str, tuple[str, str]],
) -> dict[str, str]:
    """Render diffs only for remote siblings whose source changed."""
    diffs = {}
    for output_name, rendered in siblings.items():
        basename, original = originals[output_name]
        if rendered == _strip_trailing_whitespace(original):
            continue
        diffs[f"{output_name}.diff"] = _strip_trailing_whitespace(
            unified_diff(
                _normalize_copied_source(original),
                rendered,
                fromfile=f"a/{basename}",

                fromfile=f"a/{basename}",
                tofile=f"b/{output_name}",
            )
        )
    return diffs


def _prepare_generated_python(source_text: str) -> str:
    """Mark copied source as generated and normalize comment-only URLs.

    source units. Repository lint rules for hand-maintained code therefore do
    not apply to those copied units. The markers are comments and do not alter
    imports, executable statements, or runtime behavior.
    """
    text = _normalize_copied_source(source_text)
    text = re.sub(
        r"(^[ \t]*\#.*https?://\S*[A-Za-z0-9])$",
        r"\1 (link).",
        text,
        flags=re.MULTILINE,

        r"\1 (link).",
        text,
        flags=re.MULTILINE,
    )
    if text.startswith(_GENERATED_SOURCE_HEADER):
        return text
    return (
        _GENERATED_SOURCE_HEADER
        + text.rstrip()
        + "\n"
        + _GENERATED_SOURCE_FOOTER

def _normalize_copied_source(source_text: str) -> str:
    """Normalize repository-gate-only defects in copied upstream source."""
    text = _strip_trailing_whitespace(source_text)
    for typo, correction in _COPIED_SOURCE_SPELLING_FIXES.items():
        text = text.replace(typo, correction)
    return text


def _materialize_sibling(
    source_text: str,

    *,
    source_path: str,
) -> str:
    """Apply source-owned strategy and recorded patches to one sibling."""
    source = getattr(meta, "source", None)
    architecture = (
        source.get("architecture")
        if isinstance(source, dict)
        else getattr(source, "architecture", None)
    )
    rendered = render_inline_strategies_for_source_unit(
        source_text,
        meta,
        model_type=architecture,
        patch_plan=patch_plan,

        meta,
        model_type=architecture,
        patch_plan=patch_plan,
    )
    return materialize_recorded_forwards(
        rendered,
        meta,
        patch_plan=patch_plan,
        source_path=source_path,

    )


def _generated_sibling_name(basename: str) -> str:
    stem, extension = os.path.splitext(basename)
    return f"{stem}_gen_npu{extension}"


def _rewrite_sibling_imports(
    source_text: str,

    source_text: str,
    bundled_siblings: dict[str, str],
) -> str:
    """Retarget sibling imports to source units shipped in the bundle."""
    rendered = source_text
    for original, generated in sorted(bundled_siblings.items()):
        original_module = os.path.splitext(original)[0]
        generated_module = os.path.splitext(generated)[0]
        rendered = re.sub(
            rf"(?m)^(\s*from\s+\.+){re.escape(original_module)}(\s+import\s+)",
            rf"\g<1>{generated_module}\g<2>",
            rendered,
        )
        rendered = re.sub(
            rf"(?m)^(\s*from\s+)"
            rf"transformers_modules(?:\.[A-Za-z_][A-Za-z0-9_]*)+\."
            rf"{re.escape(original_module)}(\s+import\s+)",
            rf"\g<1>.{generated_module}\g<2>",

            rf"{re.escape(original_module)}(\s+import\s+)",
            rf"\g<1>.{generated_module}\g<2>",
            rendered,
        )
    return rendered


def _strip_trailing_whitespace(text: str) -> str:
    """Keep generated artifacts clean without changing their line structure."""
    return "".join(f"{line.rstrip()}\n" for line in text.splitlines())


def emit_init_file(layout: ArtifactLayout) -> str:
    """Render the bundle's ``__init__.py`` text.

    """Render the bundle's ``__init__.py`` text.

    The bundle is an audit/export artifact, so this file carries no re-exports.
    """
    del layout
    return (
        _APACHE_LICENSE_HEADER
        + '"""Generated codegen bundle — DO NOT EDIT MANUALLY.\n'
        "\n"

    """
    unless_anchor = _relative_anchor(module_name)
    if unless_anchor is None:
        return text
    text = _rewrite_dynamic_relative_imports(text, unless_anchor)

    try:
        tokens = list(tokenize.generate_tokens(io.StringIO(text).readline))
    except tokenize.TokenError:

        return text

    edits: list[tuple[int, int, str]] = []
    i = 0
    while i < len(tokens):
        edit, i = _relative_import_edit(tokens, i, unless_anchor)
        if edit is not None:
            edits.append(edit)

    if not edits:
        return text
    return _splice(text, edits)

def _relative_import_edit(
    tokens: list[tokenize.TokenInfo], index: int, anchor: list[str]
) -> tuple[tuple[tuple[int, int], tuple[int, int], str] | None, int]:
    """Parse one token position into an absolute relative-import edit."""
    token = tokens[index]
    if token.type != tokenize.NAME or token.string != "from":
        return None, index + 1
    cursor = _skip_import_trivia(tokens, index + 1)
    if cursor >= len(tokens) or not _is_dot_token(tokens[cursor]):
        return None, index + 1
    dots_token = tokens[cursor]
    dots = 0
    while cursor < len(tokens) and _is_dot_token(tokens[cursor]):
        dots += len(tokens[cursor].string)
        cursor += 1
    path, last_token, cursor = _relative_module_path(tokens, cursor)
    import_index = _skip_import_trivia(tokens, cursor)
    if import_index >= len(tokens) or tokens[import_index].string != "import":
        return None, index + 1
    absolute = _absolute_spec(anchor, dots, path)
    if absolute is None:
        return None, index + 1
    end = last_token.end if last_token is not None else dots_token.end
    return (dots_token.start, end, absolute), cursor + 1


def _skip_import_trivia(tokens: list[tokenize.TokenInfo], index: int) -> int:
    """Skip comments and physical/logical newlines between import tokens."""
    trivia = {tokenize.NL, tokenize.NEWLINE, tokenize.COMMENT}
    while index < len(tokens) and tokens[index].type in trivia:
        index += 1
    return index


def _is_dot_token(token: tokenize.TokenInfo) -> bool:
    """Return whether a token contains only relative-import dots."""
    return token.type == tokenize.OP and bool(token.string) and set(token.string) == {"."}


def _relative_module_path(
    tokens: list[tokenize.TokenInfo], index: int

def _relative_module_path(
    tokens: list[tokenize.TokenInfo], index: int
) -> tuple[list[str], tokenize.TokenInfo | None, int]:
    """Collect the NAME (dot NAME)* portion of a relative import."""
    path: list[str] = []
    last_token = None
    while index < len(tokens):
        token = tokens[index]
        if token.type != tokenize.NAME or token.string == "import":
            break
        path.append(token.string)
        last_token = token
        index += 1
        if index >= len(tokens) or tokens[index].string != ".":
            break
        index += 1
    return path, last_token, index


def _rewrite_dynamic_relative_imports(text: str, anchor: list[str]) -> str:
    """Rewrite imports whose resolved package is not valid Python syntax."""
    try:
        tree = ast.parse(text)
    except SyntaxError:
        return text
    edits = []
    for node in ast.walk(tree):
        if not isinstance(node, ast.ImportFrom) or node.level == 0:
            continue
        absolute = _absolute_spec(anchor, node.level, (node.module or "").split("."))
        if absolute is None or _is_import_syntax_safe(absolute):
            continue
        assignments = []
        for alias in node.names:
            if alias.name == "*":
                raise ValueError(
                    f"codegen cannot rewrite a wildcard import from dynamic module {absolute!r}"
                )
            target = alias.asname or alias.name
            assignments.append(
                f"{target} = __import__({absolute!r}, fromlist=[{alias.name!r}]).{alias.name}"
            )
        indent = " " * node.col_offset
        replacement = ("\n" + indent).join(assignments)
        edits.append(
            (
                (node.lineno, node.col_offset),
                (node.end_lineno, node.end_col_offset),
                replacement,

                (node.end_lineno, node.end_col_offset),
                replacement,
            )
        )
    return _splice(text, edits) if edits else text


def _is_import_syntax_safe(module: str) -> bool:
    """Return whether every dotted component can appear in an import statement."""
    return all(
        part.isidentifier() and not keyword.iskeyword(part)
        for part in module.split(".")
    )

        if isinstance(meta.source, dict)
        else getattr(meta.source, "module_name", "")
    )
    text = rewrite_relative_imports(text, module_name)
    text = compose_nested_model_sources(text, meta)
    text, emission_meta = _apply_inline_modeling(text, meta, patch_plan=patch_plan)
    source_path = (
        meta.source.get("file_path")
        if isinstance(meta.source, dict)
        else getattr(meta.source, "file_path", None)
    )
    text = materialize_recorded_forwards(
        text,
        emission_meta,
        patch_plan=patch_plan,
        source_path=(

        source_path=(
            None if getattr(emission_meta, "nested_components", None) else source_path
        ),
    )
    text = _lower_forward_boundaries(text, emission_meta, module_name)
    _verify_recorded_patch_materialization(
        text, emission_meta, patch_plan=patch_plan
    )
    return _banner(meta) + text

    *,
    patch_plan: Any = None,
) -> None:
    """Require every trainer-installed source-visible patch to be materialized."""
    patches = patch_plan
    local_records = (
        getattr(patches, "local_computes", ()) if patches is not None else ()
    )
    replacement_records = (
        getattr(patches, "module_replacements", ()) if patches is not None else ()
    )
    wrapper_records = (
        getattr(patches, "inner_wrappers", ()) if patches is not None else ()
    )
    if not local_records and not replacement_records and not wrapper_records:
        return
    tree = ast.parse(text)
    classes = _class_index(tree)
    materialized = getattr(meta, "materialized_patches", None)
    if isinstance(materialized, list):
        materialized.clear()
    missing = _verify_local_records(tree, classes, local_records, materialized)
    missing += _verify_replacement_records(
        tree, classes, replacement_records, materialized
    )
    missing += _verify_wrapper_records(
        classes, wrapper_records, replacement_records, materialized
    )
    if missing:
        raise RuntimeError(
            "codegen: trainer-installed source-visible patches were not "
            "materialized in the generated model source: " + ", ".join(missing)
        )

def _class_index(tree: ast.AST) -> dict[str, ast.ClassDef]:
    """Index top-level classes by name."""
    return {
        node.name: node
        for node in getattr(tree, "body", ())
        if isinstance(node, ast.ClassDef)
    }

def _class_methods(host: ast.ClassDef) -> set[str]:
    """Return method names declared directly by a class."""
    return {
        node.name
        for node in host.body
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
    }

def _class_state_names(host: ast.ClassDef) -> set[str]:
    """Return simple class attributes declared by assignment."""
    names = set()
    for child in host.body:
        targets = (
            child.targets
            if isinstance(child, ast.Assign)
            else (child.target,)
            if isinstance(child, ast.AnnAssign)

            else (child.target,)
            if isinstance(child, ast.AnnAssign)
            else ()
        )
        names.update(target.id for target in targets if isinstance(target, ast.Name))
    return names


def _record_materialized(
    materialized: Any, fqn: str, host: str, kind: str, source_form: str

def _record_materialized(
    materialized: Any, fqn: str, host: str, kind: str, source_form: str
) -> None:
    """Append one materialization record when metadata is mutable."""
    if isinstance(materialized, list):
        materialized.append(
            {"fqn": fqn, "host_class": host, "kind": kind, "source_form": source_form}
        )

def _verify_local_records(
    tree: ast.AST, classes: dict[str, ast.ClassDef], records: Any, materialized: Any
) -> list[str]:
    """Verify local-compute records in one source unit."""
    missing = []
    for record in records:
        class_name = record.module_type.__name__
        if _bound_function_patch_materialized(tree, record):
            _record_materialized(
                materialized,
                record.module_fqn,
                class_name,
                "local_compute",

                class_name,
                "local_compute",
                "bound_method_toggle_forward",
            )
            continue
        host = classes.get(class_name)
        if host is None:
            continue
        valid = {"_forward_impl", "forward"} <= _class_methods(
            host
        ) and "ep_enable" in _class_state_names(host)
        if valid:
            _record_materialized(
                materialized,
                record.module_fqn,
                class_name,
                "local_compute",

                "local_compute",
                "same_class_toggle_forward",
            )
        else:
            missing.append(f"{record.module_fqn} ({class_name})")
    return missing


def _verify_replacement_records(
    tree: ast.AST, classes: dict[str, ast.ClassDef], records: Any, materialized: Any

def _verify_replacement_records(
    tree: ast.AST, classes: dict[str, ast.ClassDef], records: Any, materialized: Any
) -> list[str]:
    """Verify replacement constructors in one source unit."""
    bound_names = {
        node.id for node in ast.walk(tree) if isinstance(node, ast.Name)
    } | set(classes)
    missing = []
    for record in records:
        name = record.replacement_type.__name__
        visible = name in bound_names and any(
            isinstance(node, ast.Call) and _call_name(node.func) == name
            for node in ast.walk(tree)
        )
        if visible:
            _record_materialized(
                materialized,
                record.module_fqn,
                name,
                "module_replacement",

                "module_replacement",
                "constructor",
            )
        else:
            missing.append(f"{record.module_fqn} (replacement {name})")
    return missing


def _wrapper_identity(record: Any) -> tuple[Any, str, str]:
    """Return target type, host name, and target FQN for a wrapper record."""
    target_type = getattr(record, "target_type", None)
    host_name = (
        target_type.__name__
        if target_type is not None
        else type(record.request.target).__name__
    )
    target_fqn = (
        record.module_fqn
        if record.target_fqn == "self"
        else f"{record.module_fqn}.{record.target_fqn}"
    )
    return target_type, host_name, target_fqn


def _wrapper_source_form(
    classes: dict[str, ast.ClassDef], host: ast.ClassDef, record: Any, replacements: Any

def _wrapper_source_form(
    classes: dict[str, ast.ClassDef], host: ast.ClassDef, record: Any, replacements: Any
) -> str | None:
    """Return the concrete source form that materializes a wrapper."""
    if (
        "forward" in _class_methods(host)
        and "cp_enable" in _class_state_names(host)
        and _class_dispatch_reads_state(host, "cp_enable")
    ):
        return "same_class_toggle_forward"
    subclass = next(
        (
            node
            for node in classes.values()
            if any(

            and "forward" in _class_methods(node)
        ),
        None,
    )
    if subclass is not None:
        return "subclass_forward"
    target_type = getattr(record, "target_type", None)
    replacement_backed = any(
        item.module_fqn == record.module_fqn and item.replacement_type is target_type
        for item in replacements
    )
    if replacement_backed and {"_forward_impl", "forward"} <= _class_methods(host):
        return "replacement_forward_runtime_binding"
    return None


def _verify_wrapper_records(
    classes: dict[str, ast.ClassDef], records: Any, replacements: Any, materialized: Any

def _verify_wrapper_records(
    classes: dict[str, ast.ClassDef], records: Any, replacements: Any, materialized: Any
) -> list[str]:
    """Verify inner-wrapper records in one source unit."""
    missing = []
    for record in records:
        _, host_name, target_fqn = _wrapper_identity(record)
        host = classes.get(host_name)
        if host is None:
            continue
        source_form = _wrapper_source_form(classes, host, record, replacements)
        if source_form:
            _record_materialized(
                materialized, target_fqn, host_name, "inner_wrapper", source_form
            )
        else:
            missing.append(f"{target_fqn} (inner wrapper for {host_name})")
    return missing


def _call_name(node: ast.AST) -> str | None:
    """Return the terminal name of a constructor call expression."""
    if isinstance(node, ast.Name):
        return node.id
    if isinstance(node, ast.Attribute):
        return node.attr
    return None


def verify_recorded_patch_bundle(
    files: dict[str, str],

    *,
    patch_plan: Any = None,
) -> None:
    """Verify recorded local-compute patches across all generated source units."""
    records = (
        getattr(patch_plan, "local_computes", ()) if patch_plan is not None else ()
    )
    wrappers = (
        getattr(patch_plan, "inner_wrappers", ()) if patch_plan is not None else ()
    )
    trees = {
        name: ast.parse(text) for name, text in files.items() if name.endswith(".py")
    }
    missing = [
        f"{record.module_fqn} ({record.module_type.__name__})"
        for record in records
        if not any(_tree_has_local_record(tree, record) for tree in trees.values())
    ]
    if missing:
        raise RuntimeError(
            "codegen: recorded local-compute patches were not materialized "
            "in their owning source units: " + ", ".join(missing)
        )
    replacements = getattr(patch_plan, "module_replacements", ())
    missing_wrappers = [
        f"{record.module_fqn} ({_wrapper_identity(record)[1]})"
        for record in wrappers
        if not any(
            _tree_has_wrapper_record(tree, record, replacements)

            _tree_has_wrapper_record(tree, record, replacements)
            for tree in trees.values()
        )
    ]
    if missing_wrappers:
        raise RuntimeError(
            "codegen: recorded inner-wrapper patches were not materialized "
            "in their owning source units: " + ", ".join(missing_wrappers)
        )
    if any("HPCodegen" in text for text in files.values()):
        raise RuntimeError(
            "codegen: generated bundle contains forbidden HPCodegen class names"
        )

def _tree_has_local_record(tree: ast.AST, record: Any) -> bool:
    """Return whether one source tree materializes a local-compute record."""
    if _bound_function_patch_materialized(tree, record):
        return True
    host = _class_index(tree).get(record.module_type.__name__)
    return (
        host is not None
        and {"_forward_impl", "forward"} <= _class_methods(host)
        and "ep_enable" in _class_state_names(host)
    )

def _tree_has_wrapper_record(tree: ast.AST, record: Any, replacements: Any) -> bool:
    """Return whether one source tree materializes an inner-wrapper record."""
    _, class_name, _ = _wrapper_identity(record)
    host = _class_index(tree).get(class_name)
    if host is None or "forward" not in _class_methods(host):
        return False
    if "cp_enable" in _class_state_names(host) and _class_dispatch_reads_state(
        host, "cp_enable"
    ):
        return True
    target_type = getattr(record, "target_type", None)
    replacement_backed = any(
        item.module_fqn == record.module_fqn and item.replacement_type is target_type
        for item in replacements
    )
    return replacement_backed and {"_forward_impl", "forward"} <= _class_methods(host)


def _class_dispatch_reads_state(host: ast.ClassDef, state_name: str) -> bool:
    """Return whether a class dispatches on one state flag in either layer."""
    methods = (
        child
        for child in host.body
        if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef))
        and child.name in {"forward", "_forward_impl"}