#include <gtest/gtest.h>

#include <atomic>
#include <algorithm>
#include <chrono>
#include <cstdint>
#include <cstring>
#include <map>
#include <mutex>
#include <numeric>
#include <stdexcept>
#include <string>
#include <thread>
#include <vector>

#include "dpf/app_flow.hpp"
#include "dpf/async_protocol.hpp"
#include "dpf/compose_async.hpp"
#include "dpf/net/async_round_sink.hpp"
#include "dpf/net/async_sctp_stream_array.hpp"
#include "dpf/net/async_stream_array.hpp"
#include "dpf/net/round_lane.hpp"
#include "dpf/net/io_pool.hpp"
#include "dpf/net/party_session.hpp"
#include "dpf/net/tcp_mesh.hpp"
#include "dpf/bit_inject.hpp"
#include "dpf/circuit.hpp"
#include "dpf/compose.hpp"
#include "dpf/online_session.hpp"
#include "dpf/party_run.hpp"
#include "dpf/prep_source.hpp"
#include "dpf/buffered_prg.hpp"
#include "dpf/factory_gadgets.hpp"
#include "dpf/factory_tapes.hpp"
#include "dpf/net/dealer_cursor.hpp"
#include "dpf/protocol_factory.hpp"
#include "dpf/protocol_roles.hpp"
#include "dpf/revealing.hpp"
#include "dpf/schedule_streams.hpp"
#include "dpf/cost_pass.hpp"
#include "dpf/edabit.hpp"
#include "dpf/fixed_share.hpp"
#include "dpf/gilboa.hpp"
#include "dpf/matmul.hpp"
#include "dpf/net/edge_mesh.hpp"
#include "dpf/net/memory_sink.hpp"
#include "dpf/net/stream_array.hpp"
#include "dpf/net/stream_mesh.hpp"
#include "dpf/net/stream_edge_mesh.hpp"
#include "dpf/protocol_roles.hpp"
#include "dpf/ot_pack.hpp"
#include "dpf/protocol.hpp"
#include "dpf/random.hpp"
#include "dpf/rss_seed.hpp"
#include "dpf/share_cmp.hpp"
#include "dpf/share_expr.hpp"
#include "dpf/share_vec.hpp"
#include "dpf/shuffle.hpp"
#include "dpf/trunc.hpp"

using dpf::protocol::domain;
using dpf::protocol::phase;
using dpf::protocol::round_dir;
using dpf::protocol::schedule_round;
using dpf::protocol::schedule_session;
using dpf::protocol::receive_rule;

namespace
{

void drive_both(const dpf::protocol::plan & p,
    std::vector<std::vector<std::uint8_t>> & v0,
    std::vector<std::vector<std::uint8_t>> & v1)
{
    auto slots = p.slot_bytes_all();
    if (slots.empty())
        throw std::logic_error("drive_both: plan has no exchange");
    auto sinks = dpf::net::make_memory_sink_pair(1, slots);
    std::map<std::uint32_t, dpf::protocol::kernel_fn> kernels;
    std::mutex mu;
    std::exception_ptr err;
    auto run = [&](dpf::net::memory_sink & sink,
                   std::vector<std::vector<std::uint8_t>> & values,
                   std::size_t party) {
        try
        {
            dpf::protocol::drive(p, sink, values, kernels, party);
        }
        catch (...)
        {
            std::lock_guard<std::mutex> lock(mu);
            if (!err)
                err = std::current_exception();
        }
    };
    std::thread t0([&] { run(sinks.first, v0, 0); });
    std::thread t1([&] { run(sinks.second, v1, 1); });
    t0.join();
    t1.join();
    if (err)
        std::rethrow_exception(err);
}

void put_raw(std::vector<std::vector<std::uint8_t>> & values,
    dpf::protocol::node n, const void * src, std::size_t nbyte)
{
    values[n.id].assign(nbyte, 0);
    std::memcpy(values[n.id].data(), src, nbyte);
}

}  // namespace

// ---------------------------------------------------------------------------
// Wave 1
// ---------------------------------------------------------------------------

TEST(RssSeed, RandomReplicatedSums)
{
    auto bundle = dpf::rss::sample_seed_bundle();
    auto all = dpf::rss::random_replicated_all<std::uint64_t>(bundle, 7);
    EXPECT_EQ(all.p0.next, all.p1.own);
    EXPECT_EQ(all.p1.next, all.p2.own);
    EXPECT_EQ(all.p2.next, all.p0.own);
}

TEST(RssSeed, ZeroSharesSumToZero)
{
    auto bundle = dpf::rss::sample_seed_bundle();
    auto s0 = dpf::rss::party_seeds::from_bundle(bundle, 0);
    auto s1 = dpf::rss::party_seeds::from_bundle(bundle, 1);
    auto s2 = dpf::rss::party_seeds::from_bundle(bundle, 2);
    const auto z0 = dpf::rss::zero_share<std::uint64_t>(s0, 3);
    const auto z1 = dpf::rss::zero_share<std::uint64_t>(s1, 3);
    const auto z2 = dpf::rss::zero_share<std::uint64_t>(s2, 3);
    EXPECT_EQ(static_cast<std::uint64_t>(z0 + z1 + z2), 0u);
}

TEST(RssSeed, MulLocalThenRefresh)
{
    auto bundle = dpf::rss::sample_seed_bundle();
    const std::uint64_t x = 11, y = 13;
    auto [xs0, xs1, xs2] = dpf::make_replicated_shares(x);
    auto [ys0, ys1, ys2] = dpf::make_replicated_shares(y);
    auto s0 = dpf::rss::party_seeds::from_bundle(bundle, 0);
    auto s1 = dpf::rss::party_seeds::from_bundle(bundle, 1);
    auto s2 = dpf::rss::party_seeds::from_bundle(bundle, 2);
    const auto y0 = dpf::rss::rss_mul_local(s0, xs0.own, xs0.next, ys0.own,
        ys0.next, 0);
    const auto y1 = dpf::rss::rss_mul_local(s1, xs1.own, xs1.next, ys1.own,
        ys1.next, 0);
    const auto y2 = dpf::rss::rss_mul_local(s2, xs2.own, xs2.next, ys2.own,
        ys2.next, 0);
    EXPECT_EQ(static_cast<std::uint64_t>(y0 + y1 + y2), x * y);
}

TEST(ScheduleSession, PipelineCreditIndependentPads)
{
    auto tape = std::make_shared<std::vector<std::uint8_t>>();
    auto rounds = dpf::protocol::make_pad_rounds(2, 8, tape);
    EXPECT_EQ(rounds[0].round_phase, phase::setup);
    EXPECT_EQ(rounds[0].dir, round_dir::send_next);

    auto [a, b] = dpf::net::make_memory_sink_pair(1, {8, 8});
    schedule_session s0(1, a, rounds, /*pipeline_credit=*/1);
    schedule_session s1(1, b, rounds, /*pipeline_credit=*/1);
    s0.submit(0);
    s1.submit(0);
    s0.drive();
    s1.drive();
    EXPECT_TRUE(s0.done(0));
    EXPECT_TRUE(s1.done(0));
    EXPECT_GT(s0.tally().setup_bytes_out, 0u);
}

TEST(Compose, BinNotLocalConvertible)
{
    EXPECT_FALSE(dpf::protocol::detail::local_convertible(domain::bin,
        domain::a));
    dpf::protocol::composer c(0);
    auto a = c.input(domain::a, 8);
    EXPECT_THROW(c.as(a, domain::bin), std::invalid_argument);
}

TEST(Compose, ArithKernelsPresent)
{
    EXPECT_TRUE(dpf::protocol::detail::has_builtin_walk_kernel(
        dpf::protocol::opcodes::trunc_prob));
    EXPECT_TRUE(dpf::protocol::detail::has_builtin_walk_kernel(
        dpf::protocol::opcodes::rss_zero));
    EXPECT_TRUE(dpf::protocol::detail::has_builtin_walk_kernel(
        dpf::protocol::opcodes::bin_a2b));
    EXPECT_TRUE(dpf::protocol::detail::has_builtin_walk_kernel(
        dpf::protocol::opcodes::share_cmp));
}

TEST(Compose, RssZeroFromSeeds)
{
    auto bundle = dpf::rss::sample_seed_bundle();
    auto seeds = dpf::rss::party_seeds::from_bundle(bundle, 0);
    dpf::protocol::composer c(0);
    auto z = c.rss_zero_mask(domain::a, 8, 3);
    auto p = c.default_plan();
    std::vector<std::vector<std::uint8_t>> values(p.nodes().size());
    dpf::protocol::drive_options opt;
    opt.rss_seeds = &seeds;
    dpf::protocol::detail::run_arith_local(p, z.id, 0, 0, 1, values, opt);
    std::uint64_t got = 0;
    std::memcpy(&got, values[z.id].data(), 8);
    EXPECT_EQ(got, dpf::rss::zero_share<std::uint64_t>(seeds, 3));
}

TEST(Compose, CheckOpenRejectsCorrupt)
{
    dpf::protocol::composer c(0);
    auto x = c.input(domain::a, 8);
    auto ex = c.exchange(x);
    auto p = c.default_plan();
    auto slots = p.slot_bytes_all();
    ASSERT_FALSE(slots.empty());
    auto [sink0, sink1] = dpf::net::make_memory_sink_pair(1, slots);
    std::vector<std::vector<std::uint8_t>> v0(p.nodes().size());
    std::vector<std::vector<std::uint8_t>> v1(p.nodes().size());
    const std::uint64_t secret = 42;
    v0[x.id].assign(8, 0);
    v1[x.id].assign(8, 0);
    std::memcpy(v0[x.id].data(), &secret, 8);
    std::map<std::uint32_t, std::vector<std::uint8_t>> oracle;
    std::vector<std::uint8_t> expect(8, 0);
    std::memcpy(expect.data(), &secret, 8);
    oracle[ex.id] = expect;
    dpf::protocol::drive_options opt;
    opt.cleartext = true;
    opt.check_open = true;
    opt.clear_oracle = &oracle;
    std::uint64_t bad = 7;
    std::memcpy(v1[x.id].data(), &bad, 8);
    std::map<std::uint32_t, dpf::protocol::kernel_fn> kernels;
    std::exception_ptr err;
    std::thread t0([&] {
        try
        {
            dpf::protocol::drive(p, sink0, v0, kernels, 0, opt);
        }
        catch (...)
        {
            err = std::current_exception();
        }
    });
    std::thread t1([&] {
        try
        {
            dpf::protocol::drive(p, sink1, v1, kernels, 1, opt);
        }
        catch (...)
        {
            err = std::current_exception();
        }
    });
    t0.join();
    t1.join();
    ASSERT_TRUE(err != nullptr);
}

// ---------------------------------------------------------------------------
// Wave 2
// ---------------------------------------------------------------------------

TEST(EdaBit, A2BThenReconstruct)
{
    for (unsigned ell : {1u, 8u, 32u, 64u})
    {
        const std::uint64_t x = 0x123456789abcdef0ull
            & ((ell >= 64) ? ~0ull : ((1ull << ell) - 1ull));
        auto eda = dpf::edabit::sample_edabit_pair<std::uint64_t>(ell);
        EXPECT_EQ(eda.p0.arith + eda.p1.arith, eda.clear_r);
        auto bits = dpf::edabit::a2b_gmw_pair(eda, x, std::uint64_t{0});
        const auto got = dpf::edabit::reconstruct_bits(bits.first, bits.second, ell);
        EXPECT_EQ(got, x) << "ell=" << ell;
    }
}

TEST(EdaBit, RssArithEqualsBits)
{
    auto bundle = dpf::rss::sample_seed_bundle();
    auto all = dpf::edabit::sample_rss_all<std::uint64_t>(bundle, 16, 0);
    EXPECT_EQ(all.p0.arith + all.p1.arith + all.p2.arith, all.clear_r);
    auto bit = [](const std::vector<std::uint8_t> & p, unsigned i) {
        return static_cast<unsigned>((p[i / 8u] >> (i % 8u)) & 1u);
    };
    std::uint64_t from_bits = 0;
    for (unsigned i = 0; i < 16; ++i)
    {
        EXPECT_EQ(bit(all.p0.bits_next, i), bit(all.p1.bits_packed, i));
        EXPECT_EQ(bit(all.p1.bits_next, i), bit(all.p2.bits_packed, i));
        EXPECT_EQ(bit(all.p2.bits_next, i), bit(all.p0.bits_packed, i));
        const unsigned b = bit(all.p0.bits_packed, i) ^ bit(all.p1.bits_packed, i)
            ^ bit(all.p2.bits_packed, i);
        from_bits |= static_cast<std::uint64_t>(b) << i;
    }
    EXPECT_EQ(from_bits, all.clear_r);
}

TEST(EdaBit, GmwA2bMatchesClear)
{
    const std::uint64_t x0 = 30, x1 = 12;
    auto eda = dpf::edabit::sample_edabit_pair<std::uint64_t>(8);
    auto bits = dpf::edabit::a2b_gmw_pair(eda, x0, x1);
    EXPECT_EQ(dpf::edabit::reconstruct_bits(bits.first, bits.second, 8), x0 + x1);
}

TEST(EdaBit, DaBitPairConsistent)
{
    auto d = dpf::ot::sample_dabit_pair<std::uint64_t>();
    EXPECT_EQ(static_cast<std::uint64_t>(d.p0.arith + d.p1.arith),
        static_cast<std::uint64_t>((d.p0.bit ^ d.p1.bit) & 1u));
}

TEST(BitInject, ClearMatches)
{
    EXPECT_EQ(dpf::bit_inject::inject_clear<std::uint64_t>(1, 0, 5, 7), 12u);
    EXPECT_EQ(dpf::bit_inject::inject_clear<std::uint64_t>(1, 1, 5, 7), 0u);
}

TEST(BitInject, RssAndLocal)
{
    auto bundle = dpf::rss::sample_seed_bundle();
    auto s0 = dpf::rss::party_seeds::from_bundle(bundle, 0);
    auto s1 = dpf::rss::party_seeds::from_bundle(bundle, 1);
    auto s2 = dpf::rss::party_seeds::from_bundle(bundle, 2);
    const auto y0 = dpf::bit_inject::rss_and_local(s0, 1, 0, 1, 0, 0);
    const auto y1 = dpf::bit_inject::rss_and_local(s1, 0, 0, 0, 0, 0);
    const auto y2 = dpf::bit_inject::rss_and_local(s2, 0, 1, 0, 1, 0);
    EXPECT_EQ(static_cast<std::uint8_t>(y0 ^ y1 ^ y2), 1u);
}

