Diff Coverage

                else:
                    source_tensor = self._storage_tensor(source)
                    host_view.copy_(source_tensor.detach().reshape(-1).view(slot.shape), non_blocking=False)
                    if source_tensor.device.type != "cpu" and source is slot.tensor:
                        self.release_device_storage(slot)
                slot.host_offset = host_offset
                slot.cpu_tensor = host_view
                slot.bind_tensor(host_view)
                slot.state = "host"

    def is_swappable_tensor(self, tensor: Any, min_numel: int) -> bool:
        """Return whether ``tensor`` can participate in swap."""
        storage_tensor = self._storage_tensor(tensor)
        if not isinstance(storage_tensor, torch.Tensor):
            return False
        if not storage_tensor.is_floating_point():
            return False
        if int(storage_tensor.numel()) < int(min_numel):
            return False
        if storage_tensor.is_sparse:
            return False
        if not storage_tensor.is_contiguous():
            return False
        if storage_tensor.device.type == "cpu":
            return False
        try:
            storage_size = int(storage_tensor.untyped_storage().size())

        try:
            storage_size = int(storage_tensor.untyped_storage().size())
            expected_size = int(storage_tensor.numel()) * int(storage_tensor.element_size())
            if storage_size != expected_size:
                return False
        except RuntimeError:
            return False
        return True

    def storage_nbytes(self, tensor: Any) -> int:
        """Return storage bytes for a Torch tensor."""
        storage_tensor = self._storage_tensor(tensor)
        if not isinstance(storage_tensor, torch.Tensor):
            return 0
        try:
            return int(storage_tensor.untyped_storage().size())
        except RuntimeError:
            return int(storage_tensor.numel()) * int(storage_tensor.element_size())

    def make_cpu_tensor(self, tensor: Any) -> Any:
        """Create a CPU mirror tensor."""
        storage_tensor = self._storage_tensor(tensor)
        if isinstance(storage_tensor, torch.Tensor):
            source = storage_tensor.detach()
            try:
                cpu_tensor = torch.empty_like(source, device="cpu", pin_memory=True)
            except RuntimeError:
                cpu_tensor = torch.empty_like(source, device="cpu")
            cpu_tensor.copy_(source, non_blocking=True)
            return cpu_tensor
        raise ValueError(f"Expected torch.Tensor for CPU mirror, got {type(tensor)!r}.")

    def make_zero_cpu_tensor_like(self, tensor: Any) -> Any:
        """Create a zero-valued CPU mirror without materializing device state."""
        storage_tensor = self._storage_tensor(tensor)
        if not isinstance(storage_tensor, torch.Tensor):
            raise ValueError(f"Expected torch.Tensor for CPU mirror, got {type(tensor)!r}.")
        try:
            cpu_tensor = torch.empty_like(storage_tensor, device="cpu", pin_memory=True)
        except RuntimeError:
            cpu_tensor = torch.empty_like(storage_tensor, device="cpu")
        cpu_tensor.zero_()
        return cpu_tensor

    def make_device_tensor_like(self, param: Any, saved_tensor: Any) -> Any:
        """Create a live state tensor on the parameter device."""
        if not isinstance(saved_tensor, torch.Tensor):
            raise ValueError(f"Expected torch.Tensor in optimizer state, got {type(saved_tensor)!r}.")
        return saved_tensor.detach().to(device=param.device, dtype=saved_tensor.dtype).clone()

    def make_empty_device_tensor_like(self, param: Any, saved_tensor: Any) -> Any:
        """Create an uninitialized live state tensor shell on the parameter device."""
        if not isinstance(saved_tensor, torch.Tensor):
            raise ValueError(f"Expected torch.Tensor in optimizer state, got {type(saved_tensor)!r}.")
        return torch.empty_like(saved_tensor, device=param.device, dtype=saved_tensor.dtype)

    def copy_to_device(self, slot: SwapSlot) -> None:
        """Copy one CPU mirror to device tensor."""
        if slot.state != "host":
            return
        if slot.cpu_tensor is None:
            return
        self._storage_tensor(slot.tensor).copy_(slot.cpu_tensor, non_blocking=True)
        slot.state = "h2d"

    def wait_prefetch_slot(self, slot: SwapSlot) -> None:
        """Torch fallback copies are synchronous on CPU and stream-ordered on device."""
        slot.state = "device"

    def copy_to_cpu(self, slot: SwapSlot) -> None:
        """Copy one device tensor to CPU mirror."""
        source = self._storage_tensor(slot.tensor)
        if slot.cpu_tensor is None:
            slot.cpu_tensor = self.make_cpu_tensor(source)
        else:
            slot.cpu_tensor.copy_(source.detach(), non_blocking=True)
        slot.state = "d2h"

    def wait_offload_slot(self, slot: SwapSlot) -> None:
        """Release device storage after D2H copy completes."""
        if self._storage_tensor(slot.tensor).device.type != "cpu":
            self.release_device_storage(slot)
        slot.state = "host"

    def restore_device_storage(self, slot: SwapSlot) -> None:
        """Restore device tensor storage before H2D."""
        storage_tensor = self._storage_tensor(slot.tensor)
        if storage_tensor.device.type == "cpu":
            return
        storage = storage_tensor.untyped_storage()
        if storage.size() != slot.storage_nbytes:
            storage.resize_(slot.storage_nbytes)

    def release_device_storage(self, slot: SwapSlot) -> None:
        """Release device storage for a swappable tensor."""
        storage_tensor = self._storage_tensor(slot.tensor)
        if storage_tensor.device.type == "cpu":
            return
        storage = storage_tensor.untyped_storage()
        if storage.size() != 0:
            storage.resize_(0)

    def device_handle(self) -> Any:
        """Return the Torch device module (``torch.npu``) that owns streams and events."""
        try:
            return getattr(torch, _DEVICE_TYPE)
        except AttributeError as error:
            raise RuntimeError(
                f"Swap optimizer expects device handle: 'torch.{_DEVICE_TYPE}' failed."
            ) from error

    def current_stream(self) -> Any:

            ) from error

    def current_stream(self) -> Any:
        """Return the current compute stream."""
        return self.device_handle().current_stream()

    def new_stream(self) -> Any:
        """Create the copy stream."""
        return self.device_handle().Stream()

    def stream_context(self, stream: Any):
        """Return a stream context for ``stream``."""
        if stream is None:

    def stream_context(self, stream: Any):
        """Return a stream context for ``stream``."""
        if stream is None:
            return contextlib.nullcontext()
        return self.device_handle().stream(stream)

    def record_event(self, stream: Any = None) -> Any:
        """Record an event on ``stream``."""
        event = self.device_handle().Event()
        if stream is None:
            event.record()
        else:
            event.record(stream)
        return event

    def wait_event(self, event: Any, stream: Any = None) -> None:
        """Make ``stream`` wait for ``event``."""
        if event is None:

    def wait_event(self, event: Any, stream: Any = None) -> None:
        """Make ``stream`` wait for ``event``."""
        if event is None:
            return
        if stream is None:
            event.synchronize()
            return
        event.wait(stream)

    def supports_packed_pipeline(self, batches: Sequence[Sequence[UpdateUnit]]) -> bool:
        """Make the final, step-specific decision to use the packed pipeline."""
        if not self._packed_enabled or not batches:

            for slot in unit.slots
            if slot.swappable
        ]
        if not swappable_slots:
            return False
        devices = {slot.device for slot in swappable_slots}
        eligible = len(devices) == 1 and None not in devices and all(
            slot.packed
            and slot.cpu_tensor is not None

                max_numel_by_dtype[dtype] = max(max_numel_by_dtype.get(dtype, 0), region.numel)
                if device is None and region.slots:
                    device = region.slots[0].device
        if device is None:
            raise RuntimeError("Packed optimizer pipeline has no device-resident state metadata.")

        dtype_layouts = {}
        byte_offset = 0
        for dtype in sorted(max_numel_by_dtype, key=str):

        """Order the compute stream after the batch's packed transfer chain."""
        del staging_index
        ready_event = self._packed_ready_events.get(batch_index)
        if ready_event is None:
            raise RuntimeError(f"Packed optimizer batch {batch_index} has no ready event.")
        self.wait_event(ready_event, self.current_stream())

    def activate_packed_batch(self, batch_index: int, staging_index: int) -> None:
        """Bind each swap slot to its slice of one staging arena."""

        with self.stream_context(copy_stream):
            self.wait_event(compute_event, copy_stream)
            self._copy_packed_to_host(batch_index, staging_index)
            if next_index is not None:
                self._copy_packed_to_device(next_index, staging_index)
            chain_event = self.record_event(copy_stream)
        self._packed_offload_events[batch_index] = chain_event
        self._packed_tail_event = chain_event
        if next_index is not None:
            self._packed_ready_events[next_index] = chain_event

    def wait_packed_offload(self, batch_index: int) -> None:
        """Order the compute stream after a packed transfer chain."""
        offload_event = self._packed_offload_events.get(batch_index)

    def wait_packed_offload(self, batch_index: int) -> None:
        """Order the compute stream after a packed transfer chain."""
        offload_event = self._packed_offload_events.get(batch_index)
        if offload_event is None:
            raise RuntimeError(f"Packed optimizer batch {batch_index} has no offload event.")
        self.wait_event(offload_event, self.current_stream())

    def finish_packed_offload(self, batch_index: int) -> None:
        """Make persistent pinned views authoritative after D2H completion."""

            for slot in region.slots
            if slot.state == "device"
        }
        if active_slots:
            compute_stream = self.current_stream()
            synchronize = getattr(compute_stream, "synchronize", None)
            if synchronize is not None:
                synchronize()
            for slot in active_slots.values():
                slot.cpu_tensor.copy_(self._storage_tensor(slot.tensor).detach(), non_blocking=False)
                slot.bind_tensor(slot.cpu_tensor)
                slot.state = "host"
                slot.event = None
        for arena in self._staging_arenas:
            if arena is None:
                continue
            # Drop views before shrinking the raw storage; otherwise a view

        seen_slots = set()
        for unit in batch:
            for slot in unit.slots:
                if not slot.swappable or not slot.packed or id(slot) in seen_slots:
                    continue
                slots_by_dtype.setdefault(slot.dtype, []).append(slot)
                seen_slots.add(id(slot))

        regions = {}

            host_offset = slots[0].host_offset
            expected_offset = host_offset
            for slot in slots:
                if slot.host_offset != expected_offset:
                    raise RuntimeError(
                        f"Packed optimizer batch has a non-contiguous {dtype} host range at slot {slot.name!r}."
                    )
                expected_offset += slot.numel
            regions[dtype] = _PackedBatchRegion(dtype, host_offset, expected_offset - host_offset, slots)

            self._staging_arenas[staging_index] = arena
            self._drop_packed_views(staging_index)
            return arena

        raw_buffer = arena.raw_buffer
        storage = raw_buffer.untyped_storage()
        if storage.size() < total_bytes:
            storage.resize_(total_bytes)
            arena.dtype_views = {}
            arena.layout_signature = None
            self._drop_packed_views(staging_index)
        raw_buffer.set_(storage, 0, (total_bytes,), (1,))
        return arena

    def _drop_packed_views(self, staging_index: int) -> None:
        """Drop cached views for one arena after its storage/layout changes."""
        self._packed_device_views = {

    def _require_staging_arena(self, staging_index: int) -> _StagingArena:
        arena = self._staging_arenas[staging_index]
        if arena is None:
            raise RuntimeError(f"Packed optimizer staging arena {staging_index} is not materialized.")
        return arena

    @staticmethod
    def _align_bytes(num_bytes: int) -> int:

    @classmethod
    def matches(cls, optimizer: Any) -> bool:
        """Return whether this adapter supports ``optimizer``."""
        return isinstance(optimizer, cls.supported_cls)

    def validate(self) -> None:
        """Validate unsupported optimizer flags."""
        for group in self.optimizer.param_groups:

    def prepare_step(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
        """Initialize lazy state and collect this step's update units."""
        if args or kwargs:
            raise ValueError("Torch swap optimizer step does not support closure or extra arguments.")
        if self.runtime.packed_enabled:
            return self._prepare_packed_step()

        units = []

        units = []
        for group_index, group in enumerate(self.optimizer.param_groups):
            if self.is_new_adamw:
                group["step"] = (group.get("step") or 0) + 1
            for param in group["params"]:
                grad = getattr(param, "grad", None)
                if grad is None:
                    continue
                if getattr(grad, "is_sparse", False):
                    raise ValueError("Swap optimizer only supports dense Adam/AdamW gradients.")
                state = self.optimizer.state[param]
                self._init_param_state(param, grad, group)
                slots = self._build_slots(param, state)
                units.append(UpdateUnit(

                grad = getattr(param, "grad", None)
                state = self.optimizer.state.get(param)
                if grad is not None:
                    if getattr(grad, "is_sparse", False):
                        raise ValueError("Swap optimizer only supports dense Adam/AdamW gradients.")
                    state = self.optimizer.state[param]
                    self._init_param_state(param, grad, group)
                if state:
                    self._register_present_slots(param, state)

                if state:
                    self._register_present_slots(param, state)
                has_slots = any((id(param), key) in self._slots for key in self._configured_state_keys())
                if grad is None and not has_slots:
                    continue
                records.append((group_index, param, grad))

        self.runtime.prepare_packed_host(self._ordered_slots())
        self.publish_packed_state()

        for group_index, param, grad in records:
            state = self.optimizer.state[param]
            slots = self._build_slots(param, state)
            if grad is None and not any(slot.swappable and slot.packed for slot in slots):
                continue
            units.append(UpdateUnit(
                adapter_index=group_index,
                param=param,
                grad=grad,

        return {"units": units}

    def iter_update_units(self, step_context: Dict[str, Any]) -> List[UpdateUnit]:
        """Return units collected in ``prepare_step``."""
        return step_context["units"]

    def initial_slots(self) -> Iterable[SwapSlot]:
        """Discover optimizer states materialized before the swap wrapper was created."""
        slots = []

            exp_avg_sqs = []
            max_exp_avg_sqs = []
            for unit in units:
                if unit.grad is None:
                    continue
                state = self.optimizer.state[unit.param]
                params.append(unit.param)
                grads.append(unit.grad)
                exp_avgs.append(self._slot_tensor(unit, "exp_avg", state["exp_avg"]))

                grads.append(unit.grad)
                exp_avgs.append(self._slot_tensor(unit, "exp_avg", state["exp_avg"]))
                exp_avg_sqs.append(self._slot_tensor(unit, "exp_avg_sq", state["exp_avg_sq"]))
                if group.get("amsgrad", False):
                    max_exp_avg_sqs.append(
                        self._slot_tensor(unit, "max_exp_avg_sq", state["max_exp_avg_sq"])
                    )
                if self.is_new_adamw:
                    state_steps.append(None)
                else:
                    state_steps.append(state["step"])

            if not params:
                continue

            if self.is_new_adamw:
                if params and params[0].device.type == "cpu":
                    # torch.optim._functional.adamw increments tensor state_steps
                    # internally. New AdamW already advanced group["step"] in
                    # prepare_step(), so feed step - 1 to preserve outer-step
                    # semantics for CPU-only tests.
                    step_tensor = torch.tensor(float(group["step"] - 1), dtype=torch.float32)
                    torch.optim._functional.adamw(
                        params,
                        grads,
                        exp_avgs,
                        exp_avg_sqs,

                        found_inf=None,
                        has_complex=False,
                    )
                else:
                    _new_adamw_func()(
                        params,
                        grads,
                        exp_avgs,
                        exp_avg_sqs,

                        weight_decay=group["weight_decay"],
                        eps=group["eps"],
                        maximize=group["maximize"],
                    )
                continue

            func = getattr(torch.optim._functional, self.functional_name)
            kwargs = {
                "amsgrad": group["amsgrad"],

                "has_complex": False,
            }
            if self.functional_name == "adam":
                if "decoupled_weight_decay" in inspect.signature(func).parameters:
                    kwargs["decoupled_weight_decay"] = self.decoupled_weight_decay or group.get(
                        "decoupled_weight_decay", False
                    )
            func(params, grads, exp_avgs, exp_avg_sqs, max_exp_avg_sqs, state_steps, **kwargs)

            func(params, grads, exp_avgs, exp_avg_sqs, max_exp_avg_sqs, state_steps, **kwargs)

    def finish_step(self, step_context: Any) -> Any:
        """Finish one outer optimizer step."""
        del step_context

    def all_slots(self) -> Iterable[SwapSlot]:
        """Iterate known swap slots."""
        return tuple(self._ordered_slots())

                    slot = self._slots.get((id(param), key))
                    if slot is None or not slot.packed or slot.cpu_tensor is None:
                        continue
                    if state is None:
                        state = self.optimizer.state[param]
                    state[key] = slot.cpu_tensor

    def export_swappable_state(self, state_dict: Dict[str, Any]) -> Dict[str, Any]:
        """Build a checkpoint-safe Torch optimizer state dict.

            ids_in_order.extend(group["params"])
        for param, param_id in zip(params_in_order, ids_in_order):
            saved_state = state_dict.get("state", {}).get(param_id)
            if not saved_state:
                continue
            exported_state = {}
            for key in self._state_keys_for_param(param):
                slot = self._slots.get((id(param), key))
                if slot is not None and slot.state == "host" and slot.cpu_tensor is not None and key in saved_state:
                    exported_state[key] = slot.cpu_tensor.detach().clone()
            for key, value in saved_state.items():
                if key not in exported_state:
                    exported_state[key] = copy.deepcopy(value)
            exported["state"][param_id] = exported_state

        removed: Dict[int, Dict[str, Any]] = {}
        swappable_keys = self._configured_state_keys()
        for param_id, saved_state in list(stripped.get("state", {}).items()):
            if not isinstance(saved_state, dict):
                continue
            for key in swappable_keys:
                if key in saved_state:
                    removed.setdefault(param_id, {})[key] = saved_state.pop(key)
        return stripped, removed

            current_params.extend(current_group["params"])
        for saved_id, param in zip(saved_ids, current_params):
            key_to_tensor = removed.get(saved_id, {})
            if not key_to_tensor:
                continue
            state = self.optimizer.state[param]
            for key, saved_tensor in key_to_tensor.items():
                cpu_tensor = self._cast_swappable_tensor_to_cpu(param, saved_tensor)
                if self.runtime.packed_enabled and self.runtime.is_packable_template(param, self.config.min_numel):

                        slot = self._make_slot(key, logical_tensor)
                        state[key] = logical_tensor
                        self.runtime.release_device_storage(slot)
                    else:
                        slot = self._make_slot(key, None, template=param)
                        state[key] = cpu_tensor
                        slot.tensor = cpu_tensor
                    slot.cpu_tensor = cpu_tensor
                    slot.state = "host"
                    self._slots[(id(param), key)] = slot
                    continue

                    continue
                device_tensor = self.runtime.make_empty_device_tensor_like(param, cpu_tensor)
                slot = self._make_slot(key, device_tensor)
                if slot.swappable:
                    state[key] = device_tensor
                    slot.cpu_tensor = self.runtime.make_cpu_tensor(cpu_tensor)
                    slot.state = "host"
                    self._slots[(id(param), key)] = slot
                    self.runtime.release_device_storage(slot)
                else:
                    device_tensor = self._cast_state_tensor_like_torch(
                        param,
                        saved_tensor,

            )
            state["step"] = torch.zeros((), dtype=torch.float32, device=step_device)
        state_keys = ["exp_avg", "exp_avg_sq"]
        if group.get("amsgrad", False):
            state_keys.append("max_exp_avg_sq")
        configured_keys = set(self._configured_state_keys())
        for key in state_keys:
            if key in state or (id(param), key) in self._slots:
                continue

        slots = []
        for key in self._state_keys_for_param(param):
            tensor = state.get(key)
            if tensor is None:
                continue
            slot = self._slots.get((id(param), key))
            if slot is None:
                slot = self._make_slot(key, tensor)
                self._slots[(id(param), key)] = slot

            if slot is None:
                slot = self._make_slot(key, tensor)
                self._slots[(id(param), key)] = slot
            elif slot.tensor is not tensor and slot.cpu_tensor is not tensor:
                slot = self._make_slot(key, tensor)
                self._slots[(id(param), key)] = slot
            slots.append(slot)
        return slots

    def _make_slot(self, key: str, tensor: Any, template: Optional[Any] = None) -> SwapSlot:

    def _make_slot(self, key: str, tensor: Any, template: Optional[Any] = None) -> SwapSlot:
        """Create a swap slot for a state tensor or a packed-state template."""
        metadata_tensor = tensor if tensor is not None else template
        if metadata_tensor is None:
            raise ValueError(f"Cannot build swap slot {key!r} without a tensor or template.")
        if tensor is None:
            swappable = self.runtime.is_packable_template(metadata_tensor, self.config.min_numel)
        else:
            swappable = self.runtime.is_swappable_tensor(tensor, self.config.min_numel)

        for key in keys:
            if key in state:
                result.append(key)
            elif self.config.state_keys is not None:
                raise ValueError(f"Requested state key '{key}' is not present for parameter.")
        return tuple(result)

    @staticmethod
    def _slot_tensor(unit: UpdateUnit, key: str, fallback: Any) -> Any:

        keys = self.config.state_keys or self._default_state_keys()
        result = []
        for key in keys:
            if key == "master_param":
                if self.config.state_keys is not None:
                    raise ValueError(f"Requested state key '{key}' is not available for {type(self.optimizer)!r}.")
                continue
            result.append(key)
        return tuple(result)

    def _cast_state_tensor_like_torch(

            key: str,
    ) -> Any:
        """Cast a loaded state tensor using PyTorch optimizer load semantics."""
        if not isinstance(saved_tensor, torch.Tensor):
            raise ValueError(f"Expected torch.Tensor in optimizer state, got {type(saved_tensor)!r}.")
        process = getattr(torch.optim.Optimizer, "_process_value_according_to_param_policy", None)
        if process is not None:
            return process(param, saved_tensor, saved_id, saved_groups, key).detach().clone()
        if key == "step":
            return saved_tensor.detach().clone()
        if param.is_floating_point():
            return saved_tensor.detach().to(dtype=param.dtype, device=param.device).clone()
        return saved_tensor.detach().to(device=param.device).clone()

    def _cast_swappable_tensor_to_cpu(self, param: Any, saved_tensor: Any) -> Any:
        """Cast swappable state dtype like PyTorch while keeping values on CPU."""
        if not isinstance(saved_tensor, torch.Tensor):
            raise ValueError(f"Expected torch.Tensor in optimizer state, got {type(saved_tensor)!r}.")
        if param.is_floating_point():
            return saved_tensor.detach().to(dtype=param.dtype, device="cpu")
        return saved_tensor.detach().to(device="cpu")

    @staticmethod
    def _default_state_keys() -> Tuple[str, ...]:
        return ("exp_avg", "exp_avg_sq", "max_exp_avg_sq")

def _new_adamw_func() -> Callable[..., Any]:
    """Return hyper-parallel's functional AdamW without importing it at module load."""
    from hyper_parallel.core.optimizer.adamw import (  # pylint: disable=import-outside-toplevel
        adamw as new_adamw_func,
    )
    return new_adamw_func


class SwapOptimizer(torch.optim.Optimizer):
    """Torch optimizer wrapper for Adam/AdamW state swap."""

        self.adapter.publish_packed_state()

    def __getattr__(self, name: str) -> Any:
        """Delegate unknown attributes to the base optimizer."""
        return getattr(self.optimizer, name)

    @property
    def param_groups(self):
        """Proxy parameter groups."""
        return self.optimizer.param_groups

    @param_groups.setter
    def param_groups(self, value) -> None:
        self.optimizer.param_groups = value

    @property
    def state(self):
        """Proxy optimizer state."""

    @property
    def defaults(self):
        """Proxy optimizer defaults."""
        return self.optimizer.defaults

    def add_param_group(self, param_group: Dict[str, Any]) -> None:
        """Proxy param group addition."""
        self.optimizer.add_param_group(param_group)

    def zero_grad(self, set_to_none: bool = True) -> None:
        """Proxy gradient clearing."""
        self.optimizer.zero_grad(set_to_none=set_to_none)

    def step(self, closure: Optional[Any] = None) -> Any:
        """Run one optimizer step with pipeline state swap."""
        if closure is not None:
            raise ValueError("Swap optimizer does not support closure.")
        with self._no_grad_context():
            step_context = self.adapter.prepare_step()
            units = self.adapter.iter_update_units(step_context)
            batches = self.runtime.partition(units)
            self.runtime.run_pipeline(batches, step_context, self.adapter.step_batch)
            return self.adapter.finish_step(step_context)

    def state_dict(self) -> Dict[str, Any]:
        """Return optimizer state dict using CPU mirrors for swappable tensors."""
        return self.adapter.checkpoint_state_dict()

        )

    @contextlib.contextmanager
    def _no_grad_context(self) -> Iterable[None]:
        with torch.no_grad():
            yield