sbd_solver.py

# This code is a Qiskit project.
#
# (C) Copyright IBM 2026.
#
# This code is licensed under the Apache License, Version 2.0. You may
# obtain a copy of this license in the LICENSE.txt file in the root directory
# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
#
# Any modifications or derivative works of this code must retain this
# copyright notice, and modified files need to carry a notice indicating
# that they have been altered from the originals.

"""
SBD solver wrapper compatible with qiskit-addon-sqd interface.

This module provides functions that wrap the SBD (Selected Basis Diagonalization)
library to be compatible with the qiskit-addon-sqd diagonalize_fermionic_hamiltonian
interface, similar to how qiskit-addon-dice-solver works.
"""

from __future__ import annotations

import tempfile
import shutil
from pathlib import Path
from collections.abc import Sequence
from typing import Callable

import numpy as np
from mpi4py import MPI

# Bits packed into each size_t of SBD's ``std::vector<size_t>`` bitstring
# representation. RIKEN's documented default is 20 (see the --bit_length option
# in apps/chemistry_tpb_selected_basis_diagonalization/README.md), which is also
# what run_sbd_diag.py defaults to.
#
# This must stay <= 63. bitadvance() in framework/bit_manipulation.h computes
#     size_t d = (((size_t) 1) << bit_length) - 1;
# so bit_length == 64 shifts a 64-bit size_t by 64, which is undefined behavior;
# in practice the shift count is masked to 0 and the mask collapses to d == 0.
SBD_DEFAULT_BIT_LENGTH = 20

try:
    from pyscf import tools as pyscf_tools
except ImportError:
    pyscf_tools = None

try:
    from qiskit_addon_sqd.fermion import SCIResult, SCIState
except ImportError:
    SCIResult = None
    SCIState = None


def _resolve_backend(device_config=None):
    """Resolve a backend module from a DeviceConfig or the default.

    Reads ``device_config.device`` (string key, e.g. 'cpu', 'gpu',
    'gpu-omp') and routes through ``sbd.get_backend()`` which handles
    aliases ('cuda', 'gpu-thrust', 'gpu-nvhpc-omp', etc.).
    """
    from . import get_backend
    if device_config is not None:
        return get_backend(device_config.device)
    return get_backend()


def solve_sci(
    ci_strings: tuple[np.ndarray, np.ndarray],
    one_body_tensor: np.ndarray,
    two_body_tensor: np.ndarray,
    norb: int,
    nelec: tuple[int, int],
    *,
    spin_sq: float | None = None,
    mpi_comm: MPI.Comm | None = None,
    sbd_config: dict | None = None,
    temp_dir: str | Path | None = None,
    clean_temp_dir: bool = True,
    device_config=None,
    fcidump_path: str | Path | None = None,
) -> SCIResult:
    """
    Diagonalize Hamiltonian in subspace defined by CI strings using SBD.

    Args:
        ci_strings: Pair (strings_a, strings_b) of CI string arrays.
        one_body_tensor: The one-body tensor of the Hamiltonian.
        two_body_tensor: The two-body tensor of the Hamiltonian.
        norb: The number of spatial orbitals.
        nelec: The numbers of alpha and beta electrons.
        spin_sq: Target value for total spin squared (unused by SBD).
        mpi_comm: MPI communicator. If None, uses MPI.COMM_WORLD.
        sbd_config: Dictionary of SBD configuration parameters.
        temp_dir: Directory for temporary files.
        clean_temp_dir: Whether to delete intermediate files.
        device_config: DeviceConfig object to select CPU/GPU backend.
        fcidump_path: If set, load the FCIDUMP directly from this path on
            every rank and skip the tensor->file round-trip. MUST be on a
            filesystem visible to every rank (shared on multi-node runs).
            When None (default), rank 0 writes a regenerated FCIDUMP into
            ``temp_dir`` and every rank opens that file, which requires
            ``temp_dir`` to be shared for multi-node runs.

    Returns:
        The diagonalization result as SCIResult.
    """
    if SCIResult is None:
        raise ImportError(
            "qiskit-addon-sqd is required for solve_sci. "
            "Install with: pip install qiskit-addon-sqd"
        )
    backend = _resolve_backend(device_config)

    if mpi_comm is None:
        mpi_comm = MPI.COMM_WORLD
    mpi_rank = mpi_comm.Get_rank()

    sbd_dir, owns_sbd_dir = _make_sbd_dir(mpi_comm, mpi_rank, temp_dir)

    try:
        fcidump, ecore_offset = _load_or_regenerate_fcidump(
            backend, mpi_rank, mpi_comm, sbd_dir, fcidump_path,
            one_body_tensor, two_body_tensor, norb, nelec,
        )

        return _solve_sci_core(
            ci_strings,
            norb=norb,
            nelec=nelec,
            spin_sq=spin_sq,
            mpi_comm=mpi_comm,
            mpi_rank=mpi_rank,
            sbd_config=sbd_config,
            sbd_dir=sbd_dir,
            backend=backend,
            fcidump=fcidump,
            device_config=device_config,
            ecore_offset=ecore_offset,
        )
    finally:
        if clean_temp_dir and owns_sbd_dir and mpi_rank == 0:
            shutil.rmtree(sbd_dir, ignore_errors=True)


