chia-blockchain/chia/simulator/block_tools.py

from __future__ import annotations

import asyncio
import copy
import logging
import os
import random
import shutil
import ssl
import sys
import tempfile
import time
from collections.abc import AsyncIterator, Callable, Iterator, Sequence
from contextlib import asynccontextmanager, contextmanager
from dataclasses import dataclass, replace
from functools import lru_cache
from pathlib import Path
from random import Random
from types import TracebackType
from typing import Any

import anyio
from chia_puzzles_py.programs import CHIALISP_DESERIALISATION, ROM_BOOTSTRAP_GENERATOR
from chia_rs import (
    AugSchemeMPL,
    BlockRecord,
    ChallengeChainSubSlot,
    ConsensusConstants,
    EndOfSubSlotBundle,
    FullBlock,
    G1Element,
    G2Element,
    InfusedChallengeChainSubSlot,
    PoolTarget,
    PrivateKey,
    ProofOfSpace,
    RewardChainSubSlot,
    SpendBundle,
    SubEpochSummary,
    SubSlotProofs,
    UnfinishedBlock,
    solution_generator,
    solve_proof,
)
from chia_rs.sized_bytes import bytes32
from chia_rs.sized_ints import uint8, uint16, uint32, uint64, uint128
from filelock import FileLock
from typing_extensions import Self

from chia.consensus.block_creation import create_unfinished_block, unfinished_block_to_full_block
from chia.consensus.block_record import BlockRecordProtocol
from chia.consensus.blockchain_interface import BlockRecordsProtocol
from chia.consensus.condition_costs import ConditionCost
from chia.consensus.constants import replace_str_to_bytes
from chia.consensus.default_constants import DEFAULT_CONSTANTS
from chia.consensus.deficit import calculate_deficit
from chia.consensus.full_block_to_block_record import block_to_block_record
from chia.consensus.get_block_challenge import pre_sp_tx_block_height
from chia.consensus.make_sub_epoch_summary import next_sub_epoch_summary
from chia.consensus.pot_iterations import (
    calculate_ip_iters,
    calculate_iterations_quality,
    calculate_sp_interval_iters,
    calculate_sp_iters,
    is_overflow_block,
    validate_pospace_and_get_required_iters,
)
from chia.consensus.signage_point import SignagePoint
from chia.consensus.vdf_info_computation import get_signage_point_vdf_info
from chia.daemon.keychain_proxy import KeychainProxy, connect_to_keychain_and_validate, wrap_local_keychain
from chia.full_node.bundle_tools import simple_solution_generator, simple_solution_generator_backrefs
from chia.plotting.cache import cached_master_sk_to_local_sk
from chia.plotting.create_plots import PlotKeys, create_plots
from chia.plotting.manager import PlotManager
from chia.plotting.prover import PlotVersion, QualityProtocol, V1Prover, V2Prover, V2Quality
from chia.plotting.util import (
    Params,
    PlotRefreshEvents,
    PlotRefreshResult,
    PlotsRefreshParameter,
    add_plot_directory,
    parse_plot_info,
)
from chia.server.server import ssl_context_for_client
from chia.simulator.socket import find_available_listen_port
from chia.simulator.ssl_certs import (
    SSLTestCACertAndPrivateKey,
    SSLTestCollateralWrapper,
    SSLTestNodeCertsAndKeys,
    get_next_nodes_certs_and_keys,
    get_next_private_ca_cert_and_key,
)
from chia.simulator.vdf_prover import get_vdf_info_and_proof
from chia.simulator.wallet_tools import WalletTool
from chia.ssl.create_ssl import create_all_ssl
from chia.ssl.ssl_check import fix_ssl
from chia.types.blockchain_format.classgroup import ClassgroupElement
from chia.types.blockchain_format.coin import Coin
from chia.types.blockchain_format.program import DEFAULT_FLAGS, INFINITE_COST, Program, _run, run_with_cost
from chia.types.blockchain_format.proof_of_space import (
    calculate_pos_challenge,
    calculate_prefix_bits,
    generate_plot_public_key,
    generate_taproot_sk,
    is_v1_phased_out,
    make_pos,
    num_phase_out_epochs,
    passes_plot_filter,
)
from chia.types.blockchain_format.serialized_program import SerializedProgram
from chia.types.blockchain_format.vdf import VDFInfo, VDFProof
from chia.types.condition_opcodes import ConditionOpcode
from chia.types.generator_types import NewBlockGenerator
from chia.util.bech32m import encode_puzzle_hash
from chia.util.block_cache import BlockCache
from chia.util.config import (
    config_path_for_filename,
    create_default_chia_config,
    load_config,
    lock_config,
    override_config,
    save_config,
)
from chia.util.default_root import DEFAULT_ROOT_PATH
from chia.util.hash import std_hash
from chia.util.keychain import Keychain, bytes_to_mnemonic
from chia.util.timing import adjusted_timeout, backoff_times
from chia.wallet.derive_keys import (
    master_sk_to_farmer_sk,
    master_sk_to_pool_sk,
    master_sk_to_wallet_sk,
)
from chia.wallet.puzzles.p2_delegated_puzzle_or_hidden_puzzle import puzzle_hash_for_pk

DESERIALIZE_MOD = Program.from_bytes(CHIALISP_DESERIALISATION)

GENERATOR_MOD: SerializedProgram = SerializedProgram.from_bytes(ROM_BOOTSTRAP_GENERATOR)

test_constants = DEFAULT_CONSTANTS.replace(
    MIN_PLOT_SIZE_V1=uint8(18),
    PLOT_SIZE_V2=uint8(18),
    MIN_BLOCKS_PER_CHALLENGE_BLOCK=uint8(12),
    DIFFICULTY_STARTING=uint64(2**10),
    DISCRIMINANT_SIZE_BITS=uint16(16),
    SUB_EPOCH_BLOCKS=uint32(170),
    WEIGHT_PROOF_THRESHOLD=uint8(2),
    WEIGHT_PROOF_RECENT_BLOCKS=uint32(500),
    DIFFICULTY_CONSTANT_FACTOR=uint128(33554432),
    NUM_SPS_SUB_SLOT=uint8(16),  # Must be a power of 2
    MAX_SUB_SLOT_BLOCKS=uint32(50),
    EPOCH_BLOCKS=uint32(340),
    # the block cache must contain at least 3 epochs in order for
    # create_prev_sub_epoch_segments() to have access to all the blocks it needs
    # from the cache
    BLOCKS_CACHE_SIZE=uint32(340 * 3),  # Coordinate with the above values
    SUB_SLOT_TIME_TARGET=uint16(600),  # The target number of seconds per slot, mainnet 600
    SUB_SLOT_ITERS_STARTING=uint64(2**10),  # Must be a multiple of 64
    NUMBER_ZERO_BITS_PLOT_FILTER_V1=uint8(1),  # H(plot signature of the challenge) must start with these many zeroes
    # Allows creating blockchains with timestamps up to 10 days in the future, for testing
    MAX_FUTURE_TIME2=uint32(3600 * 24 * 10),
    MEMPOOL_BLOCK_BUFFER=uint8(6),
    PLOT_V1_PHASE_OUT_EPOCH_BITS=uint8(3),
)


@lru_cache
def cached_master_sk_to_farmer_sk(master: PrivateKey) -> PrivateKey:
    return master_sk_to_farmer_sk(master)


def compute_additions_unchecked(sb: SpendBundle) -> list[Coin]:
    ret: list[Coin] = []
    for cs in sb.coin_spends:
        parent_id = cs.coin.name()
        _, r = run_with_cost(cs.puzzle_reveal, INFINITE_COST, cs.solution)
        for cond in Program.to(r).as_iter():
            atoms = cond.as_iter()
            op = next(atoms).atom
            if op != ConditionOpcode.CREATE_COIN.value:
                continue
            puzzle_hash = next(atoms).as_atom()
            amount = uint64(next(atoms).as_int())
            ret.append(Coin(parent_id, puzzle_hash, amount))
    return ret


def compute_block_cost(
    generator: SerializedProgram, constants: ConsensusConstants, height: uint32, prev_tx_height: uint32
) -> uint64:
    # this function cannot *validate* the block or any of the transactions. We
    # deliberately create invalid blocks as parts of the tests, and we still
    # need to be able to compute the cost of it

    condition_cost = 0
    clvm_cost = 0

    if height >= constants.HARD_FORK_HEIGHT:
        blocks: list[bytes] = []
        cost, result = _run(generator, INFINITE_COST, DEFAULT_FLAGS, [DESERIALIZE_MOD, blocks])
        clvm_cost += cost

        for spend in result.first().as_iter():
            # each spend is a list of:
            # (parent-coin-id puzzle amount solution)
            puzzle = spend.at("rf")
            solution = spend.at("rrrf")

            cost, result = _run(puzzle, INFINITE_COST, DEFAULT_FLAGS, solution)
            clvm_cost += cost
            condition_cost += conditions_cost(
                result, charge_for_conditions=prev_tx_height >= constants.HARD_FORK2_HEIGHT
            )

    else:
        block_program_args = SerializedProgram.to([[]])
        clvm_cost, result = _run(GENERATOR_MOD, INFINITE_COST, DEFAULT_FLAGS, [generator, block_program_args])

        for res in result.first().as_iter():
            # each condition item is:
            # (parent-coin-id puzzle-hash amount conditions)
            conditions = res.at("rrrf")
            condition_cost += conditions_cost(conditions, charge_for_conditions=False)

    size_cost = len(bytes(generator)) * constants.COST_PER_BYTE

    return uint64(clvm_cost + size_cost + condition_cost)


