pine_strategy_host.cpp

Go to the documentation of this file.

#include <pineforge/source/pine_strategy_host.hpp>
#include <pineforge/checked_settings.hpp>
#include <pineforge/query_refusal.hpp>
#include <pineforge/source/exit_scratch.hpp>
#include <pineforge/source/pine_policy_support.hpp>
#include <pineforge/ta.hpp>
#include <pineforge/timeframe.hpp>
#include "multi_day_periods.hpp"
#include "../engine_internal.hpp"
#include "../report_replay.hpp"
#include "../timezone.hpp"
#include "../native_execution_consumer.hpp"
#include "../scratch_table_binders.hpp"
#include "pine_host_reads.hpp"
#include "pine_intent_state.hpp"
#include "pine_quiet_bar.hpp"
#include "pine_value_admit_doors.hpp"
#include "pine_run_failure.hpp"
#include "pine_value_admit_decl.hpp"
#include "pine_value_admit_risk.hpp"
#include "pine_command_scratch_production.hpp"
#include "exit_scratch_versions_production.hpp"
 
#include <algorithm>
#include <array>
#include <cerrno>
#include <cmath>
#include <cstring>
#include <cstdlib>
#include <ctime>
#include <iterator>
#include <limits>
#include <memory>
#include <numeric>
#include <optional>
#include <stdexcept>
#include <string>
#include <system_error>
#include <utility>
#include <variant>
#include <unistd.h>
 
namespace pineforge {
using namespace source;
 
namespace {
 
::pineforge::detail::ScratchStoreConfig command_scratch_config(
        std::size_t page_size, ::pineforge::detail::ScratchKeyOrder order) noexcept {
    return {2, page_size, 1, 1, order,
            ::pineforge::detail::ScratchOpenMode::create_empty};
}
 
std::size_t command_scratch_page_size() {
    errno = 0;
    const long size = ::sysconf(_SC_PAGESIZE);
    if (size <= 0)
        throw coded<std::system_error>(RunFailureCode::scratch_io_failed, {},
            std::system_error(errno ? errno : EINVAL, std::generic_category(),
                              "command scratch platform page size"));
    return static_cast<std::size_t>(size);
}
 
struct SourceExitScratchRun {
    // Backings precede their stores/context, so borrowers die first.
    // The native consumer retains this run through source-adapter destruction.
    // Geometry is fixed independently of the bar and order populations.
    const std::size_t page_size = command_scratch_page_size();
    std::unique_ptr<::pineforge::detail::ScratchBacking> exit_backing;
    ::pineforge::detail::CommandScratchStore exit{
        command_scratch_config(page_size, source::detail::exit_scratch_key_order()),
        exit_backing.get()};
    source::detail::ExitScratchContext exit_context{exit, {}};
    source::detail::ExitScratchVersionsProduction exit_versions{exit};
 