def _solve_sci_core(
    ci_strings: tuple[np.ndarray, np.ndarray],
    *,
    norb: int,
    nelec: tuple[int, int],
    spin_sq: float | None,
    mpi_comm,
    mpi_rank: int,
    sbd_config: dict | None,
    sbd_dir: Path,
    backend,
    fcidump,
    device_config=None,
    ecore_offset: float = 0.0,
) -> SCIResult:
    """
    Inner diagonalization kernel that operates on a pre-loaded FCIDUMP object.

    Separated from solve_sci so that solve_sci_batch can write and load
    the FCIDUMP only once and reuse it across all batches.
    """
    strings_a, strings_b = ci_strings

    # Build the config first: it carries the effective bit_length (possibly
    # overridden by the caller), and the determinants must be packed with the
    # same value the C++ engine will use to interpret them.
    sbd_data = _create_sbd_config(sbd_config, backend, device_config)

    # ``SCIResult.carryover`` must be ordered by DESCENDING MARGINAL WEIGHT: the loop
    # takes a solver's carryover as given (fermion.py, ``_select_carryover`` returns it
    # before any ranking) and a later truncation to ``max_dim`` keeps the leading
    # entries, so a wrong order silently drops the strings that mattered.
    #
    # Only carryover_type 1 satisfies that by itself. It ranks half-determinants by
    # p(D^alpha) = sum_beta |c|^2 -- exactly the addon's marginal weight -- and writes
    # them out in that order, so its eigenvector has no remaining consumer and the
    # |adet| x |bdet| dump can be skipped entirely.
    #
    # Types 2 and 3 singles-extend their selection, and SinglesExtendHalfdets ends in
    # sort_bitarray (upstream extend.h), which re-sorts into CANONICAL order because
    # SBD indexes subspaces by binary search. That destroys the weight order, and the
    # generated determinants never had a weight at all. So those types keep the dump
    # and we re-rank their carryover from the amplitudes below.
    skip_wf = sbd_data.carryover_type == 1 and _addon_accepts_carryover()

    adet = _ci_strings_to_sbd_dets(strings_a, norb, backend, sbd_data.bit_length)
    bdet = _ci_strings_to_sbd_dets(strings_b, norb, backend, sbd_data.bit_length)

    # Use .bin extension to trigger SBD's fast binary write path
    # (SaveMatrixFormWF in restart.h checks extension: .bin -> raw doubles)
    wf_dump_file = sbd_dir / "wavefunction.bin"
    if not skip_wf:
        sbd_data.dump_matrix_form_wf = str(wf_dump_file)

    results = backend.tpb_diag(
        mpi_comm, sbd_data, fcidump, adet, bdet, loadname="", savename=""
    )

    # Rank 0 reads the wavefunction file; Barrier ensures it's flushed.
    mpi_comm.Barrier()

    if mpi_rank != 0:
        # Placeholder: only rank 0 is given the energy and the wavefunction. With
        # skip_wf there is no SCIState to stand in, so carry an empty carryover
        # instead -- a result with neither field satisfies no caller, and
        # qiskit-addon-sqd rejects it outright.
        placeholder = (
            {"carryover": (np.array([], dtype=np.int64), np.array([], dtype=np.int64))}
            if skip_wf
            else {}
        )
        return SCIResult(
            0.0,
            None
            if skip_wf
            else SCIState(
                amplitudes=np.empty((0, 0), dtype=np.float64),
                ci_strs_a=np.array([], dtype=np.int64),
                ci_strs_b=np.array([], dtype=np.int64),
                norb=norb,
                nelec=nelec,
            ),
            orbital_occupancies=(
                np.zeros(norb, dtype=np.float64),
                np.zeros(norb, dtype=np.float64),
            ),
            **placeholder,
        )

    # --- rank 0 only ---

    # Subtract ECORE so SCIResult.energy is the pure electronic piece,
    # matching the contract callers rely on (they add nuclear_repulsion
    # separately). The regenerate path writes ECORE=0, so offset=0 there;
    # the direct-load path passes whatever ECORE lives in the user file.
    energy = results["energy"] - ecore_offset
    density = np.array(results["density"])
    occupancies_a = density[::2]
    occupancies_b = density[1::2]
    occupancies = (occupancies_a, occupancies_b)

    # Read wavefunction coefficients from the binary dump.
    #
    # SaveMatrixFormWF (upstream chemistry/tpb/restart.h) writes the FULL input
    # subspace -- adet x bdet, row-major, and it self-checks that total -- never
    # the carryover subset, whatever carryover_type is set to. Sizing this read
    # against the carryover counts therefore never matched under the default
    # carryover_type=1 / ratio=0.1, and the previous code answered the mismatch by
    # substituting a uniform array: a correctly-shaped, normalised matrix that
    # looks like a wavefunction and carries no information. That was silent, and
    # it mattered -- qiskit-addon-sqd selects the next iteration's determinants by
    # amplitude magnitude (fermion.py, _carryover_* / weights_a / weights_b), so
    # every iteration after the first was seeded from uniform weights. See #19.
    #
    # The dump is requested above whenever these amplitudes are needed at all, so
    # neither a missing file nor a size mismatch is a situation to paper over: both
    # mean the run did not do what was asked, and a loud failure is the only honest
    # response.
    # Label the amplitudes with the lists that were actually diagonalized, not with
    # the inputs. _ci_strings_to_sbd_dets ends in sort_bitarray, which sorts into
    # SBD's canonical order AND removes duplicates, so adet/bdet can differ from
    # strings_a/strings_b -- and the dump is written in adet/bdet order. Deriving the
    # labels from adet/bdet makes the amplitudes and their labels come from one list
    # by construction, instead of relying on the two orders agreeing.
    if skip_wf:
        # Report the dimension sci_state would otherwise have carried, so the
        # per-iteration series survives without the eigenvector.
        n_a, n_b = len(adet), len(bdet)
        print(f"  [sbd] solved subspace {n_a:_} x {n_b:_} = {n_a * n_b:_}")
        sci_state = None
    else:
        solved_strings_a = _sbd_dets_to_ci_strings(
            adet, norb, backend, sbd_data.bit_length
        )
        solved_strings_b = _sbd_dets_to_ci_strings(
            bdet, norb, backend, sbd_data.bit_length
        )
        n_a = len(solved_strings_a)
        n_b = len(solved_strings_b)
        if not wf_dump_file.exists():
            raise RuntimeError(
                f"SBD wrote no wavefunction dump at {wf_dump_file}. It is requested "
                "whenever the amplitudes are needed, so a missing file means the "
                "diagonalization did not complete as expected."
            )
        flat = np.fromfile(str(wf_dump_file), dtype=np.float64)
        if flat.size != n_a * n_b:
            raise RuntimeError(
                f"wavefunction dump at {wf_dump_file} holds {flat.size} amplitudes, "
                f"expected {n_a * n_b} ({n_a} alpha x {n_b} beta over the "
                "diagonalized subspace). SaveMatrixFormWF writes the full subspace; "
                "a different size means the dump and the subspace have diverged."
            )

        amplitudes = flat.reshape(n_a, n_b)
        _to_alpha_first_signs(amplitudes, solved_strings_a, solved_strings_b, norb)
        sci_state = SCIState(
            amplitudes=amplitudes,
            ci_strs_a=solved_strings_a,
            ci_strs_b=solved_strings_b,
            norb=norb,
            nelec=nelec,
        )

    rdm1, rdm2 = assemble_rdms(results, norb)

    # When SBD selected its own carryover, hand it to the loop rather than letting it
    # re-derive the same choice by thresholding our amplitudes. qiskit-addon-sqd takes
    # a solver's carryover as given -- ``_select_carryover`` returns it before applying
    # any threshold, ratio or ranking -- so ``carryover_threshold`` does not apply AND
    # the order we hand over is the order a ``max_dim`` truncation will keep.
    carryover_a, carryover_b = _extract_carryover(
        results, norb, backend, sbd_data.bit_length
    )
    if carryover_a is not None and sci_state is not None:
        # Types 2/3 come back in canonical order (see the skip_wf comment); rank them
        # the way the addon would have. Type 1 never reaches here with a sci_state.
        carryover_a, carryover_b = _rank_carryover_from_amplitudes(
            carryover_a, carryover_b, sci_state
        )
    extra = {}
    if carryover_a is not None and _addon_accepts_carryover():
        extra["carryover"] = (carryover_a, carryover_b)

    if skip_wf and "carryover" not in extra:
        raise RuntimeError(
            f"carryover_type={sbd_data.carryover_type} was requested, so the "
            "wavefunction was not dumped, but SBD selected no carryover "
            "determinants -- leaving nothing to build the next subspace from. "
            f"Either the selection was too aggressive (ratio={sbd_data.ratio}, "
            f"threshold={sbd_data.threshold}) or this SBD build does not implement "
            f"carryover_type={sbd_data.carryover_type}. Set carryover_type=0 to go "
            "back to selecting from the amplitudes."
        )

    return SCIResult(
        energy, sci_state, orbital_occupancies=occupancies, rdm1=rdm1, rdm2=rdm2, **extra
    )