// ---------------------------------------------------------------------------
// Wave 3
// ---------------------------------------------------------------------------

TEST(Trunc, ProbErrorAtMostOne)
{
    const std::uint64_t x0 = 100, x1 = 50;
    const auto got = dpf::trunc::trunc_prob_clear(x0, x1, 3);
    const auto exact = (x0 + x1) >> 3;
    EXPECT_LE(got > exact ? got - exact : exact - got, 1u);
}

TEST(Trunc, ExactPartyUsesDelta)
{
    const std::uint64_t x = 0x12345;
    const std::uint64_t x0 = 0x1111, x1 = x - x0;
    auto prep = dpf::trunc::make_trunc_exact_prep<std::uint64_t>(20, 4);
    auto [t0, t1] = dpf::trunc::trunc_exact_pair(x0, x1, prep);
    EXPECT_EQ(t0 + t1, dpf::trunc::trunc_exact_clear(x0, x1, 20, 4));
    // Changing delta (via different x) changes the party-0 share.
    auto [u0, u1] = dpf::trunc::trunc_exact_pair(x0 + 16, x1, prep);
    EXPECT_NE(t0, u0);
    (void)u1;
}

TEST(Trunc, MulTruncClear)
{
    auto z = dpf::trunc::mul_trunc_clear<std::uint64_t>(3, 0, 5, 0, 0);
    EXPECT_EQ(z.z0 + z.z1, 15u);
}

TEST(ShareCmp, ReluMaxDivParty)
{
    EXPECT_EQ(dpf::share_cmp::relu_clear<std::uint64_t>(5, 0, 64), 5u);
    EXPECT_EQ(dpf::share_cmp::relu_clear<std::uint64_t>(1ull << 63, 0, 64), 0u);
    auto [r0, r1] = dpf::share_cmp::relu_party_pair<std::uint64_t>(5, 0, 64);
    EXPECT_EQ(r0 + r1, 5u);
    auto [m0, m1] = dpf::share_cmp::max_party_pair<std::uint64_t>(3, 0, 9, 0, 64);
    EXPECT_EQ(m0 + m1, 9u);
    const std::uint64_t hi = 1ull << 63;
    auto prep = dpf::share_cmp::sample_msb_prep<std::uint64_t>(64);
    auto gt_hi = dpf::share_cmp::gt_party_pair<std::uint64_t>(
        hi + 5, 0, hi, 0, prep, 64);
    EXPECT_EQ(static_cast<unsigned>(gt_hi.first ^ gt_hi.second), 1u);
    auto gt_lo = dpf::share_cmp::gt_party_pair<std::uint64_t>(
        1, 0, hi, 0, prep, 64);
    EXPECT_EQ(static_cast<unsigned>(gt_lo.first ^ gt_lo.second), 0u);
    EXPECT_EQ(dpf::share_cmp::div_clear<std::uint64_t>(20, 4), 5u);
    auto [q0, q1] = dpf::share_cmp::div_party_pair<std::uint64_t>(20, 0, 4, 0, 16);
    EXPECT_EQ(q0 + q1, 5u);
    EXPECT_TRUE(dpf::share_cmp::range_ok_clear<std::uint64_t>(15, 4));
}

TEST(ShareCmp, Declassify)
{
    auto [a, b] = dpf::share_cmp::share_input<std::uint64_t>(42, 0);
    EXPECT_EQ(dpf::share_cmp::declassify(a, b), 42u);
}

// ---------------------------------------------------------------------------
// Wave 4
// ---------------------------------------------------------------------------

TEST(ShareVec, MulDeclassify)
{
    std::vector<std::uint64_t> clear = {1, 2, 3, 4};
    auto [x0, x1] = dpf::share_vec<std::uint64_t>::share(clear);
    auto [y0, y1] = dpf::share_vec<std::uint64_t>::share(clear);
    auto [z0, z1] = dpf::share_vec<std::uint64_t>::mul(x0, x1, y0, y1);
    auto prod = dpf::share_vec<std::uint64_t>::declassify(z0, z1);
    for (std::size_t i = 0; i < clear.size(); ++i)
        EXPECT_EQ(prod[i], clear[i] * clear[i]);
}

TEST(FixedShare, MulViaTrunc)
{
    using F = dpf::fixed<16, 16>;
    auto a = F::from_integer(3);
    auto b = F::from_integer(4);
    auto c = F::mul_clear(a, b);
    EXPECT_EQ(c.raw >> 16, 12u);
    std::vector<F::ring> ca = {a.raw}, cb = {b.raw};
    auto [x0, x1] = dpf::share_vec<F::ring>::share(ca);
    auto [y0, y1] = dpf::share_vec<F::ring>::share(cb);
    auto [z0, z1] = dpf::fixed_mul_share<16, 16>(x0, x1, y0, y1);
    auto got = dpf::share_vec<F::ring>::declassify(z0, z1);
    EXPECT_EQ(got[0], c.raw);
}

TEST(ShareExpr, MulAddDeclassify)
{
    const std::uint64_t x = 6, y = 7, w = 5;
    dpf::expr::recorder r0(0), r1(1);
    auto [x0, x1] = dpf::share_cmp::share_input(x, 0);
    auto [y0, y1] = dpf::share_cmp::share_input(y, 0);
    auto [w0, w1] = dpf::share_cmp::share_input(w, 0);
    auto hx0 = r0.bind(x0);
    auto hy0 = r0.bind(y0);
    auto hw0 = r0.bind(w0);
    auto hx1 = r1.bind(x1);
    auto hy1 = r1.bind(y1);
    auto hw1 = r1.bind(w1);
    auto pz0 = r0.mul<std::uint64_t>(hx0, hy0);
    auto pz1 = r1.mul<std::uint64_t>(hx1, hy1);
    dpf::expr::recorder::complete_mul<std::uint64_t>(r0, r1, pz0, pz1);
    auto z0 = r0.add<std::uint64_t>(pz0, hw0);
    auto z1 = r1.add<std::uint64_t>(pz1, hw1);
    auto [s0, s1] = dpf::net::make_memory_sink_pair(1, {sizeof(std::uint64_t)});
    dpf::net::edge_mesh m0;
    dpf::net::edge_mesh m1;
    m0.sinks.push_back(&s0);
    m1.sinks.push_back(&s1);
    r0.bind_mesh(std::move(m0));
    r1.bind_mesh(std::move(m1));
    EXPECT_EQ(dpf::expr::recorder::declassify<std::uint64_t>(r0, r1, z0, z1),
        47u);
}

TEST(ShareExpr, BandArity)
{
    dpf::expr::recorder r0(0), r1(1);
    // XOR shares of bits 1,1,1
    auto a0 = r0.bind<std::uint8_t>(1);
    auto b0 = r0.bind<std::uint8_t>(1);
    auto c0 = r0.bind<std::uint8_t>(1);
    auto a1 = r1.bind<std::uint8_t>(0);
    auto b1 = r1.bind<std::uint8_t>(0);
    auto c1 = r1.bind<std::uint8_t>(0);
    auto o0 = r0.band({a0, b0, c0});
    auto o1 = r1.band({a1, b1, c1});
    EXPECT_EQ(r0.last_and_arity(), 3u);
    dpf::expr::recorder::complete_band(r0, r1, o0, o1);
    EXPECT_EQ(static_cast<unsigned>(
                  r0.local_value<std::uint8_t>(o0)
                  ^ r1.local_value<std::uint8_t>(o1)),
        1u);
}

TEST(Gilboa, MulFromOtMatchesClear)
{
    auto packs = dpf::ot::sample_dealer_pair(16, 0, 16);
    const std::uint64_t x0 = 30, x1 = 12, y0 = 3, y1 = 4;
    auto prod = dpf::gilboa::mul_from_ot_pair(packs.first, packs.second, x0, x1,
        y0, y1, 16);
    EXPECT_EQ(prod.z0 + prod.z1, dpf::gilboa::mul_clear(x0, x1, y0, y1));
}

TEST(Gilboa, FillTapeOtRepeatedWire)
{
    dpf::beavers::session<std::uint64_t> s;
    auto a = s.input();
    auto out = s.product(a, a);
    s.pin(out);
    auto packs = dpf::ot::sample_dealer_pair(0, 0, 32);
    auto tapes = dpf::gilboa::fill_tape_ot_pair(s, packs.first, packs.second);
    ASSERT_FALSE(tapes.first.monomial.empty());
    ASSERT_GT(tapes.first.lambda.size(), 0u);
    const auto lam = tapes.first.lambda[a.id()] + tapes.second.lambda[a.id()];
    const auto mono = tapes.first.monomial[0] + tapes.second.monomial[0];
    EXPECT_EQ(mono, lam * lam);
}

TEST(Gilboa, FillTapeDealer)
{
    dpf::beavers::session<std::uint64_t> s;
    auto a = s.input();
    auto b = s.input();
    auto out = s.product(a, b);
    s.pin(out);
    auto tape = dpf::gilboa::fill_tape_dealer(s, 0);
    EXPECT_FALSE(tape.lambda.empty());
}

// ---------------------------------------------------------------------------
// Wave 5
// ---------------------------------------------------------------------------

TEST(Matmul, EightByEight)
{
    dpf::matmul::dims<std::uint64_t> d{8, 8, 8};
    std::vector<std::uint64_t> x(64), y(64);
    for (std::size_t i = 0; i < 64; ++i)
    {
        x[i] = i + 1;
        y[i] = (i % 8) + 1;
    }
    auto clear = dpf::matmul::clear_mul(x, y, d);
    auto [z0, z1] = dpf::matmul::mul_shared(x, y, d);
    for (std::size_t i = 0; i < clear.size(); ++i)
        EXPECT_EQ(z0[i] + z1[i], clear[i]);
}

TEST(Shuffle, CliqueSendNext)
{
    auto bundle = dpf::rss::sample_seed_bundle();
    std::vector<std::uint64_t> v(8);
    std::iota(v.begin(), v.end(), 0);
    const std::size_t n = v.size();
    dpf::shuffle::shuffle_party_view<std::uint64_t> held[3];
    for (auto & h : held)
    {
        h.own.assign(n, 0);
        h.next.assign(n, 0);
    }
    for (std::size_t i = 0; i < n; ++i)
    {
        const auto a = dpf::uniform_sample<std::uint64_t>();
        const auto b = dpf::uniform_sample<std::uint64_t>();
        const auto c = static_cast<std::uint64_t>(v[i] - a - b);
        held[0].own[i] = a;
        held[0].next[i] = b;
        held[1].own[i] = b;
        held[1].next[i] = c;
        held[2].own[i] = c;
        held[2].next[i] = a;
    }
    dpf::shuffle::shuffle_party_view<std::uint64_t> perm[3];
    for (unsigned p = 0; p < 3; ++p)
        perm[p] = dpf::shuffle::shuffle_party(held[p], bundle, 0);
    auto clique = dpf::net::make_memory_clique(3, 1, {n * sizeof(std::uint64_t)});
    auto send = [&](std::size_t from, std::size_t to,
                    const std::vector<std::uint64_t> & data) {
        std::vector<std::uint8_t> buf(n * sizeof(std::uint64_t));
        std::memcpy(buf.data(), data.data(), buf.size());
        clique.end(from, to).submit(0, 0, buf.data(), buf.size());
        clique.end(from, to).flush();
    };
    send(0, 1, perm[0].own);
    send(1, 2, perm[1].own);
    send(2, 0, perm[2].own);
    auto recv = [&](std::size_t me, std::size_t from) {
        auto & sink = clique.end(me, from);
        unsigned spins = 0;
        while (!sink.peer_ready(0, 0))
        {
            sink.poll();
            if (++spins > 100000u)
                throw std::runtime_error("shuffle peer");
        }
        std::vector<std::uint8_t> buf(n * sizeof(std::uint64_t));
        sink.read_peer(0, 0, buf.data(), buf.size());
        std::vector<std::uint64_t> out(n);
        std::memcpy(out.data(), buf.data(), buf.size());
        return out;
    };
    auto r0 = dpf::shuffle::shuffle_refresh(perm[0].own, recv(0, 2));
    auto r1 = dpf::shuffle::shuffle_refresh(perm[1].own, recv(1, 0));
    auto r2 = dpf::shuffle::shuffle_refresh(perm[2].own, recv(2, 1));
    std::vector<std::uint64_t> opened(n);
    for (std::size_t i = 0; i < n; ++i)
        opened[i] = r0.own[i] + r1.own[i] + r2.own[i];
    EXPECT_TRUE(dpf::shuffle::is_permutation_of(v, opened));
    EXPECT_EQ(opened, dpf::shuffle::shuffle_clear(v, bundle, 0));
}

TEST(Compose, CheckOpenHashRing)
{
    const std::uint64_t word = 42;
    auto run = [](std::uint64_t w0, std::uint64_t w1, std::uint64_t w2) {
        auto clique = dpf::net::make_memory_clique(3, 1, {8});
        std::uint8_t b0[8], b1[8], b2[8];
        std::memcpy(b0, &w0, 8);
        std::memcpy(b1, &w1, 8);
        std::memcpy(b2, &w2, 8);
        std::exception_ptr err;
        std::thread t0([&] {
            try
            {
                dpf::protocol::exchange_check_hash_ring(clique.end(0, 1),
                    clique.end(0, 2), b0, 8);
            }
            catch (...)
            {
                err = std::current_exception();
            }
        });
        std::thread t1([&] {
            try
            {
                dpf::protocol::exchange_check_hash_ring(clique.end(1, 2),
                    clique.end(1, 0), b1, 8);
            }
            catch (...)
            {
                err = std::current_exception();
            }
        });
        std::thread t2([&] {
            try
            {
                dpf::protocol::exchange_check_hash_ring(clique.end(2, 0),
                    clique.end(2, 1), b2, 8);
            }
            catch (...)
            {
                err = std::current_exception();
            }
        });
        t0.join();
        t1.join();
        t2.join();
        return err;
    };
    EXPECT_TRUE(run(word, word, word) == nullptr);
    EXPECT_TRUE(run(word, word ^ 1u, word) != nullptr);
}

