cr150 L17a: selected latches and anchor E

docs/design/cr150-contracts.md

@@ -210,7 +210,7 @@ references; L15a must maintain these anchors and mechanically pin every current
 | Host feed lookup/delivery and column reads | `installed_symbol_feed` `src/source/pine_security_eval.cpp:1059-1068`; `installed_symbol_feed` `src/source/pine_security_eval.cpp:1291-1301`; `installed_symbol_feed` `src/source/pine_security_eval.cpp:1320-1328` | 359-360 |
 | Feed validation: bar values/order, close length/bounds, column length; configure and consumer entry points | `feed.bars` `src/native_run_spec.cpp:259-290`; `instrument_feed_shapes` `src/native_run_spec.cpp:743`; `validate_native_instrument_feeds` `src/native_run_spec.cpp:1019-1022`; `validate_native_timeframe_subscriptions` `src/native_execution_consumer.cpp:9597-9600`; `validate_native_timeframe_subscriptions` `src/native_execution_consumer.cpp:9622-9624` | 361-363 |
 | Adapter projection validation: use the context normalizer; host stages after lending and before projection, using a context built from its store | `normalize_projected_spec` `src/source/pine_adapter.cpp:3004-3010` | 361-363; V1/V2 change 5 |
-| Instrument-feed digest and cached/uncached continuation folds | `digest_instrument_feed_views` `src/native_run_spec.cpp:1162-1205`; `native_instrument_feeds_digest` `src/native_execution_consumer.cpp:322-324`; `native_instrument_feeds_digest` `src/native_execution_consumer.cpp:2775-2776`; `native_run_spec_digest` `src/native_execution_consumer.cpp:12899-12904` | 364-365 |
+| Instrument-feed digest and cached/uncached continuation folds | `digest_instrument_feed_views` `src/native_run_spec.cpp:1162-1205`; `native_instrument_feeds_digest` `src/native_execution_consumer.cpp:322-324`; `native_instrument_feeds_digest` `src/native_execution_consumer.cpp:2775-2776`; `native_run_spec_digest` `src/native_execution_consumer.cpp:12964-12969` | 364-365 |
 | Staged feed counts in host state and its hash | `installed_symbol_feed` `src/source/pine_strategy_host.cpp:234-242`; `staged_.instrument_feeds` `src/source/pine_state_hash.cpp:293-300` | 366-367 |
 | Refresh of symbol-data digest and host hash fold | `refresh_symbol_data_digest` `src/source/pine_security_eval.cpp:1545-1553`; `symbol_data_digest_` `src/source/pine_state_hash.cpp:1151-1157` | 367-368 |
 | Restore/store upkeep and column-setter length check | `held.bars` `src/source/pine_security_eval.cpp:1091-1101`; `data.bars` `src/source/pine_security_eval.cpp:1428-1431` | 369-370 |

src/native_execution_consumer.cpp

@@ -5,7 +5,7 @@
 #include "engine_value_invalid.hpp"
 #include "native_matching.hpp"
 #include "selected_report_view.hpp"
-
+#include "report_sink_writer.hpp"
 #include <pineforge/execution_close_scope.hpp>
 #include <pineforge/market_driver.hpp>
 #include <pineforge/timeframe.hpp>