def _addon_accepts_carryover() -> bool:
    """True when the installed qiskit-addon-sqd has ``SCIResult.carryover``.

    The field arrived with qiskit-addon-sqd#369. On an older addon, passing it
    would be a TypeError, so the carryover is simply not reported and the loop
    falls back to thresholding the amplitudes as it always has.
    """
    if SCIResult is None:
        return False
    import dataclasses

    return any(f.name == "carryover" for f in dataclasses.fields(SCIResult))


def _rank_by_marginal_weight(
    carryover: np.ndarray, solved: np.ndarray, weights: np.ndarray
) -> np.ndarray:
    """Order one spin sector's carryover by descending marginal weight.

    ``SCIResult.carryover`` is required to be in that order, and the loop applies no
    ranking of its own. Carryover strings that were in the diagonalized subspace get
    their weight from the amplitudes; the rest -- the ones carryover_type 2/3 generate
    by single excitation -- were never in the subspace and so have no weight at all.
    Those go last, which is the right priority: a truncation to ``max_dim`` then sheds
    weightless additions before it sheds a ranked parent.

    Mirrors the addon's own ordering (``_rank_carryover``, a stable argsort of negated
    marginal weights) so the two agree on ties.

    Raises if none of the carryover is in the subspace, since the returned order
    would then be SBD's canonical order rather than a ranking. That should not
    happen: types 2/3 extend from the determinants SBD selected out of the
    subspace, and those parents are part of the carryover.
    """
    if carryover.size == 0:
        return carryover
    if solved.size:
        # Look each carryover string up in the solved list. searchsorted needs a
        # sorted haystack, so sort once and carry the permutation to recover
        # weight positions.
        order = np.argsort(solved, kind="stable")
        sorted_solved = solved[order]
        pos = np.minimum(np.searchsorted(sorted_solved, carryover), sorted_solved.size - 1)
        found = sorted_solved[pos] == carryover
    else:
        found = np.zeros(carryover.size, dtype=bool)
    if not found.any():
        raise RuntimeError(
            f"none of the {carryover.size} carryover determinants were in the "
            "diagonalized subspace, so none of them has a marginal weight and the "
            "order returned here would be SBD's canonical order, not a weight "
            "ranking. qiskit-addon-sqd truncates the carryover to max_dim by keeping "
            "the leading entries, so it would keep an arbitrary subset."
        )

    ranked = carryover[found]
    ranked_weights = weights[order[pos[found]]]
    # Stable argsort of the negated weights, matching the addon exactly.
    ranked = ranked[np.argsort(-ranked_weights, kind="stable")]
    return np.concatenate([ranked, carryover[~found]])