TEST(Compose, TruncExactKernel)
{
    dpf::protocol::composer c(0);
    auto x = c.input(domain::a, 8);
    auto r = c.input(domain::a, 8);
    auto delta = c.input(domain::a, 8);
    auto wrap = c.input(domain::a, 8);
    auto out = c.trunc_exact_open(x, r, delta, wrap, 4);
    auto p = c.default_plan();
    std::vector<std::vector<std::uint8_t>> values(p.nodes().size());
    auto put = [&](dpf::protocol::node n, std::uint64_t v) {
        values[n.id].assign(8, 0);
        std::memcpy(values[n.id].data(), &v, 8);
    };
    put(x, 0x20);
    put(r, 0x5);
    put(delta, 0x10);
    put(wrap, 1);
    dpf::protocol::drive_options opt;
    dpf::protocol::detail::run_arith_local(p, out.id, 0, 0, 1, values, opt);
    std::uint64_t got = 0;
    std::memcpy(&got, values[out.id].data(), 8);
    const auto expect = dpf::trunc::trunc_exact_party<std::uint64_t>(
        0x20, 0x5, 0x10, 1, 4, 0, 0);
    EXPECT_EQ(got, expect);
}

TEST(Compose, B2aOpenMatchesClear)
{
    constexpr unsigned ell = 8;
    const std::uint64_t x = 0x3c;
    dpf::protocol::composer c(0);
    auto bits = c.input(domain::bin, 1);
    auto rbits = c.input(domain::bin, 1);
    auto rarith = c.input(domain::a, ell * 8);
    auto out = c.bin_b2a(bits, rbits, rarith, ell);
    auto p = c.default_plan();
    std::vector<std::vector<std::uint8_t>> v0(p.nodes().size());
    std::vector<std::vector<std::uint8_t>> v1(p.nodes().size());
    std::uint8_t b0 = static_cast<std::uint8_t>(x), b1 = 0, rb0 = 0, rb1 = 0;
    std::uint64_t a0[ell]{}, a1[ell]{};
    for (unsigned i = 0; i < ell; ++i)
    {
        auto d = dpf::ot::sample_dabit_pair<std::uint64_t>();
        if (d.p0.bit)
            rb0 = static_cast<std::uint8_t>(rb0 | (1u << i));
        if (d.p1.bit)
            rb1 = static_cast<std::uint8_t>(rb1 | (1u << i));
        a0[i] = d.p0.arith;
        a1[i] = d.p1.arith;
    }
    put_raw(v0, bits, &b0, 1);
    put_raw(v1, bits, &b1, 1);
    put_raw(v0, rbits, &rb0, 1);
    put_raw(v1, rbits, &rb1, 1);
    put_raw(v0, rarith, a0, sizeof a0);
    put_raw(v1, rarith, a1, sizeof a1);
    drive_both(p, v0, v1);
    std::uint64_t z0 = 0, z1 = 0;
    std::memcpy(&z0, v0[out.id].data(), 8);
    std::memcpy(&z1, v1[out.id].data(), 8);
    EXPECT_EQ(z0 + z1, x);
}

TEST(Compose, MulTruncOpen)
{
    const std::uint64_t x = 6, y = 7;
    const std::uint64_t x0 = 100, x1 = x - 100, y0 = 9, y1 = y - 9;
    const std::uint64_t a0 = 3, a1 = 5, b0 = 8, b1 = 1;
    const std::uint64_t c = (a0 + a1) * (b0 + b1);
    const std::uint64_t c0 = 20, c1 = c - 20;
    dpf::protocol::composer comp(0);
    auto nx = comp.input(domain::a, 8);
    auto ny = comp.input(domain::a, 8);
    auto na = comp.input(domain::a, 8);
    auto nb = comp.input(domain::a, 8);
    auto nc = comp.input(domain::a, 8);
    auto out = comp.mul_trunc_open(nx, ny, na, nb, nc, 0);
    auto p = comp.default_plan();
    std::vector<std::vector<std::uint8_t>> v0(p.nodes().size());
    std::vector<std::vector<std::uint8_t>> v1(p.nodes().size());
    put_raw(v0, nx, &x0, 8);
    put_raw(v1, nx, &x1, 8);
    put_raw(v0, ny, &y0, 8);
    put_raw(v1, ny, &y1, 8);
    put_raw(v0, na, &a0, 8);
    put_raw(v1, na, &a1, 8);
    put_raw(v0, nb, &b0, 8);
    put_raw(v1, nb, &b1, 8);
    put_raw(v0, nc, &c0, 8);
    put_raw(v1, nc, &c1, 8);
    drive_both(p, v0, v1);
    std::uint64_t z0 = 0, z1 = 0;
    std::memcpy(&z0, v0[out.id].data(), 8);
    std::memcpy(&z1, v1[out.id].data(), 8);
    EXPECT_EQ(z0 + z1, x * y);
}

TEST(Compose, InjectAndMuxOpen)
{
    auto tp = dpf::ot::sample_bit_ring_triple_pair<std::uint64_t>();
    dpf::protocol::composer c(0);
    auto bit = c.input(domain::a, 8);
    auto x = c.input(domain::a, 8);
    auto a = c.input(domain::a, 8);
    auto b = c.input(domain::a, 8);
    auto cc = c.input(domain::a, 8);
    auto prod = c.bin_inject(bit, x, a, b, cc);
    auto plan = c.default_plan();
    std::vector<std::vector<std::uint8_t>> v0(plan.nodes().size());
    std::vector<std::vector<std::uint8_t>> v1(plan.nodes().size());
    const std::uint64_t one = 1, zero = 0;
    const std::uint64_t x0 = 20, x1 = 22;
    const std::uint64_t a0 = tp.p0.a, a1 = tp.p1.a;
    put_raw(v0, bit, &one, 8);
    put_raw(v1, bit, &zero, 8);
    put_raw(v0, x, &x0, 8);
    put_raw(v1, x, &x1, 8);
    put_raw(v0, a, &a0, 8);
    put_raw(v1, a, &a1, 8);
    put_raw(v0, b, &tp.p0.b, 8);
    put_raw(v1, b, &tp.p1.b, 8);
    put_raw(v0, cc, &tp.p0.c, 8);
    put_raw(v1, cc, &tp.p1.c, 8);
    drive_both(plan, v0, v1);
    std::uint64_t z0 = 0, z1 = 0;
    std::memcpy(&z0, v0[prod.id].data(), 8);
    std::memcpy(&z1, v1[prod.id].data(), 8);
    EXPECT_EQ(z0 + z1, x0 + x1);
}

TEST(Compose, GmwAndOpen)
{
    auto tp = dpf::ot::sample_bit_triple_pair();
    dpf::protocol::composer c(0);
    auto p0 = c.input(domain::bin, 1);
    auto q0 = c.input(domain::bin, 1);
    auto a = c.input(domain::bin, 1);
    auto b = c.input(domain::bin, 1);
    auto cc = c.input(domain::bin, 1);
    auto out = c.gmw_and(p0, q0, a, b, cc);
    auto plan = c.default_plan();
    std::vector<std::vector<std::uint8_t>> v0(plan.nodes().size());
    std::vector<std::vector<std::uint8_t>> v1(plan.nodes().size());
    const std::uint8_t one = 1, zero = 0;
    put_raw(v0, p0, &one, 1);
    put_raw(v1, p0, &zero, 1);
    put_raw(v0, q0, &one, 1);
    put_raw(v1, q0, &zero, 1);
    put_raw(v0, a, &tp.p0.a, 1);
    put_raw(v1, a, &tp.p1.a, 1);
    put_raw(v0, b, &tp.p0.b, 1);
    put_raw(v1, b, &tp.p1.b, 1);
    put_raw(v0, cc, &tp.p0.c, 1);
    put_raw(v1, cc, &tp.p1.c, 1);
    drive_both(plan, v0, v1);
    EXPECT_EQ(static_cast<unsigned>(v0[out.id][0] ^ v1[out.id][0]), 1u);
}

TEST(Gilboa, FillTapeOtReadsStash)
{
    dpf::beavers::session<std::uint64_t> s;
    auto a = s.input();
    auto out = s.product(a, a);
    s.pin(out);
    auto packs = dpf::ot::sample_dealer_pair(0, 0, 32);
    auto tapes = dpf::gilboa::fill_tape_ot_pair(s, packs.first, packs.second);
    auto again = dpf::gilboa::fill_tape_ot(s, packs.first, 0);
    EXPECT_EQ(again.lambda, tapes.first.lambda);
    EXPECT_EQ(again.monomial, tapes.first.monomial);
    dpf::ot::pack bare;
    EXPECT_THROW(dpf::gilboa::fill_tape_ot(s, bare, 0), std::runtime_error);
}

TEST(BitInject, RssAndRefresh)
{
    auto bundle = dpf::rss::sample_seed_bundle();
    auto s0 = dpf::rss::party_seeds::from_bundle(bundle, 0);
    auto s1 = dpf::rss::party_seeds::from_bundle(bundle, 1);
    auto s2 = dpf::rss::party_seeds::from_bundle(bundle, 2);
    const auto e0 = dpf::bit_inject::rss_and_local(s0, 1, 0, 1, 0, 0);
    const auto e1 = dpf::bit_inject::rss_and_local(s1, 0, 0, 0, 0, 0);
    const auto e2 = dpf::bit_inject::rss_and_local(s2, 0, 1, 0, 1, 0);
    auto clique = dpf::net::make_memory_clique(3, 1, {1});
    auto send = [&](std::size_t from, std::size_t to, std::uint8_t byte) {
        clique.end(from, to).submit(0, 0, &byte, 1);
        clique.end(from, to).flush();
    };
    send(0, 1, e0);
    send(1, 2, e1);
    send(2, 0, e2);
    auto recv = [&](std::size_t me, std::size_t from) {
        auto & sink = clique.end(me, from);
        unsigned spins = 0;
        while (!sink.peer_ready(0, 0))
        {
            sink.poll();
            if (++spins > 100000u)
                throw std::runtime_error("rss and peer");
        }
        std::uint8_t b = 0;
        sink.read_peer(0, 0, &b, 1);
        return b;
    };
    EXPECT_EQ(recv(0, 2), e2);
    EXPECT_EQ(recv(1, 0), e0);
    EXPECT_EQ(recv(2, 1), e1);
    EXPECT_EQ(static_cast<unsigned>(e0 ^ e1 ^ e2), 1u);
}

TEST(Shuffle, SelectSumCountArgmax)
{
    std::vector<std::uint8_t> pred = {1, 0, 1, 1};
    std::vector<std::uint64_t> row = {10, 20, 30, 40};
    EXPECT_EQ(dpf::shuffle::sum_if_clear(pred, row), 80u);
    EXPECT_EQ(dpf::shuffle::count_if_clear<std::uint64_t>(pred), 3u);
    auto am = dpf::shuffle::argmax_clear(row, 64);
    EXPECT_EQ(am.value, 40u);
    EXPECT_EQ(am.one_hot[3], 1);
}

TEST(Circuit, MulOpenFromDealerFile)
{
    dpf::mpc::circuit c(2);
    auto x = c.input();
    auto y = c.input();
    auto prod = c.mul(x, y);
    auto opened = c.open(prod);
    auto views = dpf::prep::deal_views(c.prep());
    const auto path0 = std::string{"/tmp/libdpf-prep-p0.bin"};
    const auto path1 = std::string{"/tmp/libdpf-prep-p1.bin"};
    dpf::prep::write_file(path0, views.first);
    dpf::prep::write_file(path1, views.second);
    auto loaded0 = dpf::prep::read_file(path0);
    auto loaded1 = dpf::prep::read_file(path1);
    EXPECT_EQ(loaded0, views.first);
    EXPECT_EQ(loaded1, views.second);
    std::uint64_t got[2]{};
    dpf::run::threads_2(c.slot_bytes(), [&](unsigned party, dpf::net::RoundSink & sink) {
        dpf::mpc::party me(c, party);
        me.bind(x, party == 0 ? 6 : 0);
        me.bind(y, party == 0 ? 7 : 0);
        auto bytes = party == 0 ? loaded0 : loaded1;
        me.run(sink, dpf::prep::cursor(std::move(bytes)));
        got[party] = me.read(opened);
    });
    EXPECT_EQ(got[0], 42u);
    EXPECT_EQ(got[1], 42u);
}

TEST(Circuit, PrivateInputAndTcpPrep)
{
    dpf::mpc::circuit c(8);
    auto secret = c.priv_input(0);
    auto opened = c.open(secret);
    std::atomic<unsigned short> port{0};
    std::vector<std::uint8_t> view0, view1;
    std::thread listener([&] {
        dpf::run::tcp_pair(0, "127.0.0.1", port, [&](dpf::net::channel & ch) {
            view0 = dpf::prep::setup_2pc_sampled(ch, 0, c.prep());
        });
    });
    while (port.load() == 0)
        std::this_thread::yield();
    std::thread dialer([&] {
        dpf::run::tcp_pair(1, "127.0.0.1", port, [&](dpf::net::channel & ch) {
            view1 = dpf::prep::setup_2pc_sampled(ch, 1, c.prep());
        });
    });
    listener.join();
    dialer.join();
    EXPECT_FALSE(view0.empty());
    EXPECT_FALSE(view1.empty());
    std::uint64_t got[2]{};
    dpf::run::threads_2(c.slot_bytes(), [&](unsigned party, dpf::net::RoundSink & sink) {
        dpf::mpc::party me(c, party);
        if (party == 0)
            me.bind_priv(secret, 42);
        auto bytes = party == 0 ? view0 : view1;
        me.run(sink, dpf::prep::cursor(std::move(bytes)));
        got[party] = me.read(opened);
    });
    EXPECT_EQ(got[0], 42u);
    EXPECT_EQ(got[1], 42u);
    EXPECT_GE(dpf::revealing::known_count, 2u);
}

TEST(Run, ThreePartyDeclassify)
{
    const std::uint64_t shares[3] = {10, 20, 12};
    std::uint64_t opened[3]{};
    dpf::run::threads_on_clique(3, {8}, [&](unsigned me, dpf::net::memory_clique & clique) {
        std::uint8_t mine[8]{}, sum[8]{};
        std::memcpy(mine, &shares[me], 8);
        dpf::run::declassify_ring(clique, me, mine, 8, sum);
        std::memcpy(&opened[me], sum, 8);
    });
    EXPECT_EQ(opened[0], 42u);
    EXPECT_EQ(opened[1], 42u);
    EXPECT_EQ(opened[2], 42u);
}

