Diff Coverage

        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)

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

        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:

    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:

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

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__ = [

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

    # 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

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

    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)

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

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

    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
    )

) -> tuple[str, Any]:
    """Render the artifact through the adapter-driven inline source pipeline."""

    source = getattr(meta, "source", None)
    architecture = (
        source.get("architecture")
        if isinstance(source, dict)
        else getattr(source, "architecture", None)
    )
    return render_inline_modeling_with_plan(
        text,
        meta,
        model_type=architecture,
        patch_plan=patch_plan,

    / ``_hp_tp`` instance attributes that ``hp_install_boundaries`` binds
    at runtime — no per-boundary global constant is emitted.
    """
    if _plan_field(frozen_plan, "status") == "not_required":
        raise ValueError("codegen emit: not_required snapshot has no parallel boundaries")
    edits: list[TextEdit] = []
    for class_name, form, emitted, func, injection, deferred_bias in iter_emitted_forms(
        source_text,
        frozen_plan,

def emission_patches_from_meta(meta: Any) -> Optional[EmissionPatchSet]:
    """Rehydrate emitter views from plain snapshot data."""
    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
            )
            continue
        if not _has_source_class(rendered, host_type.__name__):
            continue
        rendered = _materialize_source_class(
            rendered, host_type.__name__, group[0], meta
        )
        _mark_class_source_lowering_skipped(meta, host_type.__name__)

        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"):
            param_plan[module_fqn] = {**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"]

    rules or inspects a live model.
    """
    payload = snapshot._emission_view() if hasattr(snapshot, "_emission_view") else snapshot  # pylint: disable=protected-access
    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 = {

        )
        if local is not None:
            local_computes.append(local)
        if wrapper is not None:
            inner_wrappers.append(wrapper)
    return {
        "module_replacements": replacements,
        "local_computes": local_computes,
        "inner_wrappers": inner_wrappers,