def _rank_carryover_from_amplitudes(
    carryover_a: np.ndarray,
    carryover_b: np.ndarray,
    sci_state,
) -> tuple[np.ndarray, np.ndarray]:
    """Re-rank a carryover SBD returned in canonical order, using the amplitudes.

    Needed for carryover_type 2/3 only; type 1 already comes out weight-ordered.

    The ranking is PER SECTOR, where the loop would rank the two sectors jointly under
    ``symmetrize_spin``. That is safe here, and not by luck. ``symmetrize_spin`` hands
    the solver the SAME array for both sectors, so the subspace is symmetric and the
    amplitude matrix is symmetric (or antisymmetric, for a triplet at Ms=0); marginal
    weights square the amplitudes, so the two sectors' weights are equal either way
    and the per-sector rankings coincide. Singles extension is spin-agnostic, so types
    2/3 inherit it. Measured on h2o: for a symmetric subspace the two carryover lists
    come out bit-identical for types 1, 2 and 3.

    That matters because ``batch_to_ci_strings`` IGNORES ``carryover_strings_b`` under
    ``symmetrize_spin`` -- it assumes the two are equal -- so a solver whose lists
    differed there would silently lose its beta selection. Ours do not differ. When
    ``symmetrize_spin`` is off, both lists are used and may legitimately differ.
    """
    probabilities = np.abs(sci_state.amplitudes) ** 2
    weights_a = probabilities.sum(axis=1)
    weights_b = probabilities.sum(axis=0)
    return (
        _rank_by_marginal_weight(carryover_a, sci_state.ci_strs_a, weights_a),
        _rank_by_marginal_weight(carryover_b, sci_state.ci_strs_b, weights_b),
    )