TEST(CostPass, EmptyPlanNoSynthetic)
{
    dpf::protocol::composer c(0);
    auto p = c.default_plan();
    auto r_rounds = dpf::cost::annotate(p,
        dpf::beavers::schedule_objective::rounds);
    auto r_prep = dpf::cost::annotate(p, dpf::beavers::schedule_objective::prep);
    EXPECT_TRUE(r_rounds.choices.empty());
    EXPECT_TRUE(r_prep.choices.empty());
}

TEST(CostPass, RealCmpGetsChoice)
{
    dpf::protocol::composer c(0);
    auto x = c.input(domain::a, 8);
    auto y = c.input(domain::a, 8);
    (void)c.share_gt(x, y);
    auto p = c.default_plan();
    auto r = dpf::cost::annotate(p, dpf::beavers::schedule_objective::rounds);
    ASSERT_FALSE(r.choices.empty());
    EXPECT_EQ(r.choices[0].pick, dpf::cost::strategy::dcf_mask);
}

TEST(StreamArray, MemoryDealerRingTriple)
{
    auto w0 = dpf::net::make_memory_stream_pair(1);
    auto w1 = dpf::net::make_memory_stream_pair(1);
    dpf::factory::make_dealer(w0.first, w1.first, 1, [] {
        return dpf::factory::deal_ring_triple(2);
    });
    auto v0 = dpf::factory::detail::read_pod<dpf::factory::ring_triple_view>(
        w0.second, 0);
    auto v1 = dpf::factory::detail::read_pod<dpf::factory::ring_triple_view>(
        w1.second, 0);
    std::uint64_t a0 = 0, b0 = 0, c0 = 0, a1 = 0, b1 = 0, c1 = 0;
    std::memcpy(&a0, v0.a, 2);
    std::memcpy(&b0, v0.b, 2);
    std::memcpy(&c0, v0.c, 2);
    std::memcpy(&a1, v1.a, 2);
    std::memcpy(&b1, v1.b, 2);
    std::memcpy(&c1, v1.c, 2);
    const std::uint64_t mask = 0xffffu;
    EXPECT_EQ(((a0 + a1) * (b0 + b1)) & mask, (c0 + c1) & mask);
}

TEST(StreamArray, FileDealerRoundTrip)
{
    const std::string base = "/tmp/libdpf-stream-arr";
    {
        dpf::net::file_stream_array out0(base + "-p0", 1, true);
        dpf::net::file_stream_array out1(base + "-p1", 1, true);
        dpf::factory::make_dealer(out0, out1, 1, [] {
            return dpf::factory::deal_ring_triple(2);
        });
    }
    dpf::net::file_stream_array in0(base + "-p0", 1, false);
    dpf::net::file_stream_array in1(base + "-p1", 1, false);
    auto v0 = dpf::factory::detail::read_pod<dpf::factory::ring_triple_view>(
        in0, 0);
    auto v1 = dpf::factory::detail::read_pod<dpf::factory::ring_triple_view>(
        in1, 0);
    std::uint64_t a0 = 0, b0 = 0, c0 = 0, a1 = 0, b1 = 0, c1 = 0;
    std::memcpy(&a0, v0.a, 2);
    std::memcpy(&b0, v0.b, 2);
    std::memcpy(&c0, v0.c, 2);
    std::memcpy(&a1, v1.a, 2);
    std::memcpy(&b1, v1.b, 2);
    std::memcpy(&c1, v1.c, 2);
    EXPECT_EQ(((a0 + a1) * (b0 + b1)) & 0xffffu, (c0 + c1) & 0xffffu);
}

TEST(StreamArray, MuxLocalhostDealer)
{
    std::atomic<unsigned short> port{0};
    dpf::factory::ring_triple_view got0{}, got1{};
    std::thread t0([&] {
        dpf::run::tcp_pair_mux(0, "127.0.0.1", port, 1,
            [&](unsigned, dpf::net::mux_stream_array & mux) {
                // Party 0 acts as dealer write end toward party 1, and also
                // keeps a local memory view for itself.
                auto local = dpf::net::make_memory_stream_pair(1);
                dpf::factory::make_dealer(local.first, mux, 1, [] {
                    return dpf::factory::deal_ring_triple(2);
                });
                got0 = dpf::factory::detail::read_pod<
                    dpf::factory::ring_triple_view>(local.second, 0);
            });
    });
    while (port.load() == 0)
        std::this_thread::yield();
    std::thread t1([&] {
        dpf::run::tcp_pair_mux(1, "127.0.0.1", port, 1,
            [&](unsigned, dpf::net::mux_stream_array & mux) {
                got1 = dpf::factory::detail::read_pod<
                    dpf::factory::ring_triple_view>(mux, 0);
            });
    });
    t0.join();
    t1.join();
    std::uint64_t a0 = 0, b0 = 0, c0 = 0, a1 = 0, b1 = 0, c1 = 0;
    std::memcpy(&a0, got0.a, 2);
    std::memcpy(&b0, got0.b, 2);
    std::memcpy(&c0, got0.c, 2);
    std::memcpy(&a1, got1.a, 2);
    std::memcpy(&b1, got1.b, 2);
    std::memcpy(&c1, got1.c, 2);
    EXPECT_EQ(((a0 + a1) * (b0 + b1)) & 0xffffu, (c0 + c1) & 0xffffu);
}

TEST(Factory, GmwAndRoundMemory)
{
    auto peer = dpf::net::make_memory_stream_pair(1);
    auto d0 = dpf::net::make_memory_stream_pair(1);
    auto d1 = dpf::net::make_memory_stream_pair(1);
    dpf::factory::make_dealer(d0.first, d1.first, 1,
        dpf::factory::make_gmw_and_dealer_functor());
    std::uint8_t z0 = 0, z1 = 0;
    auto fac0 = dpf::factory::make_protocol_factory(
        dpf::factory::gmw_and_round0{}, dpf::factory::gmw_and_round1{0});
    auto fac1 = dpf::factory::make_protocol_factory(
        dpf::factory::gmw_and_round0{}, dpf::factory::gmw_and_round1{1});
    std::thread t0([&] {
        auto p = fac0.create(1, peer.first, d0.second);
        z0 = p(0, std::pair<std::uint8_t, std::uint8_t>{1, 1});
    });
    std::thread t1([&] {
        auto p = fac1.create(1, peer.second, d1.second);
        z1 = p(0, std::pair<std::uint8_t, std::uint8_t>{0, 0});
    });
    t0.join();
    t1.join();
    EXPECT_EQ(static_cast<unsigned>(z0 ^ z1), 1u);
}

TEST(Circuit, MulOpenFromStreamArray)
{
    dpf::mpc::circuit c(2);
    auto x = c.input();
    auto y = c.input();
    auto prod = c.mul(x, y);
    auto opened = c.open(prod);
    auto views = dpf::prep::deal_views(c.prep());
    auto w0 = dpf::net::make_memory_stream_pair(1);
    auto w1 = dpf::net::make_memory_stream_pair(1);
    w0.first.write(0, views.first.data(), views.first.size());
    w0.first.flush(0);
    w1.first.write(0, views.second.data(), views.second.size());
    w1.first.flush(0);
    const std::size_t nbytes = views.first.size();
    std::uint64_t got[2]{};
    dpf::run::threads_2(c.slot_bytes(), [&](unsigned party, dpf::net::RoundSink & sink) {
        dpf::mpc::party me(c, party);
        me.bind(x, party == 0 ? 6 : 0);
        me.bind(y, party == 0 ? 7 : 0);
        auto & dealer = party == 0 ? w0.second : w1.second;
        me.run(sink, dealer, nbytes);
        got[party] = me.read(opened);
    });
    EXPECT_EQ(got[0], 42u);
    EXPECT_EQ(got[1], 42u);
}

TEST(StreamArray, MultiRoundSinkExchange)
{
    constexpr std::size_t slot = 8;
    dpf::run::threads_2_streams(2, [&](unsigned party, dpf::net::stream_array & sa) {
        dpf::net::stream_array_sink sink(sa, std::vector<std::size_t>{slot, slot});
        const std::uint64_t mine = party == 0 ? 11u : 22u;
        std::uint8_t buf[slot]{};
        std::memcpy(buf, &mine, sizeof(mine));
        sink.submit(0, 0, buf, slot);
        sink.flush_round(0);
        EXPECT_TRUE(sink.peer_ready(0, 0));
        std::uint8_t peer0[slot]{};
        sink.read_peer(0, 0, peer0, slot);
        const std::uint64_t other0 = party == 0 ? 22u : 11u;
        std::uint64_t got0 = 0;
        std::memcpy(&got0, peer0, slot);
        EXPECT_EQ(got0, other0);
        const std::uint64_t mine1 = mine + 1u;
        std::memcpy(buf, &mine1, sizeof(mine1));
        sink.submit(1, 0, buf, slot);
        sink.flush_round(1);
        std::uint8_t peer1[slot]{};
        sink.read_peer(1, 0, peer1, slot);
        std::uint64_t got1 = 0;
        std::memcpy(&got1, peer1, slot);
        EXPECT_EQ(got1, other0 + 1u);
    });
}

TEST(Run, ThreePartyDeclassifyStreamClique)
{
    const std::uint64_t shares[3] = {10, 20, 12};
    std::uint64_t opened[3]{};
    auto clique = dpf::net::make_memory_stream_clique(3, 1);
    std::exception_ptr err;
    std::mutex mu;
    std::vector<std::thread> ts;
    for (unsigned me = 0; me < 3; ++me)
    {
        ts.emplace_back([&, me] {
            try
            {
                std::uint8_t mine[8]{}, sum[8]{};
                std::memcpy(mine, &shares[me], 8);
                dpf::run::declassify_streams(clique, me, mine, 8, sum);
                std::memcpy(&opened[me], sum, 8);
            }
            catch (...)
            {
                std::lock_guard<std::mutex> lock(mu);
                if (!err)
                    err = std::current_exception();
            }
        });
    }
    for (auto & t : ts)
        t.join();
    if (err)
        std::rethrow_exception(err);
    EXPECT_EQ(opened[0], 42u);
    EXPECT_EQ(opened[1], 42u);
    EXPECT_EQ(opened[2], 42u);
}

TEST(Factory, BeaverMulStreamMemory)
{
    auto peer = dpf::net::make_memory_stream_pair(2);
    auto d0 = dpf::net::make_memory_stream_pair(1);
    auto d1 = dpf::net::make_memory_stream_pair(1);
    dpf::factory::make_dealer(d0.first, d1.first, 1,
        dpf::factory::make_beaver_mul_dealer_functor());
    const std::uint64_t x0 = 6, x1 = 0, y0 = 7, y1 = 0;
    std::uint64_t z0 = 0, z1 = 0;
    auto fac = dpf::factory::make_beaver_mul_factory(
        dpf::factory::beaver_mul_round0{8},
        dpf::factory::beaver_mul_round1{8, 0});
    auto fac1 = dpf::factory::make_beaver_mul_factory(
        dpf::factory::beaver_mul_round0{8},
        dpf::factory::beaver_mul_round1{8, 1});
    std::thread t0([&] {
        auto p = fac.create(1, peer.first, d0.second);
        z0 = p(0, dpf::factory::beaver_mul_input{x0, y0});
    });
    std::thread t1([&] {
        auto p = fac1.create(1, peer.second, d1.second);
        z1 = p(0, dpf::factory::beaver_mul_input{x1, y1});
    });
    t0.join();
    t1.join();
    EXPECT_EQ(z0 + z1, 42u);
}

TEST(Circuit, RecordsGmwAndA2bGtTruncMux)
{
    dpf::mpc::circuit c(8);
    auto x = c.input();
    auto y = c.input();
    auto b0 = c.input();
    auto b1 = c.input();
    (void)c.gmw_and(b0, b1);
    (void)c.a2b(x, 8);
    (void)c.gt(x, y, 8);
    (void)c.trunc_exact(x, 16, 4);
    (void)c.mux(b0, x, y);
    EXPECT_GE(c.prep().bit_triples, 1u);
    EXPECT_GE(c.prep().ring_triples, 1u);
    EXPECT_GE(c.prep().dabits, 1u);
    EXPECT_FALSE(c.slot_bytes().empty());
}

TEST(Factory, MuxOnlineStreamMemory)
{
    auto peer = dpf::net::make_memory_stream_pair(2);
    auto d0 = dpf::net::make_memory_stream_pair(1);
    auto d1 = dpf::net::make_memory_stream_pair(1);
    dpf::factory::make_dealer(d0.first, d1.first, 1,
        [] { return dpf::factory::deal_ring_triple(8); });
    std::uint64_t z0 = 0, z1 = 0;
    std::thread t0([&] {
        z0 = dpf::factory::mux_online(0, 1, 40, 2, peer.first, d0.second);
    });
    std::thread t1([&] {
        z1 = dpf::factory::mux_online(1, 0, 0, 0, peer.second, d1.second);
    });
    t0.join();
    t1.join();
    EXPECT_EQ(z0 + z1, 40u);
}

namespace
{

std::uint64_t xor_bit_shares(std::uint64_t s0, std::uint64_t s1, unsigned width)
{
    const std::uint64_t mask =
        width >= 64 ? ~std::uint64_t{0} : ((std::uint64_t{1} << width) - 1u);
    return (s0 ^ s1) & mask;
}

}  // namespace

TEST(Factory, A2bTapeOnline)
{
    constexpr unsigned width = 8;
    auto peer = dpf::net::make_memory_stream_pair(2);
    auto d0 = dpf::net::make_memory_stream_pair(2);
    auto d1 = dpf::net::make_memory_stream_pair(2);
    dpf::factory::deal_a2b_tape(width, 8, d0.first, d1.first);
    const std::uint64_t x0 = 30, x1 = 12;
    std::uint64_t b0 = 0, b1 = 0;
    std::thread t0([&] {
        b0 = dpf::factory::a2b_online(0, x0, width, peer.first, d0.second);
    });
    std::thread t1([&] {
        b1 = dpf::factory::a2b_online(1, x1, width, peer.second, d1.second);
    });
    t0.join();
    t1.join();
    EXPECT_EQ(xor_bit_shares(b0, b1, width), x0 + x1);
}

