pine_state_hash.cpp

Go to the documentation of this file.

#include <pineforge/source/pine_strategy_host.hpp>
#include <pineforge/source/pine_native_host.hpp>
 
#include "../broker_state_hash_internal.hpp"
#include "../native_execution_consumer.hpp"
#include "pine_intent_state.hpp"
 
#include <algorithm>
#include <cmath>
#include <cstring>
#include <stdexcept>
#include <type_traits>
#include <variant>
#include <vector>
 
#ifndef PINEFORGE_TAIL_WORK
#define PINEFORGE_TAIL_WORK(site, count) ((void)0)
#endif
#ifndef PINEFORGE_BROKER_HASH_OBSERVE
#define PINEFORGE_BROKER_HASH_OBSERVE(rows, families, archived, digest) ((void)0)
#endif
 
namespace pineforge {
namespace {
 
// The admission journal is source-layer state (R5 lane N14): its reflected
// fields fold here, through the source host's extension, exactly as they
// did when this helper lived in the kernel-internal sink header.
void hash_admission_field(BrokerStateHashSink& f, const admission::Field& field) {
    f.s(field.path); f.u(field.value.index());
    std::visit([&](const auto& value) {
        using T = std::decay_t<decltype(value)>;
        if constexpr (std::is_same_v<T, uint64_t>) f.u(value);
        else if constexpr (std::is_same_v<T, int64_t>) f.i(value);
        else if constexpr (std::is_same_v<T, double>) f.d(value);
        else f.s(value);
    }, field.value);
}
 
void hash_source_series(BrokerStateHashSink& f, const Series<double>& series) {
    f.i(series.size());
    for (int i = 0; i < series.size(); ++i) f.d(series[i]);
}
 
} // namespace
 
namespace {
 
void hash_source_run_identity(BrokerStateHashSink& f,
                              const native_order::RunIdentity& identity) {
    f.s(identity.session_key);
    // The monotonically increasing native generation protects stale handles;
    // it does not change the source-visible state of a fresh run.
    f.u(0);
}
void hash_scratch_digest(BrokerStateHashSink& f,
                         ::pineforge::detail::ScratchStatus status, std::uint64_t digest) {
    if (status != ::pineforge::detail::ScratchStatus::ok)
        throw coded<std::logic_error>(
            status == ::pineforge::detail::ScratchStatus::io_error
                ? RunFailureCode::scratch_io_failed : RunFailureCode::engine_invariant,
            {}, "scratch state digest failed");
    f.u(digest);
}
void hash_scratch_store(BrokerStateHashSink& f,
                        const ::pineforge::detail::CommandScratchStore* store) {
    f.b(store != nullptr);
    if (!store) return;
    // One page of workspace, independent of the record population. The state
    // digest includes order.id, next_handle, record_count and every live row's
    // handle/key/value, without depending on backing or cache placement.
    std::vector<std::byte> page(store->page_size());
    std::uint64_t digest = 0;
    const auto status = store->state_digest({page.data(), page.size()}, digest);
    if (status == ::pineforge::detail::ScratchStatus::io_error)
        ::pineforge::detail::rethrow_carried_scratch_failure(store->carried_failure());
    hash_scratch_digest(f, status, digest);
}
void hash_exit_scratch(BrokerStateHashSink& f,
                       const source::detail::ExitScratchContext* scratch) {
    f.b(scratch != nullptr);
    if (!scratch) return;
    hash_source_run_identity(f, scratch->run);
    f.b(scratch->versions != nullptr);
    f.b(scratch->lots != nullptr); f.b(scratch->working != nullptr);
    hash_scratch_store(f, &scratch->store);
}
void hash_native_handle(BrokerStateHashSink& f, const native_order::RequestHandle& handle) {
    hash_source_run_identity(f, handle.run); f.u(handle.incarnation);
}
void hash_native_handle_vector(BrokerStateHashSink& f,
                               const std::vector<native_order::RequestHandle>& handles) {
    f.u(handles.size());
    for (const auto& handle : handles) hash_native_handle(f, handle);
}
void hash_placement(BrokerStateHashSink& f, const source::PlacementSnapshot& value) {
    f.i(static_cast<std::int64_t>(value.family)); f.s(value.source_id); f.s(value.from_entry);
    f.s(value.comment); f.s(value.oca_name); f.i(value.oca_type); f.i(value.qty_type);
    f.d(value.requested_qty); f.d(value.projection_remaining_qty);
    f.d(value.qty_percent); f.b(value.is_long); f.b(value.immediately);
    f.b(value.opening); f.b(value.deferred_cohort);
    // Conditional v19/v4 retirement fact: an untouched placement retains
    // its byte spelling. This receipt decides the filled-limit K1 exception.
    if (value.filled_exit) f.u(0x66696c6c6564ULL);
    f.b(value.reservation_deferred_to_pending_entry);
    // Folded only when set, so a run that never issues a void exit keeps
    // its digest (lane W3B-ENG-GRID).
    if (value.void_issue) f.b(true);
    // Folded only when set, like void_issue (MarginOpeningSwitches::
    // void_flat_global_exit).
    if (value.void_global_issue) f.u(0x766f6964676cULL);
    // Folded only when set, like void_issue: the exit is bound to its entry
    // id, not to lots (ExitBindingRuleSwitches).
    if (value.pending_bound_exit) f.u(0x70656e64ULL);
    // Folded only when set, like void_issue: one entry's bracket of a global
    // exit that covers its position entry by entry (ExitBindingRuleSwitches).
    if (value.per_entry_origin != 0) {
        f.u(0x70657265ULL); f.u(value.per_entry_origin); f.d(value.per_entry_units);
    }
    // Folded only when set, like void_issue (lane W13-ENG-MARGIN-OPP).
    if (value.crosses_zero) f.u(0x7a65726fULL);
    if (!std::isnan(value.follow_up_fill)) f.d(value.follow_up_fill);
    if (value.waypoint_margin_call) f.u(0x77617970ULL);
    f.b(value.fixed_exit_reservation);
    f.b(value.frozen_market_instruction);
    f.d(value.frozen_market_own_units); f.d(value.frozen_market_transaction_units);
    f.b(value.frozen_market_targeted_close); f.b(value.frozen_market_target_was_long);
    f.b(value.direction_gate); f.b(value.affordability_policy_active);
    f.b(value.affordability_close_only);
    f.b(value.rounded_signal_cost_close_only);
    f.b(value.affordability_keep_mc_close_surplus);
    f.b(value.reverse_to); f.b(value.replaced_opening); f.b(value.replacement_predecessor_market);
    f.b(value.terms_priced_reverse);
    f.d(value.frozen_reversal_transaction);
    f.i(value.placement_cycle); f.u(value.sequential_group); f.u(value.sequential_rank);
    f.b(value.has_full_entry_bracket);
    hash_source_run_identity(f, value.paired_reversal_parent.run);
    f.u(value.paired_reversal_parent.incarnation);
    hash_source_run_identity(f, value.preserved_by_close_all.run);
    f.u(value.preserved_by_close_all.incarnation);
    f.i(value.preserved_close_all_bar);
    hash_source_run_identity(f, value.bracket_origin.run);
    f.u(value.bracket_origin.incarnation);
    f.u(value.source_sequence);
    f.u(value.command_ordinal); f.u(value.placement_open_epoch);
    f.u(value.command_sequence);
    f.u(value.close_callsite_token); f.u(value.close_batch_calls);
    f.s(value.close_first_id); f.d(value.close_first_target);
    f.b(value.close_first_ledger_consumed); f.b(value.close_first_carry_valid);
    f.d(value.close_first_carry_qty); f.b(value.close_retire_ledger_whole);
    f.d(value.close_pending_later_qty);
    f.i(value.placement_script_open_ms);
    f.i(value.placement_sub_open_ms); f.i(value.projection_created_bar);
    f.b(value.projection_created_bar_pinned);
    f.i(value.projection_position_side); f.b(value.projection_after_close);
    // Folded only when set, so a run without a close-first entry keeps its
    // digest.
    if (value.close_first_entry) f.u(0x636c6f7365666972ULL);
    if (value.signal_price_refused) f.u(0x7369677265667573ULL);
    f.b(value.projection_over_pyramiding);
    f.b(value.projection_opposite_market_predecessor);
    f.u(value.projection_predecessor);
    f.u(value.chain_origin_sequence);
    f.u(value.recreated_after_named_cancelled_entry_incarnation);
    f.u(value.named_cancel_surviving_exit_incarnation);
    f.b(value.retained_parent_topology);
    f.b(value.defer_until_post_parent_calculation);
    f.b(value.projection_predecessor_market); f.b(value.projection_predecessor_exit);
    f.b(value.projection_created_during_coof); f.b(value.projection_coof_at_terminal);
    f.b(value.projection_coof_mid_bar); f.d(value.forced_execution_price);
    if (!std::isnan(value.forced_waypoint_print)) {
        f.u(0x7761797074ULL); f.d(value.forced_waypoint_print);
    }
    if (value.coof_open_stop_next_waypoint) f.u(0x6f70656e73746f70ULL);
    f.d(value.projection_tv_carry_qty);
    f.d(value.projection_default_stop_equity);
    f.d(value.projection_default_stop_signal_close);
    f.d(value.projection_explicit_equity); f.d(value.projection_explicit_signal_close);
    f.d(value.projection_affordability_equity);
    f.d(value.projection_affordability_signal_price);
    f.d(value.projection_affordability_held_qty);
    f.d(value.sizing.equity); f.d(value.sizing.price);
    f.d(value.sizing.fx); f.d(value.sizing.mark); f.d(value.sizing.frozen_units);
    f.b(value.sizing.at_fill);
    // Recorded only under a cash commission with a percent-of-equity default
    // quantity (the folded run configuration decides), so every other run
    // keeps its digest (R5 lane PAR-CASHFEE).
    if (!std::isnan(value.sizing.strategy_equity)) f.d(value.sizing.strategy_equity);
    f.d(value.exit_levels.limit); f.d(value.exit_levels.stop);
    f.d(value.exit_levels.trail_points); f.d(value.exit_levels.trail_offset);
    f.d(value.exit_levels.trail_price); f.d(value.exit_levels.profit_ticks);
    f.d(value.exit_levels.loss_ticks);
    f.d(value.trail_activation_level); f.d(value.retained_trail_best);
    admission::reflect(value.market_admission, "placement.market_admission",
        [&](const admission::Field& field) { hash_admission_field(f, field); });
    f.i(static_cast<std::int64_t>(value.birth.cause())); f.i(value.birth.bar());
    f.i(value.birth.timestamp()); f.i(static_cast<std::int64_t>(value.birth.cursor().domain()));
    f.i(static_cast<std::int64_t>(value.birth.cursor().position()));
    f.i(value.birth.cursor().index()); f.i(value.birth.cursor().count());
    f.d(value.birth.cursor_price()); f.u(value.birth.first_fill()); f.u(value.birth.last_fill());
    f.u(value.birth.evaluation_ordinal()); f.i(static_cast<std::int64_t>(value.birth_reach));
    f.b(value.leg_activation.bounds().has_value());
    if (value.leg_activation.bounds()) {
        f.i(value.leg_activation.bounds()->position_cycle);
        f.i(value.leg_activation.bounds()->stop_first_bar);
        f.i(value.leg_activation.bounds()->limit_first_bar);
    }
    f.b(value.exit_activation.evidence().has_value());
    if (value.exit_activation.evidence()) {
        const auto& evidence = *value.exit_activation.evidence();
        f.i(evidence.position_cycle); f.i(evidence.entry_bar); f.i(evidence.direction);
        f.d(evidence.cursor_price); f.d(evidence.stop_level); f.d(evidence.limit_level);
        f.b(evidence.limit_continuation.has_value());
        if (evidence.limit_continuation) {
            f.i(static_cast<std::int64_t>(evidence.limit_continuation->cause));
            f.u(evidence.limit_continuation->observed_fill_sequence);
        }
    }
    value.legs.visit(f);
    f.b(value.restored_after_margin);
    f.b(value.reservation_expansion.capture().has_value());
    if (value.reservation_expansion.capture()) {
        const auto& capture = *value.reservation_expansion.capture();
        f.i(capture.position_cycle); f.i(static_cast<std::int64_t>(capture.side));
        f.b(capture.first_later_admission.has_value());
        if (capture.first_later_admission) f.u(*capture.first_later_admission);
    }
    f.b(value.reservation_growth_source.reservation_owner().has_value());
    if (value.reservation_growth_source.reservation_owner())
        f.u(*value.reservation_growth_source.reservation_owner());
    f.u(value.reservation_growth_owner_incarnation);
    f.b(value.stop_limit_activated); f.i(value.coof_cascade_seg_i);
    f.b(value.coof_cascade_inflight_fires); f.b(value.paired_flat_market_candidate);
    f.d(value.paired_flat_market_own_qty); f.d(value.paired_flat_market_signal_close);
    f.d(value.paired_flat_market_signal_equity);
    f.d(value.paired_flat_market_signal_margin_pct);
    f.d(value.paired_flat_market_signal_pointvalue); f.d(value.paired_flat_market_signal_fx);
    f.i(value.paired_flat_market_peer_seq); f.d(value.paired_flat_market_transaction_qty);
    f.i(value.signal_close_mc_bar); f.u(value.signal_close_mc_entry_incarnation);
    f.u(value.signal_close_mc_fill_seq); f.d(value.signal_close_mc_remaining_qty);
    f.b(value.pooc_global_full_exit_dynamic_qty);
    f.b(value.pooc_global_full_exit_tracks_bound_adds);
    f.b(value.pooc_global_full_exit_bound_add);
    f.i(static_cast<std::int32_t>(value.cancellation.cause)); f.i(value.cancellation.state);
    f.i(value.cancellation.close_claim_release); f.u(value.cancellation.source_incarnation);
    f.i(value.cancellation.source_sequence); f.u(value.cancellation.target_incarnation);
    f.i(value.cancellation.target_owner); f.u(value.cancellation.target_revision);
    f.d(value.cancellation.close_claim_consumed); f.d(value.cancellation.close_claim_retired);
}
 
void hash_short_seed_plan(BrokerStateHashSink& f, const source::ShortSeedPlan& value) {
    hash_native_handle(f, value.long_entry); hash_native_handle(f, value.materialize_long);
    hash_native_handle(f, value.final_short); f.s(value.seed_id); f.s(value.long_entry_id);
    f.s(value.final_short_id); f.s(value.materialize_label); f.d(value.seed_qty);
    f.i(value.seed_cycle); f.b(value.active); f.b(value.report_swap_pending);
}
 
void hash_native_request(BrokerStateHashSink& f, const native_order::Request& request) {
    f.u(request.intent.index());
    if (const auto* reduce = std::get_if<native_order::Reduce>(&request.intent)) {
        f.u(reduce->size.index());
        if (const auto* units = std::get_if<native_order::ExplicitUnits>(&reduce->size)) f.d(units->units);
    } else if (const auto* transact = std::get_if<native_order::Transact>(&request.intent)) {
        f.d(transact->signed_units);
    } else if (const auto* reverse = std::get_if<native_order::ReverseTo>(&request.intent)) {
        f.d(reverse->signed_units);
    } else if (const auto* sized = std::get_if<native_order::HostSized>(&request.intent)) {
        f.i(static_cast<std::int64_t>(sized->kind)); f.b(sized->side.has_value());
        if (sized->side) f.i(static_cast<std::int64_t>(*sized->side));
    } else if (const auto* core_sized = std::get_if<native_order::Sized>(&request.intent)) {
        // A core-sized opening carries its whole quantity in the basis, so the
        // source fingerprint must separate two commands that differ only there.
        f.i(static_cast<std::int64_t>(core_sized->side));
        f.u(core_sized->basis.index());
        if (const auto* cash = std::get_if<native_order::CashValue>(&core_sized->basis)) {
            f.d(cash->cash);
        } else {
            f.d(std::get<native_order::EquityFraction>(core_sized->basis).fraction);
        }
        f.i(static_cast<std::int64_t>(core_sized->time));
        f.i(static_cast<std::int64_t>(core_sized->price));
        f.i(static_cast<std::int64_t>(core_sized->grid_policy));
        f.b(core_sized->reserve_percent_fee);
    }
    f.s(request.label); f.s(request.comment);
    f.u(request.trigger.index());
    if (const auto* limit = std::get_if<native_order::Limit>(&request.trigger)) f.d(limit->price);
    else if (const auto* stop = std::get_if<native_order::Stop>(&request.trigger)) f.d(stop->price);
    else if (const auto* stop_limit = std::get_if<native_order::StopLimit>(&request.trigger)) {
        f.d(stop_limit->stop); f.d(stop_limit->limit);
    } else if (const auto* trail = std::get_if<native_order::Trail>(&request.trigger)) {
        f.d(trail->offset); f.b(trail->arm_price.has_value());
        if (trail->arm_price) f.d(*trail->arm_price);
    }
    f.u(request.capacity.index());
    if (const auto* budget = std::get_if<native_order::PointBudget>(&request.capacity)) f.d(budget->units);
    f.u(request.owner.index());
    if (const auto* wait = std::get_if<native_order::WaitForApplied>(&request.owner)) {
        hash_native_handle(f, wait->parent);
    } else if (const auto* opening = std::get_if<native_order::BindOpening>(&request.owner)) {
        hash_native_handle(f, opening->opening); f.i(opening->cycle);
    } else if (const auto* openings = std::get_if<native_order::BindOpenings>(&request.owner)) {
        hash_native_handle_vector(f, openings->openings); f.i(openings->cycle);
    } else if (const auto* cohort = std::get_if<native_order::BindCohort>(&request.owner)) {
        f.u(cohort->cohort.value);
    }
    f.u(request.group.index());
    if (const auto* group = std::get_if<native_order::Member>(&request.group)) {
        f.u(group->group); f.i(group->cohort); f.i(static_cast<std::int64_t>(group->effect));
    }
}
 
} // namespace
 
void source::PineExecutionAdapter::hash_state(BrokerStateHashSink& f) const {
    f.s(kSourceAdapterDomain);
    hash_exit_scratch(f, exit_scratch_.get());
    f.u(source_sequence_); f.u(command_ordinal_); f.u(broker_open_epoch_);
    f.i(last_broker_open_ms_); f.u(source_command_sequence_); f.b(host_ != nullptr);
    f.b(config_.process_orders_on_close); f.b(config_.calc_on_order_fills);
    f.d(config_.initial_capital); f.i(config_.default_qty_type); f.d(config_.default_qty_value);
    f.i(config_.pyramiding); f.d(config_.commission_value); f.i(config_.commission_type);
    f.i(config_.slippage); f.d(config_.margin_long); f.d(config_.margin_short);
    f.b(config_.close_entries_rule_any); f.b(config_.src_series_active);
    f.s(staged_.syminfo.ticker); f.s(staged_.syminfo.tickerid); f.s(staged_.syminfo.currency);
    f.s(staged_.syminfo.basecurrency); f.s(staged_.syminfo.type); f.s(staged_.syminfo.timezone);
    f.s(staged_.syminfo.session); f.s(staged_.syminfo.volumetype); f.s(staged_.syminfo.description);
    f.d(staged_.syminfo.mintick); f.d(staged_.syminfo.pointvalue); f.d(staged_.syminfo.qty_step);
    f.s(staged_.chart_timezone); f.d(staged_.account_fx);
    f.u(staged_.account_fx_effective_from_ms.size());
    for (const auto timestamp : staged_.account_fx_effective_from_ms) f.i(timestamp);
    f.u(staged_.account_fx_per_quote.size());
    for (const auto rate : staged_.account_fx_per_quote) f.d(rate);
    f.b(staged_.quantity_grid.has_value()); if (staged_.quantity_grid) f.d(*staged_.quantity_grid);
    // Lane XSYM-D: other symbols' feeds, folded only when installed. Their
    // bars are the projected spec's instrument feeds, which the continuation
    // digest folds, so the keys and sizes are what this fold adds.
    if (!staged_.instrument_feeds.empty()) {
        f.u(staged_.instrument_feeds.size());
        for (const auto& feed : staged_.instrument_feeds) {
            f.s(feed.instrument);
            f.s(feed.tf);
            f.u(feed.bars);
            f.u(feed.columns);
        }
    }
    std::vector<std::string> input_keys;
    for (const auto& pair : staged_.inputs) input_keys.push_back(pair.first);
    std::sort(input_keys.begin(), input_keys.end()); f.u(input_keys.size());
    for (const auto& key : input_keys) { f.s(key); f.s(staged_.inputs.at(key)); }
    // Replay every logical cohort in the same lexicographic order as before,
    // including retired summaries, without materializing an id-sized vector.
    f.u(cohorts_by_id_.size());
    for (const auto& row : cohorts_by_id_.all_sorted()) {
        const auto& key = row.first; const auto& cohort = row.second;
        f.s(key); f.u(cohort.handle.value); f.i(cohort.cycle);
        // v4: the origin roster is append-only; its members fold once each,
        // when appended (OriginRoster::digest).
        f.u(cohort.origins.size()); f.u(cohort.origins.digest());
        hash_native_handle_vector(f, cohort.opened);
        f.u(cohort.erased_opening_sides);
        std::vector<std::uint64_t> live_origin_keys;
        live_origin_keys.reserve(cohort.live_units_by_origin.size());
        for (const auto& row : cohort.live_units_by_origin) live_origin_keys.push_back(row.first);
        std::sort(live_origin_keys.begin(), live_origin_keys.end());
        f.u(live_origin_keys.size());
        for (const auto incarnation : live_origin_keys) {
            f.u(incarnation); f.d(cohort.live_units_by_origin.at(incarnation));
        }
    }
    f.u(cohorts_by_id_.size());
    cohorts_by_id_.for_each_id([&](const auto& key) { f.s(key); });
    // v4: the rows the table still holds, in incarnation order, and what the
    // erased ones left: the high water, how many, and a running digest of
    // their identities (PlacementTable::erased_digest).
    // O1W's terminally filled, non-revivable limits use the same erase/fold
    // path. This deliberately changes values inside the retirement epoch;
    // no hash of a mutable placement is cached.
    PINEFORGE_BROKER_HASH_OBSERVE(placement_, bracket_families_,
                                settled_bracket_families_, settled_bracket_digest_);
    f.u(placement_.high_water()); f.u(placement_.erased_rows()); f.u(placement_.erased_digest());
    f.u(placement_.size());
    for (const auto& row : placement_) {
        PINEFORGE_TAIL_WORK("hash_placements", 1);
        f.u(row.first); hash_placement(f, row.second);
    }
    std::vector<std::uint64_t> live_keys;
    for (const auto& pair : live_by_source_key_) live_keys.push_back(pair.first);
    std::sort(live_keys.begin(), live_keys.end()); f.u(live_keys.size());
    for (const auto key : live_keys) { f.u(key); hash_native_handle(f, live_by_source_key_.at(key)); }
    std::vector<std::uint64_t> bracket_keys;
    for (const auto& pair : bracket_families_) bracket_keys.push_back(pair.first);
    std::sort(bracket_keys.begin(), bracket_keys.end()); f.u(bracket_keys.size());
    for (const auto key : bracket_keys) {
        // v4: the settled prefix as its count and running digest, the
        // working tail in full, and the consumed legs of erased members.
        const auto& family = bracket_families_.at(key);
        PINEFORGE_TAIL_WORK("hash_families", 1);
        f.u(key); f.u(family.settled_count()); f.u(family.settled_digest());
        hash_native_handle_vector(f, family.working_tail());
        f.u(family.consumed_facts().size());
        for (const auto& fact : family.consumed_facts()) { f.u(fact.first); f.u(fact.second); }
    }
    // Immutable wholly settled families are still addressable for a later
    // bracket cancel/re-issue. Their keyed digest changes on archive, restore
    // and erase, so reads cost only active families. This versioned extension
    // is absent when no family is archived; the v4 working-tail fold above
    // keeps its spelling. Run identity is normalized as for native handles.
    if (!settled_bracket_families_.empty()) {
        f.s("pineforge-settled-bracket-families/v1");
        f.u(settled_bracket_families_.size()); f.u(settled_bracket_digest_);
    }
    f.u(pending_bracket_legs_.size());
    for (const auto& leg : pending_bracket_legs_) {
        hash_native_request(f, leg.request); hash_placement(f, leg.snapshot);
        f.s(leg.replacement_key); f.u(leg.family_key);
    }
    f.u(pending_entries_.size());
    for (const auto& entry : pending_entries_) {
        hash_native_request(f, entry.request); hash_placement(f, entry.snapshot);
        f.s(entry.replacement_key);
    }
    f.u(delayed_market_orders_.size());
    for (const auto& order : delayed_market_orders_) {
        hash_native_request(f, order.request); hash_placement(f, order.snapshot);
        f.s(order.replacement_key); f.u(order.release_open_epoch);
        f.b(order.execute_at_open);
    }
    f.u(deferred_open_marketable_sells_.size());
    for (const auto& sell : deferred_open_marketable_sells_) {
        hash_placement(f, sell.snapshot); f.s(sell.replacement_key); f.d(sell.fill_price);
        f.d(sell.path_position); f.b(sell.open_marketable);
    }
    f.u(throttled_reopen_rearm_.size());
    for (const auto& snapshot : throttled_reopen_rearm_) hash_placement(f, snapshot);
    f.i(entry_openings_interval_index_); f.i(entry_openings_this_interval_);
    f.u(pending_same_bar_commands_.size());
    for (const auto& command : pending_same_bar_commands_) {
        hash_native_request(f, command.request); hash_placement(f, command.snapshot);
        f.s(command.replacement_key); f.b(command.opening);
        if (command.unbatched_request && command.unbatched_snapshot) {
            f.u(0x756e626174636865ULL);
            hash_native_request(f, *command.unbatched_request);
            hash_placement(f, *command.unbatched_snapshot);
            const auto& context = command.unbatched_context;
            f.d(context.current); f.d(context.limit_price); f.d(context.stop_price);
            f.b(context.default_sized); f.b(context.priced); f.b(context.reverses);
            f.b(context.opposite_opening_pending); f.b(context.pure_stop_entry);
            f.b(context.coof_market_next_open); f.b(context.coof_priced_next_open);
            f.b(context.coof_market_at_second_extreme); f.b(context.paired_all_in_reentry);
        }
        // Folded only when set, so a batch without an R1 reissue keeps its digest.
        if (command.staged_reversal) {
            f.b(command.staged_reversal); f.d(command.staged_reversal_held_units);
        }
    }
    f.u(source_shadow_pending_.size());
    for (const auto& shadow : source_shadow_pending_) {
        hash_placement(f, shadow.snapshot); f.s(shadow.label);
    }
    f.d(pending_same_bar_close_qty_);
    f.u(pending_relative_exits_.size());
    for (const auto& exit : pending_relative_exits_) {
        f.s(exit.exit_id); f.s(exit.from_entry); f.d(exit.trail_points); f.d(exit.trail_offset);
        f.d(exit.trail_price); f.d(exit.qty_percent); f.s(exit.comment); f.d(exit.qty);
        f.s(exit.oca_name); f.d(exit.profit_ticks); f.d(exit.loss_ticks);
    }
    // R5 lane R4d: folded only while a leg is anchored, so every run that
    // anchors none keeps its established hash.
    if (!anchored_relative_legs_.empty() || anchored_relative_stats_.anchored != 0
        || materializing_parent_.incarnation != 0) {
        f.s("pineforge-source-anchored-legs/v1");
        f.u(anchored_relative_stats_.anchored); f.u(anchored_relative_stats_.adopted);
        f.u(anchored_relative_stats_.withdrawn); f.u(materializing_parent_.incarnation);
        f.i(anchored_cohort_sequence_);
        f.u(anchored_relative_legs_.size());
        for (const auto& leg : anchored_relative_legs_) {
            f.s(leg.exit_id); f.s(leg.from_entry); f.u(static_cast<std::uint64_t>(leg.family));
            f.u(leg.handle.incarnation); f.u(leg.parent.incarnation); f.b(leg.parent_long);
            f.d(leg.operand_ticks); f.d(leg.trail_offset);
            hash_native_request(f, leg.request);
            // hash_native_request predates anchors; an anchored leg's own
            // spelling is folded here.
            if (const auto* anchor = std::get_if<native_order::FromOwnerFill>(
                    &leg.request.anchor)) {
                f.d(anchor->offset); f.b(anchor->ticks);
                f.u(static_cast<std::uint64_t>(anchor->rounding));
            }
            f.b(leg.armed); f.d(leg.installed_level);
        }
    }
    // Folded only while a global exit covers its position entry by entry
    // (ExitBindingRuleSwitches::global_exit_per_entry_levels), so every other
    // run keeps its established hash.
    if (!per_entry_exits_.empty()) {
        f.s("pineforge-source-per-entry-exits/v1");
        f.u(per_entry_exits_.size());
        for (const auto& exit : per_entry_exits_) {
            f.s(exit.exit_id); f.i(exit.cycle); f.b(exit.long_side);
            f.d(exit.profit_ticks); f.d(exit.loss_ticks); f.s(exit.comment);
            f.u(exit.children.size());
            for (const auto& child : exit.children) {
                f.u(child.origin); f.d(child.fill_price); f.d(child.units); f.b(child.consumed);
            }
        }
    }
    f.u(pending_coof_requests_.size());
    for (const auto& pending : pending_coof_requests_) {
        hash_native_request(f, pending.request); hash_placement(f, pending.snapshot);
        f.s(pending.replacement_key); f.b(pending.opening); f.u(pending.family_key);
        f.b(pending.next_open);
    }
    f.u(pending_margin_revivals_.size());
    for (const auto& pending : pending_margin_revivals_) {
        hash_placement(f, pending.snapshot); f.i(pending.decline_bar);
    }
    hash_native_handle_vector(f, live_handles_); hash_native_handle_vector(f, first_open_newborns_);
    // PendingIntentView is a derived read-only alias of live_handles_. Keep
    // its historical fingerprint position and bytes without copying the
    // roster at every mutation boundary.
    hash_native_handle_vector(f, live_handles_);
    // v4: a log no decision reads, folded once per receipt when recorded.
    f.u(dropped_close_count_); f.u(dropped_close_digest_);
    f.u(open_entry_fees_.size());
    for (const auto& fee : open_entry_fees_) {
        hash_native_handle(f, fee.opening); f.s(fee.source_id); f.d(fee.units);
        f.d(fee.nonpercent_fee);
    }
    // v4: the final prefix folded once, the current bar's tail in full.
    const auto trade_prefix = host_
        ? NativeExecutionConsumer::bound(*host_).closed_trade_prefix()
        : NativeExecutionConsumer::ClosedTradePrefix{};
    const std::size_t phase_end = trade_prefix.base + trade_exit_phase_.size();
    const auto* phase_consumer = host_ ? &NativeExecutionConsumer::bound(*host_) : nullptr;
    f.u(phase_end);
    if (phase_consumer && phase_consumer->exit_phases_restarted()) {
        // Preserve the sink-off hash's raw reset marker AND its phase bytes.
        // The bounded transform applies the sealed bytes to this incoming
        // hash, rather than substituting a different digest representation.
        f.u(0); f.u(BrokerStateHashSink{}.h);
        f.h = phase_consumer->replay_sealed_exit_phases(f.h);
        for (const auto phase : trade_exit_phase_) f.u(phase);
    } else {
        if (exit_phase_final_ < trade_prefix.base)
            throw coded<std::logic_error>(RunFailureCode::engine_invariant, {},
                                           "hashed exit-phase fold precedes the sealed prefix");
        f.u(exit_phase_final_); f.u(exit_phase_digest_);
        for (std::size_t index = exit_phase_final_; index < phase_end; ++index)
            f.u(trade_exit_phase_.at(index - trade_prefix.base));
    }
    // The entry-id table orders children once it holds two ids, and its size
    // matters once its peak passes 13 keys; below both it decides nothing.
    if (exit_rank_keys_.size() >= 2 || exit_rank_peak_ > 13 || exit_rank_alt_peak_ > 13) {
        f.u(exit_rank_keys_.size()); f.u(exit_rank_keys_digest_);
        f.u(exit_rank_peak_); f.u(exit_rank_alt_peak_);
    }
    std::vector<std::uint64_t> current_debit_ordinals;
    current_debit_ordinals.reserve(current_debited_applied_ordinals_.size());
    for (const auto ordinal : current_debited_applied_ordinals_) current_debit_ordinals.push_back(ordinal);
    std::sort(current_debit_ordinals.begin(), current_debit_ordinals.end());
    f.u(current_debit_ordinals.size());
    for (const auto ordinal : current_debit_ordinals) f.u(ordinal);
    std::vector<std::uint64_t> intraday_relabel_ordinals;
    intraday_relabel_ordinals.reserve(intraday_loss_relabel_ordinals_.size());
    for (const auto ordinal : intraday_loss_relabel_ordinals_)
        intraday_relabel_ordinals.push_back(ordinal);
    std::sort(intraday_relabel_ordinals.begin(), intraday_relabel_ordinals.end());
    f.u(intraday_relabel_ordinals.size());
    for (const auto ordinal : intraday_relabel_ordinals) f.u(ordinal);
    consumed_partial_exit_cycles_.fold_sorted(f);
    std::vector<std::uint64_t> shadowed_openings(
        bracket_shadowed_openings_.begin(), bracket_shadowed_openings_.end());
    std::sort(shadowed_openings.begin(), shadowed_openings.end());
    f.u(shadowed_openings.size());
    for (const auto opening : shadowed_openings) f.u(opening);
    std::vector<std::string> named_cancel_keys;
    named_cancel_keys.reserve(named_entry_cancel_tokens_.size());
    for (const auto& row : named_entry_cancel_tokens_) named_cancel_keys.push_back(row.first);
    std::sort(named_cancel_keys.begin(), named_cancel_keys.end());
    f.u(named_cancel_keys.size());
    for (const auto& key : named_cancel_keys) {
        const auto& token = named_entry_cancel_tokens_.at(key);
        f.s(key); f.u(token.entry_incarnation); f.u(token.surviving_exit_incarnation);
    }
    // Folded only when an exit's id binding outlived a cancel of its order,
    // so a run that never keeps one folds what it folded before.
    if (!id_bound_exits_.empty()) {
        f.u(0x69646278ULL);
        f.u(id_bound_exits_.size());
        for (const auto& row : id_bound_exits_) {
            f.s(row.exit_id); f.s(row.from_entry);
            f.d(row.levels.limit); f.d(row.levels.stop); f.d(row.levels.trail_points);
            f.d(row.levels.trail_offset); f.d(row.levels.trail_price);
            f.d(row.levels.profit_ticks); f.d(row.levels.loss_ticks);
            f.d(row.qty_percent); f.d(row.qty); f.s(row.comment); f.s(row.oca_name);
        }
    }
    f.u(close_logical_units_.size());
    for (const auto& row : close_logical_units_) { f.s(row.first); f.d(row.second); }
    // The ledger's records fold only when there are two or more: a single
    // record is the one id close_logical_units_ holds, and none is none, so a
    // position of one id folds what it folded before the records existed.
    if (close_ledger_records_.size() > 1) {
        f.u(close_ledger_records_.size());
        for (const auto& record : close_ledger_records_) { f.s(record.id); f.d(record.units); }
    }
    // The pyramiding records, unmerged, kept only under pyramiding above one;
    // folded only while there are any, so every other run keeps its digest.
    if (!pyramiding_records_.empty()) {
        f.s("pineforge-pyramiding-records/v1");
        f.u(pyramiding_records_.size());
        for (const auto& record : pyramiding_records_) {
            f.s(record.id); f.u(record.incarnation); f.d(record.units);
        }
    }
    // The exit tombstones, folded only while there are any.
    if (!exit_tombstones_.empty()) {
        std::vector<SourceId> keys;
        keys.reserve(exit_tombstones_.size());
        for (const auto& row : exit_tombstones_) keys.push_back(row.first);
        std::sort(keys.begin(), keys.end());
        f.s("pineforge-exit-tombstones/v1");
        f.u(keys.size());
        for (const auto& key : keys) { f.s(key); f.u(exit_tombstones_.at(key)); }
    }
    // The retired reservation model's two maps and two per-site maps fold as
    // the empty maps they always were outside a multi-call close site.
    f.u(0); f.u(0); f.u(0); f.u(0);
    f.u(close_batch_callsites_.size());
    for (const auto& row : close_batch_callsites_) {
        const auto& site = row.second;
        f.u(row.first); f.b(site.active); f.u(site.token); f.i(site.calls);
        f.s(site.first_id); f.s(site.id); f.s(site.comment);
        f.d(site.target); f.u(site.queue_sequence);
    }
    f.i(close_batch_bar_); f.u(close_batch_queue_sequence_);
    f.d(close_batch_pending_debt_); f.d(close_batch_admitted_total_);
    f.u(receipt_cursor_);
    f.u(last_applied_ordinal_);
    f.u(terminal_receipt_cursor_);
    f.i(entry_attempt_bar_); f.u(entry_attempts_on_bar_);
    f.b(materializing_relative_);
    f.i(current_position_cycle_);
    f.i(current_position_sign_);
    f.u(next_sequential_group_);
    f.b(source_batch_mutated_);
    f.b(coof_recalc_active_); f.b(coof_first_open_);
    if (coof_open_stop_next_waypoint_) f.u(0x6f70656e6e657874ULL);
    f.u(coof_market_entry_recalc_incarnation_);
    f.u(coof_market_entry_recalc_fill_seq_); f.u(coof_current_fill_seq_);
    // Live only inside a fill recalculation (NaN otherwise); folded then so
    // the idle digest keeps its prior form.
    if (coof_recalc_active_) f.d(coof_fill_cursor_t_);
    if (coof_recalc_active_ && !std::isnan(coof_fill_target_print_)) {
        f.u(0x7761797466ULL); f.d(coof_fill_target_print_);
    }
    if (coof_recalc_active_) f.b(coof_fill_forced_);
    const auto& coof_coord = coof_context_.coordinate;
    f.u(coof_coord.ordinal); f.i(coof_coord.interval_index); f.i(coof_coord.open_ms);
    f.i(coof_coord.eligible_open_ms); f.i(coof_coord.last_traded_close_ms);
    f.i(coof_coord.next_period_open_ms); f.i(coof_coord.next_input_open_ms);
    f.i(coof_coord.effective_time_ms); f.i(coof_coord.source_price_time_ms);
    f.i(static_cast<std::int64_t>(coof_coord.provenance));
    f.i(static_cast<std::int64_t>(coof_coord.path_phase));
    f.i(static_cast<std::int64_t>(coof_coord.completion)); f.i(coof_context_.decision_floor_ms);
    const auto hash_coof_interval = [&](const native_calendar::NativeInterval& interval) {
        f.i(interval.open_ms); f.i(interval.eligible_open_ms); f.i(interval.last_traded_close_ms);
        f.i(interval.next_period_open_ms); f.i(interval.next_input_open_ms);
    };
    hash_coof_interval(coof_context_.input_interval);
    hash_coof_interval(coof_context_.script_interval);
    f.i(coof_context_.sub_index); f.i(coof_context_.sub_count);
    f.b(coof_context_.is_terminal_sub_bar); f.i(coof_context_.sub_bar_open_ms);
    f.i(coof_context_.script_bar_open_ms);
    f.b(coof_context_.driver_statistics.intrabar_path_enabled);
    f.i(coof_context_.driver_statistics.sub_bars_per_script_bar);
    f.i(coof_context_.driver_statistics.samples_per_sub_bar);
    f.u(coof_context_.driver_statistics.sub_bars_processed);
    f.u(coof_context_.driver_statistics.sample_ticks_processed);
    f.d(coof_script_bar_.open); f.d(coof_script_bar_.high); f.d(coof_script_bar_.low);
    f.d(coof_script_bar_.close); f.d(coof_script_bar_.volume); f.i(coof_script_bar_.timestamp);
    f.b(coof_script_bar_valid_);
    // v4: a log no decision reads, folded once per new script bar.
    f.u(pooc_close_basis_count_); f.i(pooc_close_basis_last_bar_); f.u(pooc_close_basis_digest_);
    f.d(pooc_open_basis_); f.i(pooc_open_script_bar_); f.i(close_all_pending_script_bar_);
    f.d(last_fx_rate_); f.i(position_open_script_bar_); f.u(position_open_epoch_); f.i(position_open_bar_index_);
    f.u(static_cast<std::uint64_t>(position_open_phase_));
    f.b(position_open_priced_);
    f.i(last_margin_call_script_bar_); f.i(pooc_close_checkpoint_deferred_ms_);
    // Live only from a close to the next open; folded only then (lane
    // W13-ENG-MARGIN-OPP), so every other run keeps its digest.
    if (close_margin_open_bar_ != std::numeric_limits<std::int64_t>::min())
        f.i(close_margin_open_bar_);
    // A short's lagged follow-up owed at a path point (a process_orders_on_close
    // close owes the next open) and a close's after-script call the orders
    // placed there read at the next open (MarginScheduleSwitches); folded only
    // while set, so every other run keeps its digest.
    if (owed_follow_up_bar_ != std::numeric_limits<std::int64_t>::min()) {
        f.s("pineforge-owed-follow-up/v1");
        f.i(owed_follow_up_bar_); f.i(owed_follow_up_point_); f.d(owed_follow_up_units_);
    }
    if (close_call_after_script_bar_ != std::numeric_limits<std::int64_t>::min()) {
        f.s("pineforge-close-call-after-script/v1");
        f.i(close_call_after_script_bar_); f.d(close_call_after_script_book_);
        f.d(close_call_after_script_units_);
    }
    // An opening checkpoint owed after the exit legs marketable at an open
    // (MarginOpeningSwitches::open_marketable_exit_first); folded only while
    // set, so every other run keeps its digest.
    if (open_exit_first_bar_ != std::numeric_limits<std::int64_t>::min()) {
        f.s("pineforge-open-exit-first/v1");
        f.i(open_exit_first_bar_); f.u(open_exit_first_origin_); f.s(open_exit_first_id_);
        f.d(open_exit_first_fill_);
    }
    // The close-first fact (record_close_first); folded only once a whole
    // close of an open position was recorded, so every other run keeps its
    // digest.
    if (close_first_script_bar_ != std::numeric_limits<std::int64_t>::min()) {
        f.s("pineforge-close-first/v1");
        f.i(close_first_script_bar_); f.i(close_first_side_);
    }
    f.u(last_margin_call_event_ordinal_);
    f.u(last_margin_call_entry_incarnation_); f.i(last_margin_call_position_cycle_);
    f.b(last_margin_call_at_script_close_);
    f.d(last_margin_call_closed_units_); f.d(last_margin_call_remaining_units_);
    // The G + L money mirror (gain_loss_money), over every closed trade
    // whenever this hash is taken (the stored sums plus the trades booked
    // since their last fold), so the digest never depends on when they were
    // folded. Its sum is the engine's netprofit, already in the engine's
    // state, wherever the two agree bit for bit; the sums are folded where
    // they part, so a run they never part on keeps its digest.
    if (source::detail::margin_rule_switches().gain_loss_money) {
        if (const auto* pine = pine_view_of(host_)) {
            // A new run's ledger shorter than the stored fold starts over,
            // as fold_source_money does.
            const std::uint64_t count = pine->closed_trade_count();
            const bool restart = count < source_money_folded_;
            if (count < trade_prefix.base)
                throw coded<std::logic_error>(RunFailureCode::engine_invariant, {},
                                               "hashed source-money rewind precedes the sealed prefix");
            double gains = restart ? trade_prefix.source_gains : source_gains_;
            double losses = restart ? trade_prefix.source_losses : source_losses_;
            for (std::uint64_t index = restart ? trade_prefix.base : source_money_folded_;
                 index < count; ++index) {
                const double profit = pine->closed_trade(index).pnl;
                if (profit > 0.0) gains += profit;
                else losses += profit;
            }
            const double mirrored = gains + losses;
            const double engine_net = pine->net_profit_sum_;
            if (count != 0 && std::memcmp(&mirrored, &engine_net, sizeof(double)) != 0) {
                f.d(gains); f.d(losses); f.u(count);
            }
        }
    }
    f.i(signal_close_mc_event_bar_);
    f.i(signal_close_mc_position_cycle_); f.u(signal_close_mc_entry_incarnation_);
    f.u(signal_close_mc_fill_seq_); f.d(signal_close_mc_before_qty_);
    f.d(signal_close_mc_remaining_qty_); f.i(risk_coof_direct_script_bar_);
    f.u(cap_latest_fill_);
    f.b(source_margin_call_enabled_);
    f.d(policy_script_bar_.open); f.d(policy_script_bar_.high);
    f.d(policy_script_bar_.low); f.d(policy_script_bar_.close);
    f.d(policy_script_bar_.volume); f.i(policy_script_bar_.timestamp);
    f.b(policy_script_bar_valid_);
    // Market-add provenance. It used to ride the
    // generic lot as PyramidEntry::market_pyramid_add; it is adapter state
    // now, so the source extension folds it. Unordered container: sort the
    // keys, exactly like trail_state_at_open_ below.
    std::vector<std::uint64_t> market_add_keys(market_pyramid_adds_.begin(),
                                               market_pyramid_adds_.end());
    std::sort(market_add_keys.begin(), market_add_keys.end());
    f.u(market_add_keys.size());
    for (const auto key : market_add_keys) f.u(key);
    std::vector<std::uint64_t> trail_open_keys;
    trail_open_keys.reserve(trail_state_at_open_.size());
    for (const auto& row : trail_state_at_open_) trail_open_keys.push_back(row.first);
    std::sort(trail_open_keys.begin(), trail_open_keys.end());
    f.u(trail_open_keys.size());
    for (const auto key : trail_open_keys) {
        const auto& state = trail_state_at_open_.at(key);
        f.u(key); f.b(state.activated); f.d(state.best_price);
        f.d(state.current_level); f.u(state.activation_ordinal);
    }
    // The trails re-priced for the bar in flight (carried trail tick reach),
    // empty between bars; folded only while set, so every run without one
    // keeps its digest.
    if (!carried_trail_retunes_.empty()) {
        f.u(carried_trail_retunes_.size());
        for (const auto& [handle, distance] : carried_trail_retunes_) {
            f.u(handle.incarnation); f.d(distance);
        }
    }
    f.b(stream_mode_);
    f.i(day_ledger_.current_day); f.i(day_ledger_.last_loss_day); f.i(day_ledger_.consecutive_loss_days);
    f.i(day_ledger_.intraday_loss_day); f.d(day_ledger_.intraday_start_equity);
    f.d(day_ledger_.intraday_realized); f.u(day_ledger_.observed_applied_ordinal);
    // Decision state only under an intraday loss rule (lane W8A-SIGSTATE-1
    // R4); folded only then, so every run without one keeps its digest.
    if (risk_.max_intraday_loss > 0.0) f.i(day_ledger_.intraday_start_script_bar);
    f.i(risk_.direction); f.i(risk_.max_cons_loss_days); f.d(risk_.max_drawdown);
    f.b(risk_.max_drawdown_percent); f.d(risk_.max_intraday_loss);
    f.b(risk_.max_intraday_loss_percent); f.d(risk_.max_position_size); f.b(risk_.halted);
    f.d(risk_.observed_peak_equity); f.d(risk_.observed_max_drawdown);
    f.i(risk_.intraday_block_day); f.b(risk_.intraday_cancel_pending);
    hash_short_seed_plan(f, short_seed_);
    hash_short_seed_plan(f, pending_short_seed_.plan);
    f.u(pending_short_seed_.expected_open_epoch); f.b(pending_short_seed_.ready);
    hash_native_handle(f, short_seed_long_candidate_);
    f.i(last_bar_dual_entry_path_); f.i(last_bar_dual_entry_script_open_ms_);
    f.b(pending_view_.owner_ != nullptr);
    f.i(static_cast<std::int64_t>(cap.attachment())); f.i(cap.configuration().limit);
    f.b(cap.configuration().skip_noop_market); f.b(cap.configuration().defer_pooc_close);
    f.b(cap.configuration().count_pooc_full_close);
    // Which switches the host declared (rule CAP-ON) decides only an active
    // cap's count, so it folds only while one is active.
    if (cap.active()) f.u(cap.declared());
    const auto& cap_budget = cap.budget();
    f.b(cap_budget.day().has_value());
    if (cap_budget.day()) f.i(cap_budget.day()->key);
    f.i(cap_budget.charged_slots()); f.b(cap_budget.latched());
    f.b(cap_budget.transfer().has_value());
    if (const auto& transfer = cap_budget.transfer()) {
        f.i(transfer->day.key); f.u(transfer->close_fill);
        f.i(transfer->source_bar); f.u(transfer->inheritor);
    }
    f.b(cap.due_cause().has_value());
    if (const auto& due = cap.due_cause()) {
        f.u(due->action_id); f.i(due->charged_day.key); f.i(due->charged_slots);
        f.i(due->trigger_bar); f.u(due->trigger_order);
    }
    f.u(cap.next_action()); f.b(priority.attached());
    f.b(priority.retained_parent_first());
    // The eager boundary fold and this query share the original v4 byte loop.
    // A retired prefix substitutes for its events, without adding hash bytes.
    fold_admission_events();
    f.u(admission_journal.sequence_frontier());
    f.u(admission_events_folded_); f.u(admission_events_digest_);
}
 
void source::PineExecutionAdapter::fold_admission_events() const {
    const auto folded = admission_journal.folded_window();
    admission_events_folded_ = folded.count;
    admission_events_digest_ = folded.digest;
    admission_events_last_ = folded.last_sequence;
}
 
std::uint64_t source::PineExecutionAdapter::fold_dropped_close(
        std::uint64_t digest, const DroppedCloseReceipt& receipt) noexcept {
    BrokerStateHashSink f;
    f.h = digest;
    f.s(receipt.source_id); f.s(receipt.comment); f.d(receipt.qty);
    f.d(receipt.qty_percent); f.b(receipt.immediately); f.u(receipt.callsite_token);
    f.u(receipt.command_ordinal);
    return f.h;
}
 
std::uint64_t source::PineExecutionAdapter::fold_exit_phase(std::uint64_t digest,
                                                            std::uint8_t phase) noexcept {
    BrokerStateHashSink f;
    f.h = digest;
    f.u(phase);
    return f.h;
}
 
void source::PineScheduler::fold_consumed_digests() const noexcept {
    // ab9714be test_live_state_hash_recording: a shorter run is a hash prefix
    // of the same longer feed.  Retained future input is provider transport,
    // not broker continuation state, so fold only the consumed source prefix.
    // v3 (R5 lane V19-E): the consumed prefix as a running digest, each bar
    // folded once, the first time a read passes it; a prefix run still reads
    // like the full run up to its last bar.
    const std::size_t consumed = std::min(
        retained_n_, static_cast<std::size_t>(std::max(source_bar_count_, 0)));
    const auto extend = [](std::size_t& folded, std::uint64_t& digest, std::size_t upto,
                           const auto& element) {
        if (folded > upto) {
            folded = 0;
            digest = 1469598103934665603ULL;
        }
        for (; folded < upto; ++folded) {
            BrokerStateHashSink g;
            g.h = digest;
            element(g, folded);
            digest = g.h;
        }
    };
    extend(consumed_bars_folded_, consumed_bars_digest_, consumed,
        [&](BrokerStateHashSink& g, std::size_t index) {
            // owned is just the stream's backing for bars, not an additional
            // broker fact: either storage folds the same bar bytes once.
            const auto& bar = retained_.is_stream ? retained_.owned[index] : retained_.bars[index];
            g.d(bar.open); g.d(bar.high); g.d(bar.low); g.d(bar.close); g.d(bar.volume);
            g.i(bar.timestamp);
        });
    const std::size_t completion_prefix = std::min(input_script_size_, consumed);
    // Fold both flags here, before release_batch_input clears the caller span.
    // The independent cursor restarts with both digests for a shorter prefix.
    if (consumed_completes_folded_ > completion_prefix
        || consumed_boundaries_folded_ > completion_prefix) {
        consumed_completes_folded_ = 0;
        consumed_completes_digest_ = 1469598103934665603ULL;
        consumed_boundaries_folded_ = 0;
        consumed_boundaries_digest_ = 1469598103934665603ULL;
        if (consumed_script_cursor_) consumed_script_cursor_->reset();
    }
    extend(consumed_completes_folded_, consumed_completes_digest_, completion_prefix,
        [&](BrokerStateHashSink& g, std::size_t index) {
            const auto flags = consumed_script_cursor_
                ? consumed_script_cursor_->next(retained_.bars[index])
                : ScriptBucketCursor::Flags{};
            g.u(flags.complete);
            BrokerStateHashSink boundary;
            boundary.h = consumed_boundaries_digest_;
            boundary.u(flags.boundary);
            consumed_boundaries_digest_ = boundary.h;
            consumed_boundaries_folded_ = index + 1;
        });
}
 
void source::PineScheduler::hash_state(BrokerStateHashSink& f) const {
    fold_consumed_digests();
    // retained_n_ survives detachment. Using the now-empty span's size here
    // would rewind all three digests and change every post-run hash (F8).
    const std::size_t consumed = std::min(
        retained_n_, static_cast<std::size_t>(std::max(source_bar_count_, 0)));
    f.s("pineforge-pine-scheduler/v3");
    f.u(consumed); f.u(consumed_bars_digest_);
    f.s(retained_.input_tf); f.s(retained_.script_tf); f.b(retained_.bar_magnifier);
    f.i(retained_.magnifier_samples); f.i(static_cast<std::int64_t>(retained_.distribution));
    f.b(retained_.volume_weighted); f.i(retained_.volume_weighted_min_samples);
    f.i(retained_.volume_weighted_max_samples); f.b(retained_.is_stream); f.i(retained_.warmup_n);
    f.b(retained_.simple_run);
    f.i(language_.pos_view_freeze_bar_); f.i(static_cast<std::int64_t>(language_.pos_view_frozen_side_));
    f.d(language_.pos_view_frozen_qty_); hash_str_double_map(f, language_.pos_view_frozen_entry_qty_);
    f.b(language_._src_series_active_); hash_source_series(f, language_._src_open_);
    hash_source_series(f, language_._src_high_); hash_source_series(f, language_._src_low_);
    hash_source_series(f, language_._src_close_); hash_source_series(f, language_._src_volume_);
    hash_source_series(f, language_._src_hl2_); hash_source_series(f, language_._src_hlc3_);
    hash_source_series(f, language_._src_ohlc4_); hash_source_series(f, language_._src_hlcc4_);
    f.d(language_.prev_chart_close_); f.d(language_.last_chart_close_); f.i(language_.bar_index_offset_);
    f.b(language_.is_first_tick_); f.b(language_.is_last_tick_); f.b(language_.history_slot_is_new_);
    f.b(language_.coof_checkpoint_contains_current_bar_);
    hash_source_series(f, language_.coof_checkpoint_src_open_);
    hash_source_series(f, language_.coof_checkpoint_src_high_);
    hash_source_series(f, language_.coof_checkpoint_src_low_);
    hash_source_series(f, language_.coof_checkpoint_src_close_);
    hash_source_series(f, language_.coof_checkpoint_src_volume_);
    hash_source_series(f, language_.coof_checkpoint_src_hl2_);
    hash_source_series(f, language_.coof_checkpoint_src_hlc3_);
    hash_source_series(f, language_.coof_checkpoint_src_ohlc4_);
    hash_source_series(f, language_.coof_checkpoint_src_hlcc4_);
    f.d(language_.coof_checkpoint_prev_chart_close_); f.d(language_.coof_checkpoint_last_chart_close_);
    f.i(current_script_open_ms_);
    f.d(current_script_bar_.open); f.d(current_script_bar_.high); f.d(current_script_bar_.low);
    f.d(current_script_bar_.close); f.d(current_script_bar_.volume); f.i(current_script_bar_.timestamp);
    f.b(current_script_bar_valid_); f.b(saw_open_fill_); f.i(open_point_fills_); f.i(source_bar_count_);
    f.b(expected_source_bars_ >= source_bar_count_);
    f.u(applied_cursor_); f.i(coof_callback_script_open_);
    f.i(last_published_script_open_ms_);
    f.i(prior_input_script_open_ms_);
    f.i(awaiting_legacy_script_open_ms_);
    f.i(last_stream_input_open_ms_);
    const std::size_t completion_prefix = std::min(input_script_size_, consumed);
    f.u(completion_prefix); f.u(consumed_completes_digest_);
    const std::size_t boundary_prefix = std::min(input_script_size_, consumed);
    f.u(boundary_prefix); f.u(consumed_boundaries_digest_);
    f.b(uses_aux_security_feed_);
    if (retained_.is_stream) f.b(input_is_observed_ticks_);
    if (retained_.is_stream) {
        f.u(confirmed_input_next_);
        f.u(confirmed_input_count_);
        for (const auto& bar : confirmed_input_bars_) {
            f.d(bar.open); f.d(bar.high); f.d(bar.low); f.d(bar.close); f.d(bar.volume);
            f.i(bar.timestamp);
        }
        f.u(confirmed_open_input_next_);
        f.u(confirmed_open_input_count_);
        for (const auto& bar : confirmed_open_input_bars_) {
            f.d(bar.open); f.d(bar.high); f.d(bar.low); f.d(bar.close); f.d(bar.volume);
            f.i(bar.timestamp);
        }
    }
    f.d(deferred_boundary_input_.bar.open); f.d(deferred_boundary_input_.bar.high);
    f.d(deferred_boundary_input_.bar.low); f.d(deferred_boundary_input_.bar.close);
    f.d(deferred_boundary_input_.bar.volume); f.i(deferred_boundary_input_.bar.timestamp);
    f.i(deferred_boundary_input_.next_input_ms);
    f.i(deferred_boundary_input_.prior_script_open_ms);
    f.b(deferred_boundary_input_.calling_bar_complete);
    f.b(deferred_boundary_input_.all_security_states);
    f.b(deferred_boundary_input_.active);
}
 
namespace {
 
// The selected window's pre-roll capture (src/source/pine_intent_state.hpp), in
// its own domain and folded only while a selected run parks the state, so every
// run without one keeps the hash it had. Each payload folds its own argument
// fields in declaration order; the configuration and point folds follow the ones
// the adapter and the kernel already use.
constexpr char kSourceIntentDomain[] = "pineforge-source-intent/v1";
 
void hash_intent_config(BrokerStateHashSink& f, const source::PineStrategyConfig& config) {
    f.b(config.process_orders_on_close); f.b(config.calc_on_order_fills);
    f.d(config.initial_capital); f.i(config.default_qty_type); f.d(config.default_qty_value);
    f.i(config.pyramiding); f.d(config.commission_value); f.i(config.commission_type);
    f.i(config.slippage); f.d(config.margin_long); f.d(config.margin_short);
    f.b(config.close_entries_rule_any); f.b(config.src_series_active);
}
 
// The saved risk configuration alone: the seven setter fields with their
// sticky percent flags, and the cap's three configuration fields.
void hash_intent_risk(BrokerStateHashSink& f, const source::PineReplayRiskConfig& risk) {
    f.i(risk.direction); f.i(risk.max_cons_loss_days); f.d(risk.max_drawdown);
    f.b(risk.max_drawdown_percent); f.d(risk.max_intraday_loss);
    f.b(risk.max_intraday_loss_percent); f.d(risk.max_position_size);
    f.i(static_cast<std::int64_t>(risk.cap_attachment)); f.i(risk.cap_configuration.limit);
    f.b(risk.cap_configuration.skip_noop_market); f.b(risk.cap_configuration.defer_pooc_close);
    f.b(risk.cap_configuration.count_pooc_full_close); f.u(risk.cap_declared);
}
 
void hash_intent_interval(BrokerStateHashSink& f, const native_calendar::NativeInterval& interval) {
    f.i(interval.open_ms); f.i(interval.eligible_open_ms); f.i(interval.last_traded_close_ms);
    f.i(interval.next_period_open_ms); f.i(interval.next_input_open_ms);
}
 
// A retained point, every value field in declaration order. The four session-day
// facts are presentation to the kernel's own digests, but the replay reads the
// retained point back, so the retained value folds whole.
void hash_intent_point(BrokerStateHashSink& f, const NativeCurrentPointView& point) {
    const NativeCoordinate& c = point.decision.coordinate;
    f.u(c.ordinal); f.i(c.interval_index); f.i(c.input_interval_index); f.i(c.open_ms);
    f.i(c.eligible_open_ms); f.i(c.last_traded_close_ms); f.i(c.next_period_open_ms);
    f.i(c.next_input_open_ms); f.i(c.effective_time_ms); f.i(c.source_price_time_ms);
    f.u(static_cast<std::uint64_t>(c.provenance)); f.u(static_cast<std::uint64_t>(c.path_phase));
    f.u(static_cast<std::uint64_t>(c.completion));
    f.i(point.decision.decision_floor_ms);
    hash_intent_interval(f, point.decision.input_interval);
    hash_intent_interval(f, point.decision.script_interval);
    f.i(point.decision.sub_index); f.i(point.decision.sub_count);
    f.b(point.decision.is_terminal_sub_bar);
    f.b(point.decision.in_session); f.b(point.decision.opens_session_day);
    f.b(point.decision.closes_session_day); f.b(point.decision.closes_session_day_open_ended);
    f.i(point.decision.sub_bar_open_ms); f.i(point.decision.script_bar_open_ms);
    const NativeDriverStatistics& statistics = point.decision.driver_statistics;
    f.b(statistics.intrabar_path_enabled); f.i(statistics.sub_bars_per_script_bar);
    f.i(statistics.samples_per_sub_bar); f.u(statistics.sub_bars_processed);
    f.u(statistics.sample_ticks_processed);
    f.d(point.price); f.u(static_cast<std::uint64_t>(point.quote_kind));
    f.u(point.quote_origin_ordinal);
}
 
void hash_intent_payload(BrokerStateHashSink& f, const source::detail::IntentEntry& v) {
    f.s(v.id); f.b(v.is_long); f.d(v.limit_price); f.d(v.stop_price); f.d(v.qty);
    f.s(v.comment); f.s(v.oca_name); f.i(v.oca_type); f.i(v.qty_type);
}
void hash_intent_payload(BrokerStateHashSink& f, const source::detail::IntentClose& v) {
    f.s(v.id); f.s(v.comment); f.d(v.qty); f.d(v.qty_percent); f.b(v.immediately);
}
void hash_intent_payload(BrokerStateHashSink& f, const source::detail::IntentCloseToken& v) {
    f.s(v.id); f.s(v.comment); f.d(v.qty); f.d(v.qty_percent); f.b(v.immediately);
    f.u(v.callsite_token);
}
void hash_intent_payload(BrokerStateHashSink&, const source::detail::IntentCloseAll&) {}
void hash_intent_payload(BrokerStateHashSink& f, const source::detail::IntentExit& v) {
    f.s(v.id); f.s(v.from_entry); f.d(v.limit_price); f.d(v.stop_price);
    f.d(v.trail_points); f.d(v.trail_offset); f.d(v.trail_price); f.d(v.qty_percent);
    f.s(v.comment); f.d(v.qty); f.s(v.oca_name); f.d(v.profit_ticks); f.d(v.loss_ticks);
}
void hash_intent_payload(BrokerStateHashSink& f,
                         const source::detail::IntentExitCancelBracket& v) {
    f.s(v.exit_id); f.s(v.from_entry); f.s(v.comment);
}
void hash_intent_payload(BrokerStateHashSink& f, const source::detail::IntentCancel& v) {
    f.s(v.id);
}
void hash_intent_payload(BrokerStateHashSink&, const source::detail::IntentCancelAll&) {}
void hash_intent_payload(BrokerStateHashSink& f, const source::detail::IntentOrder& v) {
    f.s(v.id); f.b(v.is_long); f.d(v.qty); f.d(v.limit_price); f.d(v.stop_price);
    f.s(v.oca_name); f.i(v.oca_type);
}
void hash_intent_payload(BrokerStateHashSink& f,
                         const source::detail::IntentRiskDirection& v) {
    f.i(v.direction);
}
void hash_intent_payload(BrokerStateHashSink& f,
                         const source::detail::IntentRiskMaxConsLossDays& v) {
    f.i(v.value);
}
void hash_intent_payload(BrokerStateHashSink& f,
                         const source::detail::IntentRiskMaxDrawdown& v) {
    f.d(v.value); f.b(v.percent);
}
void hash_intent_payload(BrokerStateHashSink& f,
                         const source::detail::IntentRiskMaxIntradayLoss& v) {
    f.d(v.value); f.b(v.percent);
}
void hash_intent_payload(BrokerStateHashSink& f,
                         const source::detail::IntentRiskMaxIntradayFilledOrders& v) {
    f.i(v.limit);
}
void hash_intent_payload(BrokerStateHashSink& f,
                         const source::detail::IntentRiskMaxPositionSize& v) {
    f.d(v.value);
}
 
void hash_selected_intents(BrokerStateHashSink& f, const source::detail::PineIntentState& state) {
    f.s(kSourceIntentDomain);
    // The native run follows the source identity normalization of a handle's run
    // (hash_source_run_identity): it protects a stale state and is not a
    // source-visible fact, so it is never a per-run nonce. The owner, the kind
    // tag, the bracket flag and the replay pointer are addresses or transient and
    // are not folded.
    f.u(0);
    f.b(state.has_evaluation); f.b(state.replayed); f.i(state.label_ms);
    hash_intent_config(f, state.start_config); hash_intent_config(f, state.end_config);
    hash_intent_risk(f, state.start_risk); hash_intent_risk(f, state.end_risk);
    hash_intent_point(f, state.terminal_point);
    f.u(state.rows.size());
    for (const source::detail::PineIntentRow& row : state.rows) {
        f.u(row.payload.index());
        std::visit([&f](const auto& payload) { hash_intent_payload(f, payload); }, row.payload);
        hash_intent_config(f, row.config);
        hash_intent_point(f, row.point);
    }
}
 
} // namespace
 
void source::PineStrategyHost::hash_host_extension(BrokerStateHashSink& f) const {
    f.s(kSourceAdapterDomain);
    if (!trade_scratch_)
        NativeExecutionConsumer::bound(*this).rethrow_released_command_scratch_failure(
            NativeExecutionConsumer::CommandScratchBinding::Trade);
    const auto released = NativeExecutionConsumer::bound(*this).released_command_scratch(
        NativeExecutionConsumer::CommandScratchBinding::Trade);
    if (!trade_scratch_ && released) {
        // Preserve the live store's exact presence/digest bytes after its
        // run-owned backing is released. This is no additional hash field.
        f.b(true);
        hash_scratch_digest(f, released->status, released->digest);
    } else {
        hash_scratch_store(f, trade_scratch_);
    }
    f.b(config_.process_orders_on_close); f.b(config_.calc_on_order_fills);
    f.d(config_.initial_capital); f.i(config_.default_qty_type); f.d(config_.default_qty_value);
    f.i(config_.pyramiding); f.d(config_.commission_value); f.i(config_.commission_type);
    f.i(config_.slippage); f.d(config_.margin_long); f.d(config_.margin_short);
    f.b(config_.close_entries_rule_any); f.b(config_.src_series_active);
    f.d(override_.initial_capital); f.d(override_.commission_value); f.d(override_.default_qty_value);
    f.i(override_.pyramiding); f.i(override_.slippage); f.i(override_.commission_type);
    f.i(override_.default_qty_type); f.i(override_.process_orders_on_close);
    f.i(override_.calc_on_order_fills); f.i(override_.close_entries_rule);
    f.i(source_bar_index_); f.u(source_callback_count_);
    f.b(source_configuration_captured_); f.b(source_prepare_failed_);
#ifdef PINEFORGE_HAS_AUX_SECURITY_FEED_V1
    // v4: the auxiliary feed and its chart ranges never change during a run;
    // their digest is taken when they are set (refresh_aux_security_digest:
    // aux_security_bars_, aux_security_input_tf_, aux_security_chart_begin_,
    // aux_security_chart_end_).
    f.u(aux_security_bars_.size()); f.s(aux_security_input_tf_);
    f.u(aux_security_chart_begin_.size()); f.u(aux_security_chart_end_.size());
    f.u(aux_security_digest_);
#endif
    // Per-site request.security semantics, in sec_id order. Folded under
    // their own domain and only when a site is registered, so a run without
    // request.security keeps the hash it had.
    if (!pine_security_states_.empty()) {
        f.s(kSourceSecurityDomain);
        f.u(pine_security_states_.size());
        for (const auto& [sec_id, pine] : pine_security_states_) {
            f.i(sec_id);
            f.b(pine.lookahead_on);
            f.b(pine.gaps_on);
            f.i(pine.publish_gate_tf_seconds);
            f.b(pine.calling_close_completes_partial);
            // Folded only when set: every site that loads bars keeps the
            // digest it had.
            if (pine.no_loaded_bars) f.b(pine.no_loaded_bars);
            f.b(pine.heikinashi);
            f.d(pine.ha_prev_open);
            f.d(pine.ha_prev_close);
            f.b(pine.ha_seeded);
            // The projection in hand, not the prepared list: later buckets
            // are a function of input the run has not consumed yet, and a
            // prefix run must hash like the full run up to its last bar.
            f.b(!pine.historical_projections.empty());
            if (!pine.historical_projections.empty()) {
                f.u(pine.historical_projection_cursor);
                f.b(pine.historical_projection_dispatched);
                const auto& projection = pine.historical_projections[std::min(
                    pine.historical_projection_cursor,
                    pine.historical_projections.size() - 1)];
                f.d(projection.bar.open); f.d(projection.bar.high);
                f.d(projection.bar.low); f.d(projection.bar.close);
                f.d(projection.bar.volume); f.i(projection.bar.timestamp);
                f.i(projection.first_child_ms);
                f.b(projection.is_complete);
            }
            f.b(pine.lower_tf_requested);
            f.b(pine.lower_tf_emulation);
            f.i(pine.lower_tf_ratio);
            f.i(pine.lower_tf_seconds);
            f.b(pine.lower_tf_array_requested);
            f.i(pine.lower_tf_sub_bar_index);
            f.b(pine.lower_tf_use_input);
            f.i(pine.lower_tf_input_aggregation_ratio);
            f.u(pine.lower_tf_input_buffer.size());
            for (const Bar& bar : pine.lower_tf_input_buffer) {
                f.d(bar.open); f.d(bar.high); f.d(bar.low); f.d(bar.close);
                f.d(bar.volume); f.i(bar.timestamp);
            }
            f.b(pine.calling_open_latches_first);
            f.b(pine.first_bucket_published);
            f.i(pine.slice_open_label);
            f.i(pine.last_published_label);
            f.u(pine.deferred_aux.size());
            for (const auto& held : pine.deferred_aux) {
                f.d(held.bar.open); f.d(held.bar.high); f.d(held.bar.low);
                f.d(held.bar.close); f.d(held.bar.volume); f.i(held.bar.timestamp);
                f.i(held.next_input_ms);
                f.b(held.calling_bar_complete);
            }
        }
    }
    // Lane XSYM-D: the request sites of other symbols -- registration and
    // per-run cursors -- and the request data the run was handed (the feeds'
    // keys and sizes, the facts and the recorded series, through the digest
    // the setters keep; the feeds' bars are the projected spec's instrument
    // feeds, which the continuation digest folds). Under their own domain and
    // only when present, so every run without them keeps the hash it had.
    if (!foreign_security_sites_.empty() || !symbol_feeds_.empty()
        || !symbol_facts_.empty() || !recorded_series_.empty()) {
        f.s(kSourceSymbolDataDomain);
        f.u(symbol_feeds_.size());
        f.u(symbol_facts_.size());
        f.u(recorded_series_.size());
        f.u(symbol_data_digest_);
        f.i(foreign_input_index_);
        f.u(foreign_security_series_.size());
        for (const int sec_id : foreign_security_series_) f.i(sec_id);
        f.u(foreign_security_sites_.size());
        for (const auto& [sec_id, site] : foreign_security_sites_) {
            f.i(sec_id);
            f.s(site.symbol);
            f.s(site.requested_tf);
            f.b(site.lookahead);
            f.b(site.gaps);
            f.b(site.ignore_invalid);
            f.s(site.tf);
            f.b(site.invalid);
            f.i(site.subscription);
            f.u(site.feed);
            f.i(site.delivered);
            f.i(site.close_ms);
            f.i(site.last_input);
            f.i(site.handed);
        }
    }
    // The selected window's pre-roll capture, only while a selected run parks it
    // (selected_intents allocates nothing and is null for every other run).
    if (const source::detail::PineIntentState* const intents = detail::selected_intents(*this))
        hash_selected_intents(f, *intents);
    adapter_.hash_state(f); scheduler_.hash_state(f);
}
 
} // namespace pineforge