Diff Coverage

            if time.monotonic() >= deadline:
                raise TimeoutError(
                    f"codegen: timed out waiting for artifact lock {lock_path}"
                ) from None
            time.sleep(0.1)
    try:
        os.write(descriptor, str(os.getpid()).encode("ascii"))
        yield
    finally:

        TypeError: If ``snapshot`` is not a :class:`CodegenSnapshot`.
        RuntimeError: If the artifact was generated from another snapshot.
    """
    if not isinstance(snapshot, CodegenSnapshot):
        raise TypeError("snapshot must be a CodegenSnapshot")
    current_signature = signature_from_snapshot(snapshot)
    if meta.signature != current_signature:
        raise RuntimeError(
            "codegen drift: snapshot signature mismatch "

            "snapshot_schema_version and snapshot_status"
        )
    if status == CodegenStatus.NOT_REQUIRED.value:
        if not getattr(meta, "snapshot_reason", None):
            raise RuntimeError(
                "codegen preflight: not_required snapshot is missing its reason"
            )
        if (
            getattr(meta, "param_plan", None)

            or getattr(meta, "plan_application", None)
            or (getattr(meta, "entrypoints", None) or {}).get("parallelize")
            or (getattr(meta, "covered", None) or {}).get("sharding_plan")
        ):
            raise RuntimeError(
                "codegen preflight: not_required snapshot contains executable plan data"
            )
        return
    if status == CodegenStatus.FAILED.value:

        )

    # Every trainable model parameter must be covered by an explicit plan
    # entry (exact FQN or a leaf under a grouped key).
    tied = _expand_tied_param_names(meta)
    uncovered = [
        fqn
        for fqn in _iter_trainable_params(model)
        if not any(fqn == key or fqn.startswith(f"{key}.") for key in planned)

    as its canonical counterpart, which is the side the plan records.  The
    non-canonical aliases therefore have no plan entry of their own by design
    and must not be reported as uncovered.
    """
    tied: set[str] = set()
    for pair in getattr(meta, "tied_pairs", None) or []:
        if len(pair) >= 2:
            tied.add(pair[1])
    return tied