TEST(Factory, TruncTapeOnline)
{
    constexpr unsigned n = 20, s = 4;
    auto peer = dpf::net::make_memory_stream_pair(2);
    auto d0 = dpf::net::make_memory_stream_pair(2);
    auto d1 = dpf::net::make_memory_stream_pair(2);
    dpf::factory::deal_trunc_tape(n, s, 8, d0.first, d1.first);
    const std::uint64_t x = 0xabcde;
    const std::uint64_t x0 = 0x11111, x1 = x - x0;
    std::uint64_t t0 = 0, t1 = 0;
    std::thread t0th([&] {
        t0 = dpf::factory::trunc_exact_online(0, x0, n, s, peer.first,
            d0.second);
    });
    std::thread t1th([&] {
        t1 = dpf::factory::trunc_exact_online(1, x1, n, s, peer.second,
            d1.second);
    });
    t0th.join();
    t1th.join();
    EXPECT_EQ(t0 + t1, dpf::trunc::trunc_exact_clear(x0, x1, n, s));
}

TEST(Factory, GtTapeOnline)
{
    constexpr unsigned width = 8;
    auto peer = dpf::net::make_memory_stream_pair(5);
    auto d0 = dpf::net::make_memory_stream_pair(5);
    auto d1 = dpf::net::make_memory_stream_pair(5);
    dpf::factory::deal_gt_tape(width, 8, d0.first, d1.first);
    const std::uint64_t x0 = 50, x1 = 7, y0 = 10, y1 = 20;
    std::uint8_t g0 = 0, g1 = 0;
    std::thread t0([&] {
        g0 = dpf::factory::gt_online(0, x0, y0, width, peer.first, d0.second);
    });
    std::thread t1([&] {
        g1 = dpf::factory::gt_online(1, x1, y1, width, peer.second, d1.second);
    });
    t0.join();
    t1.join();
    const std::uint64_t x = x0 + x1;
    const std::uint64_t y = y0 + y1;
    const unsigned expect = x > y ? 1u : 0u;
    EXPECT_EQ(static_cast<unsigned>(g0 ^ g1), expect);
}

TEST(StreamArray, DealerCursorFromStreamPrg)
{
    constexpr std::size_t N = 8;
    auto d0 = dpf::net::make_memory_stream_pair(1);
    auto d1 = dpf::net::make_memory_stream_pair(1);
    using Prg = dpf::randomness::aes_buffered_prg<std::uint64_t, std::uint64_t>;
    const auto seed = dpf::randomness::sample_master_seed<dpf::prg::aes128>();
    Prg prg0(seed);
    Prg prg1(seed);
    dpf::factory::make_dealer_prg(d0.first, d1.first, N, prg0, prg1,
        [](Prg & p0, Prg & /*p1*/, std::size_t j) {
            auto t = dpf::factory::deal_bit_triple();
            dpf::factory::gmw_and_blind v0{t.first.a, t.first.b, t.first.c};
            dpf::factory::gmw_and_blind v1{t.second.a, t.second.b, t.second.c};
            (void)p0.at<0>(static_cast<std::uint64_t>(j));
            return std::pair{v0, v1};
        });
    dpf::net::stream_dealer_cursor cur0(d0.second, N,
        {sizeof(dpf::factory::gmw_and_blind)});
    dpf::net::stream_dealer_cursor cur1(d1.second, N,
        {sizeof(dpf::factory::gmw_and_blind)});
    for (std::size_t i = 0; i < N; ++i)
    {
        const auto v0 = cur0.at<dpf::factory::gmw_and_blind>(0, i);
        const auto v1 = cur1.at<dpf::factory::gmw_and_blind>(0, i);
        const unsigned c = static_cast<unsigned>((v0.c ^ v1.c) & 1u);
        const unsigned ab = static_cast<unsigned>(
            ((v0.a ^ v1.a) & 1u) & ((v0.b ^ v1.b) & 1u));
        EXPECT_EQ(c, ab);
    }
}

TEST(Factory, ScheduleOnStreams)
{
    auto peer = dpf::net::make_memory_stream_pair(1);
    std::vector<std::uint64_t> local0(4), local1(4);
    for (std::size_t i = 0; i < 4; ++i)
    {
        local0[i] = 10 + i;
        local1[i] = 100 + i;
    }
    auto make_rounds = [](std::vector<std::uint64_t> & local) {
        std::vector<dpf::protocol::schedule_round> rounds(1);
        rounds[0].slot_bytes = sizeof(std::uint64_t);
        rounds[0].produce = [&](std::size_t index, const std::uint8_t *,
                                std::size_t, std::uint8_t * out) {
            std::memcpy(out, &local[index], sizeof(std::uint64_t));
        };
        return rounds;
    };
    std::uint64_t sum0 = 0, sum1 = 0;
    std::thread t0([&] {
        auto s = dpf::protocol::make_owning_schedule_on_streams(4, peer.first,
            make_rounds(local0));
        for (std::size_t i = 0; i < 4; ++i)
            s.submit(i);
        s.drive();
        for (std::size_t i = 0; i < 4; ++i)
        {
            std::uint64_t v = 0;
            s.sink->read_peer(0, i, reinterpret_cast<std::uint8_t *>(&v),
                sizeof(v));
            sum0 += v;
        }
    });
    std::thread t1([&] {
        auto s = dpf::protocol::make_owning_schedule_on_streams(4, peer.second,
            make_rounds(local1));
        for (std::size_t i = 0; i < 4; ++i)
            s.submit(i);
        s.drive();
        for (std::size_t i = 0; i < 4; ++i)
        {
            std::uint64_t v = 0;
            s.sink->read_peer(0, i, reinterpret_cast<std::uint8_t *>(&v),
                sizeof(v));
            sum1 += v;
        }
    });
    t0.join();
    t1.join();
    EXPECT_EQ(sum0, 100u + 101u + 102u + 103u);
    EXPECT_EQ(sum1, 10u + 11u + 12u + 13u);
}

TEST(Roles, DealerWriteFilesRoundTrip)
{
    const std::string base = "/tmp/libdpf-roles-prep";
    dpf::prep::demand d;
    d.limb = 2;
    d.ring_triples = 1;
    dpf::roles::dealer_write_files(base, d);
    auto curs = dpf::roles::online_open_file_array(base);
    std::uint8_t a[8]{}, b[8]{}, c[8]{};
    curs.first.take_ring(a, b, c);
    curs.second.take_ring(a, b, c);
    EXPECT_EQ(curs.first.limb(), 2);
    EXPECT_EQ(curs.second.limb(), 2);
}

TEST(Compose, DrivePlanOnStreamsOpen)
{
    dpf::protocol::composer c0(0);
    dpf::protocol::composer c1(1);
    auto x0 = c0.input(domain::a, 8);
    auto x1 = c1.input(domain::a, 8);
    auto e0 = c0.exchange(x0);
    auto e1 = c1.exchange(x1);
    auto p0 = c0.schedule();
    auto p1 = c1.schedule();
    std::vector<std::vector<std::uint8_t>> v0(p0.nodes().size()),
        v1(p1.nodes().size());
    const std::uint64_t a = 3, b = 5;
    put_raw(v0, x0, &a, 8);
    put_raw(v1, x1, &b, 8);
    dpf::protocol::drive_both_on_streams(p0, p1, v0, v1);
    std::uint64_t open0 = 0, open1 = 0;
    std::memcpy(&open0, v0[e0.id].data(), 8);
    std::memcpy(&open1, v1[e1.id].data(), 8);
    EXPECT_EQ(open0, 8u);
    EXPECT_EQ(open1, 8u);
}

TEST(Compose, DrivePlanOnStreamsTwoWaves)
{
    constexpr std::uint32_t k_step = 91001;
    auto make = [](std::size_t party, std::uint64_t in) {
        dpf::protocol::composer c(party);
        auto x = c.input(domain::a, 8);
        auto e0 = c.exchange(x);
        auto y = c.compute(k_step, {e0}, domain::a, 8);
        auto e1 = c.exchange(y);
        return std::make_tuple(c.schedule(), x, e1, in);
    };
    auto [p0, x0, out0, in0] = make(0, 11);
    auto [p1, x1, out1, in1] = make(1, 19);
    ASSERT_EQ(p0.rounds(), 2u);
    dpf::protocol::kernel_fn step = [](std::uint32_t, const std::vector<dpf::protocol::node> &,
                                        const std::vector<dpf::protocol::block_span> & inputs,
                                        dpf::protocol::block_span output, std::size_t) {
        std::uint64_t v = 0;
        std::memcpy(&v, inputs[0].at(0), 8);
        v += 1;
        std::memcpy(output.at(0), &v, 8);
    };
    std::map<std::uint32_t, dpf::protocol::kernel_fn> k{{k_step, step}};
    std::vector<std::vector<std::uint8_t>> v0(p0.nodes().size()),
        v1(p1.nodes().size());
    put_raw(v0, x0, &in0, 8);
    put_raw(v1, x1, &in1, 8);
    dpf::protocol::drive_both_on_streams(p0, p1, v0, v1, k);
    std::uint64_t o0 = 0, o1 = 0;
    std::memcpy(&o0, v0[out0.id].data(), 8);
    std::memcpy(&o1, v1[out1.id].data(), 8);
    EXPECT_EQ(o0, 62u);
    EXPECT_EQ(o1, 62u);
}

TEST(Compose, DrivePlanOnStreamsGmwAnd)
{
    auto tp = dpf::ot::sample_bit_triple_pair();
    dpf::protocol::composer c(0);
    auto p0 = c.input(domain::bin, 1);
    auto q0 = c.input(domain::bin, 1);
    auto a = c.input(domain::bin, 1);
    auto b = c.input(domain::bin, 1);
    auto cc = c.input(domain::bin, 1);
    auto out = c.gmw_and(p0, q0, a, b, cc);
    auto plan = c.default_plan();
    std::vector<std::vector<std::uint8_t>> v0(plan.nodes().size());
    std::vector<std::vector<std::uint8_t>> v1(plan.nodes().size());
    const std::uint8_t one = 1, zero = 0;
    put_raw(v0, p0, &one, 1);
    put_raw(v1, p0, &zero, 1);
    put_raw(v0, q0, &one, 1);
    put_raw(v1, q0, &zero, 1);
    put_raw(v0, a, &tp.p0.a, 1);
    put_raw(v1, a, &tp.p1.a, 1);
    put_raw(v0, b, &tp.p0.b, 1);
    put_raw(v1, b, &tp.p1.b, 1);
    put_raw(v0, cc, &tp.p0.c, 1);
    put_raw(v1, cc, &tp.p1.c, 1);
    dpf::protocol::drive_both_on_streams(plan, v0, v1);
    EXPECT_EQ(static_cast<unsigned>(v0[out.id][0] ^ v1[out.id][0]), 1u);
}

TEST(Compose, DrivePlanOnStreamsDeepEightWaves)
{
    // Gap-1 stress: eight successive opens on stream arrays (FSS-depth shape).
    auto make = [](std::size_t party, std::uint64_t in) {
        dpf::protocol::composer c(party);
        auto x = c.input(domain::a, 8);
        dpf::protocol::node last = x;
        for (int i = 0; i < 8; ++i)
            last = c.exchange(last);
        return std::make_tuple(c.schedule(), x, last, in);
    };
    auto [p0, x0, out0, in0] = make(0, 1);
    auto [p1, x1, out1, in1] = make(1, 2);
    ASSERT_EQ(p0.rounds(), 8u);
    std::vector<std::vector<std::uint8_t>> v0(p0.nodes().size()),
        v1(p1.nodes().size());
    put_raw(v0, x0, &in0, 8);
    put_raw(v1, x1, &in1, 8);
    dpf::protocol::drive_options opt;
    opt.pipeline_credit = 2;
    dpf::protocol::drive_both_on_streams(p0, p1, v0, v1, {}, 1, opt);
    std::uint64_t o0 = 0, o1 = 0;
    std::memcpy(&o0, v0[out0.id].data(), 8);
    std::memcpy(&o1, v1[out1.id].data(), 8);
    // Each exchange opens the sum; after first open both hold 3, then 6, …
    EXPECT_EQ(o0, o1);
    EXPECT_EQ(o0, 3u << 7);
}

TEST(Compose, DrivePlanOnStreamsMuxTcp)
{
    dpf::protocol::composer c0(0);
    dpf::protocol::composer c1(1);
    auto x0 = c0.input(domain::a, 8);
    auto x1 = c1.input(domain::a, 8);
    auto e0 = c0.exchange(x0);
    auto e1 = c1.exchange(x1);
    auto p0 = c0.schedule();
    auto p1 = c1.schedule();
    const auto nstreams = p0.slot_bytes_all().size();
    ASSERT_EQ(nstreams, 1u);
    std::atomic<unsigned short> port{0};
    std::uint64_t open0 = 0, open1 = 0;
    std::exception_ptr err;
    std::mutex mu;
    std::thread t0([&] {
        try
        {
            dpf::run::tcp_pair_mux(0, "127.0.0.1", port, nstreams,
                [&](unsigned, dpf::net::mux_stream_array & mux) {
                    std::vector<std::vector<std::uint8_t>> v(p0.nodes().size());
                    const std::uint64_t a = 7;
                    put_raw(v, x0, &a, 8);
                    dpf::protocol::drive_plan_on_streams(p0, mux, v, {}, 0);
                    std::memcpy(&open0, v[e0.id].data(), 8);
                });
        }
        catch (...)
        {
            std::lock_guard<std::mutex> lock(mu);
            if (!err)
                err = std::current_exception();
        }
    });
    while (port.load() == 0)
        std::this_thread::yield();
    std::thread t1([&] {
        try
        {
            dpf::run::tcp_pair_mux(1, "127.0.0.1", port, nstreams,
                [&](unsigned, dpf::net::mux_stream_array & mux) {
                    std::vector<std::vector<std::uint8_t>> v(p1.nodes().size());
                    const std::uint64_t b = 35;
                    put_raw(v, x1, &b, 8);
                    dpf::protocol::drive_plan_on_streams(p1, mux, v, {}, 1);
                    std::memcpy(&open1, v[e1.id].data(), 8);
                });
        }
        catch (...)
        {
            std::lock_guard<std::mutex> lock(mu);
            if (!err)
                err = std::current_exception();
        }
    });
    t0.join();
    t1.join();
    if (err)
        std::rethrow_exception(err);
    EXPECT_EQ(open0, 42u);
    EXPECT_EQ(open1, 42u);
}