def _extract_carryover(results: dict, norb: int, backend, bit_length: int):
    """Convert SBD's carryover determinant lists to CI strings.

    ``tpb_diag`` always returns ``carryover_adet``/``carryover_bdet`` in its results
    dict, but they are empty unless the caller asked for a carryover type --
    ``_create_sbd_config`` leaves ``carryover_type`` at 0, so the common path pays
    nothing here and gets ``(None, None)``.

    SBD hands these back as packed half-determinants, the same representation
    ``_ci_strings_to_sbd_dets`` produces going in, so converting back with
    ``_sbd_dets_to_ci_strings`` yields plain CI-string ints -- the shape
    ``SCIResult.carryover`` wants.
    """
    co_a = results.get("carryover_adet")
    co_b = results.get("carryover_bdet")
    if not co_a and not co_b:
        return None, None
    empty = np.array([], dtype=np.int64)
    carryover_a = (
        _sbd_dets_to_ci_strings(co_a, norb, backend, bit_length) if co_a else empty
    )
    carryover_b = (
        _sbd_dets_to_ci_strings(co_b, norb, backend, bit_length) if co_b else empty
    )
    return carryover_a, carryover_b


def assemble_rdms(results: dict, norb: int) -> tuple[np.ndarray | None, np.ndarray | None]:
    """Build spin-summed (rdm1, rdm2) from SBD's raw one_p_rdm/two_p_rdm.

    Returns (None, None) when ``do_rdm`` was 0 (SBD leaves these keys as
    empty lists in that case -- see ``sbdiag.h``'s ``do_rdm == 0`` branch,
    which only computes the diagonal density, not the correlation
    functions), matching SCIResult's own ``rdm1``/``rdm2`` default.

    The layout conversion was verified against PySCF's ``make_rdm1`` and
    ``make_rdm2`` on all three backends (cpu, gpu, gpu-omp), both element-wise
    and through the energy identity
    ``E = einsum("pr,pr->",rdm1,hcore) + 0.5*einsum("prqs,prqs->",rdm2,eri)``
    with which ``qiskit-addon-sqd`` contracts ``SCIResult.rdm1`` and ``rdm2``.

    SBD's layout, as documented in upstream SBD's ``docs/user-guide.md``::

        one_p_rdm[s][i + L*j]                 = <c+_{i,s} c_{j,s}>
        two_p_rdm[s+2t][i + L*j + L^2*k + L^3*l] = <c+_{i,s} c+_{j,t} c_{l,t} c_{k,s}>

    A Fortran-order reshape implements those flat-index formulas directly.
    rdm1 needs no further transpose, and the spin-summed rdm2 is transposed
    with axes ``(0, 2, 1, 3)`` into the "prqs" order ``SCIResult`` expects.
    """
    one_p_rdm = results.get("one_p_rdm")
    two_p_rdm = results.get("two_p_rdm")
    if not one_p_rdm or not two_p_rdm:
        return None, None

    one_p_rdm = np.asarray(one_p_rdm)
    two_p_rdm = np.asarray(two_p_rdm)
    L = norb

    rdm1 = (np.reshape(one_p_rdm[0], (L, L), order="F")
            + np.reshape(one_p_rdm[1], (L, L), order="F"))

    spin_summed = sum(
        np.reshape(two_p_rdm[s], (L, L, L, L), order="F") for s in range(4)
    )
    rdm2 = spin_summed.transpose(0, 2, 1, 3)

    return rdm1, rdm2