# ---------------------------------------------------------------------------
# helpers

        if not filename.endswith(".py") or filename == "__init__.py":
            continue
        try:
            tree = ast.parse(source_text, filename=filename)
        except SyntaxError as exc:
            errors.append(f"{filename}: invalid Python source ({exc.msg})")
            continue
        for class_node in (
            node for node in ast.walk(tree) if isinstance(node, ast.ClassDef)
        ):
            _verify_class(filename, class_node, errors)

            preflight_integrity_check(config, yaml_path)
            layout = resolve_layout_for_config(config, yaml_path)
            meta = load_codegen_meta(layout.meta_path)
            if meta is not None:
                drifts = verify_drift_intact(meta, yaml_path, artifact_dir=layout.artifact_dir)
                if drifts:
                    failed = 1
                    logger.error("codegen: drift in %s: %s", yaml_path, "; ".join(drifts))
                else:

                    failed = 1
                    logger.error("codegen: drift in %s: %s", yaml_path, "; ".join(drifts))
                else:
                    logger.info("codegen: drift intact %s", yaml_path)
            logger.info("codegen: PASS %s", yaml_path)
        except Exception as exc:  # Aggregate failures while scanning all configs.
            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 generating artifacts."""
    status, installed, spec = verify_transformers_pin(args.transformers_pin)
    logger.info("codegen: transformers pin %s - installed %s (%s)", spec, installed, status)
    return 0


def cmd_train(args: argparse.Namespace) -> int:
    """Launch native training with rank-zero best-effort generation."""

    """Yield YAML files below root while skipping generated and environment directories."""
    if not os.path.isdir(root):
        return
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [name for name in dirnames if name not in _SKIP_DIRS]
        for name in sorted(filenames):
            if name.endswith((".yaml", ".yml")):
                yield os.path.join(dirpath, name)

    """
    if snapshot is not None:
        snapshot_payload = snapshot.to_dict()
        if getattr(meta, "snapshot_status", None) != snapshot_payload["status"]:
            raise ValueError("bundle meta and snapshot status do not match")
    if getattr(meta, "snapshot_status", None) == "not_required":
        logger.info("codegen: rendered audit-only bundle %s", layout.artifact_dir)
        modeling_name = os.path.basename(layout.modeling_path)
        return {

            os.path.basename(layout.diff_path): _NO_PATCH_DIFF,
            "__init__.py": _strip_trailing_whitespace(emit_init_file(layout)),
        }
    if patch_plan is None:
        from hyper_parallel.codegen.emit.patches import emission_patches_from_meta

        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)
    )

    from hyper_parallel.codegen.source.resolver import resolve_remote_siblings

    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)
        if siblings:
            meta.remote_siblings = sorted(siblings)
    verify_generated_source_structure(files)
    verify_recorded_patch_bundle(files, patch_plan=patch_plan)

    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.

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

    *,
    source_path: str,
) -> str:
    """Apply source-owned strategy and recorded patches to one sibling."""
    from hyper_parallel.codegen.emit.recorded import materialize_recorded_forwards
    from hyper_parallel.codegen.inline.pipeline import (
        render_inline_strategies_for_source_unit,
    )

    source = getattr(meta, "source", None)
    architecture = (
        source.get("architecture")
        if isinstance(source, dict)
        else getattr(source, "architecture", None)
    )
    model_type = getattr(meta, "model_class", None) or architecture
    rendered = render_inline_strategies_for_source_unit(
        source_text,
        meta,
        model_type=model_type,
        patch_plan=patch_plan,

        meta,
        model_type=model_type,
        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,

            i += 1
            continue
        # the spec must start with a relative (dots) operator
        j = i + 1
        while j < n and tokens[j].type in (
            tokenize.NL,
            tokenize.NEWLINE,
            tokenize.COMMENT,
        ):

            tokenize.NEWLINE,
            tokenize.COMMENT,
        ):
            j += 1
        if (
            j >= n
            or tokens[j].type != tokenize.OP
            or not set(tokens[j].string) <= {"."}
        ):

        ):
            i += 1
            continue
        dots_tok = tokens[j]
        dots = 0
        while (
            j < n
            and tokens[j].type == tokenize.OP
            and tokens[j].string
            and set(tokens[j].string) == {"."}

            and tokens[j].type == tokenize.OP
            and tokens[j].string
            and set(tokens[j].string) == {"."}
        ):
            dots += len(tokens[j].string)
            j += 1

        # collect the relative module path: NAME ('.' NAME)*
        k = j
        path: list[str] = []
        last_path_tok = None
        while k < n:
            t = tokens[k]

        # the path must be followed by the ``import`` keyword (allow a comment
        # line in between; a multiline symbol list comes *after* ``import``).
        m = k
        while m < n and tokens[m].type in (
            tokenize.NL,
            tokenize.NEWLINE,
            tokenize.COMMENT,
        ):

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

        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(

        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 = _apply_inline_modeling(text, meta, patch_plan=patch_plan)
    from hyper_parallel.codegen.emit.recorded import materialize_recorded_forwards

    source_path = (
        meta.source.get("file_path")
        if isinstance(meta.source, dict)
        else getattr(meta.source, "file_path", None)
    )
    text = materialize_recorded_forwards(
        text,
        meta,
        patch_plan=patch_plan,
        source_path=(None if getattr(meta, "nested_components", None) else source_path),

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


def _verify_recorded_patch_materialization(

    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):

                "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(

        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

    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

            _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:

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."""

def _bound_function_patch_materialized(tree: ast.AST, record: Any) -> bool:
    """Return whether a ``MethodType`` forward has an explicit EP toggle."""
    function_name = getattr(record.original_forward, "__name__", "")
    if not function_name:
        return False
    impl_name = f"{function_name}_hp_impl"
    functions = {
        node.name: node
        for node in getattr(tree, "body", ())

        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
    }
    wrapper = functions.get(function_name)
    if wrapper is None or impl_name not in functions:
        return False
    return _function_reads_ep_toggle(wrapper) and _tree_binds_method_impl(
        tree, impl_name
    )

    from hyper_parallel.codegen.inline import render_inline_modeling

    source = getattr(meta, "source", None)
    architecture = (
        source.get("architecture")
        if isinstance(source, dict)
        else getattr(source, "architecture", None)
    )

        if isinstance(source, dict)
        else getattr(source, "architecture", None)
    )
    model_type = getattr(meta, "model_class", None) or architecture
    return render_inline_modeling(
        text,
        meta,
        model_type=model_type,
        patch_plan=patch_plan,

    behavior (params sharded by the native Phase A reuse, no forward rewrite).
    When the map is absent, the source remains untouched and runtime dispatch
    handles the boundaries.
    """
    from hyper_parallel.codegen.emit.parallel import lower_forward_boundaries

    boundary_classes = getattr(meta, "boundary_classes", None) or {}
    if not boundary_classes:
        return text

def emission_patches_from_meta(meta: Any) -> Optional[EmissionPatchSet]:
    """Rehydrate emitter views from JSON-safe snapshot metadata."""
    data = getattr(meta, "emission_patches", None)
    if not data:
        return None
    replacements = tuple(
        ModuleReplacementPatch(
            module_fqn=item["module_fqn"],
            source_type=_resolve_symbol(item["source_type"]),
            replacement_type=_resolve_symbol(item["replacement_type"]),

            replacement_type=_resolve_symbol(item["replacement_type"]),
        )
        for item in data.get("module_replacements", ())
    )
    local_computes = tuple(
        LocalComputePatch(
            module_fqn=item["module_fqn"],
            module_type=_resolve_symbol(item["module_type"]),
            factory=_resolve_symbol(item["factory"]),

            installed_forward=None,
        )
        for item in data.get("local_computes", ())
    )
    inner_wrappers = tuple(
        InnerWrapperPatch(
            module_fqn=item["module_fqn"],
            target_fqn=item["target_fqn"],
            wrapper=_resolve_symbol(item["wrapper"]),

            target_type=_resolve_symbol(item["target_type"]),
        )
        for item in data.get("inner_wrappers", ())
    )
    return EmissionPatchSet(replacements, local_computes, inner_wrappers)


def local_compute_binding_name(target: Any) -> str:
    """Return the explicit instance attribute used for one compute factory."""

    """Return the explicit instance attribute used for one compute factory."""
    if isinstance(target, Mapping):
        target = target.get("_target_") or target.get("target") or target.get("name")
    else:
        target = getattr(target, "_target_", target)
    if isinstance(target, str):
        name = target.rsplit(":", 1)[-1].rsplit(".", 1)[-1]
    else:
        name = getattr(target, "__name__", "")
    if not name or not name.isidentifier() or keyword.iskeyword(name):
        raise ValueError(
            f"codegen: local-compute target {target!r} has no valid binding name"
        )
    return name

    return name


def _resolve_symbol(path: str) -> Any:
    module_name, separator, qualname = path.partition(":")
    if not separator:
        raise ValueError(f"codegen: invalid patch symbol path {path!r}")
    value = importlib.import_module(module_name)
    for segment in qualname.split("."):
        value = getattr(value, segment)
    return value


__all__ = ["EmissionPatchSet", "emission_patches_from_meta", "local_compute_binding_name"]

    for record in records:
        if source_path is not None and not _record_belongs_to_source(
            record, source_path
        ):
            continue
        original = _unbound(record.original_forward)
        groups[(record.module_type, _source_function_name(original))].append(record)

    rendered = source_text

        groups.items(), key=lambda item: (item[0][0].__name__, item[0][1])
    ):
        _require_one_contract(group, meta)
        if source_function and _has_module_function(rendered, source_function):
            rendered = _materialize_bound_module_function(
                rendered, source_function, group[0], meta
            )
            materialized = True
            continue
        if not _has_source_class(rendered, host_type.__name__):
            continue
        rendered = _materialize_source_class(
            rendered, host_type.__name__, group[0], meta
        )
        materialized = True

        local = injection.get("local_compute_fn")
        module_fqn = injection.get("match")
        class_name = boundary_classes.get(module_fqn)
        if local is None or not module_fqn or class_name not in source_classes:
            continue
        grouped[class_name].append(SimpleNamespace(
            module_fqn=module_fqn,
            factory=_callable_identity(local),
        ))

    param_plan = getattr(meta, "param_plan", None) or {}
    boundary_classes = getattr(meta, "boundary_classes", None) or {}
    for module_fqn, owner_class in boundary_classes.items():
        if owner_class != class_name:
            continue
        entry = param_plan.get(module_fqn)
        if isinstance(entry, dict) and entry.get("is_boundary"):
            entry["skip_source_lowering"] = True

def _callable_identity(value: Any) -> str:
    """Return a stable identity for a frozen callable declaration."""
    if isinstance(value, dict):
        return str(value.get("_target_") or value.get("target") or value.get("name"))
    return str(value)


def _record_belongs_to_source(record: Any, source_path: str) -> bool:
    """Return whether a recorded host is defined by this source unit."""
    candidates = (
        inspect.getsourcefile(record.module_type),
        inspect.getsourcefile(_unbound(record.original_forward)),
    )
    expected = os.path.normcase(os.path.abspath(source_path))
    return any(
        candidate is not None
        and os.path.normcase(os.path.abspath(candidate)) == expected
        for candidate in candidates
    )

        for record in records
    }
    factories = {record.factory for record in records}
    if len(contracts) != 1 or len(factories) != 1:
        fqns = ", ".join(sorted(record.module_fqn for record in records))
        raise RuntimeError(
            "codegen: recorded modules sharing one forward have incompatible "
            f"patch contracts: {fqns}"
        )