TEST(Compose, DrivePlanOnStreamMeshRssCopy)
{
    dpf::protocol::composer c0(0);
    dpf::protocol::composer c1(1);
    auto y0 = c0.input(domain::y, 8);
    auto y1 = c1.input(domain::y, 8);
    auto r0 = c0.rss_from_y(y0);
    auto r1 = c1.rss_from_y(y1);
    (void)r0;
    (void)r1;
    auto p = c0.schedule();
    ASSERT_EQ(p.exchange_waves(), 1u);
    const auto slots = dpf::protocol::slot_bytes_by_channel(p);
    ASSERT_EQ(slots.rss_next.size(), 1u);
    ASSERT_TRUE(slots.peer.empty());
    auto rss = dpf::net::make_memory_stream_pair(1);
    std::vector<std::vector<std::uint8_t>> val0(p.nodes().size()),
        val1(p.nodes().size());
    const std::uint64_t mine0 = 10, mine1 = 20;
    put_raw(val0, y0, &mine0, 8);
    put_raw(val1, y1, &mine1, 8);
    auto sinks0 = dpf::net::make_stream_edge_sinks(nullptr, {}, &rss.first,
        slots.rss_next, nullptr, {}, 1);
    auto sinks1 = dpf::net::make_stream_edge_sinks(nullptr, {}, &rss.second,
        slots.rss_next, nullptr, {}, 1);
    std::exception_ptr err;
    std::mutex mu;
    std::thread t0([&] {
        try
        {
            dpf::protocol::drive_plan_on_stream_mesh(p, sinks0, val0, {}, 0);
        }
        catch (...)
        {
            std::lock_guard<std::mutex> lock(mu);
            if (!err)
                err = std::current_exception();
        }
    });
    std::thread t1([&] {
        try
        {
            dpf::protocol::drive_plan_on_stream_mesh(p, sinks1, val1, {}, 1);
        }
        catch (...)
        {
            std::lock_guard<std::mutex> lock(mu);
            if (!err)
                err = std::current_exception();
        }
    });
    t0.join();
    t1.join();
    if (err)
        std::rethrow_exception(err);
    std::uint64_t own0 = 0, next0 = 0, own1 = 0, next1 = 0;
    std::memcpy(&own0, val0[r0.id].data(), 8);
    std::memcpy(&next0, val0[r0.id].data() + 8, 8);
    std::memcpy(&own1, val1[r1.id].data(), 8);
    std::memcpy(&next1, val1[r1.id].data() + 8, 8);
    EXPECT_EQ(own0, mine0);
    EXPECT_EQ(next0, mine1);
    EXPECT_EQ(own1, mine1);
    EXPECT_EQ(next1, mine0);
}

TEST(Gilboa, MulTapeViaStreamArrays)
{
    auto d0 = dpf::net::make_memory_stream_pair(1);
    auto d1 = dpf::net::make_memory_stream_pair(1);
    dpf::factory::deal_gilboa_mul_tape(d0.first, d1.first, 16);
    auto p0 = dpf::factory::read_ot_pack_wire(d0.second, 0);
    auto p1 = dpf::factory::read_ot_pack_wire(d1.second, 0);
    const std::uint64_t x0 = 30, x1 = 12, y0 = 3, y1 = 4;
    auto prod = dpf::gilboa::mul_from_ot_pair(p0, p1, x0, x1, y0, y1, 16);
    EXPECT_EQ(prod.z0 + prod.z1,
        dpf::gilboa::mul_clear(x0, x1, y0, y1));
}

TEST(Gilboa, MulOnlineViaStreamArrays)
{
    auto peer = dpf::net::make_memory_stream_pair(1);
    auto d0 = dpf::net::make_memory_stream_pair(1);
    auto d1 = dpf::net::make_memory_stream_pair(1);
    dpf::factory::deal_gilboa_mul_tape(d0.first, d1.first, 16);
    const std::uint64_t x0 = 30, x1 = 12, y0 = 3, y1 = 4;
    std::uint64_t z0 = 0, z1 = 0;
    std::thread t0([&] {
        z0 = dpf::factory::gilboa_mul_online(0, x0, y0, peer.first, d0.second, 16);
    });
    std::thread t1([&] {
        z1 = dpf::factory::gilboa_mul_online(1, x1, y1, peer.second, d1.second, 16);
    });
    t0.join();
    t1.join();
    EXPECT_EQ(z0 + z1, dpf::gilboa::mul_clear(x0, x1, y0, y1));
}

TEST(Factory, RssRefreshRingStreamClique)
{
    auto clique = dpf::net::make_memory_stream_clique(3, 1);
    auto bundle = dpf::rss::sample_seed_bundle();
    std::uint64_t sum_own = 0, sum_next = 0;
    std::exception_ptr err;
    std::mutex mu;
    std::vector<std::thread> ts;
    for (unsigned me = 0; me < 3; ++me)
    {
        ts.emplace_back([&, me] {
            try
            {
                const auto seeds = dpf::rss::party_seeds::from_bundle(bundle, me);
                const auto z = dpf::rss::zero_share<std::uint64_t>(seeds, 7);
                std::uint8_t mine[8]{}, own[8]{}, next[8]{};
                std::memcpy(mine, &z, 8);
                dpf::factory::rss_refresh_ring_online(clique, me, mine, 8, own,
                    next);
                std::uint64_t o = 0, n = 0;
                std::memcpy(&o, own, 8);
                std::memcpy(&n, next, 8);
                std::lock_guard<std::mutex> lock(mu);
                sum_own += o;
                sum_next += n;
            }
            catch (...)
            {
                std::lock_guard<std::mutex> lock(mu);
                if (!err)
                    err = std::current_exception();
            }
        });
    }
    for (auto & t : ts)
        t.join();
    if (err)
        std::rethrow_exception(err);
    EXPECT_EQ(sum_own, 0u);
    EXPECT_EQ(sum_next, 0u);
}

TEST(StreamArray, SctpStreamArrayStubThrows)
{
    dpf::net::sctp_stream_array arr(4);
    EXPECT_EQ(arr.size(), 4u);
    EXPECT_THROW(dpf::net::sctp_stream_array(0), std::invalid_argument);
    EXPECT_THROW(
        dpf::net::sctp_stream_array(dpf::net::sctp_association{-1}),
        std::logic_error);
    std::uint8_t byte = 1;
    EXPECT_THROW(arr.write(0, &byte, 1), std::logic_error);
    EXPECT_THROW(arr.read(0, &byte, 1), std::logic_error);
    EXPECT_THROW(arr.flush(0), std::logic_error);
}

namespace
{

// Two rounds shared by the async byte-protocol tests. Round 0 replaces the
// state with the peer's byte; round 1 adds one to the state, sends it, and
// folds in the peer's reply. Two symmetric parties starting from the empty
// state both end at 7. No blind is used (`blind_bytes == 0`).
inline std::vector<dpf::factory::async_byte_round> async_seven_rounds()
{
    dpf::factory::async_byte_round r0{};
    r0.blind_bytes = 0;
    r0.msg_bytes = 1;
    r0.produce = [](std::vector<std::uint8_t> & state, const std::uint8_t *,
                     std::size_t) {
        std::uint8_t v = state.empty() ? 3 : state[0];
        return std::vector<std::uint8_t>{v};
    };
    r0.finish = [](std::vector<std::uint8_t> & state, const std::uint8_t * peer,
                    std::size_t, const std::uint8_t *, std::size_t) {
        state.assign(1, peer[0]);
    };
    dpf::factory::async_byte_round r1 = r0;
    r1.produce = [](std::vector<std::uint8_t> & state, const std::uint8_t *,
                     std::size_t) {
        return std::vector<std::uint8_t>{static_cast<std::uint8_t>(state[0] + 1)};
    };
    r1.finish = [](std::vector<std::uint8_t> & state, const std::uint8_t * peer,
                    std::size_t, const std::uint8_t *, std::size_t) {
        state[0] = static_cast<std::uint8_t>((state[0] + peer[0]) & 0xffu);
    };
    return {r0, r1};
}

}  // namespace

TEST(Factory, AsyncByteProtocolRoundCallbacks)
{
    // Overlapped, event-driven runner over an async memory pair; io_context
    // drives both parties to completion — no threads, no busy-waiting.
    asio::io_context io;
    auto peer = dpf::net::make_async_memory_stream_pair(io, 2);
    auto rounds = async_seven_rounds();
    std::vector<std::size_t> done0;
    std::vector<std::size_t> done1;
    auto p0 = dpf::async::make_overlapped_byte_protocol(peer.first, nullptr,
        rounds, [&](std::size_t r) { done0.push_back(r); });
    auto p1 = dpf::async::make_overlapped_byte_protocol(peer.second, nullptr,
        rounds, [&](std::size_t r) { done1.push_back(r); });
    std::error_code ec0, ec1;
    std::vector<std::uint8_t> out0, out1;
    dpf::async::run_overlapped(io, [&] {
        p0->start(0, {}, [&](const std::error_code & e,
                              std::vector<std::uint8_t> s) {
            ec0 = e;
            out0 = std::move(s);
        });
        p1->start(0, {}, [&](const std::error_code & e,
                              std::vector<std::uint8_t> s) {
            ec1 = e;
            out1 = std::move(s);
        });
    });
    ASSERT_FALSE(ec0);
    ASSERT_FALSE(ec1);
    ASSERT_EQ(out0.size(), 1u);
    ASSERT_EQ(out1.size(), 1u);
    EXPECT_EQ(out0[0], 7u);
    EXPECT_EQ(out1[0], 7u);
    ASSERT_EQ(done0.size(), 2u);
    ASSERT_EQ(done1.size(), 2u);
    EXPECT_EQ(done0[0], 0u);
    EXPECT_EQ(done0[1], 1u);
    EXPECT_EQ(done1[0], 0u);
    EXPECT_EQ(done1[1], 1u);
}

// Two symmetric parties, two overlapped rounds, one io_context. Each round
// adds the peer's byte to the running state: 10/20 -> 30/30 -> 60/60. Purely
// event-driven; the io_context returns only when both parties finish.
TEST(AsyncIo, MemoryOverlapTwoRounds)
{
    asio::io_context io;
    auto peer = dpf::net::make_async_memory_stream_pair(io, 2);
    auto add_rounds = [] {
        dpf::factory::async_byte_round r{};
        r.blind_bytes = 0;
        r.msg_bytes = 1;
        r.produce = [](std::vector<std::uint8_t> & state, const std::uint8_t *,
                        std::size_t) {
            return std::vector<std::uint8_t>{state[0]};
        };
        r.finish = [](std::vector<std::uint8_t> & state, const std::uint8_t * peer,
                       std::size_t, const std::uint8_t *, std::size_t) {
            state[0] = static_cast<std::uint8_t>((state[0] + peer[0]) & 0xffu);
        };
        return std::vector<dpf::factory::async_byte_round>{r, r};
    };
    auto p0 = dpf::async::make_overlapped_byte_protocol(peer.first, nullptr,
        add_rounds());
    auto p1 = dpf::async::make_overlapped_byte_protocol(peer.second, nullptr,
        add_rounds());
    std::vector<std::uint8_t> out0, out1;
    dpf::async::run_overlapped(io, [&] {
        p0->start(0, {10}, [&](const std::error_code &,
                               std::vector<std::uint8_t> s) { out0 = std::move(s); });
        p1->start(0, {20}, [&](const std::error_code &,
                               std::vector<std::uint8_t> s) { out1 = std::move(s); });
    });
    ASSERT_EQ(out0.size(), 1u);
    ASSERT_EQ(out1.size(), 1u);
    EXPECT_EQ(out0[0], 60u);
    EXPECT_EQ(out1[0], 60u);
}

// One TCP socket per party, N logical streams multiplexed on it. Two rounds
// exchanged fully overlapped (write and peer-read issued together). Each party
// runs its own io_context; io.stop() from the done handler tears down the
// persistent read loop once the protocol finishes.
TEST(AsyncIo, MuxLocalhostOverlap)
{
    std::atomic<unsigned short> port{0};
    std::uint8_t out0 = 0, out1 = 0;
    auto add_rounds = [] {
        dpf::factory::async_byte_round r{};
        r.blind_bytes = 0;
        r.msg_bytes = 1;
        r.produce = [](std::vector<std::uint8_t> & state, const std::uint8_t *,
                        std::size_t) {
            return std::vector<std::uint8_t>{state[0]};
        };
        r.finish = [](std::vector<std::uint8_t> & state, const std::uint8_t * peer,
                       std::size_t, const std::uint8_t *, std::size_t) {
            state[0] = static_cast<std::uint8_t>((state[0] + peer[0]) & 0xffu);
        };
        return std::vector<dpf::factory::async_byte_round>{r, r};
    };
    std::thread t0([&] {
        asio::io_context io;
        asio::ip::tcp::acceptor acc(io,
            asio::ip::tcp::endpoint(asio::ip::tcp::v4(), port.load()));
        port.store(acc.local_endpoint().port());
        asio::ip::tcp::socket sock(io);
        acc.accept(sock);
        dpf::net::async_mux_stream_array mux(io, std::move(sock), 0, 1, 2);
        auto p = dpf::async::make_overlapped_byte_protocol(mux, nullptr,
            add_rounds());
        dpf::async::run_overlapped(io, [&] {
            p->start(0, {10}, [&](const std::error_code &,
                                  std::vector<std::uint8_t> s) {
                out0 = s[0];
                io.stop();
            });
        });
    });
    while (port.load() == 0)
        std::this_thread::yield();
    std::thread t1([&] {
        asio::io_context io;
        asio::ip::tcp::socket sock(io);
        asio::ip::tcp::resolver res(io);
        auto eps = res.resolve("127.0.0.1", std::to_string(port.load()));
        asio::connect(sock, eps);
        dpf::net::async_mux_stream_array mux(io, std::move(sock), 1, 0, 2);
        auto p = dpf::async::make_overlapped_byte_protocol(mux, nullptr,
            add_rounds());
        dpf::async::run_overlapped(io, [&] {
            p->start(0, {20}, [&](const std::error_code &,
                                  std::vector<std::uint8_t> s) {
                out1 = s[0];
                io.stop();
            });
        });
    });
    t0.join();
    t1.join();
    EXPECT_EQ(out0, 60u);
    EXPECT_EQ(out1, 60u);
}