def solve_sci_batch(
    ci_strings: list[tuple[np.ndarray, np.ndarray]],
    one_body_tensor: np.ndarray,
    two_body_tensor: np.ndarray,
    norb: int,
    nelec: tuple[int, int],
    *,
    spin_sq: float | None = None,
    mpi_comm: MPI.Comm | None = None,
    sbd_config: dict | None = None,
    temp_dir: str | Path | None = None,
    clean_temp_dir: bool = True,
    device_config=None,
    fcidump_path: str | Path | None = None,
) -> list[SCIResult]:
    """
    Diagonalize Hamiltonian in multiple subspaces using SBD.

    The FCIDUMP file is loaded once and reused across all batches.

    Args:
        ci_strings: List of (strings_a, strings_b) pairs.
        one_body_tensor: The one-body tensor of the Hamiltonian.
        two_body_tensor: The two-body tensor of the Hamiltonian.
        norb: The number of spatial orbitals.
        nelec: The numbers of alpha and beta electrons.
        spin_sq: Target value for total spin squared (unused by SBD).
        mpi_comm: MPI communicator. If None, uses MPI.COMM_WORLD.
        sbd_config: Dictionary of SBD configuration parameters.
        temp_dir: Directory for temporary files.
        clean_temp_dir: Whether to delete intermediate files.
        device_config: DeviceConfig object to select CPU/GPU backend.
        fcidump_path: If set, load the FCIDUMP directly from this path on
            every rank and skip the tensor->file round-trip. MUST be on a
            filesystem visible to every rank (shared on multi-node runs).
            When None (default), rank 0 writes a regenerated FCIDUMP into
            ``temp_dir`` and every rank opens that file, which requires
            ``temp_dir`` to be shared for multi-node runs.

    Returns:
        List of SCIResult for each batch.
    """
    if not ci_strings:
        return []

    backend = _resolve_backend(device_config)

    if mpi_comm is None:
        mpi_comm = MPI.COMM_WORLD
    mpi_rank = mpi_comm.Get_rank()

    sbd_dir, owns_sbd_dir = _make_sbd_dir(mpi_comm, mpi_rank, temp_dir)

    try:
        fcidump, ecore_offset = _load_or_regenerate_fcidump(
            backend, mpi_rank, mpi_comm, sbd_dir, fcidump_path,
            one_body_tensor, two_body_tensor, norb, nelec,
        )

        return [
            _solve_sci_core(
                ci_strs,
                norb=norb,
                nelec=nelec,
                spin_sq=spin_sq,
                mpi_comm=mpi_comm,
                mpi_rank=mpi_rank,
                sbd_config=sbd_config,
                sbd_dir=sbd_dir,
                backend=backend,
                fcidump=fcidump,
                device_config=device_config,
                ecore_offset=ecore_offset,
            )
            for ci_strs in ci_strings
        ]
    finally:
        if clean_temp_dir and owns_sbd_dir and mpi_rank == 0:
            shutil.rmtree(sbd_dir, ignore_errors=True)