    explicit SourceExitScratchRun(CacheCapacity capacity)
        : exit_backing(source::detail::make_command_scratch_backing(page_size, capacity)) {
        if (exit.status() != ::pineforge::detail::ScratchStatus::ok) {
            if (exit.status() == ::pineforge::detail::ScratchStatus::io_error) // the store carries a refused workspace allocation
                ::pineforge::detail::rethrow_carried_scratch_failure(exit.carried_failure() ? exit.carried_failure() : exit_backing->carried_failure());
            const auto code = exit.status() == ::pineforge::detail::ScratchStatus::io_error
                ? RunFailureCode::scratch_io_failed
                : RunFailureCode::engine_invariant;
            throw coded<std::logic_error>(code, {},
                "run command scratch store failed to open");
        }
        // This head exposes owning NativeOpenLot rows and snapshot readers,
        // rather than the reference tree's NativeOpenLotView visitor API.
        exit_context.versions = &exit_versions;
        exit_context.lots = [](const NativeStrategyHost& host, double mark, void* context,
                               bool (*visit)(void*, const NativeOpenLot&)) {
            for (const auto& lot : host.native_open_lots(mark))
                if (!visit(context, lot)) return ::pineforge::detail::ScratchStatus::cancelled;
            return ::pineforge::detail::ScratchStatus::ok;
        };
        exit_context.working = [](const NativeStrategyHost& host, void* context,
                                  bool (*visit)(void*, const NativeWorkingRequest&)) {
            for (const auto& request : host.native_working_requests())
                if (!visit(context, request)) return ::pineforge::detail::ScratchStatus::cancelled;
            return ::pineforge::detail::ScratchStatus::ok;
        };
    }
};
 
using TradeScratchKey = std::array<std::uint64_t, 4>;
using TradeScratchHeader = std::array<std::uint64_t, 23>;
enum : std::uint64_t {
    scratch_incarnation, scratch_row, scratch_sort, scratch_permutation
};
 
::pineforge::detail::ScratchBytes scratch_bytes(const void* data, std::size_t size) noexcept {
    return {static_cast<const std::byte*>(data), size};
}
 
::pineforge::detail::ScratchBuffer scratch_buffer(void* data, std::size_t size) noexcept {
    return {static_cast<std::byte*>(data), size};
}
 
::pineforge::detail::ScratchStatus compare_trade_scratch(
        void*, const ::pineforge::detail::ScratchByteSource& a, const ::pineforge::detail::ScratchByteSource& b,
        int& order) noexcept {
    TradeScratchKey ka{}, kb{};
    if (a.size() != sizeof ka || b.size() != sizeof kb)
        return ::pineforge::detail::ScratchStatus::programming_error;
    auto status = a.read(0, scratch_buffer(ka.data(), sizeof ka));
    if (status != ::pineforge::detail::ScratchStatus::ok) return status;
    status = b.read(0, scratch_buffer(kb.data(), sizeof kb));
    if (status != ::pineforge::detail::ScratchStatus::ok) return status;
    order = ka < kb ? -1 : kb < ka ? 1 : 0;
    return ::pineforge::detail::ScratchStatus::ok;
}
 
void require_trade_scratch(::pineforge::detail::ScratchStatus status,
                           const ::pineforge::detail::ScratchReadback& store) {
    if (status == ::pineforge::detail::ScratchStatus::ok) return;
    if (status == ::pineforge::detail::ScratchStatus::io_error)
        ::pineforge::detail::rethrow_carried_scratch_failure(store.carried_failure());
    const auto code = status == ::pineforge::detail::ScratchStatus::io_error
        ? RunFailureCode::scratch_io_failed
        : RunFailureCode::engine_invariant;
    throw coded<std::logic_error>(code, {},
        "trade scratch contract operation failed");
}
 
// These records are temporary, including on success. Rollback restores the
// empty store and its handle counter; no scratch state survives a callback.
class TradeScratchTransaction {
public:
    explicit TradeScratchTransaction(::pineforge::detail::CommandScratchStore& store) : store_(store) {
        require_trade_scratch(store_.begin(), store_);
    }
    ~TradeScratchTransaction() {
        if (store_.transaction_active()) (void)store_.rollback();
    }
    void finish() { require_trade_scratch(store_.rollback(), store_); }
private:
    ::pineforge::detail::CommandScratchStore& store_;
};
 
::pineforge::detail::ScratchHandle scratch_seek(const ::pineforge::detail::CommandScratchStore& store,
                                  const TradeScratchKey& key) {
    ::pineforge::detail::ScratchSpanSource probe(scratch_bytes(key.data(), sizeof key));
    ::pineforge::detail::ScratchHandle handle = ::pineforge::detail::scratch_no_handle;
    const auto status = store.lower_bound(probe, handle);
    if (status != ::pineforge::detail::ScratchStatus::end) require_trade_scratch(status, store);
    return handle;
}
 
TradeScratchKey scratch_key(const ::pineforge::detail::ScratchReadback& store,
                           ::pineforge::detail::ScratchHandle handle) {
    TradeScratchKey key{};
    require_trade_scratch(store.read(handle, ::pineforge::detail::ScratchField::key, 0,
                                     scratch_buffer(key.data(), sizeof key)), store);
    return key;
}
 
::pineforge::detail::ScratchHandle scratch_find(const ::pineforge::detail::CommandScratchStore& store,
                                  const TradeScratchKey& key) {
    const auto handle = scratch_seek(store, key);
    return handle != ::pineforge::detail::scratch_no_handle && scratch_key(store, handle) == key
        ? handle : ::pineforge::detail::scratch_no_handle;
}
 
void scratch_put(::pineforge::detail::CommandScratchStore& store, const TradeScratchKey& key,
                 const ::pineforge::detail::ScratchByteSource& value) {
    ::pineforge::detail::ScratchSpanSource source(scratch_bytes(key.data(), sizeof key));
    ::pineforge::detail::ScratchHandle handle = ::pineforge::detail::scratch_no_handle;
    require_trade_scratch(store.put(source, value, handle), store);
}
 
void scratch_put(::pineforge::detail::CommandScratchStore& store, const TradeScratchKey& key,
                 ::pineforge::detail::ScratchBytes value = {}) {
    ::pineforge::detail::ScratchSpanSource source(value);
    scratch_put(store, key, source);
}
 
template<class Visit>
void scratch_walk(const ::pineforge::detail::CommandScratchStore& store, std::uint64_t kind, Visit&& visit) {
    auto handle = scratch_seek(store, {kind, 0, 0, 0});
    while (handle != ::pineforge::detail::scratch_no_handle) {
        const auto key = scratch_key(store, handle);
        if (key[0] != kind) break;
        visit(handle, key);
        ::pineforge::detail::ScratchHandle next = ::pineforge::detail::scratch_no_handle;
        const auto status = store.next(handle, next);
        if (status != ::pineforge::detail::ScratchStatus::end) require_trade_scratch(status, store);
        handle = next;
    }
}
 
template<class T>
std::uint64_t scratch_word(const T& value) noexcept {
    static_assert(sizeof(T) <= sizeof(std::uint64_t), "trade scratch scalar");
    std::uint64_t word = 0;
    std::memcpy(&word, &value, sizeof value);
    return word;
}
 
template<class T>
void scratch_scalar(std::uint64_t word, T& value) noexcept {
    static_assert(sizeof(T) <= sizeof word, "trade scratch scalar");
    std::memcpy(&value, &word, sizeof value);
}
 
TradeScratchHeader scratch_trade_header(const Trade& row, std::size_t source,
                                       std::uint8_t phase) noexcept {
    return {source, phase, scratch_word(row.entry_time), scratch_word(row.exit_time),
            scratch_word(row.entry_price), scratch_word(row.exit_price), scratch_word(row.qty),
            scratch_word(row.pnl), scratch_word(row.pnl_pct), scratch_word(row.is_long),
            scratch_word(row.entry_bar_index), scratch_word(row.exit_bar_index),
            scratch_word(row.exit_from_bracket), scratch_word(row.max_runup),
            scratch_word(row.max_drawdown), scratch_word(row.commission),
            row.entry_incarnation, scratch_word(row.open_at_end), scratch_word(row.close_cause),
            row.entry_id.size(), row.entry_comment.size(), row.exit_comment.size(),
            row.exit_id.size()};
}
 
void scratch_trade_scalars(const TradeScratchHeader& header, Trade& row) noexcept {
    scratch_scalar(header[2], row.entry_time);
    scratch_scalar(header[3], row.exit_time);
    scratch_scalar(header[4], row.entry_price);
    scratch_scalar(header[5], row.exit_price);
    scratch_scalar(header[6], row.qty);
    scratch_scalar(header[7], row.pnl);
    scratch_scalar(header[8], row.pnl_pct);
    scratch_scalar(header[9], row.is_long);
    scratch_scalar(header[10], row.entry_bar_index);
    scratch_scalar(header[11], row.exit_bar_index);
    scratch_scalar(header[12], row.exit_from_bracket);
    scratch_scalar(header[13], row.max_runup);
    scratch_scalar(header[14], row.max_drawdown);
    scratch_scalar(header[15], row.commission);
    row.entry_incarnation = header[16];
    scratch_scalar(header[17], row.open_at_end);
    scratch_scalar(header[18], row.close_cause);
}
 
// Scalar amendments plus text borrowed from one ledger value. put streams
// all five spans without retaining the replacement population in RAM.
class TradeScratchSource final : public ::pineforge::detail::ScratchByteSource {
public:
    TradeScratchSource(TradeScratchHeader header, const Trade& text) noexcept
        : header_(header), parts_{scratch_bytes(header_.data(), sizeof header_),
            scratch_bytes(text.entry_id.data(), text.entry_id.size()),
            scratch_bytes(text.entry_comment.data(), text.entry_comment.size()),
            scratch_bytes(text.exit_comment.data(), text.exit_comment.size()),
            scratch_bytes(text.exit_id.data(), text.exit_id.size())} {}
    std::uint64_t size() const noexcept override {
        std::uint64_t result = 0;
        for (const auto part : parts_) result += part.size;
        return result;
    }
    ::pineforge::detail::ScratchStatus read(std::uint64_t offset,
                              ::pineforge::detail::ScratchBuffer destination) const noexcept override {
        if (offset > size() || destination.size > size() - offset
            || (destination.size && !destination.data))
            return ::pineforge::detail::ScratchStatus::programming_error;
        if (!destination.size) return ::pineforge::detail::ScratchStatus::ok;
        for (const auto part : parts_) {
            if (offset >= part.size) { offset -= part.size; continue; }
            const auto count = std::min(destination.size,
                                        part.size - static_cast<std::size_t>(offset));
            if (count) std::memcpy(destination.data, part.data + offset, count);
            destination.data += count;
            destination.size -= count;
            offset = 0;
            if (!destination.size) break;
        }
        return ::pineforge::detail::ScratchStatus::ok;
    }
private:
    const TradeScratchHeader header_;
    const std::array<::pineforge::detail::ScratchBytes, 5> parts_;
};
 
TradeScratchHeader scratch_read_trade(const ::pineforge::detail::ScratchReadback& store,
                                     ::pineforge::detail::ScratchHandle handle) {
    TradeScratchHeader header{};
    require_trade_scratch(store.read(handle, ::pineforge::detail::ScratchField::value, 0,
                                     scratch_buffer(header.data(), sizeof header)), store);
    return header;
}
 
void scratch_trade_text(const ::pineforge::detail::ScratchReadback& store, ::pineforge::detail::ScratchHandle handle,
                        const TradeScratchHeader& header, Trade& row) {
    const std::array<std::string*, 4> text{
        &row.entry_id, &row.entry_comment, &row.exit_comment, &row.exit_id};
    std::uint64_t offset = sizeof header;
    for (std::size_t index = 0; index < text.size(); ++index) {
        text[index]->resize(static_cast<std::size_t>(header[19 + index]));
        require_trade_scratch(store.read(handle, ::pineforge::detail::ScratchField::value, offset,
            scratch_buffer(text[index]->data(), text[index]->size())), store);
        offset += text[index]->size();
    }
}
 
[[noreturn]] void reject_begin_bar(int index, const char* field, const char* detail,
                                   const char* reason) {
    throw coded<std::invalid_argument>(
        RunFailureCode::chart_bars_rejected,
        {{"index", index}, {"field", field}, {"reason", reason}},
        "bar[" + std::to_string(index) + "]." + field + (detail ? detail : ""));
}
 
// Validate the borrowed public begin array before the source provider stages
// syminfo, inputs, adapter state, or a native run spec.  This mirrors the
// legacy chart/stream shape checks and deliberately does not impose native
// calendar or slot-label policy; those remain the generic preflight's job.
void validate_source_begin_bars(const NativeBeginArgs& args) {
    if (args.n < 0) {
        throw coded<std::invalid_argument>(RunFailureCode::engine_invariant, {},
                                           "bar count must be non-negative");
    }
    if (args.n > 0 && args.bars == nullptr) {
        throw coded<std::invalid_argument>(RunFailureCode::engine_invariant, {},
                                           "bars must be non-null for a nonempty array");
    }
    for (int i = 0; i < args.n; ++i) {
        const Bar& bar = args.bars[i];
        if (!std::isfinite(bar.open)) reject_begin_bar(i, "open", " must be finite", "not_finite");
        if (!std::isfinite(bar.high)) reject_begin_bar(i, "high", " must be finite", "not_finite");
        if (!std::isfinite(bar.low)) reject_begin_bar(i, "low", " must be finite", "not_finite");
        if (!std::isfinite(bar.close)) reject_begin_bar(i, "close", " must be finite", "not_finite");
        if (args.is_stream) {
            if (bar.timestamp < 0)
                reject_begin_bar(i, "timestamp", " must be non-negative", "negative");
            if (bar.open < 0.0) reject_begin_bar(i, "open", " must be non-negative", "negative");
            if (bar.high < 0.0) reject_begin_bar(i, "high", " must be non-negative", "negative");
            if (bar.low < 0.0) reject_begin_bar(i, "low", " must be non-negative", "negative");
            if (bar.close < 0.0) reject_begin_bar(i, "close", " must be non-negative", "negative");
            if (!std::isfinite(bar.volume) || bar.volume < 0.0)
                reject_begin_bar(i, "volume", " must be non-negative finite", "volume_invalid");
        } else if (!std::isnan(bar.volume)
                   && (!std::isfinite(bar.volume) || bar.volume < 0.0)) {
            reject_begin_bar(i, "volume", " must be non-negative finite or NaN (unavailable)",
                             "volume_invalid");
        }
        if (bar.low > std::min(bar.open, bar.close))
            reject_begin_bar(i, "low", " must not exceed open or close", "low_above_body");
        if (bar.high < std::max(bar.open, bar.close))
            reject_begin_bar(i, "high", " must not be below open or close", "high_below_body");
        if (i > 0) {
            const std::int64_t previous = args.bars[i - 1].timestamp;
            if (bar.timestamp <= previous)
                reject_begin_bar(i, "timestamp", " must be strictly increasing", "not_increasing");
            if (previous < 0
                && bar.timestamp > std::numeric_limits<std::int64_t>::max() + previous) {
                reject_begin_bar(i, "timestamp", " delta exceeds int64 range", "delta_overflow");
            }
        }
    }
    if (args.is_stream && args.n > 0
        && (!std::isfinite(args.bars[args.n - 1].close)
            || args.bars[args.n - 1].close <= 0.0)) {
        throw coded<std::invalid_argument>(RunFailureCode::stream_input_rejected, {},
                                           "stream warmup final close must be finite and positive");
    }
}
 
// AR-R4: each installed feed lent to a projected spec. The lent entry holds
// the bars, closes and column values; the store keeps the instrument, tf and
// column names. Every allocation happens before the first byte moves, and the
// moves are swaps, so a throw leaves the store whole.
std::vector<NativeInstrumentFeed> lend_feed_bytes(std::vector<NativeInstrumentFeed>& store) {
    std::vector<NativeInstrumentFeed> lent(store.size());
    for (std::size_t i = 0; i < store.size(); ++i) {
        lent[i].instrument = store[i].instrument;
        lent[i].tf = store[i].tf;
        lent[i].columns.resize(store[i].columns.size());
        for (std::size_t c = 0; c < store[i].columns.size(); ++c)
            lent[i].columns[c].name = store[i].columns[c].name;
    }
    for (std::size_t i = 0; i < store.size(); ++i) {
        lent[i].bars.swap(store[i].bars);
        lent[i].close_ms.swap(store[i].close_ms);
        for (std::size_t c = 0; c < store[i].columns.size(); ++c)
            lent[i].columns[c].values.swap(store[i].columns[c].values);
    }
    return lent;
}
 
// The bytes of a configure that did not apply go back where they came from.
void return_feed_bytes(std::vector<NativeInstrumentFeed>& store,
                       std::vector<NativeInstrumentFeed>& lent) noexcept {
    for (std::size_t i = 0; i < store.size() && i < lent.size(); ++i) {
        store[i].bars.swap(lent[i].bars);
        store[i].close_ms.swap(lent[i].close_ms);
        for (std::size_t c = 0; c < store[i].columns.size() && c < lent[i].columns.size(); ++c)
            store[i].columns[c].values.swap(lent[i].columns[c].values);
    }
}
 
// A legacy setter the generated strategy latched (a request's input or
// override it could not take) surfaces at the run's begin: the same type and
// text, coded setting_rejected unless it already carries a code.
[[noreturn]] void rethrow_latched_setting(const checked_settings::LatchedSettingsFailure& error) {
    if (dynamic_cast<const RunFailureInfo*>(&error) != nullptr) throw;
    throw coded<checked_settings::LatchedSettingsFailure>(RunFailureCode::setting_rejected, {},
                                                          error.what());
}
 
}  // namespace
 
::pineforge::detail::ScratchKeyOrder source::PineStrategyHost::trade_scratch_order() noexcept {
    return {0x5830365830370001ULL, nullptr, compare_trade_scratch};
}
 
void source::PineStrategyHost::set_trade_scratch(::pineforge::detail::CommandScratchStore& store) {
    require_trade_scratch(store.status(), store);
    if (store.transaction_active() || store.record_count() != 0
        || (trade_scratch_ && trade_scratch_->transaction_active()))
        throw coded<std::logic_error>(RunFailureCode::engine_invariant, {},
                                      "trade scratch must be exclusively owned and empty");
    trade_scratch_ = &store;
}
 
::pineforge::detail::CommandScratchStore& source::PineStrategyHost::trade_scratch() {
    if (!trade_scratch_)
        throw coded<std::logic_error>(RunFailureCode::engine_invariant, {},
                                      "run owner did not provision trade scratch");
    return *trade_scratch_;
}
 
source::PineStrategyHost::PineStrategyHost(compat::pine::CapAttachment cap)
    : NativeStrategyHost(),
      adapter_(*this, cap),
      _src_series_active_(scheduler_.language()._src_series_active_),
      _src_open_(scheduler_.language()._src_open_),
      _src_high_(scheduler_.language()._src_high_),
      _src_low_(scheduler_.language()._src_low_),
      _src_close_(scheduler_.language()._src_close_),
      _src_volume_(scheduler_.language()._src_volume_),
      _src_hl2_(scheduler_.language()._src_hl2_),
      _src_hlc3_(scheduler_.language()._src_hlc3_),
      _src_ohlc4_(scheduler_.language()._src_ohlc4_),
      _src_hlcc4_(scheduler_.language()._src_hlcc4_),
      is_last_tick_(scheduler_.language().is_last_tick_) {
    // ab9714be LegacyCompatibilityConsumer::refuse was a no-op on this handle.
    host_mutation_guard_inert_ = true;
    NativeExecutionConsumer::bound(*this).set_batch_input_release(
        this, [](void* owner) noexcept {
            static_cast<PineStrategyHost*>(owner)->scheduler_.release_batch_input();
        });
#ifdef PINEFORGE_HAS_AUX_SECURITY_FEED_V1
    // A fresh host folds its empty feed as one cleared at a run's end does.
    refresh_aux_security_digest();
#endif
}
 
std::uint64_t source::PineStrategyHost::broker_state_hash_projection() const {
    // Fold current source/generic state with the last script-point
    // continuation. Recording only controls whether the per-bar array is
    // retained; the continuation snapshot keeps the scalar independent of
    // that switch and of NativeCompleted teardown.
    if (last_script_continuation_valid_) {
        return broker_state_hash_from_execution_hash(last_script_continuation_hash_);
    }
    const std::uint64_t execution = execution_consumer().continuation_hash();
    // The fold cannot throw: a read of the spilled book that failed inside it
    // was carried by the core and is thrown here, never folded as a hash it
    // cut short.
    as_native_consumer(execution_consumer()).request_core().rethrow_book_read_failure();
    return broker_state_hash_from_execution_hash(execution);
}
 
double source::PineStrategyHost::margin_liquidation_price() const {
    // The level is the kernel's (native_liquidation_price, solved from closed
    // money alone: project() declares NativeLiquidationLevelBase::RealizedOnly),
    // and na where it has none -- a flat book, a run without a margin model.
    // TradingView's spelling is the one thing kept here (design row MG8): the
    // level rounded onto the tick grid away from the position, down for a
    // long, up for a short.
    const std::optional<double> level = native_liquidation_price();
    if (!level) return na<double>();
    double liquidation = *level;
    if (syminfo_mintick_ > 0.0) {
        liquidation = position_side_ == PositionSide::SHORT
            ? std::ceil(liquidation / syminfo_mintick_) * syminfo_mintick_
            : std::floor(liquidation / syminfo_mintick_) * syminfo_mintick_;
    }
    return liquidation;
}
 
PineStrategyConfig source::PineStrategyHost::apply_overrides(
        PineStrategyConfig config, const StrategyOverrides& overrides) {
    if (std::isinf(overrides.default_qty_value)) {
        throw coded<std::invalid_argument>(
            RunFailureCode::setting_rejected,
            {{"entrypoint", "strategy_set_override"},
             {"reason", "number_out_of_finite_range"}},
            "default_qty_value: number out of finite range");
    }
    if (!std::isnan(overrides.initial_capital)) config.initial_capital = overrides.initial_capital;
    if (!std::isnan(overrides.commission_value)) config.commission_value = overrides.commission_value;
    if (!std::isnan(overrides.default_qty_value)) config.default_qty_value = overrides.default_qty_value;
    if (overrides.pyramiding >= 0) config.pyramiding = overrides.pyramiding;
    if (overrides.slippage >= 0) config.slippage = overrides.slippage;
    if (overrides.commission_type >= 0) config.commission_type = overrides.commission_type;
    if (overrides.default_qty_type >= 0) config.default_qty_type = overrides.default_qty_type;
    if (overrides.process_orders_on_close >= 0)
        config.process_orders_on_close = overrides.process_orders_on_close != 0;
    if (overrides.calc_on_order_fills >= 0)
        config.calc_on_order_fills = overrides.calc_on_order_fills != 0;
    if (overrides.close_entries_rule >= 0)
        config.close_entries_rule_any = overrides.close_entries_rule != 0;
    return config;
}
 
StagedConfiguration source::PineStrategyHost::staged_configuration() const {
    StagedConfiguration staged;
    staged.syminfo = syminfo_;
    staged.syminfo.mintick = syminfo_mintick_;
    staged.inputs = inputs_;
    staged.chart_timezone = chart_timezone_;
    staged.account_fx = account_currency_fx_;
    staged.account_fx_effective_from_ms = account_currency_fx_timestamps_;
    staged.account_fx_per_quote = account_currency_fx_rates_;
    if (std::isfinite(qty_step_) && qty_step_ > 0.0) staged.quantity_grid = qty_step_;
    staged.instrument_feeds.reserve(symbol_feeds_.size());
    for (std::size_t i = 0; i < symbol_feeds_.size(); ++i) {
        const NativeInstrumentFeed& feed = installed_symbol_feed(i);
        const auto views = NativeExecutionConsumer::bound(*this).bar_views();
        const auto* slot = views.instrument_feed_count == 0
            ? nullptr : views.instrument_feeds[i];
        const auto data = native_instrument_feed_view(feed, slot);
        const std::size_t count = i < symbol_feed_views_.size() && symbol_feed_views_[i].present
            ? symbol_feed_views_[i].size : data.bars.size;
        staged.instrument_feeds.push_back(
            {feed.instrument, feed.tf, count, feed.columns.size()});
    }
    return staged;
}
 
void source::PineStrategyHost::prepare_native_begin(const NativeBeginArgs& args) {
    // Idle abort requests are consumed by the public begin entry even when
    // input validation refuses before a run starts.  Crucially, validation
    // runs before any provider-owned state is changed.
    abort_requested_.store(false, std::memory_order_relaxed);
    validate_source_begin_bars(args);
    // Admission owns a reentrant begin's refusal. Keep the running caller
    // span and source state intact until the outer public call unwinds.
    if (native_state().kind == NativeLifecycleKind::Running) return;
    if (args.syminfo) {
        syminfo_ = *args.syminfo;
        syminfo_mintick_ = syminfo_.mintick;
        if (std::isfinite(syminfo_.qty_step) && syminfo_.qty_step > 0.0)
            qty_step_ = syminfo_.qty_step;
    }
    if (args.inputs) inputs_ = *args.inputs;
 
    if (args.is_stream && native_security_feed_enabled()) {
        throw coded<std::runtime_error>(
            RunFailureCode::run_mode_unsupported, {{"feature", "stream_native_security_feed"}},
            "native request.security feed supports historical runs only");
    }
    // Lane XSYM-D: another symbol's feeds and the recorded request series are
    // installed whole before a run and have no realtime ingress, exactly as
    // the native feed above.
    if (args.is_stream && (!symbol_feeds_.empty() || !recorded_series_.empty())) {
        throw coded<std::runtime_error>(
            RunFailureCode::run_mode_unsupported, {{"feature", "stream_symbol_feeds"}},
            "request.security symbol feeds and recorded request series support historical "
            "runs only");
    }
 
    if (!(args.n < 2 && !args.is_stream)) {
        std::string effective_input = args.input_tf;
        if (effective_input.empty() && args.n >= 2 && args.bars != nullptr)
            effective_input = detect_timeframe(args.bars, args.n);
        const std::string effective_script = args.script_tf.empty()
            ? effective_input : args.script_tf;
        try {
            if (!effective_input.empty() && !effective_script.empty()
                && tf_ratio(effective_input, effective_script) == -2) {
                throw coded<std::runtime_error>(
                    RunFailureCode::run_options_rejected,
                    {{"option", "script_timeframe"}, {"script_tf", effective_script},
                     {"input_tf", effective_input}},
                    "script timeframe must be coarser than or equal to input timeframe: requested script_tf "
                    + effective_script + " from input timeframe " + effective_input);
            }
        } catch (const std::runtime_error&) {
            throw;
        } catch (...) {
            // The native specification validator owns malformed literals.
        }
    }
 
    if (args.is_stream && config_.calc_on_order_fills) {
        throw coded<std::runtime_error>(
            RunFailureCode::run_mode_unsupported, {{"feature", "stream_calc_on_order_fills"}},
            "native stream requires close-only calculation; calc_on_order_fills is unsupported");
    }
    if (args.is_stream && (realtime_tail_ || probe_suppress_tail_logic_)) {
        throw coded<std::runtime_error>(
            RunFailureCode::run_mode_unsupported, {{"feature", "stream_probe_overrides"}},
            "native stream cannot use historical probe/tail overrides");
    }
    PineStrategyConfig effective = config_;
    if (args.overrides_opaque) {
        const auto* overrides = static_cast<const StrategyOverrides*>(args.overrides_opaque);
        effective = apply_overrides(effective, *overrides);
    }
    const StagedConfiguration staged = staged_configuration();
    if (!staged.account_fx_effective_from_ms.empty() && effective.calc_on_order_fills)
        throw coded<std::logic_error>(
            RunFailureCode::run_mode_unsupported, {{"feature", "fx_series_calc_on_order_fills"}},
            "timestamped account-currency FX does not support calc_on_order_fills");
    if (!staged.account_fx_effective_from_ms.empty() && args.bar_magnifier)
        throw coded<std::logic_error>(
            RunFailureCode::run_mode_unsupported, {{"feature", "fx_series_bar_magnifier"}},
            "timestamped account-currency FX is not supported with bar magnifier");
 
    adapter_.reset_for_run();
    adapter_.set_cache_policy_for_run(cache_policy_for_run());
    for (auto& tape : recorded_series_)
        tape.second.rows.set_capacity_bytes(tape.second.cache_capacity_override.value_or(
            cache_policy_for_run().recorded_series_pages.capacity_bytes));
    const auto& scratch = scratch_tables_for_run();
    ::pineforge::detail::bind_tree_scratch_tables(adapter_, scratch);
    ::pineforge::detail::bind_id_pair_scratch_tables(adapter_, scratch);
    auto command_scratch = std::make_shared<SourceExitScratchRun>(cache_policy_for_run().exit_pages);
    auto& consumer = NativeExecutionConsumer::bound(*this);
    using ScratchBinding = NativeExecutionConsumer::CommandScratchBinding;
    // Reserve placement evidence by incarnation once for the disclosed batch:
    // two legs per bar, including re-issued brackets' historical projections.
    if (args.n > 0 && static_cast<std::size_t>(args.n)
        <= adapter_.placement_.max_size() / 2U) {
        adapter_.placement_.reserve(static_cast<std::size_t>(args.n) * 2U);
    }
    adapter_.set_configuration(effective);
    adapter_.set_staged_configuration(staged);
    adapter_.set_begin_mode(args.is_stream, args.bar_magnifier);
    adapter_.set_margin_call_enabled(margin_call_enabled_);
    scheduler_.capture_begin(args);
    scheduler_.set_source_series_active(effective.src_series_active);
    const NativePathOrder path_order = path_order_mode_ == 1
        ? NativePathOrder::HighFirst
        : (path_order_mode_ == 2 ? NativePathOrder::LowFirst
                                 : NativePathOrder::Auto);
    // The installed feeds' bytes move into the projected spec (project()
    // moves them in, and back out if it throws), which the kernel keeps as its
    // own when it applies it: one copy, read through installed_symbol_feed()
    // while it is lent (AR-R4). A refusal, or an exception, hands them back.
    // Nothing of this host or of the script runs inside configure.
    restore_symbol_feeds();
    std::vector<NativeInstrumentFeed> lent = lend_feed_bytes(symbol_feeds_);
    NativeRunSpec spec = [&] {
        try {
            const bool has_borrowed_feed = std::any_of(
                symbol_feed_views_.begin(), symbol_feed_views_.end(),
                [](const BorrowedFeedSlot& slot) { return slot.present; });
            if (has_borrowed_feed) {
                stage_symbol_feed_views();
            } else {
                NativeExecutionConsumer::bound(*this).stage_instrument_feed_views(nullptr, 0);
            }
            return adapter_.project(effective, staged, args, path_order, &lent);
        } catch (...) {
            return_feed_bytes(symbol_feeds_, lent);
            throw;
        }
    }();
    const bool failed_before = native_state().kind == NativeLifecycleKind::Failed;
    NativeSetupResult setup;
    try {
        // project() establishes the actual run identity and installs a
        // standalone store. Replace it with the host's service afterwards,
        // before configure or any run/bar callback can read the adapter.
        // Keeping this in the feed-return guard also covers binding failure.
        command_scratch->exit_context.run = spec.identity;
        adapter_.bind_exit_scratch(command_scratch->exit_context);
        consumer.retain_command_scratch_owner(ScratchBinding::Exit, command_scratch);
        setup = NativeExecutionConsumer::bound(*this).configure(*this, std::move(spec));
    } catch (...) {
        return_feed_bytes(symbol_feeds_, spec.instrument_feeds);
        throw;
    }
    if (setup.status == NativeSetupStatus::Applied) {
        symbol_feeds_lent_ = !symbol_feeds_.empty();
    } else {
        return_feed_bytes(symbol_feeds_, spec.instrument_feeds);
    }
    if (setup.status != NativeSetupStatus::Applied) {
        // configure() recorded the cause (its own text and code): the refusal
        // below keeps its text and carries that code.
        RunFailureValue cause = run_failure_value_of(*this);
        if (cause.code == RunFailureCode::none)
            cause = RunFailureValue{RunFailureCode::engine_invariant, nullptr};
        if (failed_before) {
            try { prepare_script_run(nullptr, 0, false); }
            catch (const checked_settings::LatchedSettingsFailure& error) {
                rethrow_latched_setting(error);
            }
            catch (...) {}
        }
        throw coded<std::logic_error>(std::move(cause),
                                      "Pine native adapter failed to configure projected run spec");
    }
    config_ = effective;
    source_configuration_captured_ = true;
}
 
void source::PineStrategyHost::on_native_run_begin() {
    NativeExecutionConsumer::bound(*this).provision_run_command_scratch(
        NativeExecutionConsumer::CommandScratchBinding::Trade,
        trade_scratch_order(), trade_scratch_);
    // The adapter reads the command journal with its receipt cursor and
    // acknowledges it as it moves (observe_terminal_receipts); acknowledging
    // "nothing read yet" here marks this host as one that polls, so the
    // kernel keeps every event above the cursor until the adapter has seen it.
    native_acknowledge_events(0);
    // The spec facts the per-bar callbacks read, once for the run (R5 lane
    // D2-C, pine_host_reads.hpp).
    detail::take_run_facts(detail::run_consumer(*this), adapter_, adapter_.run_counter_);
    // A selected window's pre-roll capture state is parked for this run alone
    // (pine_intent_state.hpp). It sits outside the try below on purpose: an
    // allocation failure fails the begin instead of becoming a script preparation
    // failure, and nothing is dropped silently.
    detail::adopt_selected_intents(*this);
    source_bar_index_ = -1;
    source_last_bar_index_ = -1;
    source_callback_count_ = 0;
    session_ismarket_ = false;
    session_isfirstbar_ = false;
    session_islastbar_ = false;
    session_isfirstbar_regular_ = false;
    session_islastbar_regular_ = false;
    source_prepare_failed_ = false;
    // Generated configure_security_evaluators() registers this run's foreign
    // sites again inside run_begin; a run that configures none has none.
    foreign_security_sites_.clear();
    foreign_security_series_.clear();
    foreign_input_index_ = -1;
    try {
        install_symbol_calendar_metadata();
        scheduler_.run_begin(*this);
    } catch (const checked_settings::LatchedSettingsFailure& error) {
        rethrow_latched_setting(error);
    } catch (const std::exception& error) {
        if (adapter_.stream_mode_) throw;
        source_prepare_failed_ = true;
        note_run_failure(*this, std::string(error.what()), classify_run_failure(error));
    } catch (...) {
        if (adapter_.stream_mode_)
            throw coded<std::runtime_error>(RunFailureCode::engine_unclassified_error, {},
                                            "unknown error during Pine script preparation");
        source_prepare_failed_ = true;
        note_run_failure(*this, "unknown error during Pine script preparation",
                         RunFailureCode::engine_unclassified_error);
    }
}
 
std::int64_t source::PineStrategyHost::symbol_calendar_open_before(std::int64_t before) const {
    if (symbol_calendar_.empty() || before <= symbol_calendar_.front().first
        || before > symbol_calendar_.back().second) {
        return na<std::int64_t>();
    }
    const bool daily = calendar_period_for(script_tf_) != CalendarPeriod::NONE;
    const std::int64_t step = daily ? 0 : static_cast<std::int64_t>(script_tf_seconds_) * 1000;
    // The last day opening before `before`, and its last instant before it.
    const auto it = std::lower_bound(
        symbol_calendar_.begin(), symbol_calendar_.end(), before,
        [](const std::pair<std::int64_t, std::int64_t>& day, std::int64_t value) {
            return day.first < value;
        });
    const auto& day = *std::prev(it);
    const std::int64_t last = std::min(before, day.second) - 1;
    if (step <= 0) return day.first;
    return day.first + (last - day.first) / step * step;
}
 
// The symbol calendar the C ABI named day by day (set_syminfo_metadata): every
// day of the count it announced, or the run fails naming what is missing.
void source::PineStrategyHost::install_symbol_calendar_metadata() {
    if (symbol_calendar_days_ < 0) return;
    // The announced days are this run's: taken whole, so a later run's
    // announcement starts from none (a day it leaves out is missing, never
    // an earlier run's).
    const long long count = symbol_calendar_days_;
    const std::map<long long, double> opens = std::move(symbol_calendar_opens_);
    const std::map<long long, double> closes = std::move(symbol_calendar_closes_);
    symbol_calendar_opens_.clear();
    symbol_calendar_closes_.clear();
    symbol_calendar_days_ = -1;
    // The announcement replaces the calendar: one it cannot install leaves
    // none behind for a later run to read.
    symbol_calendar_.clear();
    std::vector<std::pair<std::int64_t, std::int64_t>> sessions;
    for (long long i = 0; i < count; ++i) {
        const auto open = opens.find(i);
        const auto close = closes.find(i);
        if (open == opens.end() || close == closes.end()
            || !std::isfinite(open->second) || !std::isfinite(close->second)) {
            throw coded<std::runtime_error>(
                RunFailureCode::symbol_metadata_rejected, {{"field", "session_calendar"}},
                "symbol calendar: session day " + std::to_string(i) + " of "
                    + std::to_string(count) + " is missing");
        }
        sessions.emplace_back(std::llround(open->second), std::llround(close->second));
    }
    if (!set_symbol_calendar(std::move(sessions))) {
        throw coded<std::runtime_error>(RunFailureCode::symbol_metadata_rejected,
                                        {{"field", "session_calendar"}},
                                        "symbol calendar: session days out of order");
    }
}
 
bool source::PineStrategyHost::set_symbol_calendar(
        std::vector<std::pair<std::int64_t, std::int64_t>> sessions) {
    for (std::size_t i = 0; i < sessions.size(); ++i) {
        if (sessions[i].first >= sessions[i].second) return false;
        if (i + 1 < sessions.size() && sessions[i].second > sessions[i + 1].first) return false;
    }
    symbol_calendar_ = std::move(sessions);
    return true;
}
 
const std::pair<std::int64_t, std::int64_t>* source::PineStrategyHost::symbol_calendar_day(
        std::int64_t ms) const {
    const auto it = std::upper_bound(
        symbol_calendar_.begin(), symbol_calendar_.end(), ms,
        [](std::int64_t value, const std::pair<std::int64_t, std::int64_t>& day) {
            return value < day.first;
        });
    if (it == symbol_calendar_.begin()) return nullptr;
    const auto& day = *std::prev(it);
    return ms < day.second ? &day : nullptr;
}
 
std::int64_t source::PineStrategyHost::symbol_calendar_open_from(std::int64_t at) const {
    if (symbol_calendar_.empty() || at < symbol_calendar_.front().first)
        return na<std::int64_t>();
    const bool daily = calendar_period_for(script_tf_) != CalendarPeriod::NONE;
    const std::int64_t step = daily ? 0 : static_cast<std::int64_t>(script_tf_seconds_) * 1000;
    auto it = std::upper_bound(
        symbol_calendar_.begin(), symbol_calendar_.end(), at,
        [](std::int64_t value, const std::pair<std::int64_t, std::int64_t>& day) {
            return value < day.second;
        });
    for (; it != symbol_calendar_.end(); ++it) {
        if (at <= it->first) return it->first;
        if (step <= 0) continue;
        const std::int64_t slot = it->first + (at - it->first + step - 1) / step * step;
        if (slot < it->second) return slot;
    }
    return na<std::int64_t>();
}
 
void source::PineStrategyHost::capture_script_continuation_hash() {
    // Taken at once: since v19 the fold is the live state alone, which is what
    // a deferred view would have recorded word for word (R5 lane V19-A).
    const std::uint64_t execution = execution_consumer().continuation_hash();
    // A read of the spilled book that failed inside the noexcept fold is
    // thrown here, before the hash it cut short is latched as valid; the
    // latch keeps what it held.
    as_native_consumer(execution_consumer()).request_core().rethrow_book_read_failure();
    last_script_continuation_hash_ = execution;
    last_script_continuation_valid_ = true;
    if (broker_state_hash_recording_ && !broker_state_hashes_.empty()) {
        broker_state_hashes_.back() = broker_state_hash();
    }
}
 
void source::PineStrategyHost::on_native_input(
        const Bar& bar, const NativeInputContext& context) {
    if (source_prepare_failed_) return;
    if (detail::run_phase(*this) == NativeRunPhase::Realtime)
        stream_warmup_mode_ = false;
    foreign_input_index_ = context.input_index;
    scheduler_.input(bar, context, *this);
    // Aggregation can deliver leftover input after the last script callback.
    // Refresh the last recorded row (and the continuation snapshot) so the
    // scalar stays the same fold with or without recording.
    if (scheduler_.terminal_source_bar()) capture_script_continuation_hash();
}
 
void source::PineStrategyHost::on_native_tick(
        const Bar& tick, const NativeTickContext& context) {
    if (source_prepare_failed_) return;
    if (detail::run_phase(*this) == NativeRunPhase::Realtime)
        stream_warmup_mode_ = false;
    scheduler_.tick(tick, context, *this);
    adapter_.on_tick(tick, context);
}
 
void source::PineStrategyHost::on_native_bar_open(
        const Bar& bar, const NativeDecisionContext& context) {
    if (source_prepare_failed_) return;
    // Foreign sites, after the input's deliveries and before the bar's
    // calc_on_order_fills checkpoint (PineScheduler::bar_open), so every
    // calculation of the bar reads the same values: on a calendar chart the
    // bars TradingView's merge rule makes visible at this bar that the kernel
    // has not handed over yet run first; then a gaps_on site that ran on no
    // new bar reads na.
    if (!foreign_security_sites_.empty()) {
        read_ahead_foreign_security_sites(bar);
        clear_gapped_foreign_security_sites();
    }
    bar_magnifier_enabled_ = scheduler_.bar_magnifier_enabled();
    diag_magnifier_sub_bars_processed_ = bar_magnifier_enabled_
        ? static_cast<std::int64_t>(context.driver_statistics.sub_bars_processed) : 0;
    diag_magnifier_sample_ticks_processed_ = bar_magnifier_enabled_
        ? static_cast<std::int64_t>(context.driver_statistics.sample_ticks_processed) : 0;
    // A selected window's pre-roll rows are placed here, once, before the adapter
    // reads the bar's open: it must see them as the prior bar's orders
    // (selected_replay_at_open). Nothing happens for any other run.
    selected_replay_at_open(bar, context);
    adapter_.on_bar_open(bar, context);
    scheduler_.bar_open(bar, context, *this);
}
 
void source::PineStrategyHost::on_native_sub_bar(
        const Bar&, const NativeDecisionContext&) {
    if (source_prepare_failed_) return;
    scheduler_.sub_bar_complete();
}
 
void source::PineStrategyHost::on_native_bar(
        const Bar& bar, const NativeDecisionContext& context) {
    if (source_prepare_failed_) return;
    bar_magnifier_enabled_ = scheduler_.bar_magnifier_enabled();
    diag_magnifier_sub_bars_processed_ = bar_magnifier_enabled_
        ? static_cast<std::int64_t>(context.driver_statistics.sub_bars_processed) : 0;
    diag_magnifier_sample_ticks_processed_ = bar_magnifier_enabled_
        ? static_cast<std::int64_t>(context.driver_statistics.sample_ticks_processed) : 0;
    adapter_.observe_terminal_receipts();
    adapter_.on_bar_close_before_script(bar, context);
    scheduler_.bar(bar, context, *this);
    adapter_.on_bar_close(bar, context);
    if (context.is_terminal_sub_bar
        && context.coordinate.interval_index == source_last_bar_index_) {
        scheduler_record_range_end(bar);
    }
    const bool recording = broker_state_hash_recording_ && !broker_state_hashes_.empty();
    const bool last_batch = context.is_terminal_sub_bar
        && context.coordinate.interval_index == source_last_bar_index_;
    const bool stream_script = context.is_terminal_sub_bar
        && stream_phase_ == StreamPhase::REALTIME;
    if (recording || last_batch || stream_script) {
        // ab9714be pine_scheduler.cpp:1753/:1875 records after dispatch_bar,
        // including the terminal source policy updates.  The native hook
        // returns through adapter_.on_bar_close after the scheduler callback,
        // so refresh the continuation snapshot (and the just-appended row)
        // at that boundary.
        capture_script_continuation_hash();
    }
}
 
void source::PineStrategyHost::on_native_applied(
        const native_order::ExecutionAppliedEvent& event,
        const NativeDecisionContext& context) { adapter_.invalidate_exit_group(event.first_trade_index, event.closed_trade_count); // Covers later close_cause, exit_time and exit_id writes; no exit-group sort intervenes.
    // R5 lane F3: the kernel records CloseCause::Bracket for a close its
    // owner's fill armed (WaitForApplied). A Pine row keeps the cause this host
    // has always given it -- a strategy.exit leg by its order family
    // (exit_from_bracket), a strategy.close armed on a reversal parent as a
    // script close -- so the kernel's reading is not taken on these rows.
    for (std::size_t i = 0; i < event.closed_trade_count; ++i) {
        const std::size_t index = event.first_trade_index + i;
        if (index < trades_.size()
            && trades_[index].close_cause == execution::CloseCause::Bracket) {
            trades_.modify(index, [](Trade& row) {
                row.close_cause = execution::CloseCause::Unspecified;
            });
        }
    }
    if (source_prepare_failed_) return;
    const bool aggregated = detail::run_aggregates_input_bars(
        detail::run_consumer(*this), &adapter_, adapter_.run_counter_);
    // TradingView dates every fill of a chart bar at the bar's open, a
    // process_orders_on_close fill included (the e25-f-islastbar tape, lane
    // E27). The chart timeframe's zero-width interval gives exactly that:
    // every point of a confirmed bar's modeled path (its open, the segments,
    // the calculation, the close after it) sits at the bar's label. On an
    // aggregated chart the kernel dates a close point at the bucket's close,
    // and books a fill the host executes at such a point (a
    // calc_on_order_fills first-open chain, a same-bar cover) at its decision
    // floor, which the bucket's inputs have already carried to the close on
    // the no-path route. The adapter dates the rows it presents at the chart
    // bar's open: every modeled point of an aggregated chart, as ab9714be's
    // aggregation loop did, and under the magnifier too -- each sub-bar's
    // open, path and close (ModeledOHLCClose, which only a magnified run
    // presents). TradingView dates every magnified fill at its chart bar's
    // open, 218 of 218 on H-MEASURE's sixteen tapes; ab9714be kept a sub-bar
    // fill's own instant, and so did this adapter for all but the close
    // points until R5 lane PAR-ORDERS. A realtime print keeps its own
    // instant, as it does on the chart timeframe, and so does a stream's
    // quiet carried interval, whose context still names the bar before it.
    // The kernel's own instant, which its FX lookup, decision floor and
    // ordering read, is untouched (lane F1).
    const auto provenance = context.coordinate.provenance;
    const bool modeled_point = provenance == NativePriceProvenance::Confirmed
        || provenance == NativePriceProvenance::ModeledOHLCOpen
        || provenance == NativePriceProvenance::ModeledOHLCClose
        || provenance == NativePriceProvenance::Calculation
        || provenance == NativePriceProvenance::AfterCalculationClose;
    if (aggregated && modeled_point) {
        const std::int64_t script_open = context.script_bar_open_ms;
        if (event.opened_units != 0.0) {
            for (auto& lot : pyramid_entries_) {
                if (lot.entry_incarnation == event.handle().incarnation)
                    lot.time = script_open;
            }
            // A later execution of the same pass can close this fill's lot
            // before this notification arrives (the reversing second call
            // of a process_orders_on_close pair): the rows it booked after
            // this event's own still carry the kernel's instant (lane
            // FIX-E1E2, tests/fixtures/aggregated_admission_receipt).
            for (std::size_t index = event.first_trade_index + event.closed_trade_count;
                 index < trades_.size(); ++index) {
                if (trades_[index].entry_incarnation == event.handle().incarnation)
                    { adapter_.invalidate_exit_group(index); trades_.modify(index, [&](Trade& row) { row.entry_time = script_open; }); }
            }
        }
        for (std::size_t i = 0; i < event.closed_trade_count; ++i) {
            const std::size_t index = event.first_trade_index + i;
            if (index >= trades_.size()) continue;
            trades_.modify(index, [&](Trade& row) { row.exit_time = script_open; });
        }
    }
    const auto p = adapter_.placement_.find(event.handle().incarnation);
    const bool pine_priced = p != adapter_.placement_.end()
        && ((std::isfinite(p->second.exit_levels.stop) && p->second.exit_levels.stop > 0.0)
            || (std::isfinite(p->second.exit_levels.limit) && p->second.exit_levels.limit > 0.0));
    if (event.opened_units != 0.0) {
        // ab9714be pine_orders.cpp:750-753 and pine_fills.cpp:7189-7193
        // A MARKET entry that adds to a live same-side position is
        // flagged so a same-bar from_entry bracket exit covers it.
        const bool market_add = !pine_priced && event.closed_units == 0.0
            && p != adapter_.placement_.end()
            && (p->second.family == PineOrderFamily::Entry
                || p->second.family == PineOrderFamily::Order)
            && std::any_of(pyramid_entries_.begin(), pyramid_entries_.end(),
                [&](const PyramidEntry& lot) {
                    return lot.entry_incarnation != event.handle().incarnation;
                });
        if (market_add) adapter_.mark_market_pyramid_add(event.handle().incarnation);
    }
    // The legacy source observer counted one broker fill for every committed
    // execution event.  The native consumer owns those events now; mirror the
    // count at its notification boundary so restored source tests and public
    // source-side policy reads see the same monotone value.
    // ab9714be pine_fills.cpp:5954/:6300 and the margin/FX sites: one source
    // broker fill sequence is consumed per applied broker instruction, not
    // per closed trade row. Native ordinals remain the execution authority;
    // this is the generated/source-visible diagnostic projection.
    if (broker_fill_event_seq_ == std::numeric_limits<std::uint64_t>::max()) {
        throw coded<std::overflow_error>(RunFailureCode::engine_invariant, {},
                                         "source broker fill sequence exhausted");
    }
    ++broker_fill_event_seq_;
    if (position_side_ != PositionSide::FLAT) {
        // ab9714be engine_orders.cpp:531-541: settle_position_after_partial_exit resets to flat when position_qty_ <= kQtyEpsilon or empty
        bool changed = false;
        for (auto it = pyramid_entries_.begin(); it != pyramid_entries_.end(); ) {
            if (it->qty <= internal::kQtyEpsilon) {
                it = pyramid_entries_.erase(it);
                changed = true;
            } else {
                ++it;
            }
        }
        if (pyramid_entries_.empty() || position_qty_ <= internal::kQtyEpsilon) {
            reset_position_state_to_flat();
        } else if (changed) {
            double total_qty = 0.0;
            double weighted_price = 0.0;
            for (const auto& pe : pyramid_entries_) {
                total_qty += pe.qty;
                weighted_price += pe.price * pe.qty;
            }
            position_qty_ = total_qty;
            position_entry_price_ = weighted_price / total_qty;
            position_entry_count_ = static_cast<int>(pyramid_entries_.size());
        }
    }
    // Read before the adapter retires the legs this fill leaves without a lot.
    bool other_exit_here = false;
    if (event.closed_trade_count >= 2 && event.first_trade_index < trades_.size()) {
        const auto own = adapter_.placement_.find(event.handle().incarnation);
        other_exit_here = own == adapter_.placement_.end()
            || adapter_.other_exit_reaches(trades_[event.first_trade_index].exit_price,
                                           trades_[event.first_trade_index].is_long,
                                           own->second.source_id, event.handle());
    }
    adapter_.on_applied(event, context);
    precommit_held_units_ = std::numeric_limits<double>::quiet_NaN();
    if (adapter_.take_intraday_loss_relabel(event.ordinal)) {
        for (std::size_t i = 0; i < event.closed_trade_count; ++i) {
            const std::size_t index = event.first_trade_index + i;
            if (index >= trades_.size()) continue;
            trades_.modify(index, [](Trade& row) {
                row.exit_id.clear();
                row.exit_comment = source::kIntradayLossComment;
            });
        }
    }
    // R5 lane L12 (2.ii l): the kernel used to read these strings back out of
    // the row to answer strategy_closed_trade_close_cause. They are the
    // adapter's, so the adapter's host states the cause instead, on the rows
    // this event produced and after the relabel above has settled them. The
    // predicates are the retired kernel branches verbatim, including the
    // empty-exit-id gate and the filled-orders PREFIX match.
    for (std::size_t i = 0; i < event.closed_trade_count; ++i) {
        const std::size_t index = event.first_trade_index + i;
        if (index >= trades_.size()) continue;
        Trade row = trades_[index]; const auto previous_cause = row.close_cause;
        if (row.exit_id == source::kMarginCallLabel) {
            row.close_cause = execution::CloseCause::Liquidation;
        } else if (row.exit_id.empty()) {
            if (row.exit_comment.rfind(source::kFillCapCommentPrefix, 0) == 0)
                row.close_cause = execution::CloseCause::FillCap;
            else if (row.exit_comment.rfind(source::kIntradayLossComment, 0) == 0)
                row.close_cause = execution::CloseCause::RiskLimit;
        }
        adapter_.invalidate_exit_group(index); if (row.close_cause != previous_cause || index < trades_.base()) trades_.replace(index, row);
    }
    project_short_seed_report_rows(event);
    if (!other_exit_here) order_global_exit_children(event);
    scheduler_.applied(event);
    // R6: with calc_on_order_fills the kernel drives this event's
    // recalculation next, in the same drain iteration and at the same cursor
    // (NativeCalculationTrigger::BarCloseAndFills). The range-end row stays
    // after it, exactly where the adapter's own cascade used to put it.
    // Finish the same range-end mark, then any post-calculation journal tail.
    finish_applied_admission(event, context);
    // The observer seam: once per applied event, after every row this event
    // produced has settled. Empty unless a host overrides it.
    on_native_applied_event(event, context);
}
 
void source::PineStrategyHost::record_applied_range_end() {
    if (scheduler_.terminal_source_bar() || barstate_islast_) {
        const Bar terminal = scheduler_.current_script_bar()
            ? *scheduler_.current_script_bar() : current_bar_;
        scheduler_record_range_end(terminal);
    }
}
 
void source::PineStrategyHost::on_native_recalculate(
        const Bar& bar, const NativeDecisionContext& context,
        NativeCalculationReason reason,
        const native_order::ExecutionAppliedEvent* cause) {
    if (reason != NativeCalculationReason::OrderFill) {
        // BarClose is the script bar's own calculation; Tick and SubBar are
        // cadences the Pine spec never selects.
        on_native_bar(bar, context);
        return;
    }
    if (source_prepare_failed_ || cause == nullptr) return;
    // A fill Pine does not recalculate on (POOC's terminal close fill, a
    // grouped-stop sibling, a replayed ordinal) still spends a kernel
    // recalculation slot; it publishes nothing and books nothing.
    if (!scheduler_.coof_recalculation_due(*cause, context, *this)) return;
    scheduler_.recalculate(*cause, context, *this);
    record_applied_range_end();
}
 
native_order::ExecutionTerms source::PineStrategyHost::resolve_execution_terms(
        const NativeExecutionTermsFacts& facts) const {
    return adapter_.resolve_terms(facts);
}
 
bool source::PineStrategyHost::margin_check_allowed(
        const NativeMarginCheckPoint& point) const {
    return adapter_.margin_check_allowed(point);
}
 
std::optional<NativeMarginDecision> source::PineStrategyHost::resolve_margin_requirement(
        const NativeMarginRequirementView& view) const {
    return adapter_.resolve_margin_requirement(view);
}
 
std::optional<double> source::PineStrategyHost::resolve_margin_call_units(
        const NativeMarginCallView& view) const {
    return adapter_.resolve_margin_call_units(view);
}
 
std::optional<double> source::PineStrategyHost::resolve_anchored_level(
        const NativeAnchoredLevelView& view) const {
    return adapter_.resolve_anchored_level(view);
}
 
NativePrecommitVerdict source::PineStrategyHost::validate_execution_precommit(
        const NativePrecommitView& view) const {
    const auto verdict = adapter_.validate_precommit(view);
    precommit_held_units_ = std::abs(detail::run_position(*this).signed_units);
    return verdict;
}
 
void source::PineStrategyHost::configure_pine_strategy(const PineStrategyConfig& config) {
    guard_native_mutation("configure_pine_strategy");
    admit_strategy_declaration(config);
    config_ = config;
    adapter_.set_configuration(config_);
    scheduler_.set_source_series_active(config_.src_series_active);
    source_configuration_captured_ = true;
}
 
void source::PineStrategyHost::set_strategy_override(const StrategyOverrides& overrides) {
    guard_native_mutation("set_strategy_override");
    override_ = overrides;
    config_ = apply_overrides(config_, override_);
    adapter_.set_configuration(config_);
    scheduler_.set_source_series_active(config_.src_series_active);
    source_configuration_captured_ = true;
}
 
void source::PineStrategyHost::set_syminfo_session(const std::string& session) {
    BacktestEngine::set_syminfo_session(session);
}
 
// The six risk setters apply live as they always did. While a selected window's
// pre-roll evaluation captures they also append an ordered row, so the replay
// can set them again in the script's own order against the saved configuration.
// The three int arguments cannot represent NaN/infinity; their existing
// direction/count/cap behavior needs no numeric admission.
void source::PineStrategyHost::set_pine_risk_direction(int direction) {
    detail::capture_risk_row(*this, config_, detail::IntentRiskDirection{direction});
    adapter_.set_risk_direction(direction);
}
 
void source::PineStrategyHost::set_pine_risk_max_cons_loss_days(int value) {
    detail::capture_risk_row(*this, config_, detail::IntentRiskMaxConsLossDays{value});
    adapter_.set_risk_max_cons_loss_days(value);
}
 
void source::PineStrategyHost::set_pine_risk_max_drawdown(double value, bool percent) {
    admit_pine_risk_max_drawdown(value, percent);
    detail::capture_risk_row(*this, config_, detail::IntentRiskMaxDrawdown{value, percent});
    adapter_.set_risk_max_drawdown(value, percent);
}
 
void source::PineStrategyHost::set_pine_risk_max_intraday_loss(double value, bool percent) {
    admit_pine_risk_max_intraday_loss(value, percent);
    detail::capture_risk_row(*this, config_, detail::IntentRiskMaxIntradayLoss{value, percent});
    adapter_.set_risk_max_intraday_loss(value, percent);
}
 
void source::PineStrategyHost::set_pine_risk_max_intraday_filled_orders(int limit) {
    detail::capture_risk_row(*this, config_, detail::IntentRiskMaxIntradayFilledOrders{limit});
    if (limit > 0) adapter_.cap.adopt_script_count();
    adapter_.cap = limit;
}
 
void source::PineStrategyHost::set_pine_risk_max_position_size(double value) {
    admit_pine_risk_max_position_size(value);
    detail::capture_risk_row(*this, config_, detail::IntentRiskMaxPositionSize{value});
    adapter_.set_risk_max_position_size(value);
}
 
int source::PineStrategyHost::pine_bar_index() const {
    // Inside a foreign site's payload, bar_index is the requested context's.
    if (foreign_context_ != nullptr) return static_cast<int>(foreign_context_->delivered - 1);
    return source_bar_index_ + scheduler_.bar_index_offset();
}
 
int source::PineStrategyHost::pine_last_bar_index() const {
    return source_last_bar_index_ + scheduler_.bar_index_offset();
}
 
bool source::PineStrategyHost::is_first_tick() const noexcept {
    return scheduler_.is_first_tick();
}
 
bool source::PineStrategyHost::is_last_tick() const noexcept {
    return scheduler_.is_last_tick();
}
 
compat::pine::CapClock source::PineStrategyHost::fixture_cap_clock() const {
    NativeDecisionContext context;
    if (const auto point = current_execution_point()) {
        context = point->decision;
    } else {
        context.coordinate.interval_index = bar_index_;
        context.sub_bar_open_ms = current_bar_.timestamp;
        context.script_bar_open_ms = current_bar_.timestamp;
    }
    const BarTime time = fixture_chart_time(context.sub_bar_open_ms);
    return {context.sub_bar_open_ms,
            syminfo_.session.empty() ? "24x7" : syminfo_.session,
            syminfo_.timezone.empty() ? "UTC" : syminfo_.timezone,
            time.dayofmonth, time.month};
}
 
compat::pine::Calculation source::PineStrategyHost::fixture_cap_calculation() const {
    NativeDecisionContext context;
    if (const auto point = current_execution_point()) {
        context = point->decision;
    } else {
        context.coordinate.interval_index = bar_index_;
        context.sub_bar_open_ms = current_bar_.timestamp;
        context.script_bar_open_ms = current_bar_.timestamp;
    }
    return adapter_.cap_calculation(context);
}
 
bool source::PineStrategyHost::fixture_intraday_cap_latched() {
    return adapter_.cap.placement(fixture_cap_clock())
        == compat::pine::Placement::Deny;
}
 
source::PineStrategyHost::BarTime source::PineStrategyHost::fixture_chart_time(
        std::int64_t timestamp_ms) const {
    const std::time_t seconds = static_cast<std::time_t>(timestamp_ms / 1000);
    std::tm tm{};
    const auto utc = [&]() {
        return ::gmtime_r(&seconds, &tm) != nullptr;
    };
    if (chart_timezone_.empty() || chart_timezone_ == "UTC"
        || chart_timezone_ == "Etc/UTC") {
        (void)utc();
    } else {
        try {
            tz_util::ScopedTimezone guard(chart_timezone_);
            if (::localtime_r(&seconds, &tm) == nullptr) (void)utc();
        } catch (...) {
            (void)utc();
        }
    }
    BarTime result;
    result.year = tm.tm_year + 1900;
    result.month = tm.tm_mon + 1;
    result.dayofmonth = tm.tm_mday;
    result.hour = tm.tm_hour;
    result.minute = tm.tm_min;
    result.second = tm.tm_sec;
    result.dayofweek = tm.tm_wday + 1;
    result.weekofyear = (tm.tm_yday + 7 - ((tm.tm_wday + 6) % 7)) / 7;
    return result;
}
 
void source::PineStrategyHost::fixture_retain_all_events() {
    as_native_consumer(execution_consumer()).set_retention_override(NativeEventRetention::Full);
}
 
std::uint64_t source::PineStrategyHost::fixture_applied_receipt_count() const {
    // Every ExecutionAppliedEvent the run committed. The adapter acknowledges
    // the journal as it reads it, so the window no longer holds the whole
    // run; the kernel's own count of the events it committed is the same
    // number the journal scan made when it did.
    return as_native_consumer(execution_consumer()).applied_event_count();
}
 
bool source::PineStrategyHost::history_advances_new_bar() const noexcept {
    return scheduler_.history_advances_new_bar();
}
 
 
double source::PineStrategyHost::prev_chart_close() const {
    return scheduler_.previous_chart_close();
}
 
int source::PineStrategyHost::last_bar_dual_entry_path() const {
    return adapter_.pending_intent_view().last_bar_dual_entry_path();
}
 
double source::PineStrategyHost::open_entry_fees_paid() const {
    if (position_side_ == PositionSide::FLAT) return 0.0;
    double paid = 0.0;
    for (const auto& lot : pyramid_entries_) {
        if (lot.qty > internal::kQtyEpsilon) paid += open_entry_commission(lot);
    }
    return paid;
}
 
double source::PineStrategyHost::signed_position_size() const {
    return scheduler_.script_position_view(bar_index_, position_side_, position_qty_);
}
 
double source::PineStrategyHost::script_mark_price(double price) const {
    const double tick = syminfo_mintick_;
    if (!source::detail::script_rule_switches().equity_tick_mark || !std::isfinite(price)
        || !std::isfinite(tick) || !(tick > 0.0)) {
        return price;
    }
    const double ticks = price / tick;
    const double nearest = std::floor(ticks + 0.5);
    // On the grid up to the quotient's own rounding: the close itself.
    if (std::abs(ticks - nearest) <= 1e-9 * std::max(1.0, std::abs(ticks))) return price;
    return nearest * tick;
}
 
void source::PineStrategyHost::freeze_script_position_view() {
    scheduler_.freeze_script_position_view(
        bar_index_, position_side_, position_qty_, pyramid_entries_);
}
 
void source::PineStrategyHost::clear_script_position_view() {
    scheduler_.clear_script_position_view();
}
 
void source::PineStrategyHost::hold_script_average_price(double average) {
    // Only a book the call left flat: any open lot keeps its own average.
    if (position_side_ != PositionSide::FLAT || !std::isfinite(average) || !(average > 0.0))
        return;
    position_entry_price_ = average;
    script_average_held_ = true;
}
 
void source::PineStrategyHost::release_script_average_price() noexcept {
    if (!script_average_held_) return;
    script_average_held_ = false;
    // reset_position_state_to_flat's value; a fill since the hold that
    // opened a position set its own average.
    if (position_side_ == PositionSide::FLAT) position_entry_price_ = 0.0;
}
 
const Series<double>& source::PineStrategyHost::source_series(const std::string& key) const {
    return scheduler_.source_series(key);
}
 
const Series<double>& source::PineStrategyHost::source_input_series(
        const std::string& key, const Series<double>& fallback) const {
    const auto found = inputs_.find(key);
    if (found == inputs_.end() || found->second.empty()) return fallback;
    try {
        return scheduler_.source_series(found->second);
    } catch (const std::invalid_argument&) {
        return fallback;
    }
}
 
double source::PineStrategyHost::live_position_size() const {
    return detail::run_position(*this).signed_units;
}
 
int source::PineStrategyHost::pending_order_count() const {
    return pending_intent_view().size();
}
 
MarketAdmissionJournal& source::PineStrategyHost::market_admission_journal() {
    return adapter_.admission_journal;
}
 
const MarketAdmissionJournal& source::PineStrategyHost::market_admission_journal() const {
    return adapter_.admission_journal;
}
 
std::vector<admission::Field> source::PineStrategyHost::market_admission_fields() const {
    std::vector<admission::Field> fields;
    adapter_.admission_journal.reflect("journal", [&](const admission::Field& field) {
        fields.push_back(field);
    });
    return fields;
}
 
int source::PineStrategyHost::probe_fill_qty(
        int index, double fill_price, double* qty, int* close_only, int* partition) const {
    return pending_intent_view().probe_fill_qty(index, fill_price, qty, close_only, partition);
}
 
int source::PineStrategyHost::pending_order_level_resolved(int index) const {
    return pending_intent_view().level_resolved(index);
}
 
int source::PineStrategyHost::pending_order_effective_levels(
        int index, double* stop, double* limit, double* trail_activation) const {
    return pending_intent_view().effective_levels(index, stop, limit, trail_activation);
}
 
const PendingIntentView& source::PineStrategyHost::pending_intent_view() const noexcept {
    return adapter_.pending_intent_view();
}
 
int source::PineStrategyHost::short_seed_collision_role_v1(
        native_order::RequestHandle handle) const noexcept {
    return adapter_.short_seed_collision_role_v1(std::move(handle));
}
 
void source::PineStrategyHost::enable_pine_intraday_cap() {
    adapter_.enable_intraday_cap();
}
 
void source::PineStrategyHost::attach_pine_execution_adapter() {
    adapter_.attach_execution_adapter();
    // Generated constructors attach the source execution bridge before their
    // risk statements and metadata arrive.  The intraday-cap configuration is
    // part of that same source-policy attachment; leaving it detached makes a
    // later max_intraday_filled_orders declaration silently inert.
    adapter_.enable_intraday_cap();
}
 
void source::PineStrategyHost::set_syminfo_metadata(
        const std::string& key, double value) {
    if (key == "report_terminal_quote_time_ms") {
        if (NativeExecutionConsumer::selected_query_closed(*this)) {
            throw SelectedWindowQueryAfterSeal();
        }
        report_quote_time_ms_ = std::isfinite(value) && value >= 0.0
            && value <= 9007199254740991.0 && std::floor(value) == value
            ? static_cast<std::int64_t>(value) : -1;
        return;
    }
    if (key == "report_terminal_quote_close") {
        if (NativeExecutionConsumer::selected_query_closed(*this)) {
            throw SelectedWindowQueryAfterSeal();
        }
        report_quote_close_ = std::isfinite(value) && value > 0.0
            ? value : std::numeric_limits<double>::quiet_NaN();
        return;
    }
    BacktestEngine::set_syminfo_metadata(key, value);
    // TradingView's symbol calendar (set_symbol_calendar) over the C ABI:
    // "symbol_calendar_days" = N, then "symbol_calendar_open:<i>" and
    // "symbol_calendar_close:<i>" for i < N, installed (and the days consumed)
    // when the run begins. A count that is not a number >= 0 removes the
    // calendar; a day key whose index is not a decimal number is no day.
    if (key.compare(0, 16, "symbol_calendar_") == 0) {
        if (key == "symbol_calendar_days") {
            if (std::isfinite(value) && value >= 0.0 && value <= 1e9) {
                symbol_calendar_days_ = std::llround(value);
            } else {
                symbol_calendar_days_ = -1;
                symbol_calendar_opens_.clear();
                symbol_calendar_closes_.clear();
                symbol_calendar_.clear();
            }
        }
        for (const auto& [prefix, days] : {std::pair<const char*, std::map<long long, double>*>{
                                               "symbol_calendar_open:", &symbol_calendar_opens_},
                                           {"symbol_calendar_close:", &symbol_calendar_closes_}}) {
            const std::size_t n = std::char_traits<char>::length(prefix);
            if (key.size() <= n || key.size() - n > 9 || key.compare(0, n, prefix) != 0) continue;
            long long index = 0;
            bool digits = true;
            for (std::size_t i = n; i < key.size(); ++i) {
                if (key[i] < '0' || key[i] > '9') {
                    digits = false;
                    break;
                }
                index = index * 10 + (key[i] - '0');
            }
            if (digits) (*days)[index] = value;
        }
        return;
    }
    if (key == "bar_index_offset") {
        scheduler_.set_bar_index_offset(std::isfinite(value)
            ? static_cast<int>(std::llround(value)) : 0);
    }
    if (key == "security_range_start_na_warmup") {
        if (std::isfinite(value) && value > 0.0) {
            security_range_start_na_warmup_ = true;
            security_range_start_ms_ = static_cast<int64_t>(std::llround(value));
        } else {
            security_range_start_na_warmup_ = false;
            security_range_start_ms_ = 0;
        }
    }
    if (key == "chart_ema_na_warmup")
        chart_ema_na_warmup_ = std::isfinite(value) && value > 0.0;
    if (key == "historical_security_lookahead_projection")
        historical_security_lookahead_projection_ = std::isfinite(value) && value > 0.0;
    if (key == "margin_long" && config_.margin_long == 100.0)
        config_.margin_long = (std::isfinite(value) && value > 0.0) ? value : 100.0;
    if (key == "margin_short" && config_.margin_short == 100.0)
        config_.margin_short = (std::isfinite(value) && value > 0.0) ? value : 100.0;
    adapter_.set_configuration(config_);
    adapter_.priority.metadata(key, value);
    adapter_.cap.metadata(key, value);
}
 
bool source::PineStrategyHost::set_report_terminal_quote(std::int64_t time_ms, double close) {
    if (NativeExecutionConsumer::selected_query_closed(*this) || time_ms < 0
        || time_ms > 9007199254740991LL || !std::isfinite(close) || close <= 0.0) return false;
    report_quote_time_ms_ = time_ms;
    report_quote_close_ = close;
    return true;
}
 
void source::PineStrategyHost::clear_report_terminal_quote() {
    if (NativeExecutionConsumer::selected_query_closed(*this)) {
        throw SelectedWindowQueryAfterSeal();
    }
    report_quote_time_ms_ = -1;
    report_quote_close_ = std::numeric_limits<double>::quiet_NaN();
}
 
int source::PineStrategyHost::observe_last_bar_dual_entry_path_v1() const {
    return pending_intent_view().last_bar_dual_entry_path();
}
 
int source::PineStrategyHost::observe_pending_count_v1() const {
    return pending_intent_view().size();
}
 
int source::PineStrategyHost::observe_pending_copy_v1(
        int index, pf_pending_order_v1_t* out) const {
    return pending_intent_view().copy_v1(index, out);
}
 
int source::PineStrategyHost::observe_probe_fill_qty(
        int index, double fill_price, double* qty, int* close_only, int* partition) const {
    return pending_intent_view().probe_fill_qty(index, fill_price, qty, close_only, partition);
}
 
int source::PineStrategyHost::observe_pending_level_resolved(int index) const {
    return pending_intent_view().level_resolved(index);
}
 
int source::PineStrategyHost::observe_pending_effective_levels(
        int index, double* stop, double* limit, double* trail_activation) const {
    return pending_intent_view().effective_levels(index, stop, limit, trail_activation);
}
 
double source::PineStrategyHost::observe_trail_best_price_v1() const {
    return adapter_.pending_intent_view().trail_best_price();
}
 
void source::PineStrategyHost::adapter_label_bracket_trades(
        const native_order::ExecutionAppliedEvent& event, bool from_bracket) {
    // ab9714be pine_fills.cpp:6232-6252: every trade row emitted by a real
    // strategy.exit leg carries the bracket cause; strategy.close and
    // close_all requests remain script closes.
    for (std::size_t offset = 0; offset < event.closed_trade_count; ++offset) {
        const std::size_t index = event.first_trade_index + offset;
        if (index >= trades_.size()) continue;
        adapter_.invalidate_exit_group(index); trades_.modify(index, [&](Trade& trade) { trade.exit_from_bracket = from_bracket; });
    }
}
 
bool source::PineStrategyHost::adapter_has_open_entry_id(
        const std::string& id) const {
    return std::any_of(pyramid_entries_.begin(), pyramid_entries_.end(),
        [&](const PyramidEntry& row) { return row.entry_id == id && row.qty > 0.0; });
}
 
const std::vector<source::PineStrategyHost::FixtureIntentRow>&
source::PineStrategyHost::source_pending_view() const {
    source_pending_view_cache_.clear();
    source_pending_view_cache_.reserve(adapter_.pending_same_bar_commands_.size()
        + adapter_.pending_entries_.size() + adapter_.pending_bracket_legs_.size()
        + adapter_.pending_coof_requests_.size()
        + adapter_.delayed_market_orders_.size()
        + adapter_.source_shadow_pending_.size() + adapter_.live_handles_.size());
    const auto append = [&](const PlacementSnapshot& snapshot, const std::string& label) {
        FixtureIntentKind type = FixtureIntentKind::MARKET;
        switch (snapshot.family) {
        case PineOrderFamily::Close:
        case PineOrderFamily::CloseAll:
        case PineOrderFamily::ExitLimit:
        case PineOrderFamily::ExitStop:
        case PineOrderFamily::ExitTrail:
        case PineOrderFamily::Margin:
        case PineOrderFamily::Risk:
            type = FixtureIntentKind::EXIT;
            break;
        case PineOrderFamily::Order:
            type = FixtureIntentKind::RAW_ORDER;
            break;
        case PineOrderFamily::Entry:
            break;
        }
        FixtureIntentRow row;
        row.id = snapshot.family == PineOrderFamily::Close
            ? "__close__" + snapshot.source_id
            : (snapshot.frozen_market_targeted_close ? label : snapshot.source_id);
        row.type = type;
        const bool default_stop = snapshot.family == PineOrderFamily::Entry
            && !std::isfinite(snapshot.exit_levels.limit)
            && std::isfinite(snapshot.exit_levels.stop)
            && std::isnan(snapshot.requested_qty)
            && config_.default_qty_type == static_cast<int>(QtyType::PERCENT_OF_EQUITY)
            && config_.default_qty_value <= 100.0;
        const double absent = std::numeric_limits<double>::quiet_NaN();
        row.default_stop_placement_qty = default_stop ? snapshot.sizing.frozen_units : absent;
        row.default_stop_sizing_price = snapshot.sizing.price;
        row.frozen_market_own_units = snapshot.frozen_market_own_units;
        row.frozen_market_transaction_units = snapshot.frozen_market_transaction_units;
        row.from_entry = snapshot.from_entry;
        row.is_long = snapshot.is_long;
        row.qty = snapshot.family == PineOrderFamily::Close
            ? snapshot.requested_qty
            : (std::isfinite(snapshot.projection_remaining_qty)
                ? snapshot.projection_remaining_qty : snapshot.requested_qty);
        row.qty_percent = snapshot.qty_percent;
        row.created_bar = snapshot.projection_created_bar;
        row.created_seq = static_cast<std::int64_t>(snapshot.source_sequence);
        row.incarnation = snapshot.source_sequence;
        row.over_pyramiding_cap_at_placement = snapshot.projection_over_pyramiding;
        row.paired_flat_market_peer_seq = 0;
        row.paired_flat_market_transaction_qty = std::numeric_limits<double>::quiet_NaN();
        row.frozen_default_qty = default_stop ? absent : snapshot.sizing.frozen_units;
        row.default_stop_placement_equity = default_stop
            ? snapshot.projection_default_stop_equity : absent;
        row.default_stop_placement_signal_close = default_stop
            ? snapshot.projection_default_stop_signal_close : absent;
        row.affordability_placement_equity = snapshot.projection_affordability_equity;
        row.market_admission = snapshot.market_admission;
        if (!row.market_admission.observation()
            && snapshot.family == PineOrderFamily::Entry
            && snapshot.frozen_market_instruction
            && std::isfinite(snapshot.requested_qty)) {
            auto observation = std::make_shared<admission::CommandObservation>();
            observation->command = snapshot.command_ordinal;
            observation->kind = admission::CommandKind::Entry;
            observation->birth = snapshot.birth.cause() == OrderBirthCause::Unattributed
                ? OrderBirth::chart_evaluation(source_bar_index_, current_bar_.timestamp)
                : snapshot.birth;
            observation->id = snapshot.source_id;
            observation->requested_quantity = snapshot.requested_qty;
            observation->quantity_type = snapshot.qty_type;
            observation->buy = snapshot.is_long;
            observation->prices = {snapshot.exit_levels.limit, snapshot.exit_levels.stop};
            observation->oca_name = snapshot.oca_name;
            observation->oca_type = snapshot.oca_type;
            auto& configuration = observation->configuration;
            configuration.process_on_close = config_.process_orders_on_close;
            configuration.calc_on_fills = config_.calc_on_order_fills;
            configuration.slippage = config_.slippage;
            configuration.pyramiding = config_.pyramiding;
            configuration.default_quantity_type = config_.default_qty_type;
            configuration.default_quantity_value = config_.default_qty_value;
            configuration.long_margin = config_.margin_long;
            configuration.short_margin = config_.margin_short;
            configuration.commission_value = config_.commission_value;
            configuration.commission_type = config_.commission_type;
            configuration.pointvalue = staged_configuration().syminfo.pointvalue;
            configuration.fx = snapshot.sizing.fx;
            configuration.quantity_step = staged_configuration().quantity_grid
                ? *staged_configuration().quantity_grid : 0.0;
            configuration.mintick = staged_configuration().syminfo.mintick;
            observation->bar = source_bar_index_;
            observation->placement_side = static_cast<int>(PositionSide::FLAT);
            observation->placement_cycle = snapshot.placement_cycle;
            observation->held_quantity = 0.0;
            observation->held_entries = 0;
            observation->realized_equity = snapshot.sizing.equity;
            observation->placement_equity = snapshot.sizing.equity;
            observation->signal_close = snapshot.sizing.price;
            observation->quantized_fixed_quantity =
                snapshot.frozen_market_own_units;
            observation->original_sizing = admission::SizingObservation{
                snapshot.requested_qty, snapshot.sizing.equity,
                snapshot.sizing.price, snapshot.sizing.mark, snapshot.sizing.fx};
            row.market_admission.bind(std::move(observation));
        }
        source_pending_view_cache_.push_back(std::move(row));
    };
    for (const auto& command : adapter_.pending_same_bar_commands_) {
        // ab9714be strategy.close under process_orders_on_close is held in the
        // same-bar close accumulator until the callback returns; the legacy
        // pending_orders_ observer therefore sees the two entry commands but
        // not that staged close during the source body.
        if (config_.process_orders_on_close
            && command.snapshot.family == PineOrderFamily::Close
            && !command.snapshot.birth.at_terminal_fill()) {
            continue;
        }
        append(command.snapshot, command.request.label);
    }
    for (const auto& pending : adapter_.pending_entries_)
        append(pending.snapshot, pending.request.label);
    for (const auto& delayed : adapter_.delayed_market_orders_)
        append(delayed.snapshot, delayed.request.label);
    for (const auto& leg : adapter_.pending_bracket_legs_)
        append(leg.snapshot, leg.request.label);
    for (const auto& pending : adapter_.pending_coof_requests_)
        append(pending.snapshot, pending.request.label);
    for (const auto& shadow : adapter_.source_shadow_pending_)
        append(shadow.snapshot, shadow.label);
    for (const auto& handle : adapter_.live_handles_) {
        const auto found = adapter_.placement_.find(handle.incarnation);
        if (found == adapter_.placement_.end()) continue;
        if (config_.process_orders_on_close
            && found->second.family == PineOrderFamily::Close
            && found->second.projection_created_bar == source_bar_index_
            && !found->second.birth.at_terminal_fill()) {
            continue;
        }
        append(found->second, found->second.source_id);
    }
    return source_pending_view_cache_;
}
 
void source::PineStrategyHost::source_stream_entry_comment(
        const PyramidEntry&, std::string&) const {}
 
void source::PineStrategyHost::project_short_seed_report_rows(
        const native_order::ExecutionAppliedEvent& event) {
    const ShortSeedPlan plan = adapter_.short_seed_;
    if (!plan.report_swap_pending || event.closed_trade_count == 0
        || event.handle() == plan.final_short) {
        return;
    }
    std::optional<PlacementSnapshot> placement_snapshot;
    if (const auto placement = adapter_.placement_.find(event.handle().incarnation);
        placement != adapter_.placement_.end()) {
        placement_snapshot = placement->second;
    }
    if (!placement_snapshot || placement_snapshot->family != PineOrderFamily::Close
        || (placement_snapshot->from_entry != plan.seed_id
            && placement_snapshot->source_id != plan.seed_id)) {
        return;
    }
    for (std::size_t index = trades_.base(); index < trades_.size(); ++index) {
        Trade trade = trades_[index];
        if (trade.entry_incarnation == plan.materialize_long.incarnation
            && trade.entry_id == plan.materialize_label) {
            trade.entry_incarnation = plan.final_short.incarnation;
            adapter_.invalidate_exit_group(index); trades_.replace(index, trade);
        }
    }
    const std::size_t begin = event.first_trade_index;
    const std::size_t end = begin + event.closed_trade_count;
    for (std::size_t index = begin; index < end && index < trades_.size(); ++index) {
        if (trades_[index].entry_incarnation == plan.final_short.incarnation
            && trades_[index].entry_id == plan.final_short_id) {
            adapter_.invalidate_exit_group(index); trades_.modify(index, [&](Trade& row) {
                row.entry_incarnation = plan.materialize_long.incarnation;
            });
        }
    }
    adapter_.short_seed_.report_swap_pending = false;
}
 
void source::PineStrategyHost::scheduler_prepare_script_run(
        PineScheduler::InputView bars, bool static_eligible,
        int expected_script_bars, bool script_bar_geometry) {
    if (const auto state = native_state(); state.spec) {
        input_tf_ = state.spec->timeframe_undetected ? "" : state.spec->input_tf;
        script_tf_ = state.spec->timeframe_undetected ? "" : state.spec->script_tf;
        script_tf_seconds_ = tf_to_seconds(script_tf_);
    }
    prepare_script_run(bars.empty() ? nullptr : bars.data(), static_cast<int>(bars.size()),
                       static_eligible);
    last_bar_index_ = expected_script_bars - 1;
    last_bar_time_ = bars.empty() ? 0 : bars.back().timestamp;
    apply_realtime_tail_horizon(
        bars.empty() ? nullptr : bars.data(), static_cast<int>(bars.size()),
        script_bar_geometry);
    source_last_bar_index_ = last_bar_index_;
}
 
// Live-runtime tail (spec §3.1): the horizon freeze of last_bar_index_ /
// last_bar_time_. Moved verbatim from src/engine_run.cpp (R5 lane N14); the
// geometry rules are documented on the declaration.
void source::PineStrategyHost::apply_realtime_tail_horizon(const Bar* bars, int n,
                                                  bool script_bar_geometry) {
    if (!realtime_tail_ || realtime_tail_horizon_bars_ <= 0 || n <= 0 || bars == nullptr) return;
    const int horizon = realtime_tail_horizon_bars_;
    last_bar_index_ = horizon - 1;
    const int64_t script_tf_ms =
        static_cast<int64_t>(script_tf_seconds_ > 0 ? script_tf_seconds_ : 0) * 1000;
    if (script_bar_geometry) {
        if (horizon <= n) {
            last_bar_time_ = bars[horizon - 1].timestamp;
        } else {
            last_bar_time_ = bars[n - 1].timestamp
                + static_cast<int64_t>(horizon - n) * script_tf_ms;
        }
    } else {
        last_bar_time_ = bars[0].timestamp
            + static_cast<int64_t>(horizon - 1) * script_tf_ms;
    }
}
 
void source::PineStrategyHost::scheduler_configure_security_evaluators() {
    configure_security_evaluators();
    prune_pine_security_states();
}
 
bool source::PineStrategyHost::scheduler_uses_aux_security_feed() const noexcept {
#ifdef PINEFORGE_HAS_AUX_SECURITY_FEED_V1
    return aux_security_feed_enabled();
#else
    return false;
#endif
}
 
namespace {
 
// N of an N-day timeframe ("2D", "8D"), 0 for any other.
int multi_day_count(const std::string& tf) {
    const auto parsed = native_calendar::parse_timeframe(tf);
    if (!parsed || parsed->unit() != native_calendar::TimeframeUnit::Day) return 0;
    return parsed->count() >= 2 ? parsed->count() : 0;
}
 
}  // namespace
 
void source::PineStrategyHost::scheduler_prepare_security_sequence(
        PineScheduler::InputView bars) {
    security_input_tf_ = input_tf_;
#ifdef PINEFORGE_HAS_AUX_SECURITY_FEED_V1
    if (aux_security_feed_enabled()) security_input_tf_ = aux_security_input_tf_;
#endif
    validate_security_timeframes(security_input_tf_);
    security_first_chart_bar_ms_ = bars.empty() ? 0 : bars.front().timestamp;
    if (declare_security_sites_to_kernel()) {
        // The kernel registers the declared sites after this callback returns
        // and prepares their feeds itself (begin_timeframe_subscriptions).
        // The chart's day partition is the chart's, not a site's, and stays.
        security_next_input_ms_ = 0;
        security_calling_close_ms_ = 0;
        clear_historical_security_lookahead_projections();
        prepare_chart_day_partition(
            bars.empty() ? nullptr : bars.data(), static_cast<int>(bars.size()));
        return;
    }
    init_security_eval_states_for_run(security_input_tf_);
    prepare_native_security_feeds(
        bars.empty() ? nullptr : bars.data(), static_cast<int>(bars.size()));
    // An N-day request's bars are TradingView's (make_multi_day_period_source),
    // its aggregator's period partition: the calendar aggregator split them
    // into single days. The partition is generated from the chart's and the
    // auxiliary feed's bars as the run asks for it, never stored: it reads
    // them until the next prepare rebuilds this aggregator.
    for (auto& state : security_eval_states_) {
        const int days = multi_day_count(state.tf);
        if (days == 0 || state.aggregator.calendar_period() != CalendarPeriod::DAY
            || state.aggregator.has_native_periods()) {
            continue;
        }
        MultiDayFeeds feeds;
        feeds.days = days;
        feeds.chart = bars.data();
        feeds.n_chart = static_cast<int>(bars.size());
        feeds.daily_chart = tf_to_seconds(input_tf_) == 86400;
#ifdef PINEFORGE_HAS_AUX_SECURITY_FEED_V1
        if (aux_security_feed_enabled()) {
            feeds.aux = aux_security_bars_.data();
            feeds.n_aux = aux_security_bars_.size();
        }
#endif
        feeds.tz = syminfo_.timezone;
        feeds.session = syminfo_.session;
        feeds.exchange_calendar = session_template_knows_early_close();
        if (auto partition = make_multi_day_period_source(feeds))
            state.aggregator.set_native_period_source(std::move(partition));
    }
#ifdef PINEFORGE_HAS_AUX_SECURITY_FEED_V1
    if (aux_security_feed_enabled()) {
        prepare_aux_security_chart_ranges(
            bars.empty() ? nullptr : bars.data(), static_cast<int>(bars.size()), script_tf_);
    }
#endif
    prepare_historical_security_lookahead_projections(
        bars.empty() ? nullptr : bars.data(), static_cast<int>(bars.size()), input_tf_);
    prepare_chart_day_partition(
        bars.empty() ? nullptr : bars.data(), static_cast<int>(bars.size()));
}
 
bool source::PineStrategyHost::security_sites_kernel_routed() const noexcept {
    // A foreign site's InstrumentFeed series is always the kernel's; the
    // same-symbol sites are routed exactly when the running spec names one of
    // theirs.
    const NativeRunSpec* spec = detail::run_spec(*this);
    if (spec == nullptr) return false;
    for (const auto& series : spec->subscriptions) {
        if (series.source != NativeSeriesSource::InstrumentFeed) return true;
    }
    return false;
}
 
bool source::PineStrategyHost::declare_security_sites_to_kernel() {
    std::vector<NativeTimeframeSubscription> declared;
    bool routed = same_symbol_sites_routable(declared);
    const std::size_t same_symbol = declared.size();
    // Lane XSYM-D: every foreign site that reads a feed is the kernel's,
    // declared after the same-symbol ones so theirs keep sec_id = index.
    prepare_foreign_security_sites(declared);
    if (declared.empty()) return false;
    if (declared.size() == same_symbol) {
        if (!declare_timeframe_subscriptions(std::move(declared))) return false;
    } else {
        auto result = declare_timeframe_subscriptions_result(declared);
        if (result.status != NativeSetupStatus::Applied && routed) {
            // The same-symbol sites stay on this host's own drive; the
            // foreign series have no other drive, so only their refusal is
            // final.
            declared.erase(declared.begin(),
                           declared.begin() + static_cast<std::ptrdiff_t>(same_symbol));
            foreign_security_series_.erase(
                foreign_security_series_.begin(),
                foreign_security_series_.begin() + static_cast<std::ptrdiff_t>(same_symbol));
            for (auto& entry : foreign_security_sites_) {
                if (entry.second.subscription >= 0)
                    entry.second.subscription -= static_cast<std::int64_t>(same_symbol);
            }
            routed = false;
            result = declare_timeframe_subscriptions_result(declared);
        }
        if (result.status != NativeSetupStatus::Applied) {
            RunFailureArgs refused{{"reason", "kernel_refused_series"}};
            if (const char* field = instrument_feed_field_name(result.validation.field))
                refused.emplace_back("field", field);
            throw coded<std::runtime_error>(
                RunFailureCode::symbol_feeds_refused, refused,
                "request.security: the kernel refused the symbol feed series (NativeRunSpecError "
                + std::to_string(static_cast<int>(result.validation.error)) + ", field "
                + std::to_string(static_cast<int>(result.validation.field)) + ")");
        }
    }
    // The kernel registers the routed sites, sec_id by index, after this
    // callback returns; the per-site table keeps their Pine semantics.
    if (routed) security_eval_states_.clear();
    return routed;
}
 
bool source::PineStrategyHost::same_symbol_sites_routable(
        std::vector<NativeTimeframeSubscription>& out) {
    if (security_eval_states_.empty()) return false;
    // Run shapes with a Pine-only rule around the step: a stream feeds its
    // sites from realtime prints (the kernel takes confirmed bars only); an
    // aggregated chart or the bar magnifier can hold an input back until the
    // calling bar's callback has run (PineScheduler::input's deferrals); the
    // auxiliary slice is fed per chart bar; the range-start cut, the
    // historical lookahead projection and the OTC-daily pins veto or
    // re-shape inputs before the aggregator sees them.
    if (stream_warmup_mode_ || scheduler_.bar_magnifier_enabled()) return false;
    if (security_range_start_na_warmup_ || historical_security_lookahead_projection_)
        return false;
#ifdef PINEFORGE_HAS_AUX_SECURITY_FEED_V1
    if (aux_security_feed_enabled()) return false;
#endif
    if (input_tf_.empty() || input_tf_ != script_tf_ || security_input_tf_ != input_tf_)
        return false;
    const auto view = native_state();
    if (view.spec == nullptr || view.spec->timeframe_undetected
        || view.spec->input_tf != input_tf_) {
        return false;
    }
    const bool otc_daily_pins = (syminfo_.type == "forex" || syminfo_.type == "cfd")
        && native_security_feeds_.empty() && script_tf_seconds_ > 0
        && script_tf_seconds_ < 86400;
    std::vector<NativeTimeframeSubscription> declared;
    declared.reserve(security_eval_states_.size());
    for (std::size_t i = 0; i < security_eval_states_.size(); ++i) {
        const SecurityEvalState& state = security_eval_states_[i];
        const bool minute_day = calendar_period_for(state.tf) == CalendarPeriod::NONE
            && tf_to_seconds(state.tf) == 86400;
        if (minute_day && !pine_security_state(state.sec_id).lookahead_on
            && script_tf_seconds_ > 0 && script_tf_seconds_ < 86400) {
            return false;
        }
        // The kernel registers sec_id = index, which is what generated code
        // dispatches on.
        if (state.sec_id != static_cast<int>(i) || state.tf.empty()) return false;
        const PineSecurityEvalState& pine = pine_security_state(state.sec_id);
        // Site rules inside the step: the lookahead_on peeks and merge latch
        // (and the publication gate they arm), ticker.heikinashi, both
        // lower-timeframe paths, the auxiliary-slice completions.
        if (pine.lookahead_on || pine.heikinashi || pine.lower_tf_requested
            || pine.lower_tf_emulation || pine.lower_tf_use_input
            || pine.lower_tf_array_requested || pine.publish_gate_tf_seconds > 0
            || pine.calling_close_completes_partial || pine.calling_open_latches_first) {
            return false;
        }
        if (otc_daily_pins && (state.tf == "D" || state.tf == "1D")) return false;
        // TradingView's N-day bars restart every year and count the symbol's
        // trading dates (make_multi_day_period_source); the kernel's are a
        // fixed grid of calendar days.
        if (multi_day_count(state.tf) != 0) return false;
        NativeTimeframeSubscription subscription;
        subscription.tf = state.tf;
        subscription.gaps = pine.gaps_on;
        declared.push_back(std::move(subscription));
    }
    out = std::move(declared);
    return true;
}
 
bool source::PineStrategyHost::scheduler_feed_security_input(
        const Bar& bar, std::int64_t next_input_ms, bool calling_bar_complete,
        bool defer_boundary_gate) {
    security_next_input_ms_ = next_input_ms;
    security_calling_close_ms_ = 0;
    bool deferred = false;
    for (auto& state : security_eval_states_) {
        if (defer_boundary_gate
            && pine_security_state(state.sec_id).publish_gate_tf_seconds > 0) {
            deferred = true;
            continue;
        }
        pine_feed_security_eval_state(state, bar, calling_bar_complete);
    }
    return deferred;
}
 
void source::PineStrategyHost::scheduler_publish_security_boundary() {
    for (auto& state : security_eval_states_) {
        if (pine_security_state(state.sec_id).publish_gate_tf_seconds > 0)
            publish_security_eval_state_at_calling_boundary(state);
    }
}
 
void source::PineStrategyHost::scheduler_feed_deferred_security_input(
        const Bar& bar, std::int64_t next_input_ms) {
    security_next_input_ms_ = next_input_ms;
    security_calling_close_ms_ = 0;
    for (auto& state : security_eval_states_) {
        if (pine_security_state(state.sec_id).publish_gate_tf_seconds > 0)
            pine_feed_security_eval_state(state, bar, false);
    }
}
 
void source::PineStrategyHost::scheduler_feed_aux_security(int chart_index) {
#ifdef PINEFORGE_HAS_AUX_SECURITY_FEED_V1
    if (aux_security_feed_enabled() && scheduler_coof_enabled()) {
        // ab9714be pine_scheduler.cpp:1705-1708 feeds the auxiliary slice
        // before dispatch_bar, whose COOF bar takes its script checkpoint
        // only afterwards (pine_scheduler.cpp:467).  The native scheduler
        // checkpoints at the script-bar open and restores that checkpoint
        // after this feed, which would discard the fed request.security
        // values every bar.  Rebase the checkpoint on the open state plus
        // this feed so the restore keeps them.
        restore_script_state();
        feed_aux_security_for_chart_bar(chart_index);
        snapshot_script_state();
        return;
    }
    if (aux_security_feed_enabled()) feed_aux_security_for_chart_bar(chart_index);
#else
    (void)chart_index;
#endif
}
 
void source::PineStrategyHost::scheduler_feed_deferred_aux_security(int chart_index) {
#ifdef PINEFORGE_HAS_AUX_SECURITY_FEED_V1
    if (aux_security_feed_enabled()) feed_deferred_aux_security_for_chart_bar(chart_index);
#else
    (void)chart_index;
#endif
}
 
void source::PineStrategyHost::scheduler_finish_security_sequence() {
    clear_historical_security_lookahead_projections();
#ifdef PINEFORGE_HAS_AUX_SECURITY_FEED_V1
    release_aux_security_chart_ranges();
#endif
}
 
// sort_same_bar_exit_trades' first test, asked before the call (R5 lane D2-D):
// its group grows past the last row only when the row before it is a bracket
// exit of the same entry at the same exit time, as the last row is. Otherwise
// the group is one row, the sort moves nothing and the call answers nullopt.
// A quiet-bar gate (pine_quiet_bar.hpp): the ungated reference calls the sort
// at every publication.
static bool same_bar_exit_group_possible(const TradeLedger& trades) {
#if PINEFORGE_PINE_QUIET_BAR_GATES
    if (trades.size() < 2) return false;
    const auto last_index = trades.size() - 1;
    const auto previous_index = last_index - 1;
    const Trade last = trades.numeric_at(last_index);
    const Trade previous = trades.numeric_at(previous_index);
    if (!previous.exit_from_bracket || !last.exit_from_bracket
        || previous.exit_time != last.exit_time || previous.entry_time != last.entry_time)
        return false;
    // Entry identity is still exact, but comments and exit IDs are irrelevant
    // here. Do not decode even the entry IDs unless the numeric test matches.
    const auto last_entry_id = trades.entry_id_at(last_index);
    return trades.entry_id_at(previous_index) == last_entry_id;
#else
    (void)trades;
    return true;
#endif
}
 
bool source::PineStrategyHost::report_terminal_quote_eligible() const {
    if (!last_equity_point() || range_end_trades_.empty() || !native_bound()) return false;
    const auto state = as_native_consumer(execution_consumer()).view();
    const auto interval = static_cast<std::int64_t>(script_tf_seconds_) * 1000;
    const auto last_time = last_equity_point()->time_ms;
    return state.spec && state.spec->report_policy == NativeReportPolicy::KernelRecordedAtHostMarks
        && state.kind == NativeLifecycleKind::Completed && bar_magnifier_enabled_
        && script_tf_seconds_ > 60 && script_tf_seconds_ < 86400
        && report_quote_time_ms_ >= last_time && report_quote_time_ms_ - last_time <= interval
        && (report_quote_time_ms_ - last_time) % interval == 0
        && std::isfinite(report_quote_close_)
        && range_end_trades_.size() == pyramid_entries_.size();
}
 
bool source::PineStrategyHost::report_terminal_quote_applied(const ReportC& report) const {
    if (!report_terminal_quote_eligible() || report.equity_curve_len == 0
        || report.equity_curve[report.equity_curve_len - 1].time_ms != report_quote_time_ms_) return false;
    std::size_t rows = 0;
    for (int index = 0; index < report.trades_len; ++index) {
        const auto& row = report.trades[index];
        if (!row.open_at_end) continue;
        if (row.exit_time != report_quote_time_ms_ || row.exit_price != bar_fill_price(report_quote_close_))
            return false;
        ++rows;
    }
    return rows == pyramid_entries_.size();
}
 
void source::PineStrategyHost::present_report(ReportC* out) const {
    if (out->equity_curve_len == 0 || !native_bound()) return;
    const auto state = as_native_consumer(execution_consumer()).view();
    if (!state.spec
        || state.spec->report_policy != NativeReportPolicy::KernelRecordedAtHostMarks) return;
    const auto position = as_native_consumer(execution_consumer()).position(*this);
    auto& last = out->equity_curve[out->equity_curve_len - 1];
    if ((stream_phase_ == StreamPhase::REALTIME
         || state.completion == NativeCompletion::StreamEnded)
        && (!range_end_trades_.empty()
            || (position.signed_units == 0.0 && position.lot_count == 0
                && last.open_profit != 0.0))) {
        last.open_profit = 0.0;
        if (!range_end_trades_.empty()) {
            double range_end_pnl = 0.0;
            for (const auto& row : range_end_trades_) range_end_pnl += row.pnl;
            last.equity = initial_capital_ + net_profit_sum_ + range_end_pnl;
            amend_report_replay_point(last.time_ms, report::EquityBasis::range_end, range_end_pnl);
        }
    }
    if (!report_terminal_quote_eligible()) return;
    const double price = bar_fill_price(report_quote_close_);
    const double fx = account_currency_fx_at(report_quote_time_ms_);
    std::size_t lot_index = 0;
    for (int index = 0; index < out->trades_len; ++index) {
        auto& row = out->trades[index];
        if (!row.open_at_end) continue;
        const auto& lot = pyramid_entries_[lot_index++];
        execution::PhysicalExecutionContext context;
        context.effective_time_ms = report_quote_time_ms_;
        context.interval_index = row.exit_bar_index;
        context.account_fx = fx;
        const auto mark = build_close_trade_with_costs(lot, row.qty, price, row.is_long != 0,
            allocated_entry_commission(lot, row.qty), calc_commission_at(price, row.qty, fx), context);
        // settlement safety guard (an invariant), not Pine parity. The price
        // test is implied by the predicate (the mark carries it as exit_price);
        // no mutant can kill it alone.
        if (!internal::closed_row_all_finite(mark) || !std::isfinite(price))
            throw std::logic_error("non-finite terminal-quote report row");
        row.exit_time = report_quote_time_ms_;
        row.exit_price = price;
        row.pnl = mark.pnl;
        row.pnl_pct = mark.pnl_pct;
        row.commission = mark.commission;
    }
    out->net_profit = out->total_trades == out->trades_len ? 0.0 : report_closed_net_profit();
    for (int index = 0; index < out->trades_len; ++index) out->net_profit += out->trades[index].pnl;
    last.time_ms = report_quote_time_ms_;
    last.equity = initial_capital_ + out->net_profit;
    amend_report_replay_point(last.time_ms, report::EquityBasis::terminal_quote, out->net_profit);
}
 
void source::PineStrategyHost::scheduler_record_range_end(const Bar& terminal_bar) {
    range_end_trades_.clear();
    if (stream_warmup_mode_ || realtime_tail_
        || position_side_ == PositionSide::FLAT || !last_equity_point()
        || !std::isfinite(terminal_bar.close)) return;
    const double fill_price = bar_fill_price(terminal_bar.close);
    // R5 audit lane Q6 (duplicate D5): the row loop itself is the kernel's
    // generic producer, called here at TradingView's own mark instead of
    // being restated. What stays below is the part that is report SHAPE and
    // not a mark-to-market row — the reason the kernel's run-end producer is
    // gated out of every Pine run (pine_adapter.cpp project()). The rows are
    // dated on the curve's last point and convert at that instant's rate,
    // handed to the producer: the presented clock is never rewritten (R5
    // lane B-ADAPTER; E21's rule -- thread the rate, never the clock).
    const std::int64_t mark_time = last_equity_point()->time_ms;
    const double range_end_pnl = as_native_consumer(execution_consumer())
        .append_open_position_report_rows(*this, fill_price, mark_time, bar_index_,
                                          account_currency_fx_at(mark_time));
    if (stream_phase_ != StreamPhase::REALTIME) {
        auto& last = *last_equity_point();
        last.open_profit = 0.0;
        last.equity = initial_capital_ + net_profit_sum_ + range_end_pnl;
        amend_report_replay_point(last.time_ms, report::EquityBasis::range_end, range_end_pnl);
        // Restore the fold before the held point, then fold its replacement.
        // Debug checks all four resulting doubles against a complete walk.
        // The sink-off helper keeps the existing walk and initial seed.
        // Presentation later receives its own copy of this editable tail.
        refold_equity_tail();
    }
    if (same_bar_exit_group_possible(trades_)) {
        if (const auto moved = sort_same_bar_exit_trades(trades_, adapter_))
            native_closed_rows_amended(*moved);
    }
}
 
// TradingView keeps one child of a strategy.exit that names no entry for every
// open entry. When one path point triggers several, it fills them one at a
// time: the children of the entry id that comes first in its entry-id table
// first (PineExecutionAdapter::global_exit_child_ranks), and an older entry's
// child before a newer one's of the same id. Each child is a transaction of
// its entry's quantity, settled against its own entry under
// close_entries_rule = "ANY" and against the oldest lots first otherwise. Each
// settled slice is a row of its own, and its fee is its share of its lot's
// entry fee plus its share of its transaction's exit fee, each on
// TradingView's ten-significant-digit money (127 TradingView tapes,
// tests/fixtures/global_exit_children).
// The kernel fills such a leg once for the whole position, oldest lots first,
// one row per lot. Its rows are still the last ones and not final until this
// notification returns, so they are re-cut here into the children's rows.
// Every shape the pin does not reach keeps the kernel's rows:
//   - a fill that leaves part of the position open;
//   - a trailing leg;
//   - an exit filled at the same point before it, or a leg of any other
//     exit live at the fill that its price also reaches (the order of a
//     named child, or another exit's, against these; on_native_applied reads
//     it before the adapter retires that leg);
//   - a table the pin does not size (global_exit_child_ranks);
//   - calc_on_order_fills, process_orders_on_close or an intrabar feed;
//   - a non-unit account rate, a cash-per-order fee or a negative fee;
//   - a lot quantity off TradingView's ten significant digits.
void source::PineStrategyHost::order_global_exit_children(
        const native_order::ExecutionAppliedEvent& event) {
    const std::size_t first = event.first_trade_index;
    const std::size_t count = event.closed_trade_count;
    if (count < 2 || event.opened_units != 0.0 || first + count != trades_.size()
        || !pyramid_entries_.empty()) {
        return;
    }
    const auto& run = adapter_.config_;
    if (run.calc_on_order_fills || run.process_orders_on_close || bar_magnifier_enabled_
        || active_account_currency_fx() != 1.0 || commission_value_ < 0.0
        || (commission_value_ != 0.0 && commission_type_ == CommissionType::CASH_PER_ORDER)) {
        return;
    }
    const auto placement = adapter_.placement_.find(event.handle().incarnation);
    if (placement == adapter_.placement_.end()) return;
    const auto& cause = placement->second;
    if ((cause.family != PineOrderFamily::ExitLimit && cause.family != PineOrderFamily::ExitStop)
        || !cause.from_entry.empty()) {
        return;
    }
    // Ledger reads own their strings. Retain only the group keys so a wide
    // comment does not stay resident while another row is being read.
    Trade head{};
    {
        Trade row = trades_[first];
        head.exit_time = row.exit_time;
        head.exit_price = row.exit_price;
        head.exit_id = std::move(row.exit_id);
        head.is_long = row.is_long;
        head.entry_id = std::move(row.entry_id);
    }
    auto& scratch = trade_scratch();
    TradeScratchTransaction transaction(scratch);
    bool different_ids = false;
    for (std::size_t index = first; index < first + count; ++index) {
        const Trade& row = trades_[index];
        if (row.exit_time != head.exit_time || row.exit_price != head.exit_price
            || row.exit_id != head.exit_id || row.is_long != head.is_long
            || !(row.qty > 0.0) || row.open_at_end) {
            return;
        }
        const TradeScratchKey incarnation{scratch_incarnation, row.entry_incarnation, 0, 0};
        if (scratch_find(scratch, incarnation) != ::pineforge::detail::scratch_no_handle) return;
        scratch_put(scratch, incarnation);
        different_ids = different_ids || row.entry_id != head.entry_id;
    }
    if (!different_ids) return;
    if (first > 0 && trades_[first - 1].exit_time == head.exit_time
        && trades_[first - 1].exit_price == head.exit_price) {
        return;
    }
    // Validate one id and each distinct pair through the existing table rule:
    // a grown table rejects shared buckets, while the original table does
    // not. Each query owns at most two ids, irrespective of the lot count.
    // note_exit_rank_key retains at most 27 keys (the existing parity pin),
    // so distinct-id work is fixed even when one id owns arbitrarily many lots.
    std::array<std::size_t, 27> ages{};
    std::array<std::uint32_t, 27> ranks{};
    std::size_t age_count = 0;
    const auto age_of = [&](std::size_t lot) {
        const auto& keys = adapter_.exit_rank_keys_;
        const auto found = keys.find(trades_[first + lot].entry_id);
        return found ? static_cast<std::size_t>(keys.get(*found))
                     : static_cast<std::size_t>(keys.size());
    };
    const auto key_of = [&](std::size_t age) {
        SourceId id;
        adapter_.exit_rank_keys_.for_each_inserted([&](SourceId key, std::uint32_t key_age) {
            if (key_age == age) id = std::move(key);
        });
        return id;
    };
    std::vector<SourceId> ids;
    for (std::size_t lot = 0; lot < count; ++lot) {
        const auto age = age_of(lot);
        if (age == adapter_.exit_rank_keys_.size()) return;
        if (std::find(ages.begin(), ages.begin() + age_count, age) != ages.begin() + age_count)
            continue;
        ids.assign(1, trades_[first + lot].entry_id);
        const auto rank = adapter_.global_exit_child_ranks(ids);
        if (!rank) return;
        ranks[age] = rank->front();
        for (std::size_t earlier = 0; earlier < age_count; ++earlier) {
            ids.resize(1);
            ids.push_back(key_of(ages[earlier]));
            if (!adapter_.global_exit_child_ranks(ids)) return;
        }
        ages[age_count++] = age;
    }
    const auto rank_of = [&](std::size_t lot) {
        return ranks[age_of(lot)];
    };
    // Children in lot order settle exactly as the kernel's one fill did.
    bool ordered = true;
    auto previous_rank = rank_of(0);
    for (std::size_t lot = 1; lot < count; ++lot) {
        const auto rank = rank_of(lot);
        if (rank < previous_rank) ordered = false;
        previous_rank = rank;
    }
    if (ordered) return;
    auto* const replay = report_replay_state();
    std::unique_ptr<::pineforge::detail::ScratchSequence<report::ReplayTrade>> replay_rows;
    if (replay) {
        replay_rows = std::make_unique<::pineforge::detail::ScratchSequence<report::ReplayTrade>>();
        replay_rows->set_directory(replay->scratch_directory());
    }
    const auto visit_children = [&](const auto& visit) {
        for (std::size_t rank = 0; rank < adapter_.exit_rank_keys_.size(); ++rank)
            for (std::size_t child = 0; child < count; ++child)
                if (rank_of(child) == rank && !visit(child)) return false;
        return true;
    };
    std::size_t row_count = 0;
    const auto emit_row = [&](TradeScratchHeader header, const Trade& text) {
        const TradeScratchSource row(header, text);
        scratch_put(scratch, {scratch_row, row_count++, 0, 0}, row);
    };
    const auto header_for = [&](std::size_t source, const Trade& row) {
        const auto phase = adapter_.exit_phase_value(source);
        return scratch_trade_header(row, source, phase);
    };
    const auto annotate = [&](std::size_t lot, std::size_t child, double transaction_qty,
                              bool fifo, bool fees, double exit_fee, double transaction_fee) {
        if (!replay) return;
        const Trade whole = trades_.numeric_at(first + lot);
        auto row = replay->trade(first + lot);
        row.parent = event.ordinal;
        row.slice = row_count - 1;
        row.child = child;
        row.transaction_qty = transaction_qty;
        row.whole_qty = whole.qty;
        row.whole_pnl = whole.pnl;
        row.whole_commission = whole.commission;
        row.whole_runup = whole.max_runup;
        row.whole_drawdown = whole.max_drawdown;
        row.whole_exit_fee = exit_fee;
        row.transaction_fee = transaction_fee;
        row.marked = true;
        row.fifo = fifo;
        row.fees = fees;
        replay_rows->push_back(row);
    };
    // The rule the adapter runs under, not the host's live copy a body writes.
    if (run.close_entries_rule_any) {
        visit_children([&](std::size_t lot) {
            const Trade row = trades_[first + lot];
            emit_row(header_for(first + lot, row), row);
            if (replay) annotate(lot, lot, row.qty, false, false, 0.0, 0.0);
            return true;
        });
    } else {
        const bool fees = commission_value_ != 0.0;
        for (std::size_t lot = 0; lot < count; ++lot) {
            const double qty = trades_[first + lot].qty;
            if (tv_money_round(qty) != qty) return;
        }
        const auto visit_slices = [&](const auto& visit) {
            std::size_t lot = 0;
            double left = trades_[first].qty;
            if (!visit_children([&](std::size_t child) {
                    const double transaction = trades_[first + child].qty;
                    double wanted = transaction;
                    while (wanted > internal::kQtyEpsilon && lot < count) {
                        const double taken = std::min(left, wanted);
                        visit(lot, child, taken, transaction);
                        left = tv_money_round(left - taken);
                        wanted = tv_money_round(wanted - taken);
                        if (left <= internal::kQtyEpsilon) {
                            ++lot;
                            if (lot < count) left = trades_[first + lot].qty;
                        }
                    }
                    return !(wanted > internal::kQtyEpsilon);
                })) return false;
            return lot == count;
        };
        // Complete the exact quantity pass before calculating any replacement
        // money, as the old slice-vector pass did.
        if (!visit_slices([](std::size_t, std::size_t, double, double) {})) return;
        visit_slices([&](std::size_t lot, std::size_t child, double qty, double transaction) {
            const Trade& whole = trades_[first + lot];
            const double share = qty / whole.qty;
            double commission = 0.0;
            double replay_exit_fee = 0.0, replay_transaction_fee = 0.0;
            if (fees) {
                const double exit_fee = calc_commission_at(whole.exit_price, whole.qty, 1.0);
                const double entry_part =
                    tv_money_round(std::max(0.0, whole.commission - exit_fee) * share);
                const double transaction_fee =
                    tv_money_round(calc_commission_at(whole.exit_price, transaction, 1.0));
                const double exit_part = tv_money_round(transaction_fee * qty / transaction);
                commission = tv_money_round(entry_part + exit_part);
                if (replay) {
                    replay_exit_fee = exit_fee;
                    replay_transaction_fee = transaction_fee;
                }
            }
            auto header = header_for(first + lot, whole);
            Trade row{};
            scratch_trade_scalars(header, row);
            row.qty = qty;
            row.commission = commission;
            row.pnl = tv_money_round((whole.pnl + whole.commission) * share) - commission;
            const double cost = whole.entry_price * qty * syminfo_.pointvalue;
            row.pnl_pct = cost > 0.0 ? row.pnl / cost * 100.0 : 0.0;
            // The lot's excursions taken by share: an approximation where the
            // slice's lot is not its child's own (excursions are not graded).
            row.max_runup = whole.max_runup * share;
            row.max_drawdown = whole.max_drawdown * share;
            header[6] = scratch_word(row.qty);
            header[7] = scratch_word(row.pnl);
            header[8] = scratch_word(row.pnl_pct);
            header[13] = scratch_word(row.max_runup);
            header[14] = scratch_word(row.max_drawdown);
            header[15] = scratch_word(row.commission);
            emit_row(header, whole);
            if (replay) annotate(lot, child, transaction, true, fees,
                                 replay_exit_fee, replay_transaction_fee);
        });
    }
    // settlement safety guard (an invariant), not Pine parity. Validate all
    // replacements before changing the committed rows or their tallies.
    scratch_walk(scratch, scratch_row, [&](::pineforge::detail::ScratchHandle handle, const TradeScratchKey&) {
        Trade row{};
        scratch_trade_scalars(scratch_read_trade(scratch, handle), row);
        if (!internal::closed_row_all_finite(row))
            throw std::logic_error("non-finite amended trade row");
    });
    // strategy.wintrades / losstrades / eventrades and the gross sums count
    // rows: count the children's instead of the kernel's. net_profit_sum_ (the
    // equity) stays the kernel's whole-lot sum: the slices' PnLs can differ
    // from it by at most half a unit in the tenth significant digit of each
    // slice's gross and fee, which no tape has shown in strategy.netprofit or
    // in an equity-sized quantity.
    const auto tally = [this](double pnl, int sign) {
        if (pnl > 0.0) { gross_profit_sum_ += sign * pnl; win_trades_count_ += sign; }
        else if (pnl < 0.0) { gross_loss_sum_ += sign * pnl; loss_trades_count_ += sign; }
        else { eventrades_count_ += sign; }
    };
    for (std::size_t index = first; index < first + count; ++index) tally(trades_[index].pnl, -1);
    scratch_walk(scratch, scratch_row, [&](::pineforge::detail::ScratchHandle handle, const TradeScratchKey&) {
        Trade row{};
        scratch_trade_scalars(scratch_read_trade(scratch, handle), row);
        tally(row.pnl, 1);
    });
    if (replay) {
        if (replay_rows->size() != row_count)
            throw coded<std::logic_error>(RunFailureCode::engine_invariant, {},
                                          "recut replay rows differ from replacement rows");
        replay->resize_trades(first + row_count);
        for (std::size_t i = 0; i < row_count; ++i) {
            auto annotation = replay_rows->at(i);
            annotation.slices = row_count;
            replay->replace(first + i, annotation);
        }
    }
    adapter_.invalidate_exit_group(first, trades_.size() - first); trades_.resize(first);
    scratch_walk(scratch, scratch_row, [&](::pineforge::detail::ScratchHandle handle, const TradeScratchKey&) {
        const auto header = scratch_read_trade(scratch, handle);
        Trade row{};
        scratch_trade_scalars(header, row);
        scratch_trade_text(scratch, handle, header, row);
        trades_.push_back(std::move(row));
    });
    const auto none = static_cast<std::uint8_t>(NativePathPhase::None);
    scratch_walk(scratch, scratch_row, [&](::pineforge::detail::ScratchHandle handle, const TradeScratchKey& key) {
        const auto header = scratch_read_trade(scratch, handle);
        const auto target = first + static_cast<std::size_t>(key[1]);
        const auto phase = static_cast<std::uint8_t>(header[1]);
        if (phase == none && adapter_.exit_phase_value(target) == none) return;
        adapter_.set_exit_phase(target, phase);
    });
    transaction.finish();
    native_closed_rows_amended(first);
}
 
// ab9714be pine_fills.cpp:664-670: same-bar bracket exit trades sort by script command sequence created_seq
// The rows are final kernel rows by now (their applied notifications have
// returned), so a sort that moves one answers the first row it moved, which
// the caller hands to the kernel before the next hash read
// (NativeStrategyHost::native_closed_rows_amended).
std::optional<std::size_t> source::PineStrategyHost::sort_same_bar_exit_trades(
        TradeLedger& trades, source::PineExecutionAdapter& adapter) {
    ++adapter.exit_group_work_.publications; if (adapter.exit_group_certified(trades) || trades.size() < 2) return std::nullopt;
    const std::size_t end = trades.size();
    std::size_t start = end - 1;
    while (start > 0
           && trades[start - 1].exit_time == trades[end - 1].exit_time
           && trades[start - 1].entry_time == trades[end - 1].entry_time
           && trades[start - 1].entry_id == trades[end - 1].entry_id
           && trades[start - 1].exit_from_bracket
           && trades[end - 1].exit_from_bracket) {
        --start; ++adapter.exit_group_work_.tail_rows;
    }
    if (end - start > 1) {
        // ab9714be pine_scheduler.cpp:262-283 and pine_fills.cpp:3710-3730:
        // fills on the bar OPEN precede intrabar fills chronologically. Sibling
        // bracket exits that fill within the same phase tie-break by command
        // sequence (test_l10m_corpus_parity.cpp).
        const auto less = [&](std::size_t ia, std::size_t ib) {
            ++adapter.exit_group_work_.comparisons; const bool open_a = adapter.is_open_phase_exit(ia);
            const bool open_b = adapter.is_open_phase_exit(ib);
            if (open_a != open_b) return open_a;
            const auto sa = adapter.command_sequence_for_exit(trades[ia].exit_id, trades[ia].entry_id);
            const auto sb = adapter.command_sequence_for_exit(trades[ib].exit_id, trades[ib].entry_id);
            return sa < sb;
        };
        const auto none = static_cast<std::uint8_t>(NativePathPhase::None);
        bool ordered = true;
        for (std::size_t index = start + 1; index < end; ++index)
            if (less(index, index - 1)) { ordered = false; break; }
        if (ordered) { adapter.certify_exit_group(trades, start); return std::nullopt; }
        auto& scratch = trade_scratch(); adapter.certify_exit_group(trades, start, false);
        const auto scratch_before = scratch.stats(); TradeScratchTransaction transaction(scratch);
        for (std::size_t source = start; source < end; ++source) {
            const auto phase = adapter.exit_phase_value(source);
            const auto sequence = adapter.command_sequence_for_exit(
                trades[source].exit_id, trades[source].entry_id);
            scratch_put(scratch,
                {scratch_sort, adapter.is_open_phase_exit(source) ? 0U : 1U, sequence, source},
                scratch_bytes(&phase, sizeof phase));
        }
        std::size_t target = start;
        // source is the last key word, so ties retain their original order.
        // This is a seek walk, not a store callback: the fixed mapping value
        // is read out before put, and its source never reads the same store.
        scratch_walk(scratch, scratch_sort,
            [&](::pineforge::detail::ScratchHandle handle, const TradeScratchKey& key) {
                std::uint8_t phase = none;
                require_trade_scratch(scratch.read(handle, ::pineforge::detail::ScratchField::value, 0,
                                                   scratch_buffer(&phase, sizeof phase)), scratch);
                const std::array<std::uint64_t, 2> mapping{key[3], phase};
                scratch_put(scratch, {scratch_permutation, target++, 0, 0},
                            scratch_bytes(mapping.data(), sizeof mapping));
            });
        // Apply disjoint permutation cycles with one saved Trade. Erasing a
        // mapping marks that target visited in backing, without a RAM bitmap.
        for (std::size_t cycle = start; cycle < end; ++cycle) {
            auto handle = scratch_find(scratch, {scratch_permutation, cycle, 0, 0});
            if (handle == ::pineforge::detail::scratch_no_handle) continue;
            const Trade row = trades[cycle];
            auto* const replay = report_replay_state();
            const auto saved_replay = replay ? replay->trade(cycle) : report::ReplayTrade{};
            std::size_t current = cycle;
            for (;;) {
                std::array<std::uint64_t, 2> mapping{};
                require_trade_scratch(scratch.read(handle, ::pineforge::detail::ScratchField::value, 0,
                    scratch_buffer(mapping.data(), sizeof mapping)), scratch);
                const auto source = static_cast<std::size_t>(mapping[0]);
                const auto phase = static_cast<std::uint8_t>(mapping[1]);
                // Both indices are global; only the adapter translates the
                // sealed base. Clear an existing phase even when None moves
                // here, without extending an absent all-None tail.
                if (phase != none || adapter.exit_phase_value(current) != none)
                    adapter.set_exit_phase(current, phase);
                ++adapter.exit_group_work_.replacements; if (source == cycle) trades.replace(current, row);
                else trades.replace(current, trades[source]);
                if (replay)
                    replay->replace(current, source == cycle ? saved_replay : replay->trade(source));
                require_trade_scratch(scratch.erase(handle), scratch);
                if (source == cycle) break;
                current = source;
                handle = scratch_find(scratch, {scratch_permutation, current, 0, 0});
                if (handle == ::pineforge::detail::scratch_no_handle)
                    throw coded<std::logic_error>(RunFailureCode::engine_invariant, {},
                                                  "incomplete trade scratch permutation");
            }
        }
        transaction.finish(); adapter.finish_exit_group_sort(scratch, scratch_before);
        return start;
    }
    return std::nullopt;
}
 
void source::PineStrategyHost::scheduler_update_session_state() {
    // session_ismarket_ and the _regular pair are the kernel's session-day
    // facts of this script bar, selected before the source callback reads
    // them (R5 lane F5). The rule lane E26 established here -- TradingView
    // ends a session at the session DAY, the flag belonging to the last chart
    // bar whose successor is out of session or on another session day,
    // isfirstbar its dual; the `lab tv` tapes in tests/fixtures/session_islastbar
    // give it bar for bar -- is the kernel's generic reading of the run's
    // calendar and input (NativeDecisionContext, market_driver.hpp), so every
    // driving path reads the same regular day by construction. The plain pair
    // starts from it; the chart's day an extended-hours chart widens it to is
    // Pine's own, below (update_extended_session_day).
    //
    // Pine's live-probe protocol (set_realtime_tail) stays too: its batch's
    // final bar is the still-forming realtime bar, so that bar closes its
    // session day where the calendar does, not by the batch's run-end
    // convention -- the kernel's open-ended reading, which equals the ordinary
    // one on every other bar.
    //
    // The scheduler's old lookahead indexed its retained input by the count
    // of source bars published, which under calc_on_order_fills read two bars
    // ahead on a bar a fill recalculation had already published
    // (tests/test_session_day_facts_adapter.cpp); the chart day's lookahead
    // starts from the point's own input and its script bar's end instead.
    const NativeCurrentPointView* point = detail::callback_point(*this);
    if (!point) return;
    const NativeDecisionContext& facts = point->decision;
    session_ismarket_ = facts.in_session;
    session_isfirstbar_regular_ = facts.opens_session_day;
    session_islastbar_regular_ = realtime_tail_ ? facts.closes_session_day_open_ended
                                                : facts.closes_session_day;
    session_isfirstbar_ = session_isfirstbar_regular_;
    session_islastbar_ = session_islastbar_regular_;
    // TradingView's symbol calendar, where the run holds one: a regular
    // intraday bar ends its session day when the calendar's next slot does not
    // lie in it -- OANDA:XAUUSD's 16:45 ET bar (the next slot, 17:00, opens the
    // next day), not its 12:45 ET bar of 2025-07-04, where the feed stops and
    // the calendar goes on (te_session_calendar_xau15-jul4). The first bar of a
    // day stays the one the run holds (isfirstbar is the bar after a bar of
    // another day, on the calendar and off it).
    if (facts.in_session && calendar_period_for(script_tf_) == CalendarPeriod::NONE
        && script_tf_seconds_ > 0) {
        if (const auto* day = symbol_calendar_day(facts.script_bar_open_ms)) {
            const bool last = facts.script_bar_open_ms
                + static_cast<std::int64_t>(script_tf_seconds_) * 1000 >= day->second;
            session_islastbar_regular_ = last;
            session_islastbar_ = last;
        }
    }
    update_extended_session_day(facts);
}
 
// session.isfirstbar / session.islastbar mark the CHART's session day (R5
// lane K-SESSION-WINDOWS F2). With extended hours TradingView's chart holds
// the regular day's pre- and post-market bars too, and the plain pair spans
// them while the _regular pair keeps the regular day: on NASDAQ:AAPL 60 with
// extended hours isfirstbar is the 04:00 bar and islastbar the 19:00 one,
// isfirstbar_regular 10:00 and islastbar_regular 15:00
// (tests/fixtures/session_windows). A chart without extended hours holds no
// pre- or post-market bar, so there the pairs are one. The regular day is the
// kernel's session-day facts; pre- and post-market are session.ispremarket /
// ispostmarket's windows (session_time.hpp), which exist only on a session
// day that opens and closes on one date, the date the chart's day is keyed
// to. A bar neither in session nor pre- or post-market is in no chart day.
//
// The bar before is the script bar published last. The bar after is the
// first input the run holds -- a batch's, or a stream's warmup
// (PineScheduler::retained_input_from) -- past this script bar's interval,
// so the chart, magnified and aggregated paths read one answer for the same
// bars; a batch's final bar closes its day as the kernel's does unless the
// live probe tail is still forming it. Without such a bar a pre- or
// post-market bar reads the chart timeframe's next slot, while a regular bar
// keeps the kernel's close: nothing says whether the post-market bars that
// slot names will come.
void source::PineStrategyHost::update_extended_session_day(const NativeDecisionContext& facts) {
    const std::string& session = syminfo_.session;
    const std::string& tz = syminfo_.timezone;
    const std::int64_t at = facts.script_bar_open_ms;
    const bool regular = facts.in_session;
    // A regular bar that neither opens nor closes the regular day is inside
    // it, and so inside the chart's: the kernel's answer stands.
    if (regular && !facts.opens_session_day && !session_islastbar_regular_) return;
    const auto pre = [&](std::int64_t ms) {
        return pineforge::session_in_premarket(session, tz, ms, script_tf_);
    };
    const auto post = [&](std::int64_t ms) {
        return pineforge::session_in_postmarket(session, tz, ms, script_tf_);
    };
    const bool premarket = !regular && pre(at);
    const bool postmarket = !regular && !premarket && post(at);
    if (!regular && !premarket && !postmarket) return;
 
    const std::int64_t day = pineforge::calendar_day_open_local_ms(at, tz);
    const auto same_day = [&](std::int64_t ms) {
        return pineforge::calendar_day_open_local_ms(ms, tz) == day;
    };
    const auto in_chart_day = [&](std::int64_t ms) {
        return (pineforge::session_in_market(session, tz, ms) || pre(ms) || post(ms))
            && same_day(ms);
    };
    const std::optional<std::int64_t> before = scheduler_.last_published_script_open();
    // The next script bar opens at or after this one's interval end: an
    // aggregated bucket's, or -- a FeedTolerant chart bar's interval being its
    // label alone -- the instant after the label.
    const std::int64_t next_open = std::max(at + 1, facts.script_interval.next_period_open_ms);
    std::optional<std::int64_t> after;
    bool known_after = false;
    if (const auto bar = scheduler_.retained_input_from(facts.coordinate.input_interval_index,
                                                        next_open)) {
        after = bar->timestamp;
        known_after = true;
    }
    const bool run_end = !known_after && !scheduler_.retains_stream() && !realtime_tail_;
    if (!known_after && !run_end && !regular && script_tf_seconds_ > 0)
        after = at + static_cast<std::int64_t>(script_tf_seconds_) * 1000;
 
    if (regular) {
        // The regular day's first bar opens the chart's unless pre-market
        // bars of its date precede it; its last closes the chart's unless
        // post-market bars of its date follow.
        if (facts.opens_session_day)
            session_isfirstbar_ = !(before && pre(*before) && same_day(*before));
        if (session_islastbar_regular_)
            session_islastbar_ = !(known_after && post(*after) && same_day(*after));
        return;
    }
    if (premarket) {
        session_isfirstbar_ = !(before && pre(*before) && same_day(*before));
        session_islastbar_ = after ? !in_chart_day(*after) : run_end;
        return;
    }
    session_isfirstbar_ = !(before && in_chart_day(*before));
    session_islastbar_ = after ? !(post(*after) && same_day(*after)) : run_end;
}
 
void source::PineStrategyHost::scheduler_publish_source_bar(
        const Bar& bar, bool, bool advance_source_index) {
    current_bar_ = bar;
    const bool temporary_index = !advance_source_index
        && (!scheduler_.current_script_bar()
            || scheduler_.current_script_bar()->timestamp != bar.timestamp);
    const int previous_bar_index = bar_index_;
    const bool previous_barstate_islast = barstate_islast_;
    if (advance_source_index || temporary_index) ++source_bar_index_;
    ++source_callback_count_;
    bar_index_ = source_bar_index_;
    const NativeLifecycleKind lifecycle = detail::run_kind(*this);
    const NativeRunPhase phase = detail::run_phase(*this);
    if (lifecycle == NativeLifecycleKind::Running && phase == NativeRunPhase::Warmup) {
        barstate_islast_ = false;
    } else if (lifecycle == NativeLifecycleKind::Running
               && phase == NativeRunPhase::Realtime) {
        barstate_islast_ = true;
    } else {
        barstate_islast_ = source_bar_index_ == source_last_bar_index_;
    }
    // The chart day partition, the EMA seeding default and the TA bar context
    // this publication runs under, written on the pump's runtime block (R5
    // lane D2-C: no thread-local WRITE per bar; runtime_ambient.hpp). The
    // readers still reach the block through the thread's pointer, one
    // thread-local access per call: every extremum ring behind ta.highest /
    // lowest / highestbars / lowestbars / stoch / wpr / range, and the day
    // partition behind the default-anchored ta.vwap, crosses_boundary(DAY) and
    // the session-period helpers (docs/native-refactor-progress.md, "Deferred
    // performance work").
    internal::RuntimeAmbient* const ambient = NativeExecutionConsumer::bound(*this).pump_ambient();
    internal::AmbientDayPartitionScope chart_day_partition(
        ambient, chart_day_partition_.empty() ? nullptr : &chart_day_partition_);
    // A named-entry cancellation token has source-evaluation scope.  Clear a
    // prior callback before publishing receipts and entering this body.
    adapter_.begin_source_evaluation();
    // A selected window's pre-roll evaluation records its commands and risk
    // statements instead of placing them: the buffer is this evaluation's alone,
    // reset here even if the body calls nothing, and the scope closes the bracket
    // on every exit (pine_intent_state.hpp).
    const bool capturing = selected_begin_evaluation(bar.timestamp);
    const detail::IntentCaptureScope capture_scope(*this, capturing);
    // Publish terminal and group-adjustment receipts before the source body
    // reads its public pending projection at this decision boundary.
    if (same_bar_exit_group_possible(trades_)) {
        if (const auto moved = sort_same_bar_exit_trades(trades_, adapter_))
            native_closed_rows_amended(*moved);
    }
    // Publish only the immutable prefix, once at this script-bar
    // boundary. Keep the last exit-time block AND its predecessor: sorting
    // and global-child re-cutting both inspect that predecessor.
    auto& consumer = detail::run_consumer(*this);
    if (consumer.streams_closed_rows(*this) && consumer.begin_closed_row_seal(bar.timestamp)) {
        const std::size_t base = trades_.base();
        auto prefix = consumer.closed_trade_prefix();
        if (prefix.base != base)
            throw coded<std::logic_error>(RunFailureCode::engine_invariant, {},
                                           "source trade seal has a mismatched base");
        const bool can_swap = adapter_.short_seed_.final_short.incarnation != 0
            && adapter_.config_.default_qty_type != static_cast<int>(QtyType::FIXED);
        const std::size_t next = consumer.closed_rows_sealable(trades_,
            adapter_.exit_phase_final_, adapter_.source_money_folded_, bar.timestamp,
            can_swap ? std::optional<std::uint64_t>(adapter_.short_seed_.materialize_long.incarnation)
                     : std::nullopt);
        if (next - base > adapter_.trade_exit_phase_.size())
            throw coded<std::logic_error>(RunFailureCode::engine_invariant, {},
                                           "source trade seal exceeds resident exit phases");
        for (std::size_t index = base; index < next; ++index) {
            const Trade& row = trades_[index];
            prefix.exit_phase_digest = PineExecutionAdapter::fold_exit_phase(
                prefix.exit_phase_digest, adapter_.trade_exit_phase_[index - base]);
            if (row.pnl > 0.0) prefix.source_gains += row.pnl;
            else prefix.source_losses += row.pnl;
            // L18's position-tail spill consumes this row here, before the
            // seal below retires storage, including older-time barriers.
        }
        if (next > base) {
            // L18: spill the position tail of the whole prefix, in forward row
            // order, before the seal retires its storage; a throw seals nothing.
            adapter_.preserve_position_tail_before_seal(
                next, consumer.report_scratch_directory(*this));
            consumer.seal_closed_rows(*this, next, prefix.exit_phase_digest,
                                      prefix.source_gains, prefix.source_losses);
            for (std::size_t offset = 0; offset < next - base; ++offset)
                consumer.fold_sealed_exit_phase(adapter_.trade_exit_phase_[offset]);
            adapter_.trade_exit_phase_.erase(adapter_.trade_exit_phase_.begin(),
                adapter_.trade_exit_phase_.begin() + (next - base));
        }
    }
    adapter_.observe_terminal_receipts();
    // ab9714be src/source/pine_scheduler.cpp:242,258: under
    // process_orders_on_close the orders that were already resting fill at step 1
    // (process_pending_orders(before_pooc_script=true)) BEFORE the strategy body
    // runs at step 3, so a bracket leg this route parked while its parent entry
    // was pending settles on the touch bar ahead of this bar's source
    // evaluation. Draining it here keeps the body's position state, and the
    // levels it re-prices, on the same side of the fill as the owner; a leg that
    // carries a predecessor receipt stays staged for the flush below the body.
    // Quiet-bar gates (pine_quiet_bar.hpp): a drain with nothing staged or
    // queued is not called.
    using detail::QuietHook;
    using detail::skip_quiet;
    if (!skip_quiet(QuietHook::BracketLegs, adapter_.pending_bracket_legs_.empty())) {
        adapter_.flush_pending_bracket_legs({}, /*post_calculation=*/false,
                                           /*pre_script_drain=*/true);
    }
    internal::AmbientEmaSeedingScope ema_scope(ambient, chart_ema_na_warmup_);
    internal::AmbientBarContextScope bar_scope(ambient, pine_bar_index(),
                                               scheduler_.bar_index_offset());
    position_entry_count_ = detail::run_position(*this).signed_units == 0.0
        ? 0 : adapter_.source_entry_slot_count();
    on_source_bar(bar);
    // The pass is over: a pre-fill average it held for the script goes
    // before the orders it placed settle (hold_script_average_price).
    release_script_average_price();
    // Handwritten/source-generated callbacks historically read and could
    // update the live Pine configuration fields directly.  Keep the adapter's
    // source policy view synchronized at the callback boundary; the generic
    // NativeRunSpec remains immutable for the run.
    adapter_.set_configuration(config_);
    if (capturing) selected_finish_evaluation();
    if (temporary_index) {
        --source_bar_index_;
        bar_index_ = previous_bar_index;
        barstate_islast_ = previous_barstate_islast;
    }
    finish_adapter_evaluation();
    if (advance_source_index) {
        scheduler_mark_report_point(bar.timestamp);
        prev_bar_timestamp_ = bar.timestamp;
    }
}
 
// The drains that close one source evaluation once its body has run, in the
// order they always ran in. The replay of a selected window's pre-roll rows ends
// its evaluation with this same list (selected_replay_at_open).
void source::PineStrategyHost::finish_adapter_evaluation() {
    // Quiet-bar gates (pine_quiet_bar.hpp), as at the publication.
    using detail::QuietHook;
    using detail::skip_quiet;
    if (!skip_quiet(QuietHook::PendingCloses, adapter_.close_batch_callsites_.empty()))
        adapter_.flush_pending_closes();
#if PINEFORGE_PINE_QUIET_BAR_GATES
    // flush_pending_entries' own quiet gate, asked here instead of behind the
    // call (R5 lane D2-D): with nothing delayed, batched or held, the flush
    // only resets the batch's close total. The gate is asked once either way,
    // so the probe counts what it always counted.
    if (adapter_.host_ != nullptr && adapter_.delayed_market_orders_.empty()
        && adapter_.pending_same_bar_commands_.empty() && adapter_.pending_entries_.empty()) {
        (void)skip_quiet(QuietHook::PendingEntries, true);
        adapter_.pending_same_bar_close_qty_ = 0.0;
    } else {
        adapter_.flush_pending_entries();
    }
#else
    adapter_.flush_pending_entries();
#endif
    if (!skip_quiet(QuietHook::BracketLegs, adapter_.pending_bracket_legs_.empty()))
        adapter_.flush_pending_bracket_legs();
    // After every command of this evaluation is in the kernel: a queued
    // relative strategy.exit whose parent entry is now a live request becomes
    // that parent's anchored bracket child.
    if (!skip_quiet(QuietHook::RelativeExits, adapter_.anchored_relative_legs_.empty()
                                              && adapter_.pending_relative_exits_.empty())) {
        adapter_.anchor_relative_exits();
    }
    adapter_.order_same_point_entries();
}
 
void source::PineStrategyHost::scheduler_publish_suppressed_tail(const Bar& bar) {
    // ab9714be pine_scheduler.cpp:222-231: the forming probe tail advances
    // source history and settles the already-matched broker book, but does
    // not invoke generated code or synthesize a range-end close.
    current_bar_ = bar;
    ++source_bar_index_;
    bar_index_ = source_bar_index_;
    barstate_islast_ = false;
    internal::AmbientDayPartitionScope chart_day_partition(
        NativeExecutionConsumer::bound(*this).pump_ambient(),
        chart_day_partition_.empty() ? nullptr : &chart_day_partition_);
    adapter_.begin_source_evaluation();
    // A selected pre-roll tail runs no body: it leaves an empty last-present
    // evaluation, so no earlier bar's signal outlives it.
    selected_clear_suppressed_evaluation(bar.timestamp);
    adapter_.observe_terminal_receipts();
    scheduler_mark_report_point(bar.timestamp);
    prev_bar_timestamp_ = bar.timestamp;
}
 
// The selected window's pre-roll capture hooks (pine_intent_state.hpp).
bool source::PineStrategyHost::selected_begin_evaluation(std::int64_t label_ms) {
    const detail::PreRollEvaluation pre_roll = detail::selected_pre_roll_evaluation(*this, label_ms);
    if (pre_roll.state == nullptr) return false;
    detail::reset_intent_evaluation(*pre_roll.state, label_ms, config_,
                                    adapter_.capture_replay_risk_config(), *pre_roll.point);
    pre_roll.state->capturing = true;
    return true;
}
 
void source::PineStrategyHost::selected_finish_evaluation() noexcept {
    detail::PineIntentState* const state = detail::selected_intents(*this);
    if (state == nullptr || !state->capturing) return;
    state->end_config = config_;
    state->end_risk = adapter_.capture_replay_risk_config();
    if (const NativeCurrentPointView* const point = detail::intent_live_point(*this))
        state->terminal_point = *point;
    state->capturing = false;
}
 
void source::PineStrategyHost::selected_clear_suppressed_evaluation(std::int64_t label_ms) {
    const detail::PreRollEvaluation pre_roll = detail::selected_pre_roll_evaluation(*this, label_ms);
    if (pre_roll.state == nullptr) return;
    detail::reset_intent_evaluation(*pre_roll.state, label_ms, config_,
                                    adapter_.capture_replay_risk_config(), *pre_roll.point);
}
 
namespace {
 
// The selected window's replay (selected_replay_at_open) hands each retained
// row to the adapter's own door with exactly the arguments the script passed.
// The host's wrappers are never used: they would take the trading-window gate
// again and copy the live configuration over the row's. The nine command doors
// first, then the six risk setter statements.
void replay_intent(source::PineExecutionAdapter& adapter,
                   const source::detail::IntentEntry& row) {
    source::admit_strategy_entry(row.id, row.is_long, row.limit_price, row.stop_price,
                                 row.qty, row.comment, row.oca_name, row.oca_type, row.qty_type);
    adapter.entry(row.id, row.is_long, row.limit_price, row.stop_price, row.qty, row.comment,
                  row.oca_name, row.oca_type, row.qty_type);
}
void replay_intent(source::PineExecutionAdapter& adapter,
                   const source::detail::IntentClose& row) {
    source::admit_strategy_close(row.id, row.comment, row.qty, row.qty_percent, row.immediately);
    adapter.close(row.id, row.comment, row.qty, row.qty_percent, row.immediately);
}
void replay_intent(source::PineExecutionAdapter& adapter,
                   const source::detail::IntentCloseToken& row) {
    source::admit_strategy_close_token(row.id, row.comment, row.qty, row.qty_percent,
                                       row.immediately, row.callsite_token);
    adapter.close(row.id, row.comment, row.qty, row.qty_percent, row.immediately,
                  row.callsite_token);
}
void replay_intent(source::PineExecutionAdapter& adapter, const source::detail::IntentCloseAll&) {
    source::admit_strategy_close_all();
    adapter.close_all();
}
void replay_intent(source::PineExecutionAdapter& adapter,
                   const source::detail::IntentExit& row) {
    source::admit_strategy_exit(row.id, row.from_entry, row.limit_price, row.stop_price,
                                row.trail_points, row.trail_offset, row.trail_price,
                                row.qty_percent, row.comment, row.qty, row.oca_name,
                                row.profit_ticks, row.loss_ticks);
    adapter.exit(row.id, row.from_entry, row.limit_price, row.stop_price, row.trail_points,
                 row.trail_offset, row.trail_price, row.qty_percent, row.comment, row.qty,
                 row.oca_name, row.profit_ticks, row.loss_ticks);
}
void replay_intent(source::PineExecutionAdapter& adapter,
                   const source::detail::IntentExitCancelBracket& row) {
    source::admit_strategy_exit_cancel_bracket(row.exit_id, row.from_entry, row.comment);
    adapter.exit_cancel_bracket(row.exit_id, row.from_entry, row.comment);
}
void replay_intent(source::PineExecutionAdapter& adapter,
                   const source::detail::IntentCancel& row) {
    source::admit_strategy_cancel(row.id);
    adapter.cancel(row.id);
}
void replay_intent(source::PineExecutionAdapter& adapter,
                   const source::detail::IntentCancelAll&) {
    source::admit_strategy_cancel_all();
    adapter.cancel_all();
}
void replay_intent(source::PineExecutionAdapter& adapter,
                   const source::detail::IntentOrder& row) {
    source::admit_strategy_order(row.id, row.is_long, row.qty, row.limit_price, row.stop_price,
                                 row.oca_name, row.oca_type);
    adapter.order(row.id, row.is_long, row.qty, row.limit_price, row.stop_price, row.oca_name,
                  row.oca_type);
}
void replay_intent(source::PineExecutionAdapter& adapter,
                   const source::detail::IntentRiskDirection& row) {
    // The retained int direction/count/cap rows cannot carry NaN/infinity.
    adapter.set_risk_direction(row.direction);
}
void replay_intent(source::PineExecutionAdapter& adapter,
                   const source::detail::IntentRiskMaxConsLossDays& row) {
    adapter.set_risk_max_cons_loss_days(row.value);
}
void replay_intent(source::PineExecutionAdapter& adapter,
                   const source::detail::IntentRiskMaxDrawdown& row) {
    admit_pine_risk_max_drawdown(row.value, row.percent);
    adapter.set_risk_max_drawdown(row.value, row.percent);
}
void replay_intent(source::PineExecutionAdapter& adapter,
                   const source::detail::IntentRiskMaxIntradayLoss& row) {
    admit_pine_risk_max_intraday_loss(row.value, row.percent);
    adapter.set_risk_max_intraday_loss(row.value, row.percent);
}
// As the host setter: the script's own statement adopts the count switches first.
void replay_intent(source::PineExecutionAdapter& adapter,
                   const source::detail::IntentRiskMaxIntradayFilledOrders& row) {
    if (row.limit > 0) adapter.cap.adopt_script_count();
    adapter.cap = row.limit;
}
void replay_intent(source::PineExecutionAdapter& adapter,
                   const source::detail::IntentRiskMaxPositionSize& row) {
    admit_pine_risk_max_position_size(row.value);
    adapter.set_risk_max_position_size(row.value);
}
 
// A selected replay the engine cannot take, or that a native failure cut short,
// is the engine's fault and fails the run: it is never skipped quietly. After a
// native failure the consumer already holds the first cause and keeps it (its
// bar-open catch records only when nothing failed before).
[[noreturn]] void replay_invariant(const char* what) {
    throw coded<std::logic_error>(RunFailureCode::engine_invariant, {}, what);
}
 
// Ends a replay bracket on every exit, normal or by exception: the retained
// point pointer cleared, the open activation scope closed (the grants it
// recorded stay for the Open match that follows) and the adapter's
// configuration back to the host's. Two stores, a noexcept end and a plain
// copy: nothing here executes a command or allocates.
class ReplayBracket {
public:
    ReplayBracket(source::detail::PineIntentState& state, NativeExecutionConsumer& consumer,
                  source::PineExecutionAdapter& adapter,
                  const source::PineStrategyConfig& config) noexcept
        : state_(&state), consumer_(&consumer), adapter_(&adapter), config_(&config) {}
    ReplayBracket(const ReplayBracket&) = delete;
    ReplayBracket& operator=(const ReplayBracket&) = delete;
    ~ReplayBracket() noexcept {
        state_->replay_point = nullptr;
        consumer_->end_open_activation();
        adapter_->set_configuration(*config_);
    }
 
private:
    source::detail::PineIntentState* state_;
    NativeExecutionConsumer* consumer_;
    source::PineExecutionAdapter* adapter_;
    const source::PineStrategyConfig* config_;
};
 
}  // namespace
 
// The selected window's replay (pine_intent_state.hpp): the last pre-roll
// evaluation's rows, placed once at the window's first Open, before the adapter
// reads that bar's open. They are placed as the evaluation placed them, in its
// own order, each under the configuration and the retained point it was called
// with, so the Open matcher that follows sees the prior bar's orders. Nothing of
// the script runs again; nothing is published, matched or executed here.
void source::PineStrategyHost::selected_replay_at_open(const Bar& bar,
                                                      const NativeDecisionContext&) {
    pf_selected_window_config_v1 window{};
    if (!NativeExecutionConsumer::selected_window_admitted(*this, &window)
        || bar.timestamp < window.start_ms) {
        return;
    }
    detail::PineIntentState* const state = detail::selected_intents(*this);
    if (state == nullptr) replay_invariant("selected replay without its intent state");
    if (state->replayed) return;
    if (!state->has_evaluation) {
        state->replayed = true;
        return;
    }
    // The bar before the window left the book as it was: nothing held, nothing
    // working or queued in the source layer, no recalculation open.
    if (state->capturing) replay_invariant("selected replay inside a pre-roll capture bracket");
    if (state->label_ms >= window.start_ms)
        replay_invariant("selected replay of an evaluation labelled inside the window");
    if (detail::run_position(*this).signed_units != 0.0)
        replay_invariant("selected replay over a held position");
    if (adapter_.coof_recalc_active_)
        replay_invariant("selected replay inside a calc_on_order_fills recalculation");
    if (!adapter_.live_handles_.empty() || !adapter_.delayed_market_orders_.empty()
        || !adapter_.pending_same_bar_commands_.empty() || !adapter_.pending_entries_.empty()
        || !adapter_.close_batch_callsites_.empty() || !adapter_.pending_bracket_legs_.empty()
        || !adapter_.anchored_relative_legs_.empty()
        || !adapter_.pending_relative_exits_.empty()) {
        replay_invariant("selected replay over live requests or queued source commands");
    }
    // Marked before the bracket opens: a failure inside it never retries.
    state->replayed = true;
    NativeExecutionConsumer& consumer = NativeExecutionConsumer::bound(*this);
    if (!consumer.begin_open_activation(*this))
        replay_invariant("selected replay without an open activation scope");
    const ReplayBracket bracket(*state, consumer, adapter_, config_);
    const auto stop_if_failed = [this] {
        if (detail::run_kind(*this) == NativeLifecycleKind::Failed)
            replay_invariant("selected replay stopped by a native failure");
    };
    using detail::QuietHook;
    using detail::skip_quiet;
 
    // The evaluation again, from its start: the configuration it began under,
    // then its receipts and the legs staged before its body, as
    // scheduler_publish_source_bar takes them.
    state->replay_point = &state->terminal_point;
    assert_admitted_pine_replay_risk_config(state->start_risk);
    adapter_.restore_replay_risk_config(state->start_risk);
    adapter_.set_configuration(state->start_config);
    adapter_.begin_source_evaluation();
    adapter_.observe_terminal_receipts();
    stop_if_failed();
    if (!skip_quiet(QuietHook::BracketLegs, adapter_.pending_bracket_legs_.empty())) {
        adapter_.flush_pending_bracket_legs({}, /*post_calculation=*/false,
                                           /*pre_script_drain=*/true);
    }
    stop_if_failed();
 
    // Its statements, in the order the script made them.
    for (detail::PineIntentRow& row : state->rows) {
        state->replay_point = &row.point;
        adapter_.set_configuration(row.config);
        std::visit([this](const auto& payload) { replay_intent(adapter_, payload); },
                   row.payload);
        stop_if_failed();
    }
 
    // Its end: the configuration it ended under (the risk configuration alone,
    // never a counter), then the drains every source evaluation ends with.
    state->replay_point = &state->terminal_point;
    adapter_.set_configuration(state->end_config);
    assert_admitted_pine_replay_risk_config(state->end_risk);
    adapter_.restore_replay_risk_config(state->end_risk);
    finish_adapter_evaluation();
    stop_if_failed();
 
    // The terminal explicit-market policy of a process_orders_on_close pass,
    // which the close of that bar would have run: submit-only here, so the
    // reconciled entries stay working for the open matcher. The caller's own
    // quiet and count gate, as PineExecutionAdapter::on_bar_close asks it; the
    // policy keeps its own filters.
    const NativeDecisionContext& terminal = state->terminal_point.decision;
    const bool bound = adapter_.host_ != nullptr;
    const int bar_index = adapter_.projection_bar_index(terminal);
    std::size_t explicit_entries = 0;
    if (adapter_.config_.process_orders_on_close && adapter_.config_.pyramiding == 0
        && adapter_.modeled_input()) {
        for (const auto& handle : adapter_.live_handles_) {
            const auto found = adapter_.placement_.find(handle.incarnation);
            if (found != adapter_.placement_.end() && found->second.opening
                && found->second.family == PineOrderFamily::Entry
                && found->second.projection_created_bar == bar_index
                && std::isfinite(found->second.requested_qty)
                && found->second.requested_qty > 0.0) {
                ++explicit_entries;
            }
        }
    }
    if (!skip_quiet(QuietHook::TerminalExplicitMarket, bound && explicit_entries < 2))
        adapter_.apply_terminal_explicit_market_policy(terminal, false);
    stop_if_failed();
 
    // Spent only once it all succeeded. The retained configurations, the terminal
    // point and the replayed flag stay as the hash reads them.
    state->rows.clear();
}
 
// The Pine report series has one point per SOURCE slot this host published,
// which is not the kernel's per-calculation cadence: a calc_on_order_fills
// re-entry marks the slot it opened at the fill, the ordinary close
// calculation then marks nothing, and the probe's suppressed tail marks a
// slot generated code never calculated. The point also has to land inside
// this callback, before scheduler_record_broker_hash() folds the extremes it
// just moved and before scheduler_record_range_end() re-marks the curve's
// last point. So the cadence stays here and the recording does not: the
// kernel owns what a report point is — the extremes fold and the curve
// append in engine.hpp — reached through the run spec's report policy.
void source::PineStrategyHost::scheduler_mark_report_point(std::int64_t script_bar_ts) {
    detail::run_consumer(*this).mark_script_report_point(*this, script_bar_ts);
}
 
void source::PineStrategyHost::scheduler_record_broker_hash() {
    if (!broker_state_hash_recording_) return;
    const std::uint64_t execution = execution_consumer().continuation_hash();
    // As capture_script_continuation_hash: a carried read failure is thrown
    // before the cut hash is latched or recorded as a row.
    as_native_consumer(execution_consumer()).request_core().rethrow_book_read_failure();
    last_script_continuation_hash_ = execution;
    last_script_continuation_valid_ = true;
    broker_state_hashes_.push_back(broker_state_hash());
}
 
bool source::PineStrategyHost::adapter_core_sizes_default_opening(bool is_long) const {
    return adapter_.core_sizes_default_opening(is_long);
}
 
// The chart's time_close. A D/W/M chart bar closes at its period's last
// traded close, to the millisecond, whatever time of day its stamp reads, as
// time_close(timeframe.period) reads it on that chart (chart_period_close_ms,
// src/session_time.cpp, where the rule and its tapes are stated): OANDA
// stamps its XAUUSD daily bars at 17:00 ET, inside the 1800-1700 session's
// break, and the time-of-day session filter the intraday path applies read
// that stamp as out of session, so the D chart answered na.
int64_t source::PineStrategyHost::chart_time_close() const {
    return chart_bar_close_ms(current_bar_.timestamp);
}
 
int64_t source::PineStrategyHost::chart_bar_close_ms(int64_t stamp) const {
    if (calendar_period_for(script_tf_) != CalendarPeriod::NONE) {
        return pine_time_close(stamp, script_tf_, "", "", script_tf_,
                               syminfo_.timezone, syminfo_.session);
    }
    return pineforge::detail::symbol_session_time_close(stamp, script_tf_, syminfo_.session,
                                                        syminfo_.timezone, script_tf_);
}
 
// TradingView's time(tf, session, tz, bars_back, timeframe_bars_back) (lab tv
// tapes te_time_bb_chart and te_time_bb_tf, tests/fixtures/time_bars_back):
// the chart bar bars_back names -- a past one from the chart's history, na
// before its first -- then the tf bar holding it, stepped timeframe_bars_back
// of its own bars. TradingView steps forward, and steps tf bars, on its
// symbol calendar, closed days included: NASDAQ:AAPL's Thursday 2025-04-17
// 15:45 bar reads its next bar at Monday 04-21 09:30 over Good Friday, its
// daily bar of 01-08 the next at 01-10 over the 01-09 closure, time("D", 0,
// -1) on 04-17 reads 04-21 and time("60", 0, 1) on Monday's 09:30 bar
// Thursday's 15:30. The host holds no calendar of closed days; the chart's
// history holds exactly the calendar's bars on every tape but one, whose
// history lacks bars TradingView's calendar holds (OANDA:XAUUSD's 17:00-18:00
// ET slot and Good Friday, the test's no-bar slots). So the host reads the
// bars from the input it retains (a batch's, or a stream's warmup); past its
// edges it walks the symbol's session calendar (session_trades_at): a batch's
// final bar asks the calendar for the bar after it, as a live bar does.
namespace {
 
constexpr std::int64_t kTimeOffsetMinuteMs = 60'000;
// The farthest the calendar walk looks for a traded minute: past any
// weekend and any run of closed days.
constexpr int kTimeOffsetCalendarMinutes = 16 * 24 * 60;
// The most chart bars a step inside a tf bar may pass before it leaves it.
constexpr int kTimeOffsetBarsPerStep = 100'000;
 
}  // namespace
 
int64_t source::PineStrategyHost::pine_time_offset(int64_t bar_open_ms, int bars_back,
                                                   const std::string& tf,
                                                   const std::string& session,
                                                   const std::string& tz,
                                                   int timeframe_bars_back,
                                                   bool close) const {
    if (is_na(bar_open_ms) || is_na(bars_back) || is_na(timeframe_bars_back))
        return na<int64_t>();
    const std::string& sym_tz = syminfo_.timezone;
    const std::string& sym_session = syminfo_.session;
    const auto input = scheduler_.retained_input();
    const bool input_is_chart = scheduler_.retained_input_is_chart();
    const bool daily_chart = calendar_period_for(script_tf_) != CalendarPeriod::NONE;
    // A chart bar's own open (the input's stamp) and close, as the chart's
    // time and time_close read them (chart_time_close).
    const auto chart_open_of = [&](std::int64_t input_ms) -> std::int64_t {
        return input_is_chart ? input_ms
                              : pine_time(input_ms, script_tf_, "", "", script_tf_, sym_tz,
                                          sym_session);
    };
    // Where the run holds TradingView's symbol calendar (set_symbol_calendar),
    // the chart bars ahead are its slots: the chart's next bar, the first bar
    // from an instant, a chart bar's close and a day's open and close are the
    // calendar's inside its days, whatever bars the history holds there
    // (OANDA:XAUUSD's 17:00 ET slot, Good Friday's session). Bars behind stay
    // the history's. The calendar's slots are an intraday chart's bars and a
    // 1D chart's days; a W, M or multi-day chart's bars are no calendar day,
    // so it reads its history.
    const std::int64_t chart_step_ms = static_cast<std::int64_t>(script_tf_seconds_) * 1000;
    const bool calendar_chart = !symbol_calendar_.empty()
        && (!daily_chart || (calendar_period_for(script_tf_) == CalendarPeriod::DAY
                             && script_tf_seconds_ == 86400));
    const auto chart_close_of = [&](std::int64_t open_ms) -> std::int64_t {
        if (const auto* day = calendar_chart ? symbol_calendar_day(open_ms) : nullptr) {
            return daily_chart || chart_step_ms <= 0 ? day->second
                                                     : std::min(open_ms + chart_step_ms, day->second);
        }
        return daily_chart ? pine_time_close(open_ms, script_tf_, "", "", script_tf_, sym_tz,
                                             sym_session)
                           : pineforge::detail::symbol_session_time_close(
                                 open_ms, script_tf_, sym_session, sym_tz, script_tf_);
    };
    const auto chart_slot_of = [&](std::int64_t ms) -> std::int64_t {
        return pine_time(ms, script_tf_, "", "", script_tf_, sym_tz, sym_session);
    };
    // The first chart bar opening at or after `at`: the retained input's, else
    // the calendar's first traded chart slot.
    const auto chart_open_from = [&](std::int64_t at) -> std::int64_t {
        if (calendar_chart) {
            if (const std::int64_t slot = symbol_calendar_open_from(at); !is_na(slot)) return slot;
        }
        auto it = std::lower_bound(input.begin(), input.end(), at,
                                   [](const Bar& bar, std::int64_t value) {
                                       return bar.timestamp < value;
                                   });
        while (!input_is_chart && it != input.end() && chart_open_of(it->timestamp) < at)
            ++it;
        if (it != input.end()) return chart_open_of(it->timestamp);
        std::int64_t t = at;
        for (int k = 0; k < kTimeOffsetCalendarMinutes; ++k, t += kTimeOffsetMinuteMs) {
            if (session_trades_at(sym_session, sym_tz, t)) {
                const std::int64_t slot = chart_slot_of(t);
                return is_na(slot) || slot < at ? t : slot;
            }
        }
        return na<int64_t>();
    };
    // The last chart bar opening before `before`: the retained input's while
    // it holds the instant (a stream's warmup ends where its live bars
    // begin), else the calendar's last traded chart slot.
    const auto chart_open_before = [&](std::int64_t before) -> std::int64_t {
        if (calendar_chart) {
            if (const std::int64_t slot = symbol_calendar_open_before(before); !is_na(slot))
                return slot;
        }
        auto it = std::lower_bound(input.begin(), input.end(), before,
                                   [](const Bar& bar, std::int64_t value) {
                                       return bar.timestamp < value;
                                   });
        const bool held = !scheduler_.retains_stream() || it != input.end();
        if (held && it != input.begin()) return chart_open_of(std::prev(it)->timestamp);
        std::int64_t t = before - kTimeOffsetMinuteMs;
        for (int k = 0; k < kTimeOffsetCalendarMinutes; ++k, t -= kTimeOffsetMinuteMs) {
            if (session_trades_at(sym_session, sym_tz, t)) {
                const std::int64_t slot = chart_slot_of(t);
                return is_na(slot) || slot >= before ? t : slot;
            }
        }
        return na<int64_t>();
    };
    // The chart bar after the one opening at `open_ms`: the retained input's
    // next bar, else the calendar's first traded slot past its close.
    const auto next_chart_open = [&](std::int64_t open_ms) -> std::int64_t {
        if (calendar_chart && symbol_calendar_day(open_ms) != nullptr) {
            if (const std::int64_t slot = symbol_calendar_open_from(open_ms + 1); !is_na(slot))
                return slot;
        }
        auto it = std::upper_bound(input.begin(), input.end(), open_ms,
                                   [](std::int64_t value, const Bar& bar) {
                                       return value < bar.timestamp;
                                   });
        while (!input_is_chart && it != input.end() && chart_open_of(it->timestamp) <= open_ms)
            ++it;
        if (it != input.end()) return chart_open_of(it->timestamp);
        const std::int64_t end = chart_close_of(open_ms);
        return chart_open_from(is_na(end) || end <= open_ms ? open_ms + 1 : end);
    };
 
    // A chart bar before the current one is one the run has had: the
    // retained input's, na before its first (time[bars_back]). A stream's
    // live bars are past its warmup: the calendar's.
    const auto prior_chart_open = [&](std::int64_t open_ms) -> std::int64_t {
        auto it = std::lower_bound(input.begin(), input.end(), open_ms,
                                   [](const Bar& bar, std::int64_t value) {
                                       return bar.timestamp < value;
                                   });
        if (scheduler_.retains_stream() && it == input.end())
            return chart_open_before(open_ms);
        while (it != input.begin()) {
            --it;
            const std::int64_t prior = chart_open_of(it->timestamp);
            if (prior < open_ms) return prior;
        }
        return na<int64_t>();
    };
 
    std::int64_t chart_open = bar_open_ms;
    for (int k = 0; k < -bars_back && !is_na(chart_open); ++k)
        chart_open = next_chart_open(chart_open);
    for (int k = 0; k < bars_back && !is_na(chart_open); ++k)
        chart_open = prior_chart_open(chart_open);
    if (is_na(chart_open)) return na<int64_t>();
 
    // The chart's own timeframe without a session: the chart bars themselves.
    if ((tf.empty() || tf == script_tf_) && session.empty()) {
        for (int k = 0; k < -timeframe_bars_back && !is_na(chart_open); ++k)
            chart_open = next_chart_open(chart_open);
        for (int k = 0; k < timeframe_bars_back && !is_na(chart_open); ++k)
            chart_open = chart_open_before(chart_open);
        if (is_na(chart_open)) return na<int64_t>();
        return close ? chart_close_of(chart_open) : chart_open;
    }
 
    // A week or a month opens at its first traded day: the week of Martin
    // Luther King Day 2025 at Tuesday 01-21 on NASDAQ:AAPL, where the
    // calendar reads Monday (te-time-bb-tf-aapl1d). The day the chart's
    // history first trades in the period, when it trades later than the
    // calendar's first session day.
    const CalendarPeriod tf_period = calendar_period_for(tf.empty() ? script_tf_ : tf);
    // A day of the symbol calendar is its "D" bar.
    const bool calendar_days = session.empty() && tf_period == CalendarPeriod::DAY
        && tf_to_seconds(tf.empty() ? script_tf_ : tf) == 86400;
    const auto calendar_open_of = [&](std::int64_t ms) -> std::int64_t {
        if (calendar_days) {
            if (const auto* day = symbol_calendar_day(ms)) return day->first;
        }
        return pine_time(ms, tf, session, tz, script_tf_, sym_tz, sym_session);
    };
    const auto tf_close_of = [&](std::int64_t open_ms) -> std::int64_t {
        if (calendar_days) {
            if (const auto* day = symbol_calendar_day(open_ms)) return day->second;
        }
        return pine_time_close(open_ms, tf, session, tz, script_tf_, sym_tz, sym_session);
    };
    const auto tf_open_of = [&](std::int64_t ms) -> std::int64_t {
        const std::int64_t open = calendar_open_of(ms);
        if (is_na(open) || !session.empty()
            || (tf_period != CalendarPeriod::WEEK && tf_period != CalendarPeriod::MONTH)) {
            return open;
        }
        const std::int64_t first = chart_open_from(open);
        if (is_na(first) || calendar_open_of(first) != open) return open;
        const std::int64_t day = pine_time(first, "D", "", "", script_tf_, sym_tz, sym_session);
        return !is_na(day) && day > open ? day : open;
    };
    std::int64_t open = tf_open_of(chart_open);
    if (is_na(open)) return na<int64_t>();
    // A timeframe finer than the chart's steps by its own length and reads
    // the chart bar holding the instant it reaches: time("60", 0, -1) on a
    // daily bar is that bar, time("60", 0, 1) the bar before
    // (te-time-bb-tf-aapl1d / -btc1d / -eurusd1d).
    const int tf_seconds = tf_to_seconds(tf.empty() ? script_tf_ : tf);
    const int chart_seconds = tf_to_seconds(script_tf_);
    if (timeframe_bars_back != 0 && tf_seconds > 0 && chart_seconds > 0
        && tf_seconds < chart_seconds) {
        const std::int64_t at = open - static_cast<std::int64_t>(timeframe_bars_back)
                                           * tf_seconds * 1000;
        const std::int64_t held = chart_open_before(at + 1);
        if (is_na(held)) return na<int64_t>();
        return close ? chart_close_of(held) : held;
    }
    for (int k = 0; k < -timeframe_bars_back && !is_na(open); ++k) {
        // The first chart bar past this tf bar's end that a tf bar holds.
        const std::int64_t end = tf_close_of(open);
        std::int64_t at = chart_open_from(is_na(end) || end <= open ? open + 1 : end);
        std::int64_t next = na<int64_t>();
        for (int guard = 0; guard < kTimeOffsetBarsPerStep && !is_na(at); ++guard) {
            next = tf_open_of(at);
            if (!is_na(next) && next > open) break;
            next = na<int64_t>();
            at = next_chart_open(at);
        }
        open = next;
    }
    for (int k = 0; k < timeframe_bars_back && !is_na(open); ++k) {
        // The last chart bar before this tf bar's open that a tf bar holds.
        std::int64_t at = chart_open_before(open);
        std::int64_t prev = na<int64_t>();
        for (int guard = 0; guard < kTimeOffsetBarsPerStep && !is_na(at); ++guard) {
            prev = tf_open_of(at);
            if (!is_na(prev) && prev < open) break;
            prev = na<int64_t>();
            at = chart_open_before(at);
        }
        open = prev;
    }
    if (is_na(open)) return na<int64_t>();
    return close ? tf_close_of(open) : open;
}
 
void source::PineStrategyHost::finish_applied_admission(
        const native_order::ExecutionAppliedEvent& event,
        const NativeDecisionContext& context) {
    const bool recalculation_due = scheduler_.coof_recalculation_due(event, context, *this);
    if (!recalculation_due) record_applied_range_end();
    // A process-on-close fill can append admission events after on_native_bar.
    // Retire that completed tail without recapturing the broker-state row:
    // F0a preserves the existing recording and continuation boundaries.
    // A synchronous fill inside an unfinished command is folded by the
    // enclosing on_bar_close instead.
    if (!recalculation_due && context.is_terminal_sub_bar
        && context.coordinate.provenance == NativePriceProvenance::AfterCalculationClose
        && !adapter_.admission_journal.has_unfinished_events()) {
        adapter_.finish_admission_bar();
    }
}
 
} // namespace pineforge
 
namespace pineforge {
 
void source::PineStrategyHost::stage_symbol_feed_views() {
    if (symbol_feed_views_.size() != symbol_feeds_.size())
        throw coded<std::logic_error>(RunFailureCode::engine_invariant, {},
                                      "request data: symbol feed view table is out of sync");
    std::vector<NativeInstrumentFeedView> descriptors(symbol_feed_views_.size());
    std::vector<const NativeInstrumentFeedView*> slots(symbol_feed_views_.size(), nullptr);
    for (std::size_t i = 0; i < symbol_feed_views_.size(); ++i) {
        const BorrowedFeedSlot& slot = symbol_feed_views_[i];
        if (!slot.present) continue;
        descriptors[i] = {{slot.bars, slot.size}, slot.close_ms};
        slots[i] = &descriptors[i];
    }
    NativeExecutionConsumer::bound(*this).stage_instrument_feed_views(
        slots.empty() ? nullptr : slots.data(), slots.size());
}
 
}  // namespace pineforge