// One TCP connection PER stream index; concurrent exchanges on stream 0 and
// stream 1 run overlapped on a single io_context per party.
TEST(AsyncIo, ParallelSocketsTwoStreams)
{
    std::atomic<unsigned short> port{0};
    std::uint8_t p0_in0 = 0, p0_in1 = 0, p1_in0 = 0, p1_in1 = 0;
    std::thread t0([&] {
        asio::io_context io;
        auto socks = dpf::net::accept_parallel_tcp(io, port, 2);
        dpf::net::async_parallel_stream_array arr(io, std::move(socks));
        std::uint8_t o0 = 0xA0, o1 = 0xA1;
        auto pending = std::make_shared<int>(2);
        dpf::async::run_overlapped(io, [&, pending] {
            auto on = [&, pending](const std::error_code &) {
                if (--*pending == 0)
                    io.stop();
            };
            dpf::async::async_exchange(arr, 0, &o0, 1, &p0_in0, 1, on);
            dpf::async::async_exchange(arr, 1, &o1, 1, &p0_in1, 1, on);
        });
    });
    while (port.load() == 0)
        std::this_thread::yield();
    std::thread t1([&] {
        asio::io_context io;
        auto socks = dpf::net::connect_parallel_tcp(io, "127.0.0.1", port, 2);
        dpf::net::async_parallel_stream_array arr(io, std::move(socks));
        std::uint8_t o0 = 0xB0, o1 = 0xB1;
        auto pending = std::make_shared<int>(2);
        dpf::async::run_overlapped(io, [&, pending] {
            auto on = [&, pending](const std::error_code &) {
                if (--*pending == 0)
                    io.stop();
            };
            dpf::async::async_exchange(arr, 0, &o0, 1, &p1_in0, 1, on);
            dpf::async::async_exchange(arr, 1, &o1, 1, &p1_in1, 1, on);
        });
    });
    t0.join();
    t1.join();
    EXPECT_EQ(p0_in0, 0xB0u);
    EXPECT_EQ(p0_in1, 0xB1u);
    EXPECT_EQ(p1_in0, 0xA0u);
    EXPECT_EQ(p1_in1, 0xA1u);
}

// One SCTP association, index i -> SCTP stream i. Two rounds exchanged fully
// overlapped over localhost (party 0 accepts, party 1 connects). Mirrors the
// mux/parallel overlap tests but on the real SCTP backend. Requires Linux +
// libsctp AND a kernel that supports SCTP sockets; otherwise the test skips.
TEST(AsyncIo, SctpLocalhostOverlap)
{
#if DPF_HAS_LIBSCTP
    // Kernel may lack the SCTP module even when libsctp headers exist.
    {
        const int probe = ::socket(AF_INET, SOCK_STREAM, IPPROTO_SCTP);
        if (probe < 0)
            GTEST_SKIP() << "kernel SCTP unavailable (socket: "
                         << std::strerror(errno) << ")";
        ::close(probe);
    }
    std::atomic<unsigned short> port{0};
    std::uint8_t out0 = 0, out1 = 0;
    auto add_rounds = [] {
        dpf::factory::async_byte_round r{};
        r.blind_bytes = 0;
        r.msg_bytes = 1;
        r.produce = [](std::vector<std::uint8_t> & state, const std::uint8_t *,
                        std::size_t) {
            return std::vector<std::uint8_t>{state[0]};
        };
        r.finish = [](std::vector<std::uint8_t> & state, const std::uint8_t * peer,
                       std::size_t, const std::uint8_t *, std::size_t) {
            state[0] = static_cast<std::uint8_t>((state[0] + peer[0]) & 0xffu);
        };
        return std::vector<dpf::factory::async_byte_round>{r, r};
    };
    std::exception_ptr err;
    std::mutex emu;
    auto note = [&](std::exception_ptr e) {
        std::lock_guard<std::mutex> lk(emu);
        if (!err)
            err = std::move(e);
    };
    std::thread t0([&] {
        try
        {
            asio::io_context io;
            const int fd = dpf::net::accept_sctp_association(io, port, 2);
            dpf::net::async_sctp_stream_array arr(io, fd, 2);
            auto p = dpf::async::make_overlapped_byte_protocol(arr, nullptr,
                add_rounds());
            dpf::async::run_overlapped(io, [&] {
                p->start(0, {10}, [&](const std::error_code &,
                                      std::vector<std::uint8_t> s) {
                    out0 = s[0];
                    io.stop();
                });
            });
        }
        catch (...)
        {
            note(std::current_exception());
        }
    });
    while (port.load() == 0)
        std::this_thread::yield();
    std::thread t1([&] {
        try
        {
            asio::io_context io;
            const int fd =
                dpf::net::connect_sctp_association(io, "127.0.0.1", port, 2);
            dpf::net::async_sctp_stream_array arr(io, fd, 2);
            auto p = dpf::async::make_overlapped_byte_protocol(arr, nullptr,
                add_rounds());
            dpf::async::run_overlapped(io, [&] {
                p->start(0, {20}, [&](const std::error_code &,
                                      std::vector<std::uint8_t> s) {
                    out1 = s[0];
                    io.stop();
                });
            });
        }
        catch (...)
        {
            note(std::current_exception());
        }
    });
    t0.join();
    t1.join();
    if (err)
        std::rethrow_exception(err);
    EXPECT_EQ(out0, 60u);
    EXPECT_EQ(out1, 60u);
#else
    GTEST_SKIP() << "libsctp not available (Linux + <netinet/sctp.h> required)";
#endif
}

// The performance harness runs on the framework: exercise_plan honours
// DPF_TRANSPORT. Here we drive a small open plan on the synchronous stream
// framework and confirm the reported cost matches the plan.
TEST(AppFlow, ExercisePlanUsesStreamFramework)
{
    ::setenv("DPF_TRANSPORT", "stream", 1);
    dpf::protocol::composer c(0);
    auto x = c.input(domain::a, 8);
    (void)c.exchange(x);
    auto p = c.default_plan();
    const auto got = dpf::app::exercise_plan(p);
    ::unsetenv("DPF_TRANSPORT");
    EXPECT_EQ(got.rounds, p.rounds());
    std::size_t want_bytes = 0;
    for (auto n : p.slot_bytes_all())
        want_bytes += n;
    EXPECT_EQ(got.bytes, want_bytes);
}

// The default transport (async_memory) drives the same plan over the real
// event-driven framework in-process.
TEST(AppFlow, ExercisePlanAsyncMemoryFramework)
{
    ::unsetenv("DPF_TRANSPORT");  // default: async_memory
    dpf::protocol::composer c(0);
    auto x = c.input(domain::a, 8);
    (void)c.exchange(x);
    auto p = c.default_plan();
    const auto got = dpf::app::exercise_plan(p);
    EXPECT_EQ(got.rounds, p.rounds());
    EXPECT_EQ(dpf::app::transport_from_env(), dpf::app::transport_kind::async_memory);
}

// Deep dependent plan on a 2-lane pool: rounds >> streams, framed demux.
TEST(AsyncIo, LaneReuseDeepPlan)
{
    constexpr std::uint32_t k_step = 92001;
    auto make = [](std::size_t party, std::uint64_t in) {
        dpf::protocol::composer c(party);
        auto x = c.input(domain::a, 8);
        auto cur = x;
        for (int i = 0; i < 6; ++i)
        {
            auto e = c.exchange(cur);
            cur = c.compute(k_step, {e}, domain::a, 8);
        }
        return std::make_tuple(c.schedule(), x, cur, in);
    };
    auto [p0, x0, out0, in0] = make(0, 1);
    auto [p1, x1, out1, in1] = make(1, 2);
    ASSERT_GE(p0.slot_bytes_all().size(), 6u);
    EXPECT_EQ(dpf::net::lane_count_for_rounds(p0.slot_bytes_all().size(), 2), 2u);
    dpf::protocol::kernel_fn step =
        [](std::uint32_t, const std::vector<dpf::protocol::node> &,
            const std::vector<dpf::protocol::block_span> & inputs,
            dpf::protocol::block_span output, std::size_t) {
            std::uint64_t v = 0;
            std::memcpy(&v, inputs[0].at(0), 8);
            v += 1;
            std::memcpy(output.at(0), &v, 8);
        };
    std::map<std::uint32_t, dpf::protocol::kernel_fn> k{{k_step, step}};
    std::vector<std::vector<std::uint8_t>> v0(p0.nodes().size()),
        v1(p1.nodes().size());
    put_raw(v0, x0, &in0, 8);
    put_raw(v1, x1, &in1, 8);
    dpf::protocol::drive_options opt;
    opt.n_lanes = 2;
    dpf::protocol::drive_both_on_async_streams(p0, p1, v0, v1, k, 1, opt);
    std::uint64_t o0 = 0, o1 = 0;
    std::memcpy(&o0, v0[out0.id].data(), 8);
    std::memcpy(&o1, v1[out1.id].data(), 8);
    // Dependent opens double each wave (both parties hold the prior open), then +1.
    // 1+2 → 3 →+1→ 4 →open→ 8 →+1→ 9 → … → final 159 after 6 steps.
    EXPECT_EQ(o0, o1);
    EXPECT_EQ(o0, 159u);
}

// Compose plans ride the overlapped async stack (not a separate factory path).
TEST(ComposeAsync, DriveBothOnAsyncStreams)
{
    dpf::protocol::composer c0(0);
    dpf::protocol::composer c1(1);
    auto x0 = c0.input(domain::a, 8);
    auto x1 = c1.input(domain::a, 8);
    auto o0 = c0.exchange(x0);
    auto o1 = c1.exchange(x1);
    auto p0 = c0.schedule();
    auto p1 = c1.schedule();
    std::vector<std::vector<std::uint8_t>> v0(p0.nodes().size()),
        v1(p1.nodes().size());
    std::uint64_t a = 11, b = 31;
    put_raw(v0, x0, &a, 8);
    put_raw(v1, x1, &b, 8);
    dpf::protocol::drive_both_on_async_streams(p0, p1, v0, v1);
    std::uint64_t open0 = 0, open1 = 0;
    std::memcpy(&open0, v0[o0.id].data(), 8);
    std::memcpy(&open1, v1[o1.id].data(), 8);
    EXPECT_EQ(open0, 42u);
    EXPECT_EQ(open1, 42u);
}

// Overlapped factory protocol on a single peer stream (lane reuse by modulo).
TEST(AsyncIo, OverlappedLaneModulo)
{
    asio::io_context io;
    auto peer = dpf::net::make_async_memory_stream_pair(io, 1);
    using round = dpf::factory::async_byte_round;
    std::vector<round> rounds(3);
    for (std::size_t r = 0; r < 3; ++r)
    {
        rounds[r].msg_bytes = 4;
        rounds[r].blind_bytes = 0;
        rounds[r].produce = [r](const std::vector<std::uint8_t> &,
                                const std::uint8_t *, std::size_t) {
            std::vector<std::uint8_t> out(4, static_cast<std::uint8_t>(r + 1));
            return out;
        };
        rounds[r].finish = [r](std::vector<std::uint8_t> & st, const std::uint8_t * in,
                               std::size_t n, const std::uint8_t *, std::size_t) {
            ASSERT_EQ(n, 4u);
            st.push_back(in[0]);
            EXPECT_EQ(in[0], static_cast<std::uint8_t>(r + 1));
        };
    }
    std::vector<std::uint8_t> done0, done1;
    std::error_code ec0, ec1;
    auto p0 = dpf::async::make_overlapped_byte_protocol(peer.first, nullptr, rounds);
    auto p1 = dpf::async::make_overlapped_byte_protocol(peer.second, nullptr, rounds);
    dpf::async::run_overlapped(io, [&] {
        p0->start(0, {}, [&](const std::error_code & ec, std::vector<std::uint8_t> st) {
            ec0 = ec;
            done0 = std::move(st);
        });
        p1->start(0, {}, [&](const std::error_code & ec, std::vector<std::uint8_t> st) {
            ec1 = ec;
            done1 = std::move(st);
        });
    });
    EXPECT_FALSE(ec0);
    EXPECT_FALSE(ec1);
    ASSERT_EQ(done0.size(), 3u);
    ASSERT_EQ(done1.size(), 3u);
}

// Three parties join a real TCP mux clique and exchange 8 bytes on each edge.
TEST(TcpMesh, ThreePartyMuxExchange)
{
    constexpr unsigned n = 3;
    std::atomic<int> ok{0};
    dpf::run::threads_on_tcp_mesh(n, /*streams=*/1, [&](unsigned me,
                                                       dpf::net::tcp_mesh & mesh) {
        EXPECT_EQ(mesh.parties, n);
        for (unsigned peer = 0; peer < n; ++peer)
        {
            if (peer == me)
                continue;
            std::uint64_t mine = (static_cast<std::uint64_t>(me) << 8) | peer;
            std::uint64_t got = 0;
            auto & link = mesh.peer(peer);
            link.write(0, &mine, 8);
            link.flush(0);
            link.read(0, &got, 8);
            const std::uint64_t expect =
                (static_cast<std::uint64_t>(peer) << 8) | me;
            EXPECT_EQ(got, expect);
        }
        ok.fetch_add(1);
    });
    EXPECT_EQ(ok.load(), static_cast<int>(n));
}

// Composer accepts party ids beyond 0..2 (N-party meshes).
TEST(Compose, NPartyComposerAllowed)
{
    dpf::protocol::composer c5(5);
    EXPECT_EQ(c5.party(), 5u);
}