def _entry_for_record(meta: Any, record: Any) -> dict[str, Any]:
    entry = (getattr(meta, "param_plan", None) or {}).get(record.module_fqn)
    if entry is None:
        raise RuntimeError(
            f"codegen: recorded patch {record.module_fqn!r} has no frozen plan entry"
        )
    return entry

def _injection_for_record(meta: Any, record: Any) -> dict[str, Any] | None:
    for injection in getattr(meta, "injections", None) or ():
        if injection.get("match") == record.module_fqn:
            return injection
    return None


def _function_info(node: ast.FunctionDef | ast.AsyncFunctionDef, source_text: str) -> FunctionInfo:
    """Build source-index information for one recorded function."""
    lines = source_text.splitlines(keepends=True)
    offsets = [0]
    for line in lines:
        offsets.append(offsets[-1] + len(line))

    def start(line_number: int) -> int:
        return offsets[line_number - 1]

    positional = [arg.arg for arg in (*node.args.posonlyargs, *node.args.args)]
    if positional and positional[0] in ("self", "module"):
        positional.pop(0)
    first_line = min(
        [node.lineno, *(item.lineno for item in node.decorator_list)]
    )
    return FunctionInfo(
        name=node.name,
        body_start=start(node.body[0].lineno),
        body_end=offsets[node.end_lineno],
        def_offset=start(first_line),

            _render_output_redistribute(),
            form,
        )
    else:
        body = _build_forward_body(source_text, injection, func, form, emitted)
    fallback = _render_forward_call(func).replace("outputs =", "return", 1)
    return "\n".join([
        "        if self.ep_enable:",
        textwrap.indent(body, "    "),

    record: Any,
    meta: Any,
) -> str:
    """Turn an instance-bound module function into impl + explicit wrapper."""
    tree = ast.parse(source_text)
    node = next(
        item for item in tree.body
        if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef))
        and item.name == function_name
    )
    info = _function_info(node, source_text)
    impl_name = f"{function_name}_hp_impl"
    original = source_text[info.def_offset:info.body_end]
    original = original.replace(f"def {function_name}(", f"def {impl_name}(", 1)
    if not original.endswith("\n"):
        original += "\n"
    body = textwrap.dedent(_recorded_body(source_text, info, record, meta))
    body_edit = TextEdit(info.body_start, info.body_end, textwrap.indent(body, "    "))
    rendered = apply_edits(
        source_text,
        [TextEdit(info.def_offset, info.def_offset, original + "\n"), body_edit],
    )
    return _bind_module_impl(rendered, function_name, impl_name)


