diff --git a/docs/design/cr150-contracts.md b/docs/design/cr150-contracts.md index 04dabe7c6..f8548d316 100644 --- a/docs/design/cr150-contracts.md +++ b/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 | diff --git a/src/native_execution_consumer.cpp b/src/native_execution_consumer.cpp index d85449cc8..d81248121 100644 --- a/src/native_execution_consumer.cpp +++ b/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 #include #include @@ -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(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_) {} diff --git a/src/native_execution_consumer.hpp b/src/native_execution_consumer.hpp index aff6ab7f3..f70a30c62 100644 --- a/src/native_execution_consumer.hpp +++ b/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; diff --git a/src/selected_report_view.cpp b/src/selected_report_view.cpp index 9dbff9747..49a80375a 100644 --- a/src/selected_report_view.cpp +++ b/src/selected_report_view.cpp @@ -62,3 +62,71 @@ SelectedEquityCurve build_selected_equity_curve( } // namespace report } // namespace pineforge + +#include + +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::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::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 diff --git a/src/selected_report_view.hpp b/src/selected_report_view.hpp index 75cd4f4bd..569c1a471 100644 --- a/src/selected_report_view.hpp +++ b/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(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 diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 12a3679c3..725e7dd18 100644 --- a/tests/CMakeLists.txt +++ b/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 --- diff --git a/tests/cr150.d/core-cr150-l17a.cmake b/tests/cr150.d/core-cr150-l17a.cmake new file mode 100644 index 000000000..b5396d240 --- /dev/null +++ b/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() diff --git a/tests/test_report_sink_selected_l17a.cpp b/tests/test_report_sink_selected_l17a.cpp new file mode 100644 index 000000000..595fbb69e --- /dev/null +++ b/tests/test_report_sink_selected_l17a.cpp @@ -0,0 +1,485 @@ +// CR-150 L17A: selected capture emits the anchor and final window E values, +// and checks the selected invariants before a sink can acquire a trailer. +// Uses the existing L6b-1 ABI, so the first red is a runtime feature failure: +// on f00151f22 count_e is still zero after capture, until fill_report runs. +#include +#include +#include +#include +#include +#include +#include + +#include "../src/selected_report_view.hpp" +#include "../src/sha256.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace pineforge { +inline namespace engine_script_run_v19 { +int native_set_selected_window_v1(BacktestEngine*, const pf_selected_window_config_v1*) noexcept; +int native_selected_window_counts_v1(const BacktestEngine*, pf_selected_window_counts_v1*) noexcept; +int native_set_execution_observer_v1(BacktestEngine*, const pf_execution_observer_v1*) noexcept; +} +} + +namespace { +using namespace pineforge; +namespace no = pineforge::native_order; + +int checks = 0; +int failures = 0; +#define CHECK(condition) do { ++checks; if (!(condition)) { ++failures; \ + std::fprintf(stderr, "FAIL %s:%d: %s\n", __FILE__, __LINE__, #condition); } } while (0) + +constexpr std::int64_t kBase = 1704067200000LL; +constexpr std::int64_t kMinute = 60000; +constexpr double kCapital = 1000.0; + +struct Directory { + std::string path; + std::vector files; + Directory() { + char pattern[] = "/tmp/pineforge-l17a-XXXXXX"; + const char* created = mkdtemp(pattern); + if (!created) throw std::runtime_error("mkdtemp failed"); + path = created; + } + ~Directory() { + for (const auto& file : files) std::remove(file.c_str()); + rmdir(path.c_str()); + } + std::string file(const char* name) { + files.push_back(path + '/' + name); + return files.back(); + } +}; + +struct Report { + ReportC value{}; + ~Report() { BacktestEngine::free_report(&value); } +}; + +// KernelRecorded exercises actual orders, including an open range-end row. +// HostRecorded supplies deliberately inconsistent observations to exercise +// the capture's refusal precedence, without modifying the private consumer. +struct Host final : NativeStrategyHost { + std::int64_t enter_at = -1; + std::int64_t close_at = -1; + bool entered = false; + bool close_submitted = false; + bool close_accepted = false; + bool record_own = false; + bool rewrite_first = false; + double rewritten_equity = kCapital; + bool change_presented_tail = false; + bool shrink_presented_curve = false; + bool throw_in_present = false; + std::vector observations; + std::size_t next = 0; + mutable int presentations = 0; + mutable std::uint64_t broker_hash_at_capture = 0; + mutable bool broker_hash_captured = false; + + void on_native_run_begin() override { broker_hash_captured = false; } + void on_native_bar(const Bar&, const NativeDecisionContext& context) override { + if (enter_at >= 0 && !entered && context.script_bar_open_ms >= enter_at) { + entered = true; + (void)submit({no::Transact{1.0}, "entry", ""}); + } + if (close_at >= 0 && !close_submitted && context.script_bar_open_ms >= close_at) { + close_submitted = true; + const auto result = submit({no::Flatten{}, "exit", ""}); + close_accepted = result.status == no::SubmitStatus::Accepted && result.handle.has_value(); + } + if (record_own && next < observations.size()) { + equity_curve_.push_back(observations[next++]); + // A last-point rewrite is judged at capture, after the edit. + if (rewrite_first && next == observations.size()) + equity_curve_.front().equity = rewritten_equity; + } + } + void present_report(ReportC* report) const override { + ++presentations; + if (throw_in_present) throw std::runtime_error("L17A presentation refused"); + if (change_presented_tail && report->equity_curve_len > 1) { + report->equity_curve[report->equity_curve_len - 1].equity = 1023.5; + report->equity_curve[report->equity_curve_len - 1].open_profit = -0.0; + } + if (shrink_presented_curve && report->equity_curve_len > 1) + report->equity_curve_len = 1; + // Capture precedes the selected seal. Query the terminal broker state + // here; after run() returns the public hash query must refuse it. + broker_hash_at_capture = broker_state_hash(); + broker_hash_captured = true; + } +}; + +NativeRunSpec spec(bool own = false, std::uint64_t generation = 1) { + NativeRunSpec result; + result.identity = {"cr150-l17a", generation}; + result.input_tf = "1"; + result.script_tf = "1"; + result.ticker = "MOCK"; + result.tickerid = "TEST:MOCK"; + result.type = "crypto"; + result.currency = "USD"; + result.timezone = "UTC"; + result.session = "24x7"; + result.initial_capital = kCapital; + result.point_value = 1.0; + result.account_fx = 1.0; + result.price_tick = 0.01; + result.report_policy = own ? NativeReportPolicy::HostRecorded + : NativeReportPolicy::KernelRecorded; + return result; +} + +std::vector tape() { + return {{50.0, 50.0, 50.0, 50.0, 1.0, kBase}, + {50.0, 50.0, 50.0, 50.0, 1.0, kBase + kMinute}, + {100.0, 100.0, 100.0, 100.0, 1.0, kBase + 2 * kMinute}, + {100.0, 110.0, 100.0, 110.0, 1.0, kBase + 3 * kMinute}, + {110.0, 110.0, 104.5, 104.5, 1.0, kBase + 4 * kMinute}}; +} + +pf_selected_window_config_v1 window(int start, int end = 10) { + return {sizeof(pf_selected_window_config_v1), 1, + kBase + start * kMinute, kBase + end * kMinute}; +} + +void install(Host& host, const std::string& path) { + const pf_report_sink_v1 sink{sizeof sink, 1, path.c_str(), "L17A", nullptr}; + CHECK(host.set_report_sink_v1(&sink) == 0); +} + +void run(Host& host, const std::vector& bars, + const pf_selected_window_config_v1& selected, std::uint64_t generation = 1) { + CHECK(host.configure_native(spec(host.record_own, generation)).status == NativeSetupStatus::Applied); + CHECK(native_set_selected_window_v1(&host, &selected) == 0); + host.run(bars.data(), static_cast(bars.size())); +} + +pf_report_sink_receipt_v1 receipt(const Host& host) { + pf_report_sink_receipt_v1 result{}; + result.struct_size = sizeof result; + result.version = 1; + CHECK(host.report_sink_receipt_v1(&result) == 0); + return result; +} + +std::string read(const std::string& path) { + std::ifstream input(path, std::ios::binary); + return {std::istreambuf_iterator(input), std::istreambuf_iterator()}; +} + +std::vector equity_lines(const std::string& bytes) { + std::vector result; + for (std::size_t at = 0; at < bytes.size();) { + const auto end = bytes.find('\n', at); + CHECK(end != std::string::npos); + if (end == std::string::npos) break; + if (bytes.compare(at, 5, "[\"E\",") == 0) result.push_back(bytes.substr(at, end - at)); + at = end + 1; + } + return result; +} + +std::string equity_line(const pf_equity_point_t& point) { + return "[\"E\"," + std::to_string(point.time_ms) + ',' + + report_sink::f64(point.equity) + ',' + report_sink::f64(point.open_profit) + ']'; +} + +void check_hash_after_seal_is_refused(const Host& host) { + bool refused = false; + try { + (void)host.broker_state_hash(); + } catch (const SelectedWindowQueryAfterSeal& error) { + refused = true; + CHECK(std::strcmp(error.what(), "selected_window_query_after_seal") == 0); + } + CHECK(refused); +} + +void compare_selected(Directory& directory, const char* name, int start, bool trade, + bool edit = false, bool shrink = false) { + std::printf("selected fixture: %s\n", name); + Host baseline, streamed; + baseline.enter_at = streamed.enter_at = trade ? kBase + 2 * kMinute : -1; + baseline.change_presented_tail = streamed.change_presented_tail = edit; + baseline.shrink_presented_curve = streamed.shrink_presented_curve = shrink; + const auto path = directory.file(name); + install(streamed, path); + const auto selected = window(start); + auto bars = tape(); + if (start == 2 && !trade && !edit && !shrink) bars.erase(bars.begin() + 2); // first window point after T + run(baseline, bars, selected); + run(streamed, bars, selected); + CHECK(baseline.native_state().kind == NativeLifecycleKind::Completed); + CHECK(streamed.native_state().kind == NativeLifecycleKind::Completed); + Report before, after; + baseline.fill_report(&before.value); + // FIRST RED on f00151f22: capture has not written any E record yet. + CHECK(receipt(streamed).count_e == before.value.equity_curve_len); + CHECK(receipt(streamed).complete == 0); // observer success still precedes final fill + CHECK(streamed.presentations == 1); + streamed.fill_report(&after.value); + CHECK(after.value.equity_curve == nullptr && after.value.equity_curve_len == 0); + CHECK(after.value.trades == nullptr && after.value.trades_len == 0); + CHECK(before.value.total_trades == after.value.total_trades); + CHECK(before.value.net_profit == after.value.net_profit); + CHECK(std::memcmp(&before.value.metrics, &after.value.metrics, sizeof(pf_metrics_t)) == 0); + CHECK(baseline.broker_hash_captured && streamed.broker_hash_captured); + CHECK(baseline.broker_hash_at_capture == streamed.broker_hash_at_capture); + check_hash_after_seal_is_refused(baseline); + check_hash_after_seal_is_refused(streamed); + const auto bytes = read(path); + const auto lines = equity_lines(bytes); + CHECK(lines.size() == static_cast(before.value.equity_curve_len)); + for (std::size_t i = 0; i < lines.size() + && i < static_cast(before.value.equity_curve_len); ++i) + CHECK(lines[i] == equity_line(before.value.equity_curve[i])); + CHECK(!lines.empty()); + if (!lines.empty()) CHECK(lines.front() == equity_line({selected.start_ms, kCapital, 0.0})); + if (start == 5) CHECK(lines.size() == 1); // M = 0: anchor only + const auto got = receipt(streamed); + CHECK(got.complete == 1); + CHECK(got.count_e == before.value.equity_curve_len); + CHECK(got.count_t == before.value.trades_len); + const auto trailer = bytes.find("{\"trailer\":"); + CHECK(trailer != std::string::npos); + if (trailer != std::string::npos) { + report_sink::Sha256 hash; + hash.update(std::string_view(bytes).substr(0, trailer)); + CHECK(hash.hex_digest() == got.sha256_hex); + } + Report again; + streamed.fill_report(&again.value); + CHECK(read(path) == bytes); + CHECK(receipt(streamed).count_e == got.count_e); + CHECK(streamed.presentations == 1); + CHECK(std::memcmp(&after.value.metrics, &again.value.metrics, sizeof(pf_metrics_t)) == 0); +} + +void refused(Directory& directory, const char* name, Host& host, int start, + const char* message, int end = 10) { + const auto path = directory.file(name); + install(host, path); + run(host, tape(), window(start, end)); + CHECK(host.native_state().kind == NativeLifecycleKind::Failed); + CHECK(host.last_error().find(message) != std::string::npos); + CHECK(receipt(host).complete == 0); + CHECK(read(path).find("{\"trailer\":") == std::string::npos); + Report report; + host.fill_report(&report.value); + CHECK(report.value.equity_curve == nullptr && report.value.trades == nullptr); + CHECK(receipt(host).complete == 0); + CHECK(read(path).find("{\"trailer\":") == std::string::npos); +} + +void latch_failures_and_final_values(Directory& directory) { + Host account; + account.record_own = true; + account.observations = {{kBase, kCapital, 0.0}, {kBase + kMinute, kCapital, 0.0}, + {kBase + 2 * kMinute, kCapital, 0.0}, + {kBase + 3 * kMinute, kCapital, 0.0}, + {kBase + 4 * kMinute, kCapital, 0.0}}; + account.rewrite_first = true; + account.rewritten_equity = kCapital + 1.0; + refused(directory, "account.ndjson", account, 2, "the account changed before the window start"); + CHECK(account.presentations == 0); + + Host repaired; + repaired.record_own = true; + repaired.observations = {{kBase, kCapital + 1.0, 0.0}, {kBase + kMinute, kCapital, -0.0}, + {kBase + 2 * kMinute, kCapital, 0.0}, + {kBase + 3 * kMinute, kCapital, 0.0}, + {kBase + 4 * kMinute, kCapital, 0.0}}; + repaired.rewrite_first = true; + const auto repair_path = directory.file("repaired.ndjson"); + install(repaired, repair_path); + run(repaired, tape(), window(2)); + CHECK(repaired.native_state().kind == NativeLifecycleKind::Completed); + CHECK(receipt(repaired).count_e == 4); + Report repaired_report; + repaired.fill_report(&repaired_report.value); + CHECK(receipt(repaired).complete == 1); + + Host contiguity; + contiguity.record_own = true; + contiguity.observations = {{kBase, kCapital, 0.0}, + {kBase + 2 * kMinute, kCapital, 0.0}, + {kBase + 10 * kMinute, kCapital, 0.0}, + {kBase + 3 * kMinute, kCapital, 0.0}, + {kBase + 4 * kMinute, kCapital, 0.0}}; + refused(directory, "contiguity.ndjson", contiguity, 2, "observations are not one run"); + + Host count; + count.record_own = true; + refused(directory, "count.ndjson", count, 2, "observations differ from its script bars"); + + // HostRecorded synthesizes no range-end row. Close a real trade before T + // so its carry-in violation precedes the malformed account/count facts. + Host carry; + carry.enter_at = kBase; + carry.close_at = kBase + kMinute; + carry.record_own = true; + carry.observations = {{kBase, kCapital + 1.0, 0.0}}; + refused(directory, "trade-first.ndjson", carry, 3, "a trade was entered before the window start"); + CHECK(carry.close_accepted); + CHECK(carry.presentations == 0); + + Host presentation; + presentation.throw_in_present = true; + refused(directory, "presentation.ndjson", presentation, 2, "L17A presentation refused"); + CHECK(presentation.presentations == 1); +} + +int reject_observer(void*, const pf_execution_boundary_v1*, pf_boundary_receipt_v1*) { return 1; } + +void failed_observer_and_reuse(Directory& directory) { + Host host; + const auto path = directory.file("reuse.ndjson"); + install(host, path); + const pf_execution_observer_v1 observer{sizeof observer, 1, nullptr, reject_observer}; + CHECK(host.configure_native(spec()).status == NativeSetupStatus::Applied); + CHECK(native_set_execution_observer_v1(&host, &observer) == 0); + CHECK(native_set_selected_window_v1(&host, nullptr) == 0); + // Admit with a selected window after the observer, as the host harness does. + const auto first = window(2); + CHECK(native_set_selected_window_v1(&host, &first) == 0); + const auto bars = tape(); + host.run(bars.data(), static_cast(bars.size())); + CHECK(host.native_state().kind == NativeLifecycleKind::Failed); + CHECK(receipt(host).complete == 0); + CHECK(read(path).find("{\"trailer\":") == std::string::npos); + CHECK(native_set_execution_observer_v1(&host, nullptr) == 0); + // An observer fault is a permanent failure, not a recoverable abort. + // Clearing the pending observer cannot reopen that host's lifecycle. + const auto refused_setup = host.configure_native(spec(false, 2)); + CHECK(refused_setup.status == NativeSetupStatus::Failed); + CHECK(refused_setup.validation.error == NativeRunSpecError::WrongPhase); + CHECK(refused_setup.validation.field == NativeRunSpecField::None); + CHECK(host.native_state().kind == NativeLifecycleKind::Failed); + Report failed_report; + host.fill_report(&failed_report.value); + CHECK(failed_report.value.trades == nullptr && failed_report.value.equity_curve == nullptr); + CHECK(receipt(host).complete == 0); + CHECK(read(path).find("{\"trailer\":") == std::string::npos); + + // Reuse is legal after Completed, with the same session and a higher run + // number. First populate a nonempty selected sink, then prove that the + // same host resets it to the new T's anchor-only (M = 0) curve. + Host reusable; + const auto reuse_path = directory.file("reuse-completed.ndjson"); + install(reusable, reuse_path); + run(reusable, bars, first); + CHECK(reusable.native_state().kind == NativeLifecycleKind::Completed); + CHECK(receipt(reusable).count_e == 4); + Report before_reuse; + reusable.fill_report(&before_reuse.value); + CHECK(receipt(reusable).complete == 1 && receipt(reusable).count_e == 4); + CHECK(equity_lines(read(reuse_path)).size() == 4); + run(reusable, bars, window(5), 2); + CHECK(reusable.native_state().kind == NativeLifecycleKind::Completed); + CHECK(receipt(reusable).count_e == 1); + Report report; + reusable.fill_report(&report.value); + CHECK(receipt(reusable).complete == 1 && receipt(reusable).count_e == 1); + CHECK(equity_lines(read(reuse_path)).size() == 1); + CHECK(equity_lines(read(reuse_path)) == (std::vector{ + equity_line({kBase + 5 * kMinute, kCapital, 0.0})})); +} + +// These rows hold the new O(1) latch boundary itself to the contract; they +// cannot pass merely because the old array capture did its checks first. +// Committed separately before the implementation: on the L6b-1 header this +// half is a compile-red for the missing report::SelectedReportStream. +void direct_latches(Directory& directory) { + const auto selected = window(2, 5); + const char* causes[] = { + "a trade was entered before the window start", + "a position was carried in before the window start", + "the account changed before the window start", + "the window's equity observations are not one run", + "the window's equity observations differ from its script bars"}; + for (int first = 0; first < 5; ++first) { + const std::string name = "latch-" + std::to_string(first) + ".ndjson"; + const auto path = directory.file(name.c_str()); + { + const pf_report_sink_v1 sink{sizeof sink, 1, path.c_str(), "latches", nullptr}; + report_sink::LineWriter writer(sink); + writer.write_header({false, false, false, 0, 0}); + report::SelectedReportStream stream(selected.start_ms, selected.end_ms, kCapital); + stream.write_point(writer, {kBase, first <= 2 ? kCapital + 1.0 : kCapital, -0.0}, true); + stream.write_point(writer, {selected.start_ms, kCapital, 0.0}, true); + if (first <= 3) stream.write_point(writer, {selected.end_ms, kCapital, 0.0}, true); + stream.write_point(writer, {selected.start_ms + kMinute, kCapital + 1.0, 1.0}, true); + // Higher-priority evidence can arrive after the account/point + // latches. Capture order decides the refusal, not arrival order. + if (first <= 1) stream.note_open_lot(kBase); + if (first == 0) stream.note_trade(kBase); + bool caught = false; + try { + stream.finish(writer, 3, true); // the count mismatch is always present + } catch (const std::logic_error& error) { + caught = true; + CHECK(classify_run_failure(error).code == RunFailureCode::engine_invariant); + CHECK(std::string(error.what()).find(causes[first]) != std::string::npos); + } + CHECK(caught); + } + CHECK(read(path).find("{\"trailer\":") == std::string::npos); + } + const auto path = directory.file("latch-values.ndjson"); + { + const pf_report_sink_v1 sink{sizeof sink, 1, path.c_str(), "values", nullptr}; + report_sink::LineWriter writer(sink); + writer.write_header({false, false, false, 0, 0}); + report::SelectedReportStream stream(selected.start_ms, selected.end_ms, kCapital); + stream.note_trade(selected.start_ms); + stream.note_open_lot(selected.start_ms); + stream.write_point(writer, {kBase, kCapital, -0.0}, true); + stream.write_point(writer, {selected.start_ms, 1010.0, -0.0}, true); + stream.write_point(writer, {selected.end_ms - kMinute, 1004.5, 4.5}, true); + stream.write_point(writer, {selected.end_ms, 2000.0, 1000.0}, true); + stream.finish(writer, 2, true); + CHECK(stream.observations() == 2); + CHECK(stream.equity_count() == 3); + CHECK(writer.write_trailer().count_e == 3); + } + const auto lines = equity_lines(read(path)); + CHECK(lines == (std::vector{ + equity_line({selected.start_ms, kCapital, 0.0}), + equity_line({selected.start_ms, 1010.0, -0.0}), + equity_line({selected.end_ms - kMinute, 1004.5, 4.5})})); +} +} + +int main() { + try { + Directory directory; + compare_selected(directory, "flat.ndjson", 2, false); + compare_selected(directory, "open.ndjson", 2, true); + compare_selected(directory, "empty.ndjson", 5, false); + compare_selected(directory, "presented.ndjson", 2, true, true); + compare_selected(directory, "shrunk.ndjson", 2, true, false, true); + latch_failures_and_final_values(directory); + failed_observer_and_reuse(directory); + direct_latches(directory); + } catch (const std::exception& error) { + ++failures; + std::fprintf(stderr, "L17A exception: %s\n", error.what()); + } + std::printf("test_report_sink_selected_l17a: %d checks, %d failures\n", checks, failures); + return failures ? 1 : 0; +}