def _make_sbd_dir(mpi_comm, mpi_rank, temp_dir):
    """Create a per-run tempdir on rank 0 and broadcast its path.

    Used to hold the wavefunction.bin written by rank 0 (and the
    regenerated fcidump.txt when ``fcidump_path`` is not provided).
    Returns (path, owns_sbd_dir) where owns_sbd_dir is True on rank 0
    (so the caller knows to rmtree it on exit).

    ``temp_dir`` is created along with any missing parents, so a caller may
    pass a path that does not exist yet (e.g. ``--temp_dir`` pointing at a
    fresh scratch location). Creation happens only on rank 0; if it fails, the
    error is broadcast so that every rank raises together, rather than the
    other ranks deadlocking in the broadcast below while rank 0 unwinds.
    """
    temp_dir = temp_dir or tempfile.gettempdir()
    # (ok, payload): payload is the created directory when ok is True, else a
    # message describing why rank 0 could not create it.
    if mpi_rank == 0:
        try:
            Path(temp_dir).mkdir(parents=True, exist_ok=True)
            created = tempfile.mkdtemp(prefix="sbd_files_", dir=temp_dir)
            result = (True, str(Path(created)))
        except OSError as exc:
            result = (False, f"{type(exc).__name__}: {exc}")
    else:
        result = None
    ok, payload = mpi_comm.bcast(result, root=0)
    if not ok:
        raise RuntimeError(
            f"rank 0 could not create an SBD temporary directory under "
            f"{temp_dir!r}: {payload}"
        )
    return Path(payload), (mpi_rank == 0)


def _load_or_regenerate_fcidump(
    backend, mpi_rank, mpi_comm, sbd_dir, fcidump_path,
    one_body_tensor, two_body_tensor, norb, nelec,
):
    """Return ``(fcidump, ecore_offset)``.

    ``ecore_offset`` is the ECORE constant baked into whatever FCIDUMP
    SBD is about to load, and is subtracted from the raw energy so that
    the SCIResult contract (``energy`` is the pure electronic piece) is
    preserved in both the regenerate and direct-load paths.

    If ``fcidump_path`` is given, every rank loads that file directly
    (must be on a filesystem visible to every rank). Otherwise rank 0
    regenerates an FCIDUMP with ECORE=0 into ``sbd_dir/fcidump.txt`` and
    every rank opens that file.
    """
    if fcidump_path is not None:
        ecore_offset = _read_fcidump_ecore(fcidump_path)
        return backend.LoadFCIDump(str(fcidump_path)), ecore_offset

    fcidump_path = sbd_dir / "fcidump.txt"
    if mpi_rank == 0:
        pyscf_tools.fcidump.from_integrals(
            str(fcidump_path), one_body_tensor, two_body_tensor, norb, nelec,
        )
    mpi_comm.Barrier()
    return backend.LoadFCIDump(str(fcidump_path)), 0.0


def _read_fcidump_ecore(fcidump_path):
    """Pull the ECORE constant out of an FCIDUMP file.

    ECORE is the line whose four orbital indices are all zero. Return
    0.0 if no such line exists (some FCIDUMPs simply omit it).
    """
    with open(fcidump_path) as f:
        for line in f:
            # FCIDUMP integral lines look like: "<val> i j k l"
            parts = line.split()
            if len(parts) == 5:
                try:
                    val = float(parts[0])
                    i, j, k, l = (int(x) for x in parts[1:])
                except ValueError:
                    continue
                if i == j == k == l == 0:
                    return val
    return 0.0


def _ci_strings_to_sbd_dets(
    ci_strings: np.ndarray, norb: int, backend,
    bit_length: int = SBD_DEFAULT_BIT_LENGTH,
) -> list[list[int]]:
    """Convert CI strings (integers) to SBD determinant format.

    Determinants are sorted in canonical order (matching C++ sort_bitarray)
    which is required by the GPU Correlation kernel (do_rdm=1).
    """
    dets = []
    for ci_str in ci_strings:
        binary_str = format(int(ci_str), f'0{norb}b')
        det = backend.from_string(binary_str, bit_length, norb)
        dets.append(det)
    return backend.sort_bitarray(dets)