@@ -1531,7 +1531,7 @@ void NativeExecutionConsumer::terminal_success(BacktestEngine& engine,
         seen.phase = kEpochCapturing;
         {
             const EpochDispatchScope dispatch(epoch);
-            capture_presentation(engine);
+            if (!capture_selected_sink(engine)) capture_presentation(engine);
         }
         // The barrier is released. Abort and first cause first (check_abort
         // answers false for either), then the reentry latch, which is consumed
@@ -12410,7 +12410,7 @@ void NativeExecutionConsumer::drop_presentation() noexcept {
     presentation_ = ReportC{};
     presentation_metric_equity_len_ = 0;
     presentation_metric_observations_ = 0;
-    presentation_ready_ = false;
+    presentation_ready_ = presentation_sink_equity_written_ = false;
     selected_results_open_ = false;
 }
 
@@ -12596,17 +12596,17 @@ bool NativeExecutionConsumer::fill_selected_report(const BacktestEngine& engine,
         return true;
     }
 
-    // The capture's scalars come across by value; every array pointer starts
-    // null, so what a throw frees below is only what this call allocated and
-    // never the capture's own. The two arrays the capture owns are copied; the
-    // others are the suffix's to build, as fill_report builds them after
-    // present_report.
+    // The capture's scalars come across by value; array pointers start null,
+    // so a throw frees only storage this call allocated. A sink's E records
+    // already exist; its metrics read the captured curve without another copy.
+    // Sink-off reads copy both arrays, and the suffix builds the other arrays.
+    // present_report is never called at read.
     const ReportC& captured = consumer->presentation_;
-    // Two equity lengths, as OFF has them: the presented one
-    // (captured.equity_curve_len, whatever present_report left, published as it
-    // stands) and the metrics' own, the engine vector's size from before that
-    // call, kept beside the capture. OFF's buffer holds the second and a reader
-    // may take either, so the copy covers the larger.
+    const bool sink_equity = consumer->selected_sink_equity_written(engine);
+    // The presentation may publish a different length than metrics use.
+    // Sink-off storage covers their maximum; a sink reads the capture's
+    // metric storage directly. The captured lengths keep their preexisting
+    // roles even when a custom presentation shrinks the published curve.
     const int64_t metric_equity_len = consumer->presentation_metric_equity_len_;
     const int64_t equity_storage_len = std::max(captured.equity_curve_len, metric_equity_len);
     ReportC local = captured;
@@ -12621,12 +12621,12 @@ bool NativeExecutionConsumer::fill_selected_report(const BacktestEngine& engine,
             local.trades = new TradeC[captured.trades_len];
             std::copy(captured.trades, captured.trades + captured.trades_len, local.trades);
         }
-        if (captured.equity_curve != nullptr && equity_storage_len > 0) {
+        if (captured.equity_curve != nullptr && equity_storage_len > 0 && !sink_equity) {
             local.equity_curve = new pf_equity_point_t[equity_storage_len];
             std::copy(captured.equity_curve, captured.equity_curve + equity_storage_len,
                       local.equity_curve);
         }
-
+        const auto* metric_curve = sink_equity ? captured.equity_curve : local.equity_curve;
         // The suffix, in fill_report's order. The metrics get the original
         // length, as fill_metrics_section passes its `n`: never the presented
         // length, script_bars_processed, or the engine's vector read now.
@@ -12638,7 +12638,7 @@ bool NativeExecutionConsumer::fill_selected_report(const BacktestEngine& engine,
             // input's close, undefined (NaN) while the window has none. The
             // trade blocks and every other equity formula are the ordinary ones.
             internal::compute_report_metrics(local.metrics, local.trades, local.trades_len,
-                                             local.equity_curve, metric_equity_len,
+                                             metric_curve, metric_equity_len,
                                              local.net_profit, engine.initial_capital_,
                                              engine.chart_timezone_,
                                              consumer->selected_first_open_,
@@ -12654,7 +12654,6 @@ bool NativeExecutionConsumer::fill_selected_report(const BacktestEngine& engine,
         }
         engine.fill_security_diag_section(&local);
         engine.fill_trace_section(&local);
-
         const int64_t hn = (int64_t)engine.broker_state_hashes_.size();
         local.broker_state_hash_len = hn;
         local.broker_state_hash = hn > 0 ? new uint64_t[hn] : nullptr;
@@ -12666,6 +12665,7 @@ bool NativeExecutionConsumer::fill_selected_report(const BacktestEngine& engine,
         BacktestEngine::free_report(&local);
         throw;
     }
+    if (sink_equity) local.equity_curve_len = 0;
     *out = local;
     return true;
 }
@@ -12861,6 +12861,71 @@ int native_selected_window_counts_v1(const BacktestEngine* engine,
     return NativeExecutionConsumer::selected_window_counts_v1(engine, out);
 }
 
+// CR-150 L17A: L6b-1 still owns the full recorded curve. Walk its FINAL values
+// at capture through the same O(1) latch/writer boundary that the hold-back
+// release will feed when L6b-2 and L17B are integrated. Nothing below records
+// provisional values or mutates the engine's point through a presentation.
+
+bool NativeExecutionConsumer::selected_sink_equity_written(
+        const BacktestEngine& engine) const noexcept {
+    return presentation_sink_equity_written_ && engine.report_sink_state_.enabled
+        && engine.report_sink_state_.writer != nullptr;
+}
+
+bool NativeExecutionConsumer::capture_selected_sink(BacktestEngine& engine) {
+    auto& sink = engine.report_sink_state_;
+    if (!selected_epoch_.admitted_selected || !sink.enabled) return false;
+    if (!sink.writer || sink.receipt.complete)
+        refuse_selected_capture("native selected report: no active report sink at capture");
+    try {
+        // Preserve the existing pre-hook refusal order, the single virtual
+        // presentation, and its independent array ownership. These are the
+        // L6b-1 metrics inputs; L17B replaces them with the scratch point source.
+        capture_presentation(engine);
+        const auto& window = selected_epoch_.admitted_window;
+        report::SelectedReportStream stream(window.start_ms, window.end_ms,
+                                            engine.initial_capital_);
+        for (int i = 0; i < presentation_.trades_len; ++i)
+            stream.note_trade(presentation_.trades[i].entry_time);
+        for (const auto& lot : engine.pyramid_entries_)
+            stream.note_open_lot(lot.time);
+
+        std::int64_t selected_index = 1; // the synthetic anchor is not an observation
+        for (const auto& recorded : engine.equity_curve_) {
+            pf_equity_point_t final_point = recorded;
+            bool publish = false;
+            if (recorded.time_ms >= window.start_ms && recorded.time_ms < window.end_ms) {
+                // The pre-hook count check proved this storage exists, even
+                // when a custom hook leaves its presented length behind.
+                if (selected_index >= presentation_metric_equity_len_)
+                    refuse_selected_capture("native selected report: missing captured observation");
+                final_point = presentation_.equity_curve[selected_index];
+                publish = selected_index < presentation_.equity_curve_len;
+                ++selected_index;
+            }
+            // Every value supplied here is final: range-end edits already ran
+            // on recorded, and present_report edited only the captured copy.
+            stream.write_point(*sink.writer, final_point, publish);
+            sink.receipt.count_e = stream.equity_count();
+        }
+        stream.finish(*sink.writer, selected_epoch_.counts.window_script_bars,
+                      presentation_.equity_curve_len > 0);
+        sink.receipt.count_e = stream.equity_count();
+        presentation_sink_equity_written_ = true;
+        // L6b-1 finishes T and the trailer at the successful report fill. An
+        // observer fault after this capture therefore cannot publish a trailer.
+    } catch (const report_sink::WriterError& error) {
+        sink.writer.reset();
+        const auto code = error.reason() == report_sink::WriterFailure::io
+            ? RunFailureCode::report_sink_failed : RunFailureCode::engine_invariant;
+        throw coded<std::runtime_error>(code, {}, error.what());
+    } catch (...) {
+        sink.writer.reset();
+        throw;
+    }
+    return true;
+}
+
 ScriptBucketCursor::ScriptBucketCursor(const NativeRunSpec& spec)
     : script_tf_(spec.script_tf), input_tf_(spec.input_tf), timezone_(spec.timezone),
       session_(spec.session), aggregator_(script_tf_, input_tf_, timezone_, session_) {}

src/native_execution_consumer.hpp

@@ -2434,6 +2434,15 @@ class NativeExecutionConsumer final : public IExecutionConsumer {
     void* batch_input_owner_ = nullptr;
     void (*batch_input_release_)(void*) noexcept = nullptr;
 
+    // CR-150 L17A: selected E records have already been emitted at capture.
+    // Output bookkeeping, like presentation_ready_: not hashed and never read
+    // by a broker decision, a continuation fold or the source adapter. Reset
+    // with the presentation; appended so every pre-existing member keeps its
+    // offset. The L6b-1 array remains the metrics source until L17B supplies the
+    // scratch curve; no second per-read equity allocation is needed with a sink.
+    bool presentation_sink_equity_written_ = false;
+    bool capture_selected_sink(BacktestEngine& engine);
+    bool selected_sink_equity_written(const BacktestEngine& engine) const noexcept;
 public:
     void set_batch_input_release(void* owner, void (*release)(void*) noexcept) noexcept {
         batch_input_owner_ = owner;

src/selected_report_view.cpp

@@ -62,3 +62,71 @@ SelectedEquityCurve build_selected_equity_curve(
 
 }  // namespace report
 }  // namespace pineforge
+
+#include <pineforge/report_sink_writer.hpp>
+
+namespace pineforge {
+namespace report {
+
+SelectedReportStream::SelectedReportStream(std::int64_t start_ms, std::int64_t end_ms,
+                                          double initial_capital) noexcept
+    : start_ms_(start_ms), end_ms_(end_ms), initial_capital_(initial_capital) {}
+
+void SelectedReportStream::note_trade(std::int64_t entry_time) noexcept {
+    trade_before_window_ = trade_before_window_ || entry_time < start_ms_;
+}
+
+void SelectedReportStream::note_open_lot(std::int64_t entry_time) noexcept {
+    lot_before_window_ = lot_before_window_ || entry_time < start_ms_;
+}
+
+void SelectedReportStream::write_anchor(report_sink::LineWriter& writer) {
+    if (anchor_written_) return;
+    writer.write_e({start_ms_, initial_capital_, 0.0});
+    anchor_written_ = true;
+    ++equity_count_;
+}
+
+void SelectedReportStream::write_point(report_sink::LineWriter& writer,
+                                      const pf_equity_point_t& point, bool publish) {
+    if (point.time_ms < start_ms_) {
+        account_changed_ = account_changed_
+            || point.equity != initial_capital_ || point.open_profit != 0.0;
+    }
+    const bool in_window = point.time_ms >= start_ms_ && point.time_ms < end_ms_;
+    if (!in_window) {
+        if (observations_ != 0) gap_after_window_ = true;
+        return;
+    }
+    not_contiguous_ = not_contiguous_ || gap_after_window_;
+    if (observations_ == std::numeric_limits<std::uint64_t>::max())
+        refuse_view("native selected report: the observation count overflowed");
+    ++observations_;
+    if (!publish) return;
+    write_anchor(writer);
+    // The writer already enforces the JSON-safe count bound; guard the C++
+    // counter too, before incrementing a signed value.
+    if (equity_count_ == std::numeric_limits<std::int64_t>::max())
+        refuse_view("native selected report: the equity count overflowed");
+    writer.write_e(point);
+    ++equity_count_;
+}
+
+void SelectedReportStream::finish(report_sink::LineWriter& writer,
+                                 std::uint64_t expected_observations,
+                                 bool publish_anchor) {
+    if (trade_before_window_)
+        refuse_view("native selected report: a trade was entered before the window start");
+    if (lot_before_window_)
+        refuse_view("native selected report: a position was carried in before the window start");
+    if (account_changed_)
+        refuse_view("native selected report: the account changed before the window start");
+    if (not_contiguous_)
+        refuse_view("native selected report: the window's equity observations are not one run");
+    if (observations_ != expected_observations)
+        refuse_view("native selected report: the window's equity observations differ from its script bars");
+    if (publish_anchor) write_anchor(writer);
+}
+
+}  // namespace report
+}  // namespace pineforge

src/selected_report_view.hpp

@@ -64,3 +64,51 @@ SelectedEquityCurve build_selected_equity_curve(
 
 }  // namespace report
 }  // namespace pineforge
+
+// CR-150 L17A. Kept after the existing view so its cited lines do not move.
+namespace pineforge {
+namespace report_sink { class LineWriter; }
+namespace report {
+
+// O(1) selected-window validation and E emission. A caller supplies each FINAL
+// point once, only when the point can no longer be rewritten. Pre-T points are
+// checked but not emitted; points at/after E are neither selected nor emitted.
+// The first emitted observation is preceded by the anchor at T, including when
+// that observation is itself stamped T. finish() emits the anchor for M = 0.
+// No point, trade row, engine pointer or writer is retained here.
+class SelectedReportStream {
+public:
+    SelectedReportStream(std::int64_t start_ms, std::int64_t end_ms,
+                         double initial_capital) noexcept;
+    void note_trade(std::int64_t entry_time) noexcept;
+    void note_open_lot(std::int64_t entry_time) noexcept;
+    // publish distinguishes a host's presented length from its metric storage
+    // length (the existing selected-report convention). All final observations
+    // are judged and counted, including a suffix hidden by a presentation hook.
+    void write_point(report_sink::LineWriter& writer,
+                     const pf_equity_point_t& final_point, bool publish);
+    // Refuse in capture order: trades, lots, account, contiguity, count. A
+    // refusal throws coded<logic_error>(engine_invariant) with the established
+    // selected-capture text; it never writes a trailer. publish_anchor is false
+    // only when a custom presentation publishes an empty curve.
+    void finish(report_sink::LineWriter& writer, std::uint64_t expected_observations,
+                bool publish_anchor);
+    std::int64_t equity_count() const noexcept { return equity_count_; }
+    std::uint64_t observations() const noexcept { return observations_; }
+
+private:
+    void write_anchor(report_sink::LineWriter& writer);
+    std::int64_t start_ms_, end_ms_;
+    double initial_capital_;
+    std::uint64_t observations_ = 0;
+    std::int64_t equity_count_ = 0;
+    bool trade_before_window_ = false;
+    bool lot_before_window_ = false;
+    bool account_changed_ = false;
+    bool gap_after_window_ = false;
+    bool not_contiguous_ = false;
+    bool anchor_written_ = false;
+};
+
+}  // namespace report
+}  // namespace pineforge

tests/CMakeLists.txt

@@ -7164,3 +7164,9 @@ if(PINEFORGE_BUILD_SOURCE_LAYER AND UNIX)
     add_dependencies(test_report_sink_l6b1 report_sink_l6b1_strategy)
 endif()
 # --- CR-150 core-cr150-l6b1 end ---
+
+# --- CR-150 core-cr150-l17a begin ---
+# CR-150 L17A: a kernel-only native host and a real version-1 path sink.
+# The UNIX-only plain registration is in cr150.d/core-cr150-l17a.cmake.
+# No additional target or test settings are required here.
+# --- CR-150 core-cr150-l17a end ---

tests/cr150.d/core-cr150-l17a.cmake

@@ -0,0 +1,4 @@
+if(UNIX)
+    list(APPEND TEST_SOURCES test_report_sink_selected_l17a)
+    pineforge_add_unit_test(test_report_sink_selected_l17a)
+endif()