def _materialize_source_class(
    source_text: str,

) -> str:
    """Rewrite a source-defined host even when old meta normalization skips it."""
    func = build_source_index(source_text).find_forward(class_name)
    if func is None:
        raise RuntimeError(
            f"codegen: recorded source host {class_name!r} defines no forward"
        )
    body = _recorded_body(source_text, func, record, meta)
    impl = _build_forward_impl_edit(

def _bind_module_impl(source_text: str, wrapper_name: str, impl_name: str) -> str:
    """Add the instance ``_forward_impl`` binding beside MethodType(forward)."""
    tree = ast.parse(source_text)
    lines = source_text.splitlines(keepends=True)
    offsets = [0]
    for line in lines:
        offsets.append(offsets[-1] + len(line))
    edits: list[TextEdit] = []
    for node in ast.walk(tree):
        if not isinstance(node, ast.Assign) or len(node.targets) != 1:
            continue
        target = node.targets[0]
        value = node.value
        if not (
            isinstance(target, ast.Attribute)
            and target.attr == "forward"
            and isinstance(value, ast.Call)
            and isinstance(value.func, ast.Name)

            and value.args
            and isinstance(value.args[0], ast.Name)
            and value.args[0].id == wrapper_name
        ):
            continue
        owner = ast.unparse(target.value)
        indent = " " * node.col_offset
        binding = f"{indent}{owner}._forward_impl = MethodType({impl_name}, {owner})\n"
        edits.append(TextEdit(offsets[node.lineno - 1], offsets[node.lineno - 1], binding))
    if not edits:
        raise RuntimeError(
            f"codegen: source function {wrapper_name!r} is not installed with MethodType"
        )
    return apply_edits(source_text, edits)


__all__ = ["materialize_recorded_forwards"]

    never recompiles rules or inspects a live model.
    """
    payload = snapshot.to_dict() if hasattr(snapshot, "to_dict") else snapshot
    if not isinstance(payload, dict):
        raise TypeError("replacement snapshot must be a CodegenSnapshot or dict")
    replacements = payload.get("replacements", {})
    declarations = replacements.get("declarations", ())
    applications = replacements.get("applications", ())
    declarations_by_id = {item["rule_id"]: item for item in declarations}

    declarations = replacements.get("declarations", ())
    applications = replacements.get("applications", ())
    declarations_by_id = {item["rule_id"]: item for item in declarations}
    if len(declarations_by_id) != len(declarations):
        raise ValueError("replacement snapshot contains duplicate rule_id values")
    records = []
    for application in applications:
        rule_id = application.get("rule_id")
        declaration = declarations_by_id.get(rule_id)

    for application in applications:
        rule_id = application.get("rule_id")
        declaration = declarations_by_id.get(rule_id)
        if declaration is None:
            raise ValueError(
                f"replacement application references unknown rule_id {rule_id!r}"
            )
        source_identity = application.get("source_type") or declaration.get("source_type")
        if not isinstance(source_identity, str) or not source_identity:
            raise ValueError(
                f"replacement application {rule_id!r} has no resolved source_type"
            )
        source_type = source_identity.replace(":", ".", 1)
        record = {

def canonical_snapshot_payload(snapshot: Any) -> str:
    """Serialize the stable identity projection of a build snapshot."""
    signature_payload = getattr(snapshot, "signature_payload", None)
    if not callable(signature_payload):
        raise TypeError("snapshot must provide signature_payload()")
    return canonical_json(signature_payload())


def signature_from_snapshot(snapshot: Any, *, prefix: int = 12) -> str:

    except ImportError:
        return None
    component = getattr(module, attr, None)
    if not isinstance(component, type):
        return None
    if (
        target.startswith(_COMPONENTS_PREFIX)
        or getattr(component, "_hp_module_replacement", False)
    ):

        target.startswith(_COMPONENTS_PREFIX)
        or getattr(component, "_hp_module_replacement", False)
    ):
        return component
    return None


def _component_replacement_target(component: type) -> ReplacementTarget | None:
    """Read a wrap-style component's own constructor as its replacement target.