TEST(Session, ShipPrepAsync)
{
    dpf::prep::demand d;
    d.limb = 8;
    d.ring_triples = 2;
    const auto shipped = dpf::session::ship_prep(d);
    EXPECT_EQ(shipped.bytes0, shipped.bytes1);
    EXPECT_GT(shipped.bytes0, 18u);
    std::uint8_t a0[8]{}, b0[8]{}, c0[8]{}, a1[8]{}, b1[8]{}, c1[8]{};
    auto p0 = shipped.party0;
    auto p1 = shipped.party1;
    p0.take_ring(a0, b0, c0);
    p1.take_ring(a1, b1, c1);
    std::uint64_t sa = 0, sb = 0, sc = 0, ta = 0, tb = 0, tc = 0;
    std::memcpy(&sa, a0, 8);
    std::memcpy(&sb, b0, 8);
    std::memcpy(&sc, c0, 8);
    std::memcpy(&ta, a1, 8);
    std::memcpy(&tb, b1, 8);
    std::memcpy(&tc, c1, 8);
    EXPECT_EQ((sa + ta) * (sb + tb), sc + tc);
}

TEST(Session, HushmapAsyncDealerAndPeer)
{
    EXPECT_NO_THROW(dpf::session::drive_hushmap_add(3));
}

TEST(Session, PirsonaAsyncStar)
{
    constexpr std::size_t L = 1;
    constexpr std::size_t n = 2;
    auto seeds = std::make_shared<std::vector<std::vector<std::uint8_t>>>(n);
    auto answers = std::make_shared<std::vector<std::vector<std::uint8_t>>>(n);
    for (std::size_t i = 0; i < n; ++i)
    {
        (*seeds)[i].assign(16, static_cast<std::uint8_t>(i + 1));
        (*answers)[i].assign(8, static_cast<std::uint8_t>(0x40 + i));
    }
    auto client = dpf::protocol::pirsona_bitmore_fetch(L, 16, 8, seeds, answers);
    EXPECT_NO_THROW(dpf::session::drive_async_star(n, {16u, 8u}, std::move(client),
        [&](std::size_t i) {
            return dpf::protocol::star_server_reply_rounds(16, 8, (*answers)[i]);
        }));
}

TEST(WireWindow, MuxTinyWindowStillCompletes)
{
    asio::io_context io0;
    asio::io_context io1;
    dpf::net::wire_policy pol;
    pol.window_bytes = 64;
    std::atomic<unsigned short> port{0};
    std::atomic<int> ok{0};
    std::exception_ptr err;
    std::mutex mu;
    std::thread t0([&] {
        try
        {
            asio::ip::tcp::acceptor acc(io0,
                asio::ip::tcp::endpoint(asio::ip::tcp::v4(), 0));
            port.store(acc.local_endpoint().port());
            asio::ip::tcp::socket sock(io0);
            acc.accept(sock);
            dpf::net::async_mux_stream_array mux(io0, std::move(sock), 0, 1, 1, pol);
            EXPECT_EQ(mux.window_bytes(), 64u);
            std::vector<std::uint8_t> out(200, 0xab);
            std::vector<std::uint8_t> in(200);
            std::atomic<int> left{2};
            mux.async_write(0, out.data(), out.size(), [&](const std::error_code & ec) {
                EXPECT_FALSE(ec);
                left.fetch_sub(1);
            });
            mux.async_read(0, in.data(), in.size(), [&](const std::error_code & ec) {
                EXPECT_FALSE(ec);
                left.fetch_sub(1);
            });
            while (left.load() != 0)
                io0.run_one();
            EXPECT_EQ(in, out);
            ok.fetch_add(1);
        }
        catch (...)
        {
            std::lock_guard<std::mutex> lock(mu);
            err = std::current_exception();
        }
    });
    std::thread t1([&] {
        try
        {
            while (port.load() == 0)
                std::this_thread::yield();
            asio::ip::tcp::socket sock(io1);
            asio::ip::tcp::resolver res(io1);
            auto eps = res.resolve("127.0.0.1", std::to_string(port.load()));
            asio::connect(sock, eps);
            dpf::net::async_mux_stream_array mux(io1, std::move(sock), 1, 0, 1);
            std::vector<std::uint8_t> out(200, 0xab);
            std::vector<std::uint8_t> in(200);
            std::atomic<int> left{2};
            mux.async_write(0, out.data(), out.size(), [&](const std::error_code & ec) {
                EXPECT_FALSE(ec);
                left.fetch_sub(1);
            });
            mux.async_read(0, in.data(), in.size(), [&](const std::error_code & ec) {
                EXPECT_FALSE(ec);
                left.fetch_sub(1);
            });
            while (left.load() != 0)
                io1.run_one();
            ok.fetch_add(1);
        }
        catch (...)
        {
            std::lock_guard<std::mutex> lock(mu);
            err = std::current_exception();
        }
    });
    t0.join();
    t1.join();
    if (err)
        std::rethrow_exception(err);
    EXPECT_EQ(ok.load(), 2);
}

TEST(IoPool, PostsComputeOnSharedWorkers)
{
    dpf::net::io_pool pool(2);
    EXPECT_GE(pool.size(), 2u);
    constexpr int n = 32;
    std::atomic<int> sum{0};
    for (int i = 0; i < n; ++i)
        pool.post([&] { sum.fetch_add(1); });
    for (int spins = 0; sum.load() != n && spins < 1000000; ++spins)
        std::this_thread::yield();
    EXPECT_EQ(sum.load(), n);
}

namespace
{

void exchange_u64(dpf::net::async_stream_array & link, std::uint64_t mine,
    std::uint64_t expect)
{
    std::uint64_t got = 0;
    std::atomic<int> left{2};
    std::error_code wec, rec;
    link.async_write(0, &mine, 8, [&](const std::error_code & ec) {
        wec = ec;
        left.fetch_sub(1);
    });
    link.async_read(0, &got, 8, [&](const std::error_code & ec) {
        rec = ec;
        left.fetch_sub(1);
    });
    for (int spins = 0; left.load() != 0 && spins < 1000000; ++spins)
        std::this_thread::yield();
    ASSERT_EQ(left.load(), 0);
    EXPECT_FALSE(wec);
    EXPECT_FALSE(rec);
    EXPECT_EQ(got, expect);
}

}  // namespace

TEST(PartySession, JoinReconnectAndDealerOnOnePool)
{
    dpf::net::io_pool pool(2);
    auto ports = dpf::net::make_mesh_ports(2);
    std::exception_ptr err;
    std::mutex mu;
    std::thread t0([&] {
        try
        {
            dpf::net::party_session me(pool.context(), 0, 2, 1);
            me.join("127.0.0.1", ports);
            exchange_u64(me.peer(1), 0x10, 0x21);
            me.reconnect(1);
            exchange_u64(me.peer(1), 0x30, 0x41);
        }
        catch (...)
        {
            std::lock_guard<std::mutex> lock(mu);
            err = std::current_exception();
        }
    });
    std::thread t1([&] {
        try
        {
            dpf::net::party_session me(pool.context(), 1, 2, 1);
            me.join("127.0.0.1", ports);
            exchange_u64(me.peer(0), 0x21, 0x10);
            me.reconnect(0);
            exchange_u64(me.peer(0), 0x41, 0x30);
        }
        catch (...)
        {
            std::lock_guard<std::mutex> lock(mu);
            err = std::current_exception();
        }
    });
    t0.join();
    t1.join();
    if (err)
        std::rethrow_exception(err);

    // Dealer is its own link, not a mesh neighbor.
    std::atomic<unsigned short> dport{0};
    std::thread dealer([&] {
        try
        {
            dpf::net::party_session d(pool.context(), 0, 2, 1);
            dport.store(d.listen());
            d.accept_dealer();
            exchange_u64(d.dealer(), 7, 9);
        }
        catch (...)
        {
            std::lock_guard<std::mutex> lock(mu);
            if (!err)
                err = std::current_exception();
        }
    });
    std::thread party([&] {
        try
        {
            while (dport.load() == 0)
                std::this_thread::yield();
            dpf::net::party_session p(pool.context(), 1, 2, 1);
            p.connect_dealer("127.0.0.1", dport.load());
            EXPECT_TRUE(p.has_dealer());
            exchange_u64(p.dealer(), 9, 7);
        }
        catch (...)
        {
            std::lock_guard<std::mutex> lock(mu);
            if (!err)
                err = std::current_exception();
        }
    });
    dealer.join();
    party.join();
    if (err)
        std::rethrow_exception(err);
}

TEST(SocketTune, NoDelayOnConnectedSocket)
{
    asio::io_context io;
    std::atomic<unsigned short> port{0};
    asio::ip::tcp::no_delay server_delay(false);
    asio::ip::tcp::no_delay client_delay(false);
    std::thread acc([&] {
        asio::ip::tcp::acceptor a(io,
            asio::ip::tcp::endpoint(asio::ip::tcp::v4(), 0));
        port.store(a.local_endpoint().port());
        asio::ip::tcp::socket s(io);
        a.accept(s);
        dpf::net::tune_tcp(s);
        s.get_option(server_delay);
    });
    while (port.load() == 0)
        std::this_thread::yield();
    asio::ip::tcp::socket c(io);
    asio::ip::tcp::resolver res(io);
    asio::connect(c, res.resolve("127.0.0.1", std::to_string(port.load())));
    dpf::net::tune_tcp(c);
    c.get_option(client_delay);
    acc.join();
    EXPECT_TRUE(server_delay.value());
    EXPECT_TRUE(client_delay.value());
}

TEST(IoPool, ComputeDoesNotUseSocketContext)
{
    dpf::net::io_pool pool(2, 2);
    EXPECT_GE(pool.compute_size(), 2u);
    std::atomic<bool> done{false};
    std::thread::id compute_id;
    const auto caller = std::this_thread::get_id();
    pool.post_compute([&] {
        compute_id = std::this_thread::get_id();
        done.store(true);
    });
    for (int i = 0; !done.load() && i < 1000000; ++i)
        std::this_thread::yield();
    EXPECT_TRUE(done.load());
    EXPECT_NE(compute_id, caller);
}

TEST(DriveOptions, LaneCountAndWaitTimeout)
{
    dpf::protocol::drive_options opt;
    opt.n_lanes = 2;
    opt.wait_timeout = std::chrono::milliseconds(50);
    EXPECT_EQ(dpf::protocol::lanes_for_plan(6, opt), 2u);

    dpf::protocol::composer c0(0);
    auto x = c0.input(domain::a, 8);
    (void)c0.exchange(x);
    auto p = c0.schedule();
    asio::io_context io0;
    asio::io_context io1;
    auto ends = dpf::net::make_async_dual_memory_stream_pair(io0, io1, 1);
    auto work = asio::make_work_guard(io0);
    std::vector<std::vector<std::uint8_t>> values(p.nodes().size());
    std::uint64_t in = 1;
    put_raw(values, x, &in, 8);
    EXPECT_THROW(
        dpf::protocol::drive_plan_on_async_streams(p, ends.first, values, {}, 0,
            1, opt),
        std::runtime_error);
}

TEST(DriveOptions, KernelRunsOnComputePool)
{
    constexpr std::uint32_t k_step = 93001;
    dpf::net::io_pool pool(2, 2);
    std::thread::id kid;
    dpf::protocol::kernel_fn step =
        [&](std::uint32_t, const std::vector<dpf::protocol::node> &,
            const std::vector<dpf::protocol::block_span> & inputs,
            dpf::protocol::block_span output, std::size_t) {
            kid = std::this_thread::get_id();
            std::uint64_t v = 0;
            std::memcpy(&v, inputs[0].at(0), 8);
            v += 1;
            std::memcpy(output.at(0), &v, 8);
        };
    auto make = [&](std::size_t party, std::uint64_t in) {
        dpf::protocol::composer c(party);
        auto xin = c.input(domain::a, 8);
        auto e = c.exchange(xin);
        auto y = c.compute(k_step, {e}, domain::a, 8);
        return std::make_tuple(c.schedule(), xin, y, in);
    };
    auto [p0, x0, out0, in0] = make(0, 4);
    auto [p1, x1, out1, in1] = make(1, 6);
    std::map<std::uint32_t, dpf::protocol::kernel_fn> k{{k_step, step}};
    std::vector<std::vector<std::uint8_t>> v0(p0.nodes().size()),
        v1(p1.nodes().size());
    put_raw(v0, x0, &in0, 8);
    put_raw(v1, x1, &in1, 8);
    dpf::protocol::drive_options opt;
    opt.workers = &pool;
    const auto caller = std::this_thread::get_id();
    dpf::protocol::drive_both_on_async_streams(p0, p1, v0, v1, k, 1, opt);
    EXPECT_NE(kid, caller);
    std::uint64_t o0 = 0;
    std::memcpy(&o0, v0[out0.id].data(), 8);
    EXPECT_EQ(o0, 11u);
}

TEST(PartySession, WindowAndLaneOverride)
{
    dpf::net::io_pool pool(2);
    auto ports = dpf::net::make_mesh_ports(2);
    std::atomic<std::size_t> win0{0};
    std::atomic<std::size_t> streams0{0};
    std::thread t0([&] {
        dpf::net::party_session me(pool.context(), 0, 2, 1);
        me.set_window_bytes(4096);
        me.join("127.0.0.1", ports, 4);
        win0.store(me.peer(1).window_bytes());
        streams0.store(me.streams());
        std::uint64_t mine = 1, got = 0;
        std::atomic<int> left{2};
        me.peer(1).async_write(0, &mine, 8, [&](const std::error_code &) {
            left.fetch_sub(1);
        });
        me.peer(1).async_read(0, &got, 8, [&](const std::error_code &) {
            left.fetch_sub(1);
        });
        for (int i = 0; left.load() != 0 && i < 1000000; ++i)
            std::this_thread::yield();
        EXPECT_EQ(got, 2u);
    });
    std::thread t1([&] {
        dpf::net::party_session me(pool.context(), 1, 2, 1);
        me.set_window_bytes(4096);
        me.join("127.0.0.1", ports, 4);
        std::uint64_t mine = 2, got = 0;
        std::atomic<int> left{2};
        me.peer(0).async_write(0, &mine, 8, [&](const std::error_code &) {
            left.fetch_sub(1);
        });
        me.peer(0).async_read(0, &got, 8, [&](const std::error_code &) {
            left.fetch_sub(1);
        });
        for (int i = 0; left.load() != 0 && i < 1000000; ++i)
            std::this_thread::yield();
        EXPECT_EQ(got, 1u);
    });
    t0.join();
    t1.join();
    EXPECT_EQ(win0.load(), 4096u);
    EXPECT_EQ(streams0.load(), 4u);
}