def make_spend_bundle(coins: list[Coin], wallet: WalletTool, rng: Random) -> tuple[SpendBundle, list[Coin]]:
    """
    makes a new spend bundle (block generator) spending some of the coins in the
    list of coins. The list will be updated to have spent coins removed and new
    coins appended.
    """
    new_coins: list[Coin] = []
    spend_bundles: list[SpendBundle] = []
    to_spend = rng.sample(coins, min(5, len(coins)))
    receiver = wallet.get_new_puzzlehash()
    for c in to_spend:
        bundle = wallet.generate_signed_transaction(uint64(c.amount // 2), receiver, c)
        new_coins.extend(bundle.additions())
        spend_bundles.append(bundle)

    return SpendBundle.aggregate(spend_bundles), new_coins


class BlockTools:
    """
    Tools to generate blocks for testing.
    """

    _block_cache_header: bytes32
    _block_cache_height_to_hash: dict[uint32, bytes32]
    _block_cache_difficulty: uint64
    _block_cache: dict[bytes32, BlockRecord]

    def __init__(
        self,
        constants: ConsensusConstants = test_constants,
        root_path: Path | None = None,
        keychain: Keychain | None = None,
        config_overrides: dict[str, Any] | None = None,
        automated_testing: bool = True,
        plot_dir: str = "test-plots",
        log: logging.Logger = logging.getLogger(__name__),
    ) -> None:
        self._block_cache_header = bytes32.zeros

        self._tempdir = None
        if root_path is None:
            self._tempdir = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
            root_path = Path(self._tempdir.name)

        self.root_path = root_path
        self.log = log
        self.local_keychain = keychain
        self.local_sk_cache: dict[bytes32, tuple[PrivateKey, Any]] = {}
        self.automated_testing = automated_testing
        self.plot_dir_name = plot_dir

        if automated_testing:
            # Hold onto the wrappers so that they can keep track of whether the certs/keys
            # are in use by another BlockTools instance.
            self.ssl_ca_cert_and_key_wrapper: SSLTestCollateralWrapper[SSLTestCACertAndPrivateKey] = (
                get_next_private_ca_cert_and_key()
            )
            self.ssl_nodes_certs_and_keys_wrapper: SSLTestCollateralWrapper[SSLTestNodeCertsAndKeys] = (
                get_next_nodes_certs_and_keys()
            )
            create_default_chia_config(root_path)
            create_all_ssl(
                root_path,
                private_ca_crt_and_key=self.ssl_ca_cert_and_key_wrapper.collateral.cert_and_key,
                node_certs_and_keys=self.ssl_nodes_certs_and_keys_wrapper.collateral.certs_and_keys,
            )
            fix_ssl(root_path)
            with lock_config(root_path=root_path, filename="config.yaml"):
                path = config_path_for_filename(root_path=root_path, filename="config.yaml")
                path.write_text(path.read_text().replace("localhost", "127.0.0.1"))
        self._config = load_config(self.root_path, "config.yaml")
        if automated_testing:
            if config_overrides is None:
                config_overrides = {}
            config_overrides["logging.log_stdout"] = True
            config_overrides["selected_network"] = "testnet0"
            for service in [
                "harvester",
                "farmer",
                "full_node",
                "wallet",
                "introducer",
                "timelord",
                "pool",
                "simulator",
            ]:
                config_overrides[service + ".selected_network"] = "testnet0"

            # some tests start the daemon, make sure it's on a free port
            config_overrides["daemon_port"] = find_available_listen_port("BlockTools daemon")

        self._config = override_config(self._config, config_overrides)

        with lock_config(self.root_path, "config.yaml"):
            save_config(self.root_path, "config.yaml", self._config)
        overrides = self._config["network_overrides"]["constants"][self._config["selected_network"]]
        updated_constants = replace_str_to_bytes(constants, **overrides)
        self.constants = updated_constants

        self.plot_dir: Path = get_plot_dir(self.plot_dir_name, self.automated_testing)
        self.temp_dir: Path = get_plot_tmp_dir(self.plot_dir_name, self.automated_testing)
        self.plot_dir.mkdir(parents=True, exist_ok=True)
        self.temp_dir.mkdir(parents=True, exist_ok=True)
        self.expected_plots: dict[bytes32, Path] = {}
        self.created_plots: int = 0
        self.total_result = PlotRefreshResult()

        def test_callback(event: PlotRefreshEvents, update_result: PlotRefreshResult) -> None:
            assert update_result.duration < 120
            if event == PlotRefreshEvents.started:
                self.total_result = PlotRefreshResult()

            if event == PlotRefreshEvents.batch_processed:
                self.total_result.loaded += update_result.loaded
                self.total_result.processed += update_result.processed
                self.total_result.duration += update_result.duration
                assert update_result.remaining >= len(self.expected_plots) - self.total_result.processed
                assert len(update_result.loaded) <= self.plot_manager.refresh_parameter.batch_size

            if event == PlotRefreshEvents.done:
                assert self.total_result.loaded == update_result.loaded
                assert self.total_result.processed == update_result.processed
                assert self.total_result.duration == update_result.duration
                assert update_result.remaining == 0

                expected_plots: set[str] = set()
                found_plots: set[str] = set()
                if len(self.plot_manager.plots) != len(self.expected_plots):  # pragma: no cover
                    for pid, filename in self.expected_plots.items():
                        expected_plots.add(filename.name)
                    for filename, _ in self.plot_manager.plots.items():
                        found_plots.add(filename.name)
                    print(f"directory: {self.plot_dir}")
                    print(f"expected: {len(expected_plots)}")
                    for f in expected_plots:
                        print(f)
                    print(f"plot manager: {len(found_plots)}")
                    for f in found_plots:
                        print(f)
                    diff = found_plots.difference(expected_plots)
                    print(f"found unexpected: {len(diff)}")
                    for f in diff:
                        print(f)
                    diff = expected_plots.difference(found_plots)
                    print(f"not found: {len(diff)}")
                    for f in diff:
                        print(f)
                assert len(self.plot_manager.plots) == len(self.expected_plots)

        self.plot_manager: PlotManager = PlotManager(
            self.root_path,
            refresh_parameter=PlotsRefreshParameter(batch_size=uint32(2)),
            refresh_callback=test_callback,
            constants=self.constants,
            match_str=str(self.plot_dir.relative_to(DEFAULT_ROOT_PATH.parent)) if not automated_testing else None,
        )

    def __enter__(self) -> Self:
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        traceback: TracebackType | None,
    ) -> None:
        if self._tempdir is not None:
            self._tempdir.cleanup()

    def setup_new_gen(
        self,
        generator_block_heights: list[uint32],
        curr: BlockRecordProtocol,
        wallet: WalletTool | None,
        rng: random.Random | None,
        available_coins: list[Coin],
        *,
        prev_tx_height: uint32,
        dummy_block_references: bool,
        include_transactions: bool,
        block_generator: NewBlockGenerator | None,
        block_refs: list[uint32],
    ) -> NewBlockGenerator | None:
        if prev_tx_height >= self.constants.HARD_FORK2_HEIGHT:
            assert block_refs == [], "block references are not allowed after hard fork 2"
            dummy_block_references = False

        # we don't know if the new block will be a transaction
        # block or not, so even though we prepare a block
        # generator, we can't update our state (like,
        # available_coins) until it's confirmed the block
        # generator made it into the block.
        dummy_refs: list[uint32]
        if dummy_block_references and len(generator_block_heights) > 4:
            dummy_refs = [
                generator_block_heights[1],
                generator_block_heights[len(generator_block_heights) // 2],
                generator_block_heights[-2],
            ]
        else:
            dummy_refs = []

        if block_generator is not None:
            # this means the caller passed in block_generator
            # to be included in the block.
            assert block_refs == [], "block references cannot be combined with block_generator"
            assert not dummy_block_references, "(dummy) block references cannot be combined with block_generator"
            return block_generator

        if include_transactions:
            # if the caller did not pass in specific
            # transactions, this parameter means we just want
            # some transactions
            assert wallet is not None
            assert rng is not None
            bundle, additions = make_spend_bundle(available_coins, wallet, rng)
            removals = bundle.removals()
            program = simple_solution_generator(bundle).program
            cost = compute_block_cost(program, self.constants, uint32(curr.height + 1), prev_tx_height)
            return NewBlockGenerator(
                program,
                [],
                block_refs + dummy_refs,
                bundle.aggregated_signature,
                additions,
                removals,
                cost,
            )

        if dummy_block_references:
            program = SerializedProgram.from_bytes(solution_generator([]))
            cost = compute_block_cost(program, self.constants, uint32(curr.height + 1), prev_tx_height)
            return NewBlockGenerator(program, [], block_refs + dummy_refs, G2Element(), [], [], cost)

        return None

    async def setup_keys(self, fingerprint: int | None = None, reward_ph: bytes32 | None = None) -> None:
        keychain_proxy: KeychainProxy | None
        try:
            if self.local_keychain:
                keychain_proxy = wrap_local_keychain(self.local_keychain, log=self.log)
            elif not self.automated_testing and fingerprint is not None:
                keychain_proxy = await connect_to_keychain_and_validate(self.root_path, self.log)
            else:  # if we are automated testing or if we don't have a fingerprint.
                keychain_proxy = await connect_to_keychain_and_validate(
                    self.root_path, self.log, user="testing-1.8.0", service="chia-testing-1.8.0"
                )
            assert keychain_proxy is not None
            if fingerprint is None:  # if we are not specifying an existing key
                await keychain_proxy.delete_all_keys()
                self.farmer_master_sk_entropy = std_hash(b"block_tools farmer key")  # both entropies are only used here
                self.pool_master_sk_entropy = std_hash(b"block_tools pool key")
                self.farmer_master_sk = await keychain_proxy.add_key(bytes_to_mnemonic(self.farmer_master_sk_entropy))
                self.pool_master_sk = await keychain_proxy.add_key(
                    bytes_to_mnemonic(self.pool_master_sk_entropy),
                )
            else:
                sk = await keychain_proxy.get_key_for_fingerprint(fingerprint)
                assert sk is not None
                self.farmer_master_sk = sk
                sk = await keychain_proxy.get_key_for_fingerprint(fingerprint)
                assert sk is not None
                self.pool_master_sk = sk

            self.farmer_pk = cached_master_sk_to_farmer_sk(self.farmer_master_sk).get_g1()
            self.pool_pk = master_sk_to_pool_sk(self.pool_master_sk).get_g1()

            if reward_ph is None:
                self.farmer_ph: bytes32 = puzzle_hash_for_pk(
                    master_sk_to_wallet_sk(self.farmer_master_sk, uint32(0)).get_g1()
                )
                self.pool_ph: bytes32 = puzzle_hash_for_pk(
                    master_sk_to_wallet_sk(self.pool_master_sk, uint32(0)).get_g1()
                )
            else:
                self.farmer_ph = reward_ph
                self.pool_ph = reward_ph
            if self.automated_testing:
                self.all_sks: list[PrivateKey] = [sk for sk, _ in await keychain_proxy.get_all_private_keys()]
            else:
                self.all_sks = [self.farmer_master_sk]  # we only want to include plots under the same fingerprint
            self.pool_pubkeys: list[G1Element] = [master_sk_to_pool_sk(sk).get_g1() for sk in self.all_sks]

            self.farmer_pubkeys: list[G1Element] = [cached_master_sk_to_farmer_sk(sk).get_g1() for sk in self.all_sks]
            if len(self.pool_pubkeys) == 0 or len(self.farmer_pubkeys) == 0:
                raise RuntimeError("Keys not generated. Run `chia keys generate`")

            self.plot_manager.set_public_keys(self.farmer_pubkeys, self.pool_pubkeys)
        finally:
            if keychain_proxy is not None:
                await keychain_proxy.close()  # close the keychain proxy

    def change_config(self, new_config: dict[str, Any]) -> None:
        self._config = new_config
        overrides = self._config["network_overrides"]["constants"][self._config["selected_network"]]
        updated_constants = replace_str_to_bytes(self.constants, **overrides)
        self.constants = updated_constants
        with lock_config(self.root_path, "config.yaml"):
            save_config(self.root_path, "config.yaml", self._config)

    def add_plot_directory(self, path: Path) -> None:
        # don't add to config if block_tools is user run and the directory is already in the config.
        if str(path.resolve()) not in self._config["harvester"]["plot_directories"] or self.automated_testing:
            self._config = add_plot_directory(self.root_path, str(path))

    async def setup_plots(
        self,
        num_og_plots: int = 15,
        num_pool_plots: int = 5,
        num_non_keychain_plots: int = 3,
        plot_size: int = 20,
        bitfield: bool = True,
        testrun_uid: str | None = None,
    ) -> bool:
        if testrun_uid is None:
            lock_file_name = self.plot_dir / ".lockfile"
        else:
            lock_file_name = self.plot_dir / (testrun_uid + ".lockfile")

        with FileLock(lock_file_name):
            self.add_plot_directory(self.plot_dir)
            assert self.created_plots == 0
            existing_plots: bool = True
            # OG Plots
            for i in range(num_og_plots):
                plot = await self.new_plot(plot_size=plot_size, bitfield=bitfield)
                if plot.new_plot:
                    existing_plots = False
            # Pool Plots
            for i in range(num_pool_plots):
                plot = await self.new_plot(self.pool_ph, plot_size=plot_size, bitfield=bitfield)
                if plot.new_plot:
                    existing_plots = False
            # Some plots with keys that are not in the keychain
            for i in range(num_non_keychain_plots):
                plot = await self.new_plot(
                    path=self.plot_dir / "not_in_keychain",
                    plot_keys=PlotKeys(G1Element(), G1Element(), None),
                    exclude_plots=True,
                    plot_size=plot_size,
                    bitfield=bitfield,
                )
                if plot.new_plot:
                    existing_plots = False
            await self.refresh_plots()
            assert len(self.plot_manager.plots) == len(self.expected_plots)
            return existing_plots

    async def new_plot(
        self,
        pool_contract_puzzle_hash: bytes32 | None = None,
        path: Path | None = None,
        tmp_dir: Path | None = None,
        plot_keys: PlotKeys | None = None,
        exclude_plots: bool = False,
        plot_size: int = 20,
        bitfield: bool = True,
    ) -> BlockToolsNewPlotResult:
        final_dir = self.plot_dir
        if path is not None:
            final_dir = path
            final_dir.mkdir(parents=True, exist_ok=True)
        if tmp_dir is None:
            tmp_dir = self.temp_dir
        params = Params(
            # Can't go much lower than 20, since plots start having no solutions and more buggy
            size=plot_size,
            # Uses many plots for testing, in order to guarantee proofs of space at every height
            num=1,
            buffer=100,
            tmp_dir=Path(tmp_dir),
            tmp2_dir=Path(tmp_dir),
            final_dir=Path(final_dir),
            plotid=None,
            memo=None,
            buckets=0,
            stripe_size=2000,
            num_threads=0,
            nobitfield=not bitfield,
        )
        try:
            if plot_keys is None:
                pool_pk: G1Element | None = None
                pool_address: str | None = None
                if pool_contract_puzzle_hash is None:
                    pool_pk = self.pool_pk
                else:
                    pool_address = encode_puzzle_hash(pool_contract_puzzle_hash, "xch")

                plot_keys = PlotKeys(self.farmer_pk, pool_pk, pool_address)
            # No datetime in the filename, to get deterministic filenames and not re-plot
            created, existed = await create_plots(
                params,
                plot_keys,
                use_datetime=False,
                test_private_keys=[AugSchemeMPL.key_gen(std_hash(self.created_plots.to_bytes(2, "big")))],
            )
            self.created_plots += 1

            plot_id_new: bytes32 | None = None
            path_new: Path | None = None
            new_plot: bool = True

            if len(created):
                assert len(existed) == 0
                plot_id_new, path_new = next(iter(created.items()))

            if len(existed):
                assert len(created) == 0
                plot_id_new, path_new = next(iter(existed.items()))
                new_plot = False
            assert plot_id_new is not None
            assert path_new is not None

            if not exclude_plots:
                self.expected_plots[plot_id_new] = path_new

            return BlockToolsNewPlotResult(plot_id_new, new_plot)

        except KeyboardInterrupt:
            shutil.rmtree(self.temp_dir, ignore_errors=True)
            sys.exit(1)

    async def refresh_plots(self) -> None:
        self.plot_manager.refresh_parameter = replace(
            self.plot_manager.refresh_parameter, batch_size=uint32(4 if len(self.expected_plots) % 3 == 0 else 3)
        )  # Make sure we have at least some batches + a remainder
        self.plot_manager.trigger_refresh()
        assert self.plot_manager.needs_refresh()
        self.plot_manager.start_refreshing(sleep_interval_ms=1)

        with anyio.fail_after(delay=adjusted_timeout(120)):
            for backoff in backoff_times():
                if not self.plot_manager.needs_refresh():
                    break

                await asyncio.sleep(backoff)

        self.plot_manager.stop_refreshing()
        assert not self.plot_manager.needs_refresh()

    async def delete_plot(self, plot_id: bytes32) -> None:
        assert plot_id in self.expected_plots
        self.expected_plots[plot_id].unlink()
        del self.expected_plots[plot_id]
        await self.refresh_plots()

    @property
    def config(self) -> dict[str, Any]:
        return copy.deepcopy(self._config)

    def get_daemon_ssl_context(self) -> ssl.SSLContext:
        crt_path = self.root_path / self.config["daemon_ssl"]["private_crt"]
        key_path = self.root_path / self.config["daemon_ssl"]["private_key"]
        ca_cert_path = self.root_path / self.config["private_ssl_ca"]["crt"]
        ca_key_path = self.root_path / self.config["private_ssl_ca"]["key"]
        return ssl_context_for_client(ca_cert_path, ca_key_path, crt_path, key_path)

    def get_plot_signature(self, m: bytes32, plot_pk: G1Element) -> G2Element:
        """
        Returns the plot signature of the header data.
        """
        farmer_sk = cached_master_sk_to_farmer_sk(self.all_sks[0])
        for plot_info in self.plot_manager.plots.values():
            if plot_pk == plot_info.plot_public_key:
                # Look up local_sk from plot to save locked memory
                if plot_info.prover.get_id() in self.local_sk_cache:
                    local_master_sk, pool_pk_or_ph = self.local_sk_cache[plot_info.prover.get_id()]
                else:
                    pool_pk_or_ph, _, local_master_sk = parse_plot_info(plot_info.prover.get_memo())
                    self.local_sk_cache[plot_info.prover.get_id()] = (local_master_sk, pool_pk_or_ph)
                if isinstance(pool_pk_or_ph, G1Element):
                    include_taproot = False
                else:
                    assert isinstance(pool_pk_or_ph, bytes32)
                    include_taproot = True
                local_sk = cached_master_sk_to_local_sk(local_master_sk)
                agg_pk = generate_plot_public_key(local_sk.get_g1(), farmer_sk.get_g1(), include_taproot)
                assert agg_pk == plot_pk
                harv_share = AugSchemeMPL.sign(local_sk, m, agg_pk)
                farm_share = AugSchemeMPL.sign(farmer_sk, m, agg_pk)
                if include_taproot:
                    taproot_sk: PrivateKey = generate_taproot_sk(local_sk.get_g1(), farmer_sk.get_g1())
                    taproot_share: G2Element = AugSchemeMPL.sign(taproot_sk, m, agg_pk)
                else:
                    taproot_share = G2Element()
                return AugSchemeMPL.aggregate([harv_share, farm_share, taproot_share])

        raise ValueError(f"Do not have key {plot_pk}")

    def get_pool_key_signature(self, pool_target: PoolTarget, pool_pk: G1Element | None) -> G2Element | None:
        # Returns the pool signature for the corresponding pk. If no pk is provided, returns None.
        if pool_pk is None:
            return None

        for sk in self.all_sks:
            sk_child = master_sk_to_pool_sk(sk)
            if sk_child.get_g1() == pool_pk:
                return AugSchemeMPL.sign(sk_child, bytes(pool_target))
        raise ValueError(f"Do not have key {pool_pk}")

    def get_farmer_wallet_tool(self) -> WalletTool:
        return WalletTool(self.constants, self.farmer_master_sk)

    def get_pool_wallet_tool(self) -> WalletTool:
        return WalletTool(self.constants, self.pool_master_sk)

    def get_consecutive_blocks(
        self,
        num_blocks: int,
        block_list_input: list[FullBlock] | None = None,
        *,
        farmer_reward_puzzle_hash: bytes32 | None = None,
        pool_reward_puzzle_hash: bytes32 | None = None,
        transaction_data: SpendBundle | None = None,
        seed: bytes = b"",
        time_per_block: float | None = None,
        force_overflow: bool = False,
        skip_slots: int = 0,  # Force at least this number of empty slots before the first SB
        guarantee_transaction_block: bool = False,  # Force that this block must be a tx block
        keep_going_until_tx_block: bool = False,  # keep making new blocks until we find a tx block
        normalized_to_identity_cc_eos: bool = False,
        normalized_to_identity_icc_eos: bool = False,
        normalized_to_identity_cc_sp: bool = False,
        normalized_to_identity_cc_ip: bool = False,
        current_time: bool = False,
        block_refs: list[uint32] = [],
        genesis_timestamp: uint64 | None = None,
        force_plot_id: bytes32 | None = None,
        dummy_block_references: bool = False,
        include_transactions: bool = False,
        skip_overflow: bool = False,
        min_signage_point: int = -1,
    ) -> list[FullBlock]:
        # make a copy to not have different invocations affect each other
        block_refs = block_refs[:]
        assert num_blocks > 0
        if block_list_input is not None:
            block_list = block_list_input.copy()
        else:
            block_list = []

        # these are heights of blocks that have transactions generators. Note
        # that there may be transactions blocks without generators. These are
        # used to generate dummy block references. Block references require
        # generators, not just transaction blocks.
        generator_block_heights: list[uint32] = []
        if dummy_block_references:
            # block references can only point to transaction blocks, so we need
            # to record which ones are
            for b in block_list:
                if b.transactions_generator is not None:
                    generator_block_heights.append(b.height)

        constants = self.constants

        if time_per_block is None:
            time_per_block = float(constants.SUB_SLOT_TIME_TARGET) / float(constants.SLOT_BLOCKS_TARGET)

        available_coins: list[Coin] = []
        # award coins aren't available to spend until the transaction block
        # after the one they were created by, so we "stage" them here to move
        # them into available_coins at the next transaction block
        pending_rewards: list[Coin] = []
        wallet: WalletTool | None = None
        rng: Random | None = None
        if include_transactions:
            # when we generate transactions in the chain, the caller cannot also
            # have ownership of the rewards and control the transactions
            assert farmer_reward_puzzle_hash is None
            assert pool_reward_puzzle_hash is None
            assert transaction_data is None

            for b in block_list:
                for coin in b.get_included_reward_coins():
                    if coin.puzzle_hash == self.farmer_ph:
                        available_coins.append(coin)
            print(
                f"found {len(available_coins)} reward coins in existing chain."
                "for simplicity, we assume the rewards are all unspent in the original chain"
            )
            wallet = self.get_farmer_wallet_tool()
            rng = Random()
            rng.seed(seed)

        if farmer_reward_puzzle_hash is None:
            farmer_reward_puzzle_hash = self.farmer_ph

        if len(block_list) == 0:
            if force_plot_id is not None:
                raise ValueError("Cannot specify plot_id for genesis block")
            initial_block_list_len = 0
            genesis = self.create_genesis_block(
                constants,
                seed,
                force_overflow=force_overflow,
                skip_slots=skip_slots,
                timestamp=(uint64(time.time()) if genesis_timestamp is None else genesis_timestamp),
            )
            self.log.info(f"Created block 0 iters: {genesis.total_iters}")
            num_empty_slots_added = skip_slots
            block_list = [genesis]
            num_blocks -= 1
        else:
            initial_block_list_len = len(block_list)
            num_empty_slots_added = uint32(0)  # Allows forcing empty slots in the beginning, for testing purposes

        if num_blocks == 0:
            return block_list

        blocks: dict[bytes32, BlockRecord]
        if block_list[-1].header_hash == self._block_cache_header:
            height_to_hash = self._block_cache_height_to_hash
            difficulty = self._block_cache_difficulty
            blocks = self._block_cache
        else:
            height_to_hash, difficulty, blocks = load_block_list(block_list, constants)

        latest_block: BlockRecord = blocks[block_list[-1].header_hash]
        curr = latest_block
        while not curr.is_transaction_block:
            curr = blocks[curr.prev_hash]
        assert curr.timestamp is not None
        last_timestamp = float(curr.timestamp)
        prev_tx_height = uint32(0)

        curr = latest_block
        blocks_added_this_sub_slot = 1

        while not curr.first_in_sub_slot:
            curr = blocks[curr.prev_hash]
            blocks_added_this_sub_slot += 1

        finished_sub_slots_at_sp: list[EndOfSubSlotBundle] = []  # Sub-slots since last block, up to signage point
        finished_sub_slots_at_ip: list[EndOfSubSlotBundle] = []  # Sub-slots since last block, up to infusion point
        sub_slot_iters: uint64 = latest_block.sub_slot_iters  # The number of iterations in one sub-slot
        same_slot_as_last = True  # Only applies to first slot, to prevent old blocks from being added
        sub_slot_start_total_iters: uint128 = latest_block.ip_sub_slot_total_iters(constants)
        sub_slots_finished = 0
        # this variable is true whenever there is a pending sub-epoch or epoch that needs to be added in the next block.
        pending_ses: bool = False

        block_generator: NewBlockGenerator | None = None
        if transaction_data is not None:
            additions = compute_additions_unchecked(transaction_data)
            removals = transaction_data.removals()
            if curr.height >= self.constants.HARD_FORK_HEIGHT:
                program = simple_solution_generator_backrefs(transaction_data).program
            else:
                program = simple_solution_generator(transaction_data).program
            block_refs = []
            cost = compute_block_cost(program, self.constants, uint32(curr.height + 1), prev_tx_height)
            block_generator = NewBlockGenerator(
                program,
                [],
                block_refs,
                transaction_data.aggregated_signature,
                additions,
                removals,
                cost,
            )

        # Start at the last block in block list
        # Get the challenge for that slot
        while True:
            slot_cc_challenge, slot_rc_challenge = get_challenges(
                constants,
                blocks,
                finished_sub_slots_at_sp,
                latest_block.header_hash,
            )
            prev_num_of_blocks = num_blocks
            if num_empty_slots_added < skip_slots:
                # If did not reach the target slots to skip, don't make any proofs for this sub-slot
                num_empty_slots_added += 1
            else:
                # Loop over every signage point (Except for the last ones, which are used for overflows)
                for signage_point_index in range(constants.NUM_SPS_SUB_SLOT - constants.NUM_SP_INTERVALS_EXTRA):
                    curr = latest_block
                    while curr.total_iters > sub_slot_start_total_iters + calculate_sp_iters(
                        constants, sub_slot_iters, uint8(signage_point_index)
                    ):
                        if curr.height == 0:
                            break
                        curr = blocks[curr.prev_hash]
                    if curr.total_iters > sub_slot_start_total_iters:
                        finished_sub_slots_at_sp = []

                    if same_slot_as_last:
                        if signage_point_index < latest_block.signage_point_index:
                            # Ignore this signage_point because it's in the past
                            continue

                        if signage_point_index <= min_signage_point:
                            # start farming blocks after min_signage_point
                            continue

                    signage_point: SignagePoint = get_signage_point(
                        constants,
                        BlockCache(blocks),
                        latest_block,
                        sub_slot_start_total_iters,
                        uint8(signage_point_index),
                        finished_sub_slots_at_sp,
                        sub_slot_iters,
                        normalized_to_identity_cc_sp,
                    )
                    if signage_point_index == 0:
                        cc_sp_output_hash: bytes32 = slot_cc_challenge
                    else:
                        assert signage_point.cc_vdf is not None
                        cc_sp_output_hash = signage_point.cc_vdf.output.get_hash()

                    prev_tx_height = pre_sp_tx_block_height(
                        constants=constants,
                        blocks=BlockCache(blocks),
                        prev_b_hash=latest_block.header_hash,
                        sp_index=uint8(signage_point_index),
                        finished_sub_slots=len(finished_sub_slots_at_ip),
                    )

                    qualified_proofs: list[tuple[uint64, ProofOfSpace]] = self.get_pospaces_for_challenge(
                        constants,
                        slot_cc_challenge,
                        cc_sp_output_hash,
                        seed,
                        difficulty,
                        sub_slot_iters,
                        curr.height,
                        prev_tx_height,
                        force_plot_id=force_plot_id,
                    )

                    for required_iters, proof_of_space in sorted(qualified_proofs, key=lambda t: t[0]):
                        if blocks_added_this_sub_slot == constants.MAX_SUB_SLOT_BLOCKS or force_overflow:
                            break
                        if same_slot_as_last:
                            if signage_point_index == latest_block.signage_point_index:
                                # Ignore this block because it's in the past
                                if required_iters <= latest_block.required_iters:
                                    continue

                        assert latest_block.header_hash in blocks

                        assert last_timestamp is not None
                        if proof_of_space.pool_contract_puzzle_hash is not None:
                            if pool_reward_puzzle_hash is not None:
                                # The caller wants to be paid to a specific address, but this PoSpace is tied to an
                                # address, so continue until a proof of space tied to a pk is found
                                continue
                            pool_target = PoolTarget(proof_of_space.pool_contract_puzzle_hash, uint32(0))
                        elif pool_reward_puzzle_hash is not None:
                            pool_target = PoolTarget(pool_reward_puzzle_hash, uint32(0))
                        else:
                            pool_target = PoolTarget(self.pool_ph, uint32(0))

                        new_gen = self.setup_new_gen(
                            generator_block_heights,
                            curr,
                            wallet,
                            rng,
                            available_coins,
                            prev_tx_height=prev_tx_height,
                            dummy_block_references=dummy_block_references,
                            block_generator=block_generator,
                            include_transactions=include_transactions,
                            block_refs=block_refs,
                        )

                        (
                            full_block,
                            block_record,
                            new_timestamp,
                        ) = get_full_block_and_block_record(
                            constants,
                            blocks,
                            sub_slot_start_total_iters,
                            uint8(signage_point_index),
                            proof_of_space,
                            slot_cc_challenge,
                            slot_rc_challenge,
                            farmer_reward_puzzle_hash,
                            pool_target,
                            last_timestamp,
                            time_per_block,
                            new_gen,
                            height_to_hash,
                            difficulty,
                            required_iters,
                            sub_slot_iters,
                            self.get_plot_signature,
                            self.get_pool_key_signature,
                            finished_sub_slots_at_ip,
                            signage_point,
                            latest_block,
                            seed,
                            normalized_to_identity_cc_ip=normalized_to_identity_cc_ip,
                            current_time=current_time,
                            overflow_cc_challenge=None,
                            overflow_rc_challenge=None,
                        )
                        if block_record.is_transaction_block:
                            block_generator = None
                            block_refs = []
                            keep_going_until_tx_block = False
                            assert full_block.foliage_transaction_block is not None
                        elif guarantee_transaction_block:
                            continue
                        self.log.info(
                            f"Created Block {full_block.height:4} "
                            f"prev-tx: {prev_tx_height:4} "
                            f"diff: {difficulty} "
                            f"time: {new_timestamp - last_timestamp:0.2f} "
                            f"additions: {len(new_gen.additions) if new_gen else 0:2} "
                            f"removals: {len(new_gen.removals) if new_gen else 0:2} "
                            f"refs: {len(full_block.transactions_generator_ref_list):3} "
                            f"iters: {block_record.total_iters} "
                            f"[{'TransactionBlock ' if block_record.is_transaction_block else ''}"
                            f"{'V1' if proof_of_space.param().size_v1 else 'V2'}]"
                        )
                        last_timestamp = new_timestamp
                        block_list.append(full_block)

                        if include_transactions:
                            for coin in full_block.get_included_reward_coins():
                                if coin.puzzle_hash == self.farmer_ph:
                                    pending_rewards.append(coin)
                            if full_block.is_transaction_block():
                                available_coins.extend(pending_rewards)
                                pending_rewards = []
                                if new_gen is not None:
                                    for rem in new_gen.removals:
                                        available_coins.remove(rem)
                                    available_coins.extend(new_gen.additions)

                        if full_block.transactions_generator is not None:
                            generator_block_heights.append(full_block.height)

                        blocks_added_this_sub_slot += 1
                        blocks[full_block.header_hash] = block_record
                        num_blocks -= 1

                        height_to_hash[uint32(full_block.height)] = full_block.header_hash
                        latest_block = blocks[full_block.header_hash]
                        finished_sub_slots_at_ip = []
                        # Reset pending_ses when a new block is created
                        pending_ses = False

                        if num_blocks <= 0 and not keep_going_until_tx_block:
                            self._block_cache_header = block_list[-1].header_hash
                            self._block_cache_height_to_hash = height_to_hash
                            self._block_cache_difficulty = difficulty
                            self._block_cache = blocks
                            return block_list

            # Finish the end of sub-slot and try again next sub-slot
            # End of sub-slot logic
            if len(finished_sub_slots_at_ip) == 0:
                # Block has been created within this sub-slot
                eos_iters: uint64 = uint64(sub_slot_iters - (latest_block.total_iters - sub_slot_start_total_iters))
                cc_input: ClassgroupElement = latest_block.challenge_vdf_output
                rc_challenge: bytes32 = latest_block.reward_infusion_new_challenge
            else:
                # No blocks were successfully created within this sub-slot
                eos_iters = sub_slot_iters
                cc_input = ClassgroupElement.get_default_element()
                rc_challenge = slot_rc_challenge
            cc_vdf, cc_proof = get_vdf_info_and_proof(
                constants,
                cc_input,
                slot_cc_challenge,
                eos_iters,
            )
            rc_vdf, rc_proof = get_vdf_info_and_proof(
                constants,
                ClassgroupElement.get_default_element(),
                rc_challenge,
                eos_iters,
            )

            eos_deficit: uint8 = (
                latest_block.deficit if latest_block.deficit > 0 else constants.MIN_BLOCKS_PER_CHALLENGE_BLOCK
            )
            icc_eos_vdf, icc_ip_proof = get_icc(
                constants,
                uint128(sub_slot_start_total_iters + sub_slot_iters),
                finished_sub_slots_at_ip,
                latest_block,
                blocks,
                sub_slot_start_total_iters,
                eos_deficit,
            )
            # End of slot vdf info for icc and cc have to be from challenge block or start of slot, respectively,
            # in order for light clients to validate.
            cc_vdf = VDFInfo(cc_vdf.challenge, sub_slot_iters, cc_vdf.output)
            if normalized_to_identity_cc_eos:
                _, cc_proof = get_vdf_info_and_proof(
                    constants,
                    ClassgroupElement.get_default_element(),
                    cc_vdf.challenge,
                    sub_slot_iters,
                    True,
                )
            # generate sub_epoch_summary, and if the last block was the last block of the sub-epoch or epoch
            # include the hash in the next sub-slot
            sub_epoch_summary: SubEpochSummary | None = None
            if not pending_ses:  # if we just created a sub-epoch summary, we can at least skip another sub-slot
                sub_epoch_summary = next_sub_epoch_summary(
                    constants,
                    BlockCache(blocks),
                    latest_block.required_iters,
                    block_list[-1],
                    False,
                )
            if sub_epoch_summary is not None:  # the previous block is the last block of the sub-epoch or epoch
                pending_ses = True
                ses_hash: bytes32 | None = sub_epoch_summary.get_hash()
                # if the last block is the last block of the epoch, we set the new sub-slot iters and difficulty
                new_sub_slot_iters: uint64 | None = sub_epoch_summary.new_sub_slot_iters
                new_difficulty: uint64 | None = sub_epoch_summary.new_difficulty

                self.log.info(f"Sub epoch summary: {sub_epoch_summary} for block {latest_block.height + 1}")
            else:  # the previous block is not the last block of the sub-epoch or epoch
                # Don't reset pending_ses to False here - it will be reset when a new block is created
                ses_hash = None
                new_sub_slot_iters = None
                new_difficulty = None

            if icc_eos_vdf is not None:
                # Icc vdf (Deficit of latest block is <= 4)
                if len(finished_sub_slots_at_ip) == 0:
                    # This means there are blocks in this sub-slot
                    curr = latest_block
                    while not curr.is_challenge_block(constants) and not curr.first_in_sub_slot:
                        curr = blocks[curr.prev_hash]
                    if curr.is_challenge_block(constants):
                        icc_eos_iters = uint64(sub_slot_start_total_iters + sub_slot_iters - curr.total_iters)
                    else:
                        icc_eos_iters = sub_slot_iters
                else:
                    # This means there are no blocks in this sub-slot
                    icc_eos_iters = sub_slot_iters
                icc_eos_vdf = VDFInfo(
                    icc_eos_vdf.challenge,
                    icc_eos_iters,
                    icc_eos_vdf.output,
                )
                if normalized_to_identity_icc_eos:
                    _, icc_ip_proof = get_vdf_info_and_proof(
                        constants,
                        ClassgroupElement.get_default_element(),
                        icc_eos_vdf.challenge,
                        icc_eos_iters,
                        True,
                    )
                icc_sub_slot: InfusedChallengeChainSubSlot | None = InfusedChallengeChainSubSlot(icc_eos_vdf)
                assert icc_sub_slot is not None
                icc_sub_slot_hash = icc_sub_slot.get_hash() if latest_block.deficit == 0 else None
                cc_sub_slot = ChallengeChainSubSlot(
                    cc_vdf,
                    icc_sub_slot_hash,
                    ses_hash,
                    new_sub_slot_iters,
                    new_difficulty,
                )
            else:
                # No icc
                icc_sub_slot = None
                cc_sub_slot = ChallengeChainSubSlot(cc_vdf, None, ses_hash, new_sub_slot_iters, new_difficulty)

            finished_sub_slots_at_ip.append(
                EndOfSubSlotBundle(
                    cc_sub_slot,
                    icc_sub_slot,
                    RewardChainSubSlot(
                        rc_vdf,
                        cc_sub_slot.get_hash(),
                        icc_sub_slot.get_hash() if icc_sub_slot is not None else None,
                        eos_deficit,
                    ),
                    SubSlotProofs(cc_proof, icc_ip_proof, rc_proof),
                )
            )

            finished_sub_slots_eos = finished_sub_slots_at_ip.copy()
            latest_block_eos = latest_block
            overflow_cc_challenge = finished_sub_slots_at_ip[-1].challenge_chain.get_hash()
            overflow_rc_challenge = finished_sub_slots_at_ip[-1].reward_chain.get_hash()
            sub_slots_finished += 1
            self.log.info(
                f"Sub slot finished. blocks included: {blocks_added_this_sub_slot} blocks_per_slot: "
                f"{(len(block_list) - initial_block_list_len) / sub_slots_finished} "
                f"Sub Epoch Summary Included: {sub_epoch_summary is not None} "
            )
            blocks_added_this_sub_slot = 0  # Sub slot ended, overflows are in next sub slot

            # Handle overflows: No overflows on new epoch or sub-epoch

            if (
                new_sub_slot_iters is None
                and num_empty_slots_added >= skip_slots
                and not pending_ses
                and not skip_overflow
            ):
                for signage_point_index in range(
                    constants.NUM_SPS_SUB_SLOT - constants.NUM_SP_INTERVALS_EXTRA,
                    constants.NUM_SPS_SUB_SLOT,
                ):
                    if same_slot_as_last and signage_point_index <= min_signage_point:
                        # start farming blocks after min_signage_point
                        continue
                    # note that we are passing in the finished slots which include the last slot
                    signage_point = get_signage_point(
                        constants,
                        BlockCache(blocks),
                        latest_block_eos,
                        sub_slot_start_total_iters,
                        uint8(signage_point_index),
                        finished_sub_slots_eos,
                        sub_slot_iters,
                        normalized_to_identity_cc_sp,
                    )
                    if signage_point_index == 0:
                        cc_sp_output_hash = slot_cc_challenge
                    else:
                        assert signage_point is not None
                        assert signage_point.cc_vdf is not None
                        cc_sp_output_hash = signage_point.cc_vdf.output.get_hash()

                    # If did not reach the target slots to skip, don't make any proofs for this sub-slot
                    prev_tx_height = pre_sp_tx_block_height(
                        constants=constants,
                        blocks=BlockCache(blocks),
                        prev_b_hash=latest_block.header_hash,
                        sp_index=uint8(signage_point_index),
                        finished_sub_slots=len(finished_sub_slots_at_ip),
                    )

                    qualified_proofs = self.get_pospaces_for_challenge(
                        constants,
                        slot_cc_challenge,
                        cc_sp_output_hash,
                        seed,
                        difficulty,
                        sub_slot_iters,
                        curr.height,
                        prev_tx_height,
                        force_plot_id=force_plot_id,
                    )
                    for required_iters, proof_of_space in sorted(qualified_proofs, key=lambda t: t[0]):
                        if blocks_added_this_sub_slot == constants.MAX_SUB_SLOT_BLOCKS:
                            break
                        assert last_timestamp is not None
                        if proof_of_space.pool_contract_puzzle_hash is not None:
                            if pool_reward_puzzle_hash is not None:
                                # The caller wants to be paid to a specific address, but this PoSpace is tied to an
                                # address, so continue until a proof of space tied to a pk is found
                                continue
                            pool_target = PoolTarget(proof_of_space.pool_contract_puzzle_hash, uint32(0))
                        elif pool_reward_puzzle_hash is not None:
                            pool_target = PoolTarget(pool_reward_puzzle_hash, uint32(0))
                        else:
                            pool_target = PoolTarget(self.pool_ph, uint32(0))

                        new_gen = self.setup_new_gen(
                            generator_block_heights,
                            curr,
                            wallet,
                            rng,
                            available_coins,
                            prev_tx_height=prev_tx_height,
                            dummy_block_references=dummy_block_references,
                            block_generator=block_generator,
                            include_transactions=include_transactions,
                            block_refs=block_refs,
                        )

                        (
                            full_block,
                            block_record,
                            new_timestamp,
                        ) = get_full_block_and_block_record(
                            constants,
                            blocks,
                            sub_slot_start_total_iters,
                            uint8(signage_point_index),
                            proof_of_space,
                            slot_cc_challenge,
                            slot_rc_challenge,
                            farmer_reward_puzzle_hash,
                            pool_target,
                            last_timestamp,
                            time_per_block,
                            new_gen,
                            height_to_hash,
                            difficulty,
                            required_iters,
                            sub_slot_iters,
                            self.get_plot_signature,
                            self.get_pool_key_signature,
                            finished_sub_slots_at_ip,
                            signage_point,
                            latest_block,
                            seed,
                            normalized_to_identity_cc_ip=normalized_to_identity_cc_ip,
                            current_time=current_time,
                            overflow_cc_challenge=overflow_cc_challenge,
                            overflow_rc_challenge=overflow_rc_challenge,
                        )
                        if block_record.is_transaction_block:
                            block_generator = None
                            block_refs = []
                            keep_going_until_tx_block = False
                            assert full_block.foliage_transaction_block is not None
                        elif guarantee_transaction_block:
                            continue
                        self.log.info(
                            f"Created Block {full_block.height:4} "
                            f"prev-tx: {prev_tx_height:4} "
                            f"diff: {difficulty} "
                            f"time: {new_timestamp - last_timestamp:0.2f} "
                            f"additions: {len(new_gen.additions) if new_gen else 0:2} "
                            f"removals: {len(new_gen.removals) if new_gen else 0:2} "
                            f"refs: {len(full_block.transactions_generator_ref_list):3} "
                            f"iters: {block_record.total_iters} "
                            f"[{'TransactionBlock ' if block_record.is_transaction_block else ''}"
                            f"{'V1 ' if proof_of_space.param().size_v1 else 'V2 '}"
                            "Overflow]"
                        )
                        last_timestamp = new_timestamp
                        block_list.append(full_block)

                        if include_transactions:
                            for coin in full_block.get_included_reward_coins():
                                if coin.puzzle_hash == self.farmer_ph:
                                    pending_rewards.append(coin)
                            if full_block.is_transaction_block():
                                available_coins.extend(pending_rewards)
                                pending_rewards = []
                                if new_gen is not None:
                                    for rem in new_gen.removals:
                                        available_coins.remove(rem)
                                    available_coins.extend(new_gen.additions)

                        if full_block.transactions_generator is not None:
                            generator_block_heights.append(full_block.height)

                        blocks_added_this_sub_slot += 1
                        blocks[full_block.header_hash] = block_record
                        num_blocks -= 1

                        height_to_hash[uint32(full_block.height)] = full_block.header_hash
                        latest_block = blocks[full_block.header_hash]
                        finished_sub_slots_at_ip = []
                        # Reset pending_ses when a new block is created
                        pending_ses = False

                        if num_blocks <= 0 and not keep_going_until_tx_block:
                            self._block_cache_header = block_list[-1].header_hash
                            self._block_cache_height_to_hash = height_to_hash
                            self._block_cache_difficulty = difficulty
                            self._block_cache = blocks
                            return block_list

            finished_sub_slots_at_sp = finished_sub_slots_eos.copy()
            same_slot_as_last = False
            sub_slot_start_total_iters = uint128(sub_slot_start_total_iters + sub_slot_iters)
            if num_blocks < prev_num_of_blocks:
                num_empty_slots_added += 1

            if new_sub_slot_iters is not None and new_difficulty is not None:  # new epoch
                sub_slot_iters = new_sub_slot_iters
                difficulty = new_difficulty

    def create_genesis_block(
        self,
        constants: ConsensusConstants,
        seed: bytes = b"",
        timestamp: uint64 | None = None,
        force_overflow: bool = False,
        skip_slots: int = 0,
    ) -> FullBlock:
        if timestamp is None:
            timestamp = uint64(time.time())

        finished_sub_slots: list[EndOfSubSlotBundle] = []
        unfinished_block: UnfinishedBlock | None = None
        ip_iters: uint64 = uint64(0)
        sub_slot_total_iters: uint128 = uint128(0)

        # Keep trying until we get a good proof of space that also passes sp filter
        while True:
            cc_challenge, rc_challenge = get_challenges(constants, {}, finished_sub_slots, None)
            for signage_point_index in range(constants.NUM_SPS_SUB_SLOT):
                signage_point: SignagePoint = get_signage_point(
                    constants,
                    BlockCache({}),
                    None,
                    sub_slot_total_iters,
                    uint8(signage_point_index),
                    finished_sub_slots,
                    constants.SUB_SLOT_ITERS_STARTING,
                )
                if signage_point_index == 0:
                    cc_sp_output_hash: bytes32 = cc_challenge
                else:
                    assert signage_point is not None
                    assert signage_point.cc_vdf is not None
                    cc_sp_output_hash = signage_point.cc_vdf.output.get_hash()
                    # If did not reach the target slots to skip, don't make any proofs for this sub-slot
                # we're creating the genesis block, its height is always 0
                qualified_proofs: list[tuple[uint64, ProofOfSpace]] = self.get_pospaces_for_challenge(
                    constants,
                    cc_challenge,
                    cc_sp_output_hash,
                    seed,
                    constants.DIFFICULTY_STARTING,
                    constants.SUB_SLOT_ITERS_STARTING,
                    uint32(0),
                    uint32(0),
                )

                # Try each of the proofs of space
                for required_iters, proof_of_space in qualified_proofs:
                    sp_iters: uint64 = calculate_sp_iters(
                        constants,
                        uint64(constants.SUB_SLOT_ITERS_STARTING),
                        uint8(signage_point_index),
                    )
                    ip_iters = calculate_ip_iters(
                        constants,
                        uint64(constants.SUB_SLOT_ITERS_STARTING),
                        uint8(signage_point_index),
                        required_iters,
                    )
                    is_overflow = is_overflow_block(constants, uint8(signage_point_index))
                    if force_overflow and not is_overflow:
                        continue
                    if len(finished_sub_slots) < skip_slots:
                        continue

                    unfinished_block = create_unfinished_block(
                        constants,
                        sub_slot_total_iters,
                        constants.SUB_SLOT_ITERS_STARTING,
                        uint8(signage_point_index),
                        sp_iters,
                        ip_iters,
                        proof_of_space,
                        cc_challenge,
                        constants.GENESIS_PRE_FARM_FARMER_PUZZLE_HASH,
                        PoolTarget(constants.GENESIS_PRE_FARM_POOL_PUZZLE_HASH, uint32(0)),
                        self.get_plot_signature,
                        self.get_pool_key_signature,
                        signage_point,
                        timestamp,
                        BlockCache({}),
                        seed=seed,
                        finished_sub_slots_input=finished_sub_slots,
                        compute_fees=compute_fee_test,
                    )
                    assert unfinished_block is not None
                    if not is_overflow:
                        cc_ip_vdf, cc_ip_proof = get_vdf_info_and_proof(
                            constants,
                            ClassgroupElement.get_default_element(),
                            cc_challenge,
                            ip_iters,
                        )
                        cc_ip_vdf = cc_ip_vdf.replace(number_of_iterations=ip_iters)
                        rc_ip_vdf, rc_ip_proof = get_vdf_info_and_proof(
                            constants,
                            ClassgroupElement.get_default_element(),
                            rc_challenge,
                            ip_iters,
                        )
                        assert unfinished_block is not None
                        total_iters_sp = uint128(sub_slot_total_iters + sp_iters)
                        return unfinished_block_to_full_block(
                            unfinished_block,
                            cc_ip_vdf,
                            cc_ip_proof,
                            rc_ip_vdf,
                            rc_ip_proof,
                            None,
                            None,
                            finished_sub_slots,
                            None,
                            BlockCache({}),
                            total_iters_sp,
                            constants.DIFFICULTY_STARTING,
                        )

                if signage_point_index == constants.NUM_SPS_SUB_SLOT - constants.NUM_SP_INTERVALS_EXTRA - 1:
                    # Finish the end of sub-slot and try again next sub-slot
                    cc_vdf, cc_proof = get_vdf_info_and_proof(
                        constants,
                        ClassgroupElement.get_default_element(),
                        cc_challenge,
                        constants.SUB_SLOT_ITERS_STARTING,
                    )
                    rc_vdf, rc_proof = get_vdf_info_and_proof(
                        constants,
                        ClassgroupElement.get_default_element(),
                        rc_challenge,
                        constants.SUB_SLOT_ITERS_STARTING,
                    )
                    cc_slot = ChallengeChainSubSlot(cc_vdf, None, None, None, None)
                    finished_sub_slots.append(
                        EndOfSubSlotBundle(
                            cc_slot,
                            None,
                            RewardChainSubSlot(
                                rc_vdf,
                                cc_slot.get_hash(),
                                None,
                                uint8(constants.MIN_BLOCKS_PER_CHALLENGE_BLOCK),
                            ),
                            SubSlotProofs(cc_proof, None, rc_proof),
                        )
                    )

                if unfinished_block is not None:
                    cc_ip_vdf, cc_ip_proof = get_vdf_info_and_proof(
                        constants,
                        ClassgroupElement.get_default_element(),
                        finished_sub_slots[-1].challenge_chain.get_hash(),
                        ip_iters,
                    )
                    rc_ip_vdf, rc_ip_proof = get_vdf_info_and_proof(
                        constants,
                        ClassgroupElement.get_default_element(),
                        finished_sub_slots[-1].reward_chain.get_hash(),
                        ip_iters,
                    )
                    total_iters_sp = uint128(
                        sub_slot_total_iters
                        + calculate_sp_iters(
                            self.constants,
                            self.constants.SUB_SLOT_ITERS_STARTING,
                            unfinished_block.reward_chain_block.signage_point_index,
                        )
                    )
                    return unfinished_block_to_full_block(
                        unfinished_block,
                        cc_ip_vdf,
                        cc_ip_proof,
                        rc_ip_vdf,
                        rc_ip_proof,
                        None,
                        None,
                        finished_sub_slots,
                        None,
                        BlockCache({}),
                        total_iters_sp,
                        constants.DIFFICULTY_STARTING,
                    )
            sub_slot_total_iters = uint128(sub_slot_total_iters + constants.SUB_SLOT_ITERS_STARTING)

    def get_pospaces_for_challenge(
        self,
        constants: ConsensusConstants,
        challenge_hash: bytes32,
        signage_point: bytes32,
        seed: bytes,
        difficulty: uint64,
        sub_slot_iters: uint64,
        height: uint32,
        prev_tx_height: uint32,
        force_plot_id: bytes32 | None = None,
    ) -> list[tuple[uint64, ProofOfSpace]]:
        found_proofs: list[tuple[uint64, ProofOfSpace]] = []
        rng = random.Random()
        rng.seed(seed)

        sp_interval_iters = calculate_sp_interval_iters(constants, sub_slot_iters)
        phase_out_epochs = num_phase_out_epochs(constants)

        for plot_info in self.plot_manager.plots.values():
            plot_id: bytes32 = plot_info.prover.get_id()
            if force_plot_id is not None and plot_id != force_plot_id:
                continue
            prefix_bits = calculate_prefix_bits(constants, height, plot_info.prover.get_param())
            if not passes_plot_filter(prefix_bits, plot_id, challenge_hash, signage_point):
                continue

            if plot_info.prover.get_version() == PlotVersion.V2:
                # v2 plots aren't valid until after the hard fork
                if prev_tx_height < constants.HARD_FORK2_HEIGHT:
                    continue

                if plot_info.prover.get_strength() < constants.MIN_PLOT_STRENGTH:
                    self.log.warn(
                        f"Plot strength ({plot_info.prover.get_strength()}) too low, "
                        f"cannot be used for farming: {plot_info.prover.get_filename()}"
                    )
                    continue
                if plot_info.prover.get_strength() > constants.MAX_PLOT_STRENGTH:
                    self.log.warn(
                        f"Plot strength ({plot_info.prover.get_strength()}) too high, "
                        f"cannot be used for farming: {plot_info.prover.get_filename()}"
                    )
                    continue
            elif prev_tx_height >= constants.HARD_FORK2_HEIGHT + phase_out_epochs * constants.EPOCH_BLOCKS:
                continue

            new_challenge: bytes32 = calculate_pos_challenge(plot_id, challenge_hash, signage_point)

            qualities: Sequence[QualityProtocol] = plot_info.prover.get_qualities_for_challenge(
                new_challenge, constants.QUALITY_PROOF_SCAN_FILTER
            )

            for idx, quality in enumerate(qualities):
                required_iters = calculate_iterations_quality(
                    constants,
                    quality.get_string(),
                    plot_info.prover.get_param(),
                    difficulty,
                    signage_point,
                )
                if required_iters >= sp_interval_iters:
                    continue

                proof = b""
                if isinstance(plot_info.prover, V1Prover):
                    proof = plot_info.prover.get_full_proof(new_challenge, idx)
                    if is_v1_phased_out(proof, prev_tx_height, constants):
                        continue
                elif isinstance(plot_info.prover, V2Prover):
                    assert isinstance(quality, V2Quality)
                    partial_proof = plot_info.prover.get_partial_proof(quality)
                    strength = plot_info.prover.get_strength()
                    proof = solve_proof(partial_proof, plot_id, strength, constants.PLOT_SIZE_V2)

                # Look up local_sk from plot to save locked memory
                (
                    pool_public_key_or_puzzle_hash,
                    farmer_public_key,
                    local_master_sk,
                ) = parse_plot_info(plot_info.prover.get_memo())
                local_sk = cached_master_sk_to_local_sk(local_master_sk)

                if isinstance(pool_public_key_or_puzzle_hash, G1Element):
                    include_taproot = False
                else:
                    assert isinstance(pool_public_key_or_puzzle_hash, bytes32)
                    include_taproot = True
                plot_pk = generate_plot_public_key(local_sk.get_g1(), farmer_public_key, include_taproot)
                proof_of_space: ProofOfSpace = make_pos(
                    new_challenge,
                    plot_info.pool_public_key,
                    plot_info.pool_contract_puzzle_hash,
                    plot_pk,
                    plot_info.prover.get_param(),
                    proof,
                )
                found_proofs.append((required_iters, proof_of_space))
        random_sample = found_proofs
        if len(found_proofs) >= 1:
            if rng.random() < 0.1:
                # Removes some proofs of space to create "random" chains, based on the seed
                random_sample = rng.sample(found_proofs, len(found_proofs) - 1)
        return random_sample


def get_signage_point(
    constants: ConsensusConstants,
    blocks: BlockRecordsProtocol,
    latest_block: BlockRecord | None,
    sub_slot_start_total_iters: uint128,
    signage_point_index: uint8,
    finished_sub_slots: list[EndOfSubSlotBundle],
    sub_slot_iters: uint64,
    normalized_to_identity_cc_sp: bool = False,
) -> SignagePoint:
    if signage_point_index == 0:
        return SignagePoint(None, None, None, None)
    sp_iters = calculate_sp_iters(constants, sub_slot_iters, signage_point_index)
    overflow = is_overflow_block(constants, signage_point_index)
    sp_total_iters = uint128(
        sub_slot_start_total_iters + calculate_sp_iters(constants, sub_slot_iters, signage_point_index)
    )

    (
        cc_vdf_challenge,
        rc_vdf_challenge,
        cc_vdf_input,
        rc_vdf_input,
        cc_vdf_iters,
        rc_vdf_iters,
    ) = get_signage_point_vdf_info(
        constants,
        finished_sub_slots,
        overflow,
        latest_block,
        blocks,
        sp_total_iters,
        sp_iters,
    )

    cc_sp_vdf, cc_sp_proof = get_vdf_info_and_proof(
        constants,
        cc_vdf_input,
        cc_vdf_challenge,
        cc_vdf_iters,
    )
    rc_sp_vdf, rc_sp_proof = get_vdf_info_and_proof(
        constants,
        rc_vdf_input,
        rc_vdf_challenge,
        rc_vdf_iters,
    )
    cc_sp_vdf = cc_sp_vdf.replace(number_of_iterations=sp_iters)
    if normalized_to_identity_cc_sp:
        _, cc_sp_proof = get_vdf_info_and_proof(
            constants,
            ClassgroupElement.get_default_element(),
            cc_sp_vdf.challenge,
            sp_iters,
            True,
        )
    return SignagePoint(cc_sp_vdf, cc_sp_proof, rc_sp_vdf, rc_sp_proof)


def finish_block(
    constants: ConsensusConstants,
    blocks: dict[bytes32, BlockRecord],
    height_to_hash: dict[uint32, bytes32],
    finished_sub_slots: list[EndOfSubSlotBundle],
    sub_slot_start_total_iters: uint128,
    signage_point_index: uint8,
    unfinished_block: UnfinishedBlock,
    required_iters: uint64,
    ip_iters: uint64,
    slot_cc_challenge: bytes32,
    slot_rc_challenge: bytes32,
    latest_block: BlockRecord,
    sub_slot_iters: uint64,
    difficulty: uint64,
    normalized_to_identity_cc_ip: bool = False,
) -> tuple[FullBlock, BlockRecord]:
    is_overflow = is_overflow_block(constants, signage_point_index)
    cc_vdf_challenge = slot_cc_challenge
    if len(finished_sub_slots) == 0:
        new_ip_iters = uint64(unfinished_block.total_iters - latest_block.total_iters)
        cc_vdf_input = latest_block.challenge_vdf_output
        rc_vdf_challenge = latest_block.reward_infusion_new_challenge
    else:
        new_ip_iters = ip_iters
        cc_vdf_input = ClassgroupElement.get_default_element()
        rc_vdf_challenge = slot_rc_challenge
    cc_ip_vdf, cc_ip_proof = get_vdf_info_and_proof(
        constants,
        cc_vdf_input,
        cc_vdf_challenge,
        new_ip_iters,
    )
    cc_ip_vdf = cc_ip_vdf.replace(number_of_iterations=ip_iters)
    if normalized_to_identity_cc_ip:
        _, cc_ip_proof = get_vdf_info_and_proof(
            constants,
            ClassgroupElement.get_default_element(),
            cc_ip_vdf.challenge,
            ip_iters,
            True,
        )
    deficit = calculate_deficit(
        constants,
        uint32(latest_block.height + 1),
        latest_block,
        is_overflow,
        len(finished_sub_slots),
    )

    icc_ip_vdf, icc_ip_proof = get_icc(
        constants,
        unfinished_block.total_iters,
        finished_sub_slots,
        latest_block,
        blocks,
        uint128(sub_slot_start_total_iters + sub_slot_iters) if is_overflow else sub_slot_start_total_iters,
        deficit,
    )

    rc_ip_vdf, rc_ip_proof = get_vdf_info_and_proof(
        constants,
        ClassgroupElement.get_default_element(),
        rc_vdf_challenge,
        new_ip_iters,
    )
    assert unfinished_block is not None
    sp_total_iters = uint128(
        sub_slot_start_total_iters + calculate_sp_iters(constants, sub_slot_iters, signage_point_index)
    )
    full_block: FullBlock = unfinished_block_to_full_block(
        unfinished_block,
        cc_ip_vdf,
        cc_ip_proof,
        rc_ip_vdf,
        rc_ip_proof,
        icc_ip_vdf,
        icc_ip_proof,
        finished_sub_slots,
        latest_block,
        BlockCache(blocks),
        sp_total_iters,
        difficulty,
    )

    block_record = block_to_block_record(
        constants, BlockCache(blocks), required_iters, full_block, sub_slot_iters=sub_slot_iters
    )
    return full_block, block_record


def get_challenges(
    constants: ConsensusConstants,
    blocks: dict[bytes32, BlockRecord],
    finished_sub_slots: list[EndOfSubSlotBundle],
    prev_header_hash: bytes32 | None,
) -> tuple[bytes32, bytes32]:
    if len(finished_sub_slots) == 0:
        if prev_header_hash is None:
            return constants.GENESIS_CHALLENGE, constants.GENESIS_CHALLENGE
        curr: BlockRecord = blocks[prev_header_hash]
        while not curr.first_in_sub_slot:
            curr = blocks[curr.prev_hash]
        assert curr.finished_challenge_slot_hashes is not None
        assert curr.finished_reward_slot_hashes is not None
        cc_challenge = curr.finished_challenge_slot_hashes[-1]
        rc_challenge = curr.finished_reward_slot_hashes[-1]
    else:
        cc_challenge = finished_sub_slots[-1].challenge_chain.get_hash()
        rc_challenge = finished_sub_slots[-1].reward_chain.get_hash()
    return cc_challenge, rc_challenge


def get_plot_dir(plot_dir_name: str = "test-plots", automated_testing: bool = True) -> Path:
    root_dir = DEFAULT_ROOT_PATH.parent
    if not automated_testing:  # make sure we don't accidentally stack directories.
        root_dir = (
            root_dir.parent
            if root_dir.parts[-1] == plot_dir_name.split("/", maxsplit=1)[0]
            or root_dir.parts[-1] == plot_dir_name.split("\\", maxsplit=1)[0]
            else root_dir
        )
    cache_path = root_dir.joinpath(plot_dir_name)

    ci = os.environ.get("CI")
    if ci is not None and not cache_path.exists() and automated_testing:
        raise Exception(f"Running in CI and expected path not found: {cache_path!r}")

    cache_path.mkdir(parents=True, exist_ok=True)
    return cache_path


def get_plot_tmp_dir(plot_dir_name: str = "test-plots", automated_testing: bool = True) -> Path:
    return get_plot_dir(plot_dir_name, automated_testing) / "tmp"


def load_block_list(
    block_list: list[FullBlock], constants: ConsensusConstants
) -> tuple[dict[uint32, bytes32], uint64, dict[bytes32, BlockRecord]]:
    difficulty = uint64(constants.DIFFICULTY_STARTING)
    sub_slot_iters = uint64(constants.SUB_SLOT_ITERS_STARTING)
    height_to_hash: dict[uint32, bytes32] = {}
    blocks: dict[bytes32, BlockRecord] = {}
    for full_block in block_list:
        if full_block.height != 0:
            if len(full_block.finished_sub_slots) > 0:
                if full_block.finished_sub_slots[0].challenge_chain.new_difficulty is not None:
                    difficulty = full_block.finished_sub_slots[0].challenge_chain.new_difficulty
                if full_block.finished_sub_slots[0].challenge_chain.new_sub_slot_iters is not None:
                    sub_slot_iters = full_block.finished_sub_slots[0].challenge_chain.new_sub_slot_iters
        if full_block.reward_chain_block.signage_point_index == 0:
            challenge = full_block.reward_chain_block.pos_ss_cc_challenge_hash
            sp_hash = challenge
        else:
            assert full_block.reward_chain_block.challenge_chain_sp_vdf is not None
            challenge = full_block.reward_chain_block.challenge_chain_sp_vdf.challenge
            sp_hash = full_block.reward_chain_block.challenge_chain_sp_vdf.output.get_hash()

        cache = BlockCache(blocks)
        prev_transaction_b_height = pre_sp_tx_block_height(
            constants=constants,
            blocks=cache,
            prev_b_hash=full_block.prev_header_hash,
            sp_index=full_block.reward_chain_block.signage_point_index,
            finished_sub_slots=len(full_block.finished_sub_slots),
        )
        required_iters = validate_pospace_and_get_required_iters(
            constants,
            full_block.reward_chain_block.proof_of_space,
            challenge,
            sp_hash,
            full_block.height,
            uint64(difficulty),
            prev_transaction_b_height,
        )
        assert required_iters is not None

        blocks[full_block.header_hash] = block_to_block_record(
            constants,
            cache,
            required_iters,
            full_block,
            sub_slot_iters,
        )
        height_to_hash[uint32(full_block.height)] = full_block.header_hash
    return height_to_hash, uint64(difficulty), blocks


def get_icc(
    constants: ConsensusConstants,
    vdf_end_total_iters: uint128,
    finished_sub_slots: list[EndOfSubSlotBundle],
    latest_block: BlockRecord,
    blocks: dict[bytes32, BlockRecord],
    sub_slot_start_total_iters: uint128,
    deficit: uint8,
) -> tuple[VDFInfo | None, VDFProof | None]:
    if len(finished_sub_slots) == 0:
        prev_deficit = latest_block.deficit
    else:
        prev_deficit = finished_sub_slots[-1].reward_chain.deficit

    if deficit == prev_deficit == constants.MIN_BLOCKS_PER_CHALLENGE_BLOCK:
        # new slot / overflow sb to new slot / overflow sb
        return None, None

    if deficit == (prev_deficit - 1) == (constants.MIN_BLOCKS_PER_CHALLENGE_BLOCK - 1):
        # new slot / overflow sb to challenge sb
        return None, None

    if len(finished_sub_slots) != 0:
        last_ss = finished_sub_slots[-1]
        assert last_ss.infused_challenge_chain is not None
        assert finished_sub_slots[-1].reward_chain.deficit <= (constants.MIN_BLOCKS_PER_CHALLENGE_BLOCK - 1)
        return get_vdf_info_and_proof(
            constants,
            ClassgroupElement.get_default_element(),
            last_ss.infused_challenge_chain.get_hash(),
            uint64(vdf_end_total_iters - sub_slot_start_total_iters),
        )

    curr = latest_block  # curr deficit is 0, 1, 2, 3, or 4
    while not curr.is_challenge_block(constants) and not curr.first_in_sub_slot:
        curr = blocks[curr.prev_hash]
    icc_iters = uint64(vdf_end_total_iters - latest_block.total_iters)
    if latest_block.is_challenge_block(constants):
        icc_input: ClassgroupElement | None = ClassgroupElement.get_default_element()
    else:
        icc_input = latest_block.infused_challenge_vdf_output
    assert icc_input is not None

    if curr.is_challenge_block(constants):  # Deficit 4
        icc_challenge_hash = curr.challenge_block_info_hash
    else:
        assert curr.finished_infused_challenge_slot_hashes is not None
        # First block in sub slot has deficit 0,1,2 or 3
        icc_challenge_hash = curr.finished_infused_challenge_slot_hashes[-1]
    return get_vdf_info_and_proof(
        constants,
        icc_input,
        icc_challenge_hash,
        icc_iters,
    )


def get_full_block_and_block_record(
    constants: ConsensusConstants,
    blocks: dict[bytes32, BlockRecord],
    sub_slot_start_total_iters: uint128,
    signage_point_index: uint8,
    proof_of_space: ProofOfSpace,
    slot_cc_challenge: bytes32,
    slot_rc_challenge: bytes32,
    farmer_reward_puzzle_hash: bytes32,
    pool_target: PoolTarget,
    last_timestamp: float,
    time_per_block: float,
    new_gen: NewBlockGenerator | None,
    height_to_hash: dict[uint32, bytes32],
    difficulty: uint64,
    required_iters: uint64,
    sub_slot_iters: uint64,
    get_plot_signature: Callable[[bytes32, G1Element], G2Element],
    get_pool_signature: Callable[[PoolTarget, G1Element | None], G2Element | None],
    finished_sub_slots: list[EndOfSubSlotBundle],
    signage_point: SignagePoint,
    prev_block: BlockRecord,
    seed: bytes = b"",
    *,
    overflow_cc_challenge: bytes32 | None = None,
    overflow_rc_challenge: bytes32 | None = None,
    normalized_to_identity_cc_ip: bool = False,
    current_time: bool = False,
) -> tuple[FullBlock, BlockRecord, float]:
    # we're simulating time between blocks here. The more VDF iterations the
    # blocks advances, the longer it should have taken (and vice versa). This
    # formula is meant to converge at 1024 iters per the specified
    # time_per_block (which defaults to 18.75 seconds)
    time_per_block *= (((sub_slot_iters / 1024) - 1) * 0.2) + 1
    if current_time is True:
        timestamp = max(int(time.time()), last_timestamp + time_per_block)
    else:
        timestamp = last_timestamp + time_per_block
    sp_iters = calculate_sp_iters(constants, sub_slot_iters, signage_point_index)
    ip_iters = calculate_ip_iters(constants, sub_slot_iters, signage_point_index, required_iters)

    unfinished_block = create_unfinished_block(
        constants,
        sub_slot_start_total_iters,
        sub_slot_iters,
        signage_point_index,
        sp_iters,
        ip_iters,
        proof_of_space,
        slot_cc_challenge,
        farmer_reward_puzzle_hash,
        pool_target,
        get_plot_signature,
        get_pool_signature,
        signage_point,
        uint64(timestamp),
        BlockCache(blocks),
        seed,
        new_gen,
        prev_block,
        finished_sub_slots,
        compute_fees=compute_fee_test,
    )

    if (overflow_cc_challenge is not None) and (overflow_rc_challenge is not None):
        slot_cc_challenge = overflow_cc_challenge
        slot_rc_challenge = overflow_rc_challenge

    full_block, block_record = finish_block(
        constants,
        blocks,
        height_to_hash,
        finished_sub_slots,
        sub_slot_start_total_iters,
        signage_point_index,
        unfinished_block,
        required_iters,
        ip_iters,
        slot_cc_challenge,
        slot_rc_challenge,
        prev_block,
        sub_slot_iters,
        difficulty,
        normalized_to_identity_cc_ip,
    )

    return full_block, block_record, timestamp


# these are the costs of unknown conditions, as defined chia_rs here:
# https://github.com/Chia-Network/chia_rs/pull/181
def compute_cost_table() -> list[int]:
    A = 17
    B = 16
    s = []
    NUM = 100
    DEN = 1
    MAX = 1 << 59
    for i in range(256):
        v = str(NUM // DEN)
        v1 = v[:3] + ("0" * (len(v) - 3))
        s.append(int(v1))
        NUM *= A
        DEN *= B
        assert NUM < 1 << 64
        assert DEN < 1 << 64
        if NUM > MAX:
            NUM >>= 5
            DEN >>= 5
    return s


CONDITION_COSTS = compute_cost_table()


def conditions_cost(conds: Program, *, charge_for_conditions: bool) -> uint64:
    free_conditions = 100

    condition_cost = 0
    for cond in conds.as_iter():
        condition = cond.first().as_atom()

        # this is new in hard fork 2
        if free_conditions > 0:
            free_conditions -= 1
        elif charge_for_conditions:
            condition_cost += ConditionCost.GENERIC_CONDITION_COST.value

        if condition == ConditionOpcode.CREATE_COIN:
            condition_cost += ConditionCost.CREATE_COIN.value
        # after the 2.0 hard fork, two byte conditions (with no leading 0)
        # have costs. Account for that.
        elif len(condition) == 2 and condition[0] != 0:
            condition_cost += CONDITION_COSTS[condition[1]]
        elif condition == ConditionOpcode.SOFTFORK.value:
            arg = cond.rest().first().as_int()
            condition_cost += arg * 10000
        elif condition in {
            ConditionOpcode.AGG_SIG_UNSAFE,
            ConditionOpcode.AGG_SIG_ME,
            ConditionOpcode.AGG_SIG_PARENT,
            ConditionOpcode.AGG_SIG_PUZZLE,
            ConditionOpcode.AGG_SIG_AMOUNT,
            ConditionOpcode.AGG_SIG_PUZZLE_AMOUNT,
            ConditionOpcode.AGG_SIG_PARENT_AMOUNT,
            ConditionOpcode.AGG_SIG_PARENT_PUZZLE,
        }:
            condition_cost += ConditionCost.AGG_SIG.value
    return uint64(condition_cost)


def compute_fee_test(additions: Sequence[Coin], removals: Sequence[Coin]) -> uint64:
    removal_amount = 0
    addition_amount = 0
    for coin in removals:
        removal_amount += coin.amount
    for coin in additions:
        addition_amount += coin.amount

    ret = removal_amount - addition_amount
    # in order to allow creating blocks that mint coins, clamp the fee
    # to 0, if it ends up being negative
    ret = max(ret, 0)
    return uint64(ret)


@dataclass
class BlockToolsNewPlotResult:
    plot_id: bytes32
    new_plot: bool


# Remove these counters when `create_block_tools` and `create_block_tools_async` are removed
create_block_tools_async_count = 0
create_block_tools_count = 0

# Note: tests that still use `create_block_tools` and `create_block_tools_async` should probably be
# moved to the bt fixture in conftest.py. Take special care to find out if the users of these functions
# need different BlockTools instances

# All tests need different root directories containing different config.yaml files.
# The daemon's listen port is configured in the config.yaml, and the only way a test can control which
# listen port it uses is to write it to the config file.


@asynccontextmanager
async def create_block_tools_async(
    constants: ConsensusConstants = test_constants,
    root_path: Path | None = None,
    keychain: Keychain | None = None,
    config_overrides: dict[str, Any] | None = None,
    num_og_plots: int = 15,
    num_pool_plots: int = 5,
    num_non_keychain_plots: int = 3,
    testrun_uid: str | None = None,
) -> AsyncIterator[BlockTools]:
    global create_block_tools_async_count
    create_block_tools_async_count += 1
    print(f"  create_block_tools_async called {create_block_tools_async_count} times")

    with BlockTools(constants, root_path, keychain, config_overrides=config_overrides) as bt:
        await bt.setup_keys()
        await bt.setup_plots(
            num_og_plots=num_og_plots,
            num_pool_plots=num_pool_plots,
            num_non_keychain_plots=num_non_keychain_plots,
            testrun_uid=testrun_uid,
        )

        yield bt


@contextmanager
def create_block_tools(
    constants: ConsensusConstants = test_constants,
    root_path: Path | None = None,
    keychain: Keychain | None = None,
    config_overrides: dict[str, Any] | None = None,
) -> Iterator[BlockTools]:
    global create_block_tools_count
    create_block_tools_count += 1
    print(f"  create_block_tools called {create_block_tools_count} times")
    with BlockTools(constants, root_path, keychain, config_overrides=config_overrides) as bt:
        asyncio.get_event_loop().run_until_complete(bt.setup_keys())
        asyncio.get_event_loop().run_until_complete(bt.setup_plots())
        yield bt


def make_unfinished_block(
    block: FullBlock, constants: ConsensusConstants, *, force_overflow: bool = False
) -> UnfinishedBlock:
    if force_overflow or is_overflow_block(constants, block.reward_chain_block.signage_point_index):
        finished_ss = block.finished_sub_slots[:-1]
    else:
        finished_ss = block.finished_sub_slots

    return UnfinishedBlock(
        finished_ss,
        block.reward_chain_block.get_unfinished(),
        block.challenge_chain_sp_proof,
        block.reward_chain_sp_proof,
        block.foliage,
        block.foliage_transaction_block,
        block.transactions_info,
        block.transactions_generator,
        block.transactions_generator_ref_list,
    )