pine_scheduler_native.cpp

Go to the documentation of this file.

#include <pineforge/source/pine_scheduler.hpp>
 
#include <pineforge/source/pine_strategy_host.hpp>
#include <pineforge/timeframe.hpp>
 
#include "pine_host_reads.hpp"
#include <pineforge/run_failure.hpp>
#include <pineforge/selected_window.h>
#include <pineforge/selected_window_plan.hpp>
 
#include <algorithm>
#include <cstdint>
#include <limits>
#include <stdexcept>
#include <utility>
 
namespace pineforge::source {
 
PineScheduler::PineScheduler() = default;
PineScheduler::~PineScheduler() = default;
 
void PineScheduler::capture_begin(const NativeBeginArgs& args) {
    RetainedBegin next;
    if (args.bars && args.n > 0) {
        if (args.is_stream) next.owned.assign(args.bars, args.bars + args.n);
        next.bars = InputView(args.is_stream ? next.owned.data() : args.bars,
                              static_cast<std::size_t>(args.n));
    }
    next.input_tf = args.input_tf; next.script_tf = args.script_tf;
    next.bar_magnifier = args.bar_magnifier; next.magnifier_samples = args.magnifier_samples;
    next.distribution = args.magnifier_distribution;
    next.volume_weighted = args.magnifier_volume_weighted;
    next.volume_weighted_min_samples = args.magnifier_volume_weighted_min_samples;
    next.volume_weighted_max_samples = args.magnifier_volume_weighted_max_samples;
    next.is_stream = args.is_stream; next.warmup_n = args.warmup_n;
    next.simple_run = args.simple_run;
    retained_ = std::move(next);
    retained_n_ = retained_.bars.size();
    confirmed_input_bars_ = {};
    confirmed_input_next_ = 0;
    confirmed_input_count_ = 0;
    confirmed_open_input_bars_ = {};
    confirmed_open_input_next_ = 0;
    confirmed_open_input_count_ = 0;
    reset_consumed_digests();
}
 
void PineScheduler::release_batch_input() noexcept {
    if (retained_.is_stream) return;
    fold_consumed_digests();
    retained_.bars = {};
}
 
void PineScheduler::reset_consumed_digests() noexcept {
    consumed_script_cursor_.reset();
    consumed_bars_folded_ = 0;
    consumed_bars_digest_ = 1469598103934665603ULL;
    consumed_completes_folded_ = 0;
    consumed_completes_digest_ = 1469598103934665603ULL;
    consumed_boundaries_folded_ = 0;
    consumed_boundaries_digest_ = 1469598103934665603ULL;
}
 
void PineScheduler::reset_language() {
    language_.reset_for_run();
    language_.pos_view_freeze_bar_ = -1;
    language_.pos_view_frozen_side_ = PositionSide::FLAT;
    language_.pos_view_frozen_qty_ = 0.0;
    language_.pos_view_frozen_entry_qty_.clear();
    language_.is_first_tick_ = true; language_.is_last_tick_ = true;
    language_.history_slot_is_new_ = true; language_.coof_checkpoint_contains_current_bar_ = false;
    language_.coof_checkpoint_src_open_.clear(); language_.coof_checkpoint_src_high_.clear();
    language_.coof_checkpoint_src_low_.clear(); language_.coof_checkpoint_src_close_.clear();
    language_.coof_checkpoint_src_volume_.clear(); language_.coof_checkpoint_src_hl2_.clear();
    language_.coof_checkpoint_src_hlc3_.clear(); language_.coof_checkpoint_src_ohlc4_.clear();
    language_.coof_checkpoint_src_hlcc4_.clear();
    current_script_open_ms_ = 0; saw_open_fill_ = false; open_point_fills_ = 0;
    current_script_bar_ = {}; current_script_bar_valid_ = false;
    source_bar_count_ = 0; expected_source_bars_ = 0; applied_cursor_ = 0;
    coof_callback_script_open_ = std::numeric_limits<std::int64_t>::min();
    last_published_script_open_ms_ = std::numeric_limits<std::int64_t>::min();
    prior_input_script_open_ms_ = std::numeric_limits<std::int64_t>::min();
    awaiting_legacy_script_open_ms_ = std::numeric_limits<std::int64_t>::min();
    last_stream_input_open_ms_ = std::numeric_limits<std::int64_t>::min();
    input_script_cursor.reset();
    input_script_size_ = 0;
    uses_aux_security_feed_ = false;
    input_is_observed_ticks_ = false;
    confirmed_input_bars_ = {};
    confirmed_input_next_ = 0;
    confirmed_input_count_ = 0;
    confirmed_open_input_bars_ = {};
    confirmed_open_input_next_ = 0;
    confirmed_open_input_count_ = 0;
    deferred_boundary_input_ = {};
    reset_consumed_digests();
}
 
void PineScheduler::snapshot_coof_script_state(PineStrategyHost& host) {
    if (language_._src_series_active_) {
        language_.coof_checkpoint_src_open_ = language_._src_open_;
        language_.coof_checkpoint_src_high_ = language_._src_high_;
        language_.coof_checkpoint_src_low_ = language_._src_low_;
        language_.coof_checkpoint_src_close_ = language_._src_close_;
        language_.coof_checkpoint_src_volume_ = language_._src_volume_;
        language_.coof_checkpoint_src_hl2_ = language_._src_hl2_;
        language_.coof_checkpoint_src_hlc3_ = language_._src_hlc3_;
        language_.coof_checkpoint_src_ohlc4_ = language_._src_ohlc4_;
        language_.coof_checkpoint_src_hlcc4_ = language_._src_hlcc4_;
    }
    language_.coof_checkpoint_prev_chart_close_ = language_.prev_chart_close_;
    language_.coof_checkpoint_last_chart_close_ = language_.last_chart_close_;
    language_.coof_checkpoint_contains_current_bar_ = false;
    host.snapshot_script_state();
}
 
void PineScheduler::restore_coof_script_state(PineStrategyHost& host) {
    if (language_._src_series_active_) {
        language_._src_open_ = language_.coof_checkpoint_src_open_;
        language_._src_high_ = language_.coof_checkpoint_src_high_;
        language_._src_low_ = language_.coof_checkpoint_src_low_;
        language_._src_close_ = language_.coof_checkpoint_src_close_;
        language_._src_volume_ = language_.coof_checkpoint_src_volume_;
        language_._src_hl2_ = language_.coof_checkpoint_src_hl2_;
        language_._src_hlc3_ = language_.coof_checkpoint_src_hlc3_;
        language_._src_ohlc4_ = language_.coof_checkpoint_src_ohlc4_;
        language_._src_hlcc4_ = language_.coof_checkpoint_src_hlcc4_;
    }
    language_.prev_chart_close_ = language_.coof_checkpoint_prev_chart_close_;
    language_.last_chart_close_ = language_.coof_checkpoint_last_chart_close_;
    host.restore_script_state();
}
 
void PineScheduler::commit_coof_script_state(PineStrategyHost& host) {
    if (language_._src_series_active_) {
        language_.coof_checkpoint_src_open_ = language_._src_open_;
        language_.coof_checkpoint_src_high_ = language_._src_high_;
        language_.coof_checkpoint_src_low_ = language_._src_low_;
        language_.coof_checkpoint_src_close_ = language_._src_close_;
        language_.coof_checkpoint_src_volume_ = language_._src_volume_;
        language_.coof_checkpoint_src_hl2_ = language_._src_hl2_;
        language_.coof_checkpoint_src_hlc3_ = language_._src_hlc3_;
        language_.coof_checkpoint_src_ohlc4_ = language_._src_ohlc4_;
        language_.coof_checkpoint_src_hlcc4_ = language_._src_hlcc4_;
    }
    language_.coof_checkpoint_prev_chart_close_ = language_.prev_chart_close_;
    language_.coof_checkpoint_last_chart_close_ = language_.last_chart_close_;
    language_.coof_checkpoint_contains_current_bar_ = true;
    host.commit_script_state();
}
 
namespace {
// The script bars a SELECTED admitted batch run executes over its retained
// primary rows: the kernel's count_selected_executed_primary, which walks them as
// the native consumer does (the groups that open before T, the last pending one
// completed at the accepted T, and the window groups its seal rules execute; an
// unsealed window tail adds none). The request is the run's own: its window (E
// unchecked: it bounds no such group), its spec's timeframes, timezone and session,
// the host's retained chart timezone and its slot-label policy. A refusal of the
// count after admission, or a count the scheduler's int cannot hold, is the
// engine's fault and fails as engine_invariant; no count is guessed.
int selected_executed_source_bars(const PineStrategyHost& host, const NativeRunSpec* spec,
                                  const pf_selected_window_config_v1& window,
                                  PineScheduler::InputView retained) {
    if (spec == nullptr) {
        throw coded<std::logic_error>(RunFailureCode::engine_invariant, {},
                                      "selected run begin without a running spec for its "
                                      "source tail count");
    }
    SelectedRetainedRequest request;
    request.start_ms = window.start_ms;
    request.end_ms = window.end_ms;
    request.input_tf = spec->input_tf;
    request.script_tf = spec->script_tf;
    request.chart_timezone = host.chart_timezone();
    request.engine_timezone = spec->timezone;
    request.session = spec->session;
    request.feed_tolerant = spec->slot_label_policy == NativeSlotLabelPolicy::FeedTolerant;
    const SelectedRetainedCount counted =
        count_selected_executed_primary(retained.data(), retained.size(), request);
    if (counted.status != SelectedPlanStatus::Ok
        || counted.script_bars > static_cast<std::uint64_t>(std::numeric_limits<int>::max())) {
        throw coded<std::logic_error>(RunFailureCode::engine_invariant, {},
                                      "selected run source tail count refused or out of range "
                                      "after admission");
    }
    return static_cast<int>(counted.script_bars);
}
}  // namespace
 
void PineScheduler::run_begin(PineStrategyHost& host) {
    reset_language();
    const bool static_eligible = !retained_.is_stream && !retained_.bar_magnifier
        && retained_.input_tf.empty() && retained_.script_tf.empty();
    const auto state = host.native_state();
    const bool undetected = state.spec && state.spec->timeframe_undetected;
    const int ratio = undetected || !state.spec ? 1
        : tf_ratio(state.spec->input_tf, state.spec->script_tf);
    const bool needs_aggregation = ratio > 1 || ratio == -1;
    host.diag_script_tf_ratio_ = ratio;
    host.diag_needs_aggregation_ = needs_aggregation;
    input_script_size_ = retained_.bars.size();
    expected_source_bars_ = static_cast<int>(input_script_size_);
    // Count once without retaining flags. Input delivery and consumed-prefix
    // hashing each get their own walk of the very same kernel producer.
    // Ratio 1 uses complete=1/boundary=0 without constructing any cursor.
    if (needs_aggregation && state.spec) {
        ScriptBucketCursor counting(*state.spec);
        expected_source_bars_ = 0;
        for (std::size_t i = 0; i < input_script_size_; ++i)
            expected_source_bars_ += counting.next(retained_.bars[i]).complete != 0U ? 1 : 0;
        input_script_cursor = std::make_unique<ScriptBucketCursor>(*state.spec);
        consumed_script_cursor_ = std::make_unique<ScriptBucketCursor>(*state.spec);
    }
    // A selected admitted batch run executes the script bars of the consumer's
    // own bucket walk, which the aggregator's per-input hint above does not
    // model: the last pending pre-T group is completed at the accepted T and an
    // unsealed window tail is never executed. The tail expectations
    // (terminal_source_bar, the probe tail) read the executed count. An OFF run
    // and a stream never reach this and keep the hint; completion cursors are
    // unchanged either way.
    if (!retained_.is_stream && state.phase == NativeRunPhase::Batch) {
        pf_selected_window_config_v1 window{};
        if (NativeExecutionConsumer::selected_window_admitted(host, &window)) {
            expected_source_bars_ =
                selected_executed_source_bars(host, state.spec, window, retained_.bars);
        }
    }
    host.stream_warmup_mode_ = retained_.is_stream;
    host.scheduler_prepare_script_run(retained_.bars, static_eligible,
                                      expected_source_bars_, !needs_aggregation);
    // ab9714be pine_scheduler.cpp:717-804 (legacy_run_simple) never configures
    // the request.security evaluator surface; :1249-1323 (run_tf_impl, reached
    // by every timeframe-aware, magnified or stream overload, including the
    // generated wrapper's run_backtest / run_backtest_full with EMPTY
    // timeframes) configures it unconditionally after prepare_script_run()
    // received the static-eligibility flag.  Empty timeframes only request
    // auto-detection; only the bare run(bars, n) overload skips the surface,
    // and the kernel reports that overload as NativeBeginArgs::simple_run.
    // (R4-D L10o: gating on static eligibility left every request.security
    // series na whenever a caller auto-detected the timeframe.)
    if (!retained_.simple_run) host.scheduler_configure_security_evaluators();
    uses_aux_security_feed_ = host.scheduler_uses_aux_security_feed();
    host.scheduler_prepare_security_sequence(retained_.bars);
}
 
int PineScheduler::source_bar_index_for(const NativeDecisionContext& context) const noexcept {
    // Matching at a new script-bar open precedes the terminal source callback;
    // all fills after that callback (including COOF/POOC notifications) belong
    // to the already-published source index.  This is the same cadence the
    // legacy aggregation loop used for Trade.entry_bar_index/exit_bar_index.
    // A stream's quiet carried interval is matched at its own open before it
    // joins a script bar, while the context still names the bar before it:
    // that fill belongs to the bar the carried interval opens into (lane F1).
    const std::int64_t script_open_ms =
        context.coordinate.provenance == NativePriceProvenance::CarriedOpen
        ? context.coordinate.open_ms : context.script_bar_open_ms;
    const bool published = last_published_script_open_ms_ == script_open_ms;
    const bool coof_published = coof_callback_script_open_ == script_open_ms;
    if (published || coof_published)
        return std::max(0, source_bar_count_ - 1);
    return source_bar_count_;
}
 
const Bar* PineScheduler::lower_path_bar_at(
        const PineStrategyHost& host, const NativeDecisionContext& context) const noexcept {
    const NativeRunSpec* spec = detail::run_spec(host);
    const auto* lower = spec ? spec->intrabar.lower() : nullptr;
    if (!lower) return nullptr;
    const auto views = detail::run_consumer(host).bar_views();
    if (views.lower_path != nullptr) {
        const auto segment = views.lower_segment;
        if (segment.size == 0) return nullptr;
        const auto found = std::lower_bound(
            segment.data, segment.data + segment.size, context.sub_bar_open_ms,
            [](const Bar& bar, std::int64_t time) { return bar.timestamp < time; });
        return found != segment.data + segment.size && found->timestamp == context.sub_bar_open_ms
            ? found : nullptr;
    }
    // preflight_intrabar_path admitted the path only with strictly
    // increasing stamps.
    const auto found = std::lower_bound(
        lower->bars.begin(), lower->bars.end(), context.sub_bar_open_ms,
        [](const Bar& bar, std::int64_t time) { return bar.timestamp < time; });
    if (found == lower->bars.end() || found->timestamp != context.sub_bar_open_ms)
        return nullptr;
    return &*found;
}
 
void PineScheduler::sub_bar_complete() noexcept {
    if (!retained_.bar_magnifier) return;
    saw_open_fill_ = false;
    open_point_fills_ = 0;
}
 
namespace {
// A one-price bar that traded nothing: a print of a Pine magnified path (the
// chart bar's own open or close, source::tradingview_magnifier_bars).
bool single_print(const Bar& bar) noexcept {
    return bar.volume == 0.0 && bar.high == bar.open && bar.low == bar.open
        && bar.close == bar.open;
}
}  // namespace
 
std::optional<PineScheduler::InputWaypoint> PineScheduler::next_input_waypoint(
        const PineStrategyHost& host, const NativeDecisionContext& context) const noexcept {
    // The sub-bar being walked is a bar of the run's lower path -- one of
    // TradingView's intrabars since R5 lane MAG-INTRABAR, no longer always an
    // input bar -- so it is looked up there.
    const Bar* found = lower_path_bar_at(host, context);
    if (!found) return std::nullopt;
    // TradingView's magnified broker fills a fill recalculation's order at
    // the next of the path's fill points: the chart bar's own open (a print,
    // walked once), then each intrabar's open, first extreme and second
    // extreme -- an intrabar's close is not one, the next intrabar's open
    // follows its second extreme (tests/fixtures/magnifier_intrabars/
    // mi-coof-refill*, 72 of 72 cascades; R5 lane MAG-INTRABAR).
    const auto* spec = detail::run_spec(host);
    const auto views = detail::run_consumer(host).bar_views();
    const Bar* successor = nullptr;
    if (views.lower_path != nullptr) {
        successor = found + 1 < views.lower_segment.data + views.lower_segment.size
            ? found + 1 : views.lower_path->next_after_segment();
    } else {
        const auto& path = spec->intrabar.lower()->bars;
        if (found + 1 < path.data() + path.size()) successor = found + 1;
    }
    const InputWaypoint next_open{successor ? successor->open
        : std::numeric_limits<double>::quiet_NaN(), true};
    if (single_print(*found)) return next_open;
    const bool high_first = host.adapter_.source_path_uses_high_first(*found);
    const NativePathPhase phase[] = {
        NativePathPhase::Open,
        high_first ? NativePathPhase::High : NativePathPhase::Low,
        high_first ? NativePathPhase::Low : NativePathPhase::High,
        NativePathPhase::Close,
    };
    const double price[] = {
        found->open,
        high_first ? found->high : found->low,
        high_first ? found->low : found->high,
        found->close,
    };
    for (int index = 0; index < 4; ++index) {
        if (phase[index] != context.coordinate.path_phase) continue;
        // A fill AT an extreme books that print's tick, which can sit a
        // binary64 ulp off the print (coof_fill_at_path_point): it is at the
        // waypoint, not short of it.
        if (index > 0 && index < 3 && !host.adapter_.coof_fill_at_path_point(price[index]))
            return InputWaypoint{price[index], false};
        // The matcher's fill of a resting request at its own level AT the
        // intrabar's second extreme ends the leg there, and the
        // recalculation's market order fills at that extreme, as on a chart
        // bar (PineExecutionAdapter::coof_fill_at_second_extreme, H-MEASURE
        // Finding 6d): TradingView's hm-mag-diff-v2 / -v3 bar 19 book ML at
        // 11.72, where PX's limit filled on the 2-minute intrabar's high. A
        // fill the adapter forced there -- a cascade's market order -- still
        // takes the next intrabar's open (R5 lane INT27).
        if (index == 2 && !host.adapter_.coof_fill_forced_)
            return InputWaypoint{price[index], false};
        if (index < 2) return InputWaypoint{price[index + 1], false};
        return next_open;
    }
    return std::nullopt;
}
 
void PineScheduler::publish_series(const Bar& bar, PineStrategyHost& host) {
    (void)host;
    update_source_series(bar);
}
 
void PineScheduler::update_source_series(const Bar& bar) {
    if (language_.history_slot_is_new_)
        language_.prev_chart_close_ = language_.last_chart_close_;
    language_.last_chart_close_ = bar.close;
    if (!language_._src_series_active_) return;
    const double hl2 = (bar.high + bar.low) / 2.0;
    const double hlc3 = (bar.high + bar.low + bar.close) / 3.0;
    const double ohlc4 = (bar.open + bar.high + bar.low + bar.close) / 4.0;
    const double hlcc4 = (bar.high + bar.low + bar.close + bar.close) / 4.0;
    if (language_.history_slot_is_new_) {
        language_._src_open_.push(bar.open);
        language_._src_high_.push(bar.high);
        language_._src_low_.push(bar.low);
        language_._src_close_.push(bar.close);
        language_._src_volume_.push(bar.volume);
        language_._src_hl2_.push(hl2);
        language_._src_hlc3_.push(hlc3);
        language_._src_ohlc4_.push(ohlc4);
        language_._src_hlcc4_.push(hlcc4);
        return;
    }
    language_._src_open_.update(bar.open);
    language_._src_high_.update(bar.high);
    language_._src_low_.update(bar.low);
    language_._src_close_.update(bar.close);
    language_._src_volume_.update(bar.volume);
    language_._src_hl2_.update(hl2);
    language_._src_hlc3_.update(hlc3);
    language_._src_ohlc4_.update(ohlc4);
    language_._src_hlcc4_.update(hlcc4);
}
 
double PineScheduler::script_position_view(
        int bar_index, PositionSide side, double quantity) const noexcept {
    if (language_.pos_view_freeze_bar_ == bar_index) {
        if (language_.pos_view_frozen_side_ == PositionSide::LONG)
            return language_.pos_view_frozen_qty_;
        if (language_.pos_view_frozen_side_ == PositionSide::SHORT)
            return -language_.pos_view_frozen_qty_;
        return 0.0;
    }
    if (side == PositionSide::LONG) return quantity;
    if (side == PositionSide::SHORT) return -quantity;
    return 0.0;
}
 
void PineScheduler::freeze_script_position_view(
        int bar_index, PositionSide side, double quantity,
        const std::vector<PyramidEntry>& lots) {
    if (language_.pos_view_freeze_bar_ == bar_index) return;
    language_.pos_view_freeze_bar_ = bar_index;
    language_.pos_view_frozen_side_ = side;
    language_.pos_view_frozen_qty_ = quantity;
    language_.pos_view_frozen_entry_qty_.clear();
    for (const auto& lot : lots)
        language_.pos_view_frozen_entry_qty_[lot.entry_id] += lot.qty;
}
 
void PineScheduler::clear_script_position_view() noexcept {
    language_.pos_view_freeze_bar_ = -1;
}
 
const Series<double>& PineScheduler::source_series(const std::string& key) const {
    if (key == "open") return language_._src_open_;
    if (key == "high") return language_._src_high_;
    if (key == "low") return language_._src_low_;
    if (key == "close") return language_._src_close_;
    if (key == "volume") return language_._src_volume_;
    if (key == "hl2") return language_._src_hl2_;
    if (key == "hlc3") return language_._src_hlc3_;
    if (key == "ohlc4") return language_._src_ohlc4_;
    if (key == "hlcc4") return language_._src_hlcc4_;
    throw coded<std::invalid_argument>(RunFailureCode::engine_invariant, {},
                                       "unknown source series");
}
 
void PineScheduler::fixture_publish_source_series(const Bar& bar, bool new_history_slot) {
    language_.history_slot_is_new_ = new_history_slot;
    language_.is_first_tick_ = new_history_slot;
    update_source_series(bar);
}
 
void PineScheduler::input(
    const Bar& bar, const NativeInputContext& context, PineStrategyHost& host) {
    input_is_observed_ticks_ = false;
    retain_confirmed_input(bar, context.script_interval.open_ms);
    const bool retained_input = context.input_index >= 0
        && static_cast<std::size_t>(context.input_index) < input_script_size_;
    ScriptBucketCursor::Flags flags{};
    if (retained_input && input_script_cursor) {
        const auto index = static_cast<std::size_t>(context.input_index);
        auto& cursor = *input_script_cursor;
        if (cursor.position() > index + 1) cursor.reset();
        while (cursor.position() <= index)
            (void)cursor.next(retained_.bars[cursor.position()]);
        flags = cursor.last();
    }
    // Advance before routed, auxiliary and deferred-input early returns too.
    struct InputBarIndexScope {
        PineStrategyHost& host;
        int previous;
        explicit InputBarIndexScope(PineStrategyHost& value, int index)
            : host(value), previous(value.bar_index_) { host.bar_index_ = index; }
        ~InputBarIndexScope() { host.bar_index_ = previous; }
    } input_bar_index(host, context.input_index);
    // The auxiliary slice is fed per chart bar (scheduler_feed_aux_security);
    // sites the kernel steps (declare_security_sites_to_kernel) are fed by
    // its own pump, right after this callback returns.
    if (uses_aux_security_feed_ || host.security_sites_kernel_routed()) {
        prior_input_script_open_ms_ = context.script_interval.open_ms;
        return;
    }
    std::int64_t next_input_ms = 0;
    if (context.input_index >= 0
        && context.input_index + 1 < static_cast<int>(retained_.bars.size())) {
        next_input_ms = retained_.bars[static_cast<std::size_t>(context.input_index + 1)].timestamp;
    }
    if (retained_.bar_magnifier && deferred_boundary_input_.active
        && deferred_boundary_input_.prior_script_open_ms != context.script_interval.open_ms) {
        // A sparse lower feed reveals the completed caller only when the next
        // child arrives. Feed the retained final child first, then let the
        // new caller's child proceed in timestamp order.
        const std::int64_t completed_script_open =
            deferred_boundary_input_.prior_script_open_ms;
        (void)host.scheduler_feed_security_input(
            deferred_boundary_input_.bar, deferred_boundary_input_.next_input_ms,
            true, false);
        // Keep the new caller's first child out of request.security until the
        // completed caller's chart callback has observed the published value.
        deferred_boundary_input_.bar = bar;
        deferred_boundary_input_.next_input_ms = next_input_ms;
        deferred_boundary_input_.prior_script_open_ms = completed_script_open;
        deferred_boundary_input_.calling_bar_complete = false;
        deferred_boundary_input_.all_security_states = true;
        deferred_boundary_input_.active = true;
        prior_input_script_open_ms_ = context.script_interval.open_ms;
        return;
    }
    // The generic calendar may wait for a later tradable opening before it
    // seals a script interval.  The source chart aggregator can have already
    // completed that interval on the prior raw bar.  Keep the new raw input
    // out of request.security until the pending script callback observes the
    // same legacy point; then feed it immediately after that callback.
    if (!retained_.bar_magnifier && awaiting_legacy_script_open_ms_
        != std::numeric_limits<std::int64_t>::min()) {
        deferred_boundary_input_.bar = bar;
        deferred_boundary_input_.next_input_ms = next_input_ms;
        deferred_boundary_input_.prior_script_open_ms = awaiting_legacy_script_open_ms_;
        deferred_boundary_input_.calling_bar_complete = false;
        deferred_boundary_input_.all_security_states = true;
        deferred_boundary_input_.active = true;
        prior_input_script_open_ms_ = context.script_interval.open_ms;
        return;
    }
 
    bool calling_bar_complete = context.completes_script_interval;
    bool boundary = prior_input_script_open_ms_
        != std::numeric_limits<std::int64_t>::min()
        && prior_input_script_open_ms_ != context.script_interval.open_ms;
    if (retained_input) {
        calling_bar_complete = flags.complete != 0U;
        boundary = flags.boundary != 0U;
    } else if (retained_.is_stream && context.input_index >= 0) {
        calling_bar_complete = confirmed_script_interval_complete(context);
        boundary = false;
    }
    // The final sparse magnifier child is a partial requested bucket.  The
    // legacy lower-TF pump does not promote that tail to a completed
    // request.security value merely because the input array ended.
    if (retained_.bar_magnifier && next_input_ms == 0)
        calling_bar_complete = false;
    bool security_boundary_ahead = false;
    if (retained_.bar_magnifier && next_input_ms != 0) {
        const bool caller_boundary = tf_change(
            bar.timestamp, next_input_ms, retained_.script_tf,
            host.syminfo_.timezone, host.syminfo_.session);
        if (caller_boundary) {
            for (const auto& state : host.security_eval_states_) {
                if (host.pine_security_state(state.sec_id).publish_gate_tf_seconds > 0) {
                    security_boundary_ahead = true;
                    break;
                }
            }
        }
    }
    if (security_boundary_ahead) {
        deferred_boundary_input_.bar = bar;
        deferred_boundary_input_.next_input_ms = next_input_ms;
        deferred_boundary_input_.prior_script_open_ms = context.script_interval.open_ms;
        deferred_boundary_input_.calling_bar_complete = true;
        deferred_boundary_input_.all_security_states = false;
        deferred_boundary_input_.active = true;
        prior_input_script_open_ms_ = context.script_interval.open_ms;
        return;
    }
    const bool deferred_gate = host.scheduler_feed_security_input(
        bar, next_input_ms, calling_bar_complete,
        boundary && !retained_.bar_magnifier);
    if (deferred_gate) {
        deferred_boundary_input_.bar = bar;
        deferred_boundary_input_.next_input_ms = next_input_ms;
        deferred_boundary_input_.prior_script_open_ms = prior_input_script_open_ms_;
        deferred_boundary_input_.calling_bar_complete = false;
        deferred_boundary_input_.all_security_states = false;
        deferred_boundary_input_.active = true;
    }
    if (calling_bar_complete && !retained_.bar_magnifier) {
        awaiting_legacy_script_open_ms_ = boundary
            ? prior_input_script_open_ms_ : context.script_interval.open_ms;
    }
    prior_input_script_open_ms_ = context.script_interval.open_ms;
}
 
void PineScheduler::tick(const Bar& bar, const NativeTickContext& context,
                         PineStrategyHost& host) {
    if (!retained_.is_stream) return;
    input_is_observed_ticks_ = true;
    const auto& interval = context.decision.input_interval;
    if (last_stream_input_open_ms_ == interval.open_ms) return;
    if (prior_input_script_open_ms_ != std::numeric_limits<std::int64_t>::min()
        && prior_input_script_open_ms_ != context.decision.script_interval.open_ms) {
        // A realtime tick can be the first child of the next caller after a
        // sparse warmup.  The legacy stream replays the retained final child
        // at this boundary (without advancing the requested-context slot).
        host.scheduler_publish_security_boundary();
    }
    last_stream_input_open_ms_ = interval.open_ms;
    const auto& script = context.decision.script_interval;
    const bool complete = interval.next_period_open_ms >= script.next_period_open_ms;
    (void)host.scheduler_feed_security_input(
        bar, interval.next_input_open_ms, complete, false);
    prior_input_script_open_ms_ = script.open_ms;
}
 
void PineScheduler::bar_open(const Bar& value, const NativeDecisionContext& context,
                             PineStrategyHost& host) {
    if (context.script_bar_open_ms != current_script_open_ms_) {
        current_script_open_ms_ = context.script_bar_open_ms;
        saw_open_fill_ = false;
        open_point_fills_ = 0;
        coof_callback_script_open_ = std::numeric_limits<std::int64_t>::min();
        if (host.scheduler_coof_enabled()) snapshot_coof_script_state(host);
    }
    if (!retained_.is_stream) {
        current_script_bar_ = value;
        current_script_bar_.timestamp = context.script_bar_open_ms;
        current_script_bar_valid_ = true;
    }
}
 
void PineScheduler::bar(const Bar& value, const NativeDecisionContext& context, PineStrategyHost& host) {
    // P7c: matching advances over every sub-bar; the language callback occurs
    // only at the terminal sub-bar with the script bar timestamp restored.
    language_.is_first_tick_ = context.is_terminal_sub_bar;
    language_.is_last_tick_ = context.is_terminal_sub_bar;
    language_.history_slot_is_new_ = context.is_terminal_sub_bar;
    if (!context.is_terminal_sub_bar) return;
    // A FeedTolerant native stream can emit a quiet carried callback for
    // the calendar-aligned slot immediately preceding the last raw-label
    // warmup bar. The retired source stream starts its realtime cadence from
    // that warmup bar's next source slot, so it never published this stale
    // callback to generated code.
    if (retained_.is_stream && current_script_bar_valid_
        && context.script_bar_open_ms <= current_script_bar_.timestamp) {
        // The generic driver still dispatched a script point. Preserve the
        // public one-hash-per-dispatch accounting without exposing it to the
        // source callback cadence.
        host.scheduler_record_broker_hash();
        return;
    }
    const bool coof = host.scheduler_coof_enabled();
    const bool had_coof_recalc = coof_callback_script_open_
        == context.script_bar_open_ms;
    Bar script_bar = value;
    script_bar.timestamp = context.script_bar_open_ms;
    if (retained_.is_stream && input_is_observed_ticks_
        && !retained_.bars.empty() && calendar_period_for(retained_.script_tf) == CalendarPeriod::DAY
        && tf_to_seconds(retained_.input_tf) == 86400
        && host.syminfo_.session == "1800-1700"
        && host.syminfo_.timezone == "America/New_York") {
        const auto stamp = session_period_open_ms(script_bar.timestamp,
            host.syminfo_.timezone, host.syminfo_.session, CalendarPeriod::DAY);
        const auto next_stamp = session_period_open_ms(retained_.bars.back().timestamp + 86400000LL,
            host.syminfo_.timezone, host.syminfo_.session, CalendarPeriod::DAY);
        script_bar.timestamp = std::max(stamp, next_stamp);
    }
    current_script_bar_ = script_bar;
    current_script_bar_valid_ = true;
    const bool completes_awaiting_legacy_script = awaiting_legacy_script_open_ms_
        == context.script_bar_open_ms;
    if (deferred_boundary_input_.active
        && !deferred_boundary_input_.all_security_states
        && deferred_boundary_input_.prior_script_open_ms == context.script_bar_open_ms) {
        // The legacy magnifier feeds the boundary-triggering lower bar after
        // the completed script callback; it does not replay the caller at the
        // callback boundary.  The ordinary batch loop does replay it.
        if (!retained_.bar_magnifier)
            host.scheduler_publish_security_boundary();
    }
    const int chart_index = context.coordinate.input_interval_index;
    // A calc_on_order_fills recalculation of this bar before this callback
    // fed its slice already (recalculate).
    if (uses_aux_security_feed_ && !had_coof_recalc) host.scheduler_feed_aux_security(chart_index);
    if (coof) {
        restore_coof_script_state(host);
        language_.is_first_tick_ = true;
        language_.is_last_tick_ = true;
        language_.history_slot_is_new_ =
            !language_.coof_checkpoint_contains_current_bar_;
    }
    publish_series(script_bar, host);
    // The session flags belong to the script bar, not to a callback: a
    // calc_on_order_fills recalculation that already published this bar set
    // them (recalculate below) and its close reads the same answer (lane F1).
    // They are the kernel's session-day facts of the bar (lane F5), which
    // every callback of the bar carries alike.
    if (const NativeRunSpec* spec = detail::run_spec(host); spec
        && !spec->timeframe_undetected && !had_coof_recalc) {
        host.scheduler_update_session_state();
    }
    const bool suppress_probe_tail = host.probe_suppress_tail_logic()
        && expected_source_bars_ > 0
        && source_bar_count_ + 1 >= expected_source_bars_;
    if (suppress_probe_tail) host.scheduler_publish_suppressed_tail(script_bar);
    else {
        host.scheduler_publish_source_bar(script_bar, true, !had_coof_recalc);
        // The bar a recalculation published early becomes the previous bar
        // of timeframe.change() only now, after its own close (recalculate).
        if (had_coof_recalc) host.prev_bar_timestamp_ = script_bar.timestamp;
        last_published_script_open_ms_ = context.script_bar_open_ms;
    }
    if (coof) commit_coof_script_state(host);
    if (uses_aux_security_feed_) host.scheduler_feed_deferred_aux_security(chart_index);
    if (deferred_boundary_input_.active
        && deferred_boundary_input_.prior_script_open_ms == context.script_bar_open_ms) {
        if (deferred_boundary_input_.all_security_states) {
            (void)host.scheduler_feed_security_input(
                deferred_boundary_input_.bar, deferred_boundary_input_.next_input_ms,
                deferred_boundary_input_.calling_bar_complete, false);
        } else {
            host.scheduler_feed_deferred_security_input(
                deferred_boundary_input_.bar, deferred_boundary_input_.next_input_ms);
        }
        deferred_boundary_input_ = {};
    }
    if (completes_awaiting_legacy_script) {
        awaiting_legacy_script_open_ms_ = std::numeric_limits<std::int64_t>::min();
    }
    if (!had_coof_recalc) ++source_bar_count_;
    if (terminal_source_bar()) {
        if (!suppress_probe_tail) host.scheduler_record_range_end(current_script_bar_);
        if (!retained_.is_stream) host.scheduler_finish_security_sequence();
    }
    host.scheduler_record_broker_hash();
}
 
void PineScheduler::applied(const native_order::ExecutionAppliedEvent& event) {
    if (event.ordinal <= applied_cursor_) return;
    applied_cursor_ = event.ordinal;
}
 
bool PineScheduler::coof_recalculation_due(
        const native_order::ExecutionAppliedEvent& event,
        const NativeDecisionContext& context, PineStrategyHost& host) const {
    // applied() advanced the cursor onto this event a moment ago, in the same
    // drain iteration. A replayed ordinal never scheduled a source callback.
    if (event.ordinal != applied_cursor_) return false;
    if (!host.scheduler_coof_enabled()) return false;
    if (host.config_.process_orders_on_close
        && (context.coordinate.provenance == NativePriceProvenance::Calculation
            || context.coordinate.provenance == NativePriceProvenance::AfterCalculationClose
            || context.coordinate.path_phase == NativePathPhase::None)) {
        // ab9714be pine_scheduler.cpp terminal POOC dispatch: a fill at the
        // already-consumed close is final for that script bar and does not
        // schedule a calc_on_order_fills source callback.
        return false;
    }
    // So is a strategy.close / close_all with immediately = true that the
    // bar's own calculation executed: TradingView recalculates after it
    // neither -- an entry sent before it while the pyramiding limit held
    // never fills, one sent after it fills once at the next open (lab tv
    // tapes te-coof-immediate-reentry-*, tests/fixtures/coof_immediate_close;
    // lane TAIL-E). A recalculation there re-sent the entry, which the kernel
    // filled at the same close -- and, when it also closed again, looped
    // without end.
    if (host.adapter_.immediate_calculation_close(event, context)) return false;
    if (host.adapter_.suppress_grouped_stop_recalc(event, context)) return false;
    return true;
}
 
// One calc_on_order_fills recalculation at the kernel's OrderFill cursor
// (NativeCalculationTrigger::BarCloseAndFills): the consumer has already
// delivered this event's on_native_applied and re-entered the same point
// frame, so everything below is Pine's own language-state rollback,
// publication and first-open execution chain.
void PineScheduler::recalculate(const native_order::ExecutionAppliedEvent& event,
                                const NativeDecisionContext& context, PineStrategyHost& host) {
    // ab9714be pine_scheduler.cpp:531-537: the historical O point admits the
    // carried order's open fill and one refill; every later fill advances
    // along the path.  The native matcher may book such a fill at the start
    // (t == 0) of the O->W1 segment instead of at the Open phase; both are
    // the same O point, so its recalculation carries Open provenance.  Once
    // the O point is spent, a further fill at that price lies on the O->W1
    // leg (pine_scheduler.cpp:571-606) and its recalculation says so.
    const bool bar_known = current_script_bar_valid_
        && current_script_bar_.timestamp == context.script_bar_open_ms;
    const NativePathPhase first_extreme = bar_known
        && host.adapter_.source_path_uses_high_first(current_script_bar_)
        ? NativePathPhase::High : NativePathPhase::Low;
    // A print of a magnified path (the chart bar's own open) admits the
    // carried fill alone: TradingView's refill waits for the first intrabar's
    // open (sub_bar_complete, next_input_waypoint).
    // On a lower path a point's phase is its sub-bar's, so the open point is
    // that sub-bar's own Open: the chart bar's first extreme says nothing
    // about it.
    const Bar* path_bar = lower_path_bar_at(host, context);
    const bool open_point = path_bar
        ? !single_print(*path_bar) && context.coordinate.path_phase == NativePathPhase::Open
        : context.coordinate.path_phase == NativePathPhase::Open
            || (bar_known && context.coordinate.path_phase == first_extreme
                && event.cursor.t == 0.0);
    const bool at_open = open_point && open_point_fills_ < 2;
    open_point_fills_ = at_open ? open_point_fills_ + 1 : 2;
    const bool first_open = at_open && !saw_open_fill_;
    if (at_open) saw_open_fill_ = true;
    const bool first_callback = coof_callback_script_open_
        != context.script_bar_open_ms;
    const bool callback_advances_source_bar = first_callback
        && !language_.coof_checkpoint_contains_current_bar_;
    // The bar's first recalculation before its close callback feeds the
    // bar's auxiliary slice. TradingView runs a recalculation of a historical
    // bar on the whole bar, its finer request.security reads as well as its
    // close, high and low: the 5m close read after the open fill, and after a
    // later fill of the bar, is the chart bar's close, a lookahead_on 5m site
    // reads the bar's first 5m bar and a 5m lower_tf array all three (lab tv
    // int28fix-rc, -rc2, -rc3; lane INT28-FIX rule RC). The slice waited for
    // the close callback, so the recalculation read the previous bar's. A
    // recalculation after the close callback (a fill the close booked) finds
    // the slice that callback fed.
    if (uses_aux_security_feed_ && callback_advances_source_bar)
        host.scheduler_feed_aux_security(context.coordinate.input_interval_index);
    // COOF re-evaluates the source script against the full script bar while
    // the native current-execution coordinate still supplies the fill price
    // for sizing/placement.  A one-price synthetic callback erases high/low,
    // volume and barstate facts that the legacy scheduler retained.
    Bar callback_bar = current_script_bar_valid_
        && current_script_bar_.timestamp == context.script_bar_open_ms
        ? current_script_bar_ : host.current_bar_;
    callback_bar.timestamp = context.script_bar_open_ms;
    restore_coof_script_state(host);
    language_.is_first_tick_ = true;
    language_.is_last_tick_ = true;
    language_.history_slot_is_new_ =
        !language_.coof_checkpoint_contains_current_bar_;
    publish_series(callback_bar, host);
    // A fill recalculation runs the script ON the bar the fill is on, so it
    // reads that bar's session flags, as ab9714be's loop set them before any
    // of the bar's fills. The first publication of the bar asks; the close
    // callback and later recalculations of the bar read the same answer
    // (lane F1).
    if (const NativeRunSpec* spec = detail::run_spec(host); spec
        && !spec->timeframe_undetected
        && last_published_script_open_ms_ != context.script_bar_open_ms) {
        host.scheduler_update_session_state();
    }
    NativeDecisionContext coof_context = context;
    if (at_open) coof_context.coordinate.path_phase = NativePathPhase::Open;
    else if (open_point && bar_known) coof_context.coordinate.path_phase = first_extreme;
    // timeframe.change() compares the script bar with the bar before it. A
    // recalculation publishes its bar ahead of the bar's close, and every
    // later calculation of the bar still compares with that same previous
    // bar: TradingView resets a day's counter on the close of a day's first
    // bar that a fill recalculated (lab tv tape
    // tests/fixtures/coof_timeframe_change td-m3a; lane TAIL-D).
    const std::int64_t previous_bar_timestamp = host.prev_bar_timestamp_;
    host.adapter_.begin_coof_recalc(
        event, coof_context, first_open, host.broker_fill_event_seq_);
    try {
        host.scheduler_publish_source_bar(
            callback_bar, true, callback_advances_source_bar);
        // Low/high/close recalculations stage source requests until the
        // callback has finished so their statement order is complete.  Drain
        // that source queue while the fill coordinate is still current; the
        // accepted MARKET newborns below then execute at this same broker
        // point, matching calc_on_order_fills chronology.
        host.adapter_.flush_coof_tail(/*openings_only=*/true);
        host.adapter_.flush_coof_tail();
    } catch (...) {
        host.prev_bar_timestamp_ = previous_bar_timestamp;
        host.adapter_.end_coof_recalc();
        throw;
    }
    host.prev_bar_timestamp_ = previous_bar_timestamp;
    host.adapter_.end_coof_recalc();
    restore_coof_script_state(host);
    coof_callback_script_open_ = context.script_bar_open_ms;
    last_published_script_open_ms_ = context.script_bar_open_ms;
    if (callback_advances_source_bar) ++source_bar_count_;
    if (!first_open) {
        auto newborns = host.adapter_.take_first_open_newborns();
        for (const auto& handle : newborns) {
            (void)host.execute_current({handle, NativeCurrentPriceRule::NearestTick});
        }
        return;
    }
    constexpr std::uint64_t kNoFillEventBudget = std::numeric_limits<std::uint64_t>::max();
    // The same TradingView literal project() hands the kernel as
    // max_recalculations_per_point; here it still bounds Pine's own first-open
    // execution chain inside one recalculation.
    constexpr std::size_t kFirstOpenLoopGuard = kCoofLoopGuard;
    std::uint64_t budget = kNoFillEventBudget;
    std::size_t executed = 0;
    for (;;) {
        auto newborns = host.adapter_.take_first_open_newborns();
        if (newborns.empty()) break;
        for (const auto& handle : newborns) {
            if (budget == 0 || executed == kFirstOpenLoopGuard) {
                throw coded<std::overflow_error>(RunFailureCode::recalc_cap, {},
                                                 "Pine COOF first-open loop guard exhausted");
            }
            --budget;
            ++executed;
            (void)host.execute_current({handle, NativeCurrentPriceRule::NearestTick});
        }
    }
}
 
} // namespace pineforge::source