def _to_alpha_first_signs(amplitudes, strings_a, strings_b, norb, block_elements=1 << 24):
    """Convert SBD's amplitudes, in place, to the sign convention SCIState assumes.

    The two conventions order the creation operators of a determinant differently.
    SBD interleaves the spins orbital by orbital (alpha 0, beta 0, alpha 1, ...);
    PySCF, and so ``SCIState.rdm()`` and ``SCIState.spin_square()``, put every alpha
    operator before every beta one. Moving each beta operator past the alpha
    operators of higher orbitals gives the determinant the phase

        (-1) ** sum over occupied beta orbitals p of (occupied alpha orbitals above p)

    The magnitudes are the same, so energies, SBD's own RDMs, occupancies and any
    selection by amplitude magnitude are unaffected, which is why this went unnoticed:
    only what PySCF computes *from the amplitudes* was wrong -- e.g. ``spin_square()``
    of 0.036 for an exact singlet, and an energy from ``rdm()`` 89 mHa off.

    Args:
        amplitudes: The ``(len(strings_a), len(strings_b))`` matrix, modified in place.
        strings_a: Alpha CI strings labelling the rows; bit ``p`` is orbital ``p``.
        strings_b: Beta CI strings labelling the columns.
        norb: Number of spatial orbitals.
        block_elements: Rows are converted in blocks of about this many elements, so
            the temporary sign matrix stays bounded at large subspaces.
    """
    orbitals = np.arange(norb, dtype=np.int64)
    occupied_b = ((np.asarray(strings_b, dtype=np.int64)[:, None] >> orbitals) & 1).astype(np.int32)
    occupied_a = (np.asarray(strings_a, dtype=np.int64)[:, None] >> orbitals) & 1
    # alpha electrons in orbitals strictly above p, kept only as a parity
    above_a = ((np.cumsum(occupied_a[:, ::-1], axis=1)[:, ::-1] - occupied_a) & 1).astype(np.int32)
    rows = max(1, block_elements // max(1, len(strings_b)))
    for start in range(0, len(strings_a), rows):
        parity = (above_a[start:start + rows] @ occupied_b.T) & 1
        amplitudes[start:start + rows][parity == 1] *= -1.0


def _sbd_dets_to_ci_strings(
    dets: list[list[int]], norb: int, backend,
    bit_length: int = SBD_DEFAULT_BIT_LENGTH,
) -> np.ndarray:
    """Convert SBD determinants to CI strings (integers)."""
    ci_strings = []
    for det in dets:
        binary_str = backend.makestring(det, bit_length, norb)
        ci_str = int(binary_str, 2)
        ci_strings.append(ci_str)
    return np.array(ci_strings, dtype=np.int64)


def _create_sbd_config(config_dict: dict | None = None, backend=None, device_config=None):
    """Create SBD configuration object from dictionary."""
    if backend is None:
        backend = _resolve_backend(device_config)

    sbd_data = backend.TPB_SBD()

    # Defaults
    sbd_data.method = 0  # Davidson
    sbd_data.max_it = 100
    sbd_data.max_nb = 50
    sbd_data.eps = 1e-8
    sbd_data.max_time = 3600
    sbd_data.init = 0
    sbd_data.do_shuffle = 0
    sbd_data.do_rdm = 0
    # Off by default, but no longer discarded: since qiskit-addon-sqd#369 the loop
    # accepts a solver's own carryover (SCIResult.carryover) instead of thresholding
    # the amplitudes, and _solve_sci_core reports it when SBD computed one. Default
    # stays 0 because computing it is real work -- carryover_type=2 builds
    # singles-extended determinant lists -- and because the amplitude-threshold path
    # remains the behaviour callers expect. Set carryover_type through sbd_config to
    # let SBD choose the next iteration's determinants instead.
    sbd_data.carryover_type = 0
    sbd_data.ratio = 0.1
    sbd_data.threshold = 1e-4
    sbd_data.bit_length = SBD_DEFAULT_BIT_LENGTH

    if config_dict:
        for key, value in config_dict.items():
            if hasattr(sbd_data, key):
                setattr(sbd_data, key, value)

    if device_config is not None:
        device_config.apply(sbd_data)

    return sbd_data


def create_sbd_solver(
    mpi_comm: MPI.Comm | None = None,
    sbd_config: dict | None = None,
    temp_dir: str | Path | None = None,
    clean_temp_dir: bool = True,
    device_config=None,
) -> Callable:
    """
    Create a configured SBD solver function for use with
    diagonalize_fermionic_hamiltonian.

    Example:
        >>> from functools import partial
        >>> sbd_solver = create_sbd_solver(sbd_config={"method": 0, "eps": 1e-10})
        >>> result = diagonalize_fermionic_hamiltonian(
        ...     hcore, eri, bit_array, sci_solver=sbd_solver, ...
        ... )
    """
    from functools import partial

    return partial(
        solve_sci_batch,
        mpi_comm=mpi_comm,
        sbd_config=sbd_config,
        temp_dir=temp_dir,
        clean_temp_dir=clean_temp_dir,
        device_config=device_config,
    )