engine: preserve cohort admission ordering and capacity

ci/cr150-source-bounds.yaml

@@ -83,11 +83,13 @@ ceilings:
   # CR-559: one map Store resize group and one group of three error-string temporaries.
   # Both stay OPEN; an optional pair limit with default zero establishes no fixed bound.
   # Ten time-format string groups use per-call pattern and written-width storage.
-  # Forty-seven more unreviewed groups on the integrated tree (51 emitted by
+  # Forty-six more unreviewed groups on the integrated tree (50 emitted by
   # --emit-seed, 4 stale entries removed): the bounded drawings, the owned
   # storage and runtime context service, the ledger cache
   # and the selected-capacity driver, reader and core. None is bounded.
-  unreviewed: 1407
+  # One materialize_relative_exits local-list group and three
+  # SeriesPercentileNearestRank groups remain OPEN as well.
+  unreviewed: 1406
   pending-flat-memory: 184
 owners:
   - {name: "generated-cpp", globs: ["tutorial/**"]}
@@ -923,7 +925,6 @@ entries:
   - {id: "U0828", file: "src/source/pine_adapter.cpp", symbol: "close_logical_units_", ops: ["subscript"], container: "unresolved by the index: close_logical_units_[...]", owner: "source-layer", phase: "unclassified", state: "unreviewed", line: 9488}
   - {id: "U0830", file: "src/source/pine_adapter.cpp", symbol: "closes", ops: ["assign", "decl", "grow"], container: "std::vector<std::pair<native_order::RequestHandle, PlacementSnapshot>>", owner: "source-layer", phase: "unclassified", state: "unreviewed", line: 1885}
   - {id: "U0833", file: "src/source/pine_adapter.cpp", symbol: "cohort_sides", ops: ["decl", "subscript"], container: "std::unordered_map<const void*, CohortSides>", owner: "source-layer", phase: "unclassified", state: "unreviewed", line: 1551}
-  - {id: "U0834", file: "src/source/pine_adapter.cpp", symbol: "cohorts_by_id_", ops: ["grow", "subscript"], container: "unresolved by the index: cohorts_by_id_.emplace(...)", owner: "source-layer", phase: "unclassified", state: "unreviewed", line: 3020}
   - {id: "U0835", file: "src/source/pine_adapter.cpp", symbol: "consumed_partial_exit_cycles_", ops: ["grow", "subscript"], container: "unresolved by the index: consumed_partial_exit_cycles_[...]", owner: "source-layer", phase: "unclassified", state: "unreviewed", line: 24790}
   - {id: "U0836", file: "src/source/pine_adapter.cpp", symbol: "cross_side_by_entry", ops: ["decl", "grow"], container: "std::unordered_map<SourceId, bool>", owner: "source-layer", phase: "unclassified", state: "unreviewed", line: 1974}
   - {id: "U0837", file: "src/source/pine_adapter.cpp", symbol: "current_debited_applied_ordinals_", ops: ["grow"], container: "unresolved by the index: current_debited_applied_ordinals_.insert(...)", owner: "source-layer", phase: "unclassified", state: "unreviewed", line: 10535}

ci/cr150-source-reading.tsv

@@ -429,7 +429,6 @@ U0829	src/source/pine_adapter.cpp	closed_cohorts	std::vector<SourceId>	grow=1	lo
 U0830	src/source/pine_adapter.cpp	closes	std::vector<std::pair<native_order::RequestHandle, PlacementSnapshot>>	assign=6,decl=1,grow=1	mixed:file+local	NativeExecutionConsumer* native = nullptr;	adapter state	source-layer	stale 8 / new 8	SITES	READ:growth-op+assign-count/source		
 U0831	src/source/pine_adapter.cpp	cohort_ids	std::vector<SourceId>	decl=2,grow=4	local	}	adapter state	source-layer	stale 6 / new 6	SITES	READ:growth-op	DEFECT:G2	reader (second pass); first pass DEFECT; evidence pine_adapter.cpp:7511-7513,8719-8721; Per-call copies of every resident cohort id (two sites). Resident rows are ids with exposure or a pinned roster window (U1225); reserve() is no cap.
 U0833	src/source/pine_adapter.cpp	cohort_sides	std::unordered_map<const void*, CohortSides>	decl=1,subscript=1	mixed:file+member	|| snapshot.family == PineOrderFamily::ExitTrail)	adapter state	source-layer	stale 2 / new 2	SITES	READ:growth-op+decl-member/ns/file	DEFECT:G2	reader (second pass); first pass DEFECT; evidence pine_adapter.cpp:1705,1939-1952,28778-28780; One entry per resident cohort with >16 lifetime origins and a non-empty window, erased only when that window empties (28778-28780). Follows the resident rows.
-U0834	src/source/pine_adapter.cpp	cohorts_by_id_	unresolved by the index: cohorts_by_id_.emplace(...)	grow=1	local	// is a named divergence in docs/design/native-feature-parity.md §3.7:	adapter state	source-layer	stale 2 / new 1	SITES	READ:growth-op	REJECT:scratch index	Second pass; first pass DEFECT; evidence pine_adapter.cpp:3196-3209; disk_id_table.hpp:18-23,340-343; disk_resident_id_table.hpp:454-466; The per-id index node lives in scratch (DiskIdTable RAM = 5 scalars + a ScratchFile), and a fresh row is spillable at the next spill. Resident growth is U1225's.
 U0835	src/source/pine_adapter.cpp	consumed_partial_exit_cycles_	unresolved by the index: consumed_partial_exit_cycles_[...]	grow=1	local	// `id`) released onto its new book: bound to that request, bound to the id	adapter state	source-layer	stale 1 / new 1	SITES	READ:growth-op	REJECT:scratch RAM fixed	reader, re-read at source by reviewer (second pass); first pass DEFECT; evidence pine_adapter.cpp:25396-25400; id_pair_store.hpp:3,34-36,811-830; IdPairStore keeps no per-key RAM directory: records and index are scratch files, with RAM static_assert <= 32 KiB (:830). Disk grows with distinct keys; RAM does not.
 U0836	src/source/pine_adapter.cpp	cross_side_by_entry	std::unordered_map<SourceId, bool>	decl=1,grow=1	local	// A leg whose row was erased answers through its fact (`erased`).	adapter state	source-layer	stale 2 / new 2	SITES	READ:growth-op		
 U0837	src/source/pine_adapter.cpp	current_debited_applied_ordinals_	unresolved by the index: current_debited_applied_ordinals_.insert(...)	grow=1	local	// A partial source close breaks the exact ShortSeed transaction book.	adapter state	source-layer	stale 1 / new 1	SITES	READ:growth-op		

ci/cr150-source-sites.json

@@ -1127,6 +1127,30 @@
    "spelling": "at = . . .",
    "ordinal": 1
   },
+  {
+   "file": "include/pineforge/detail/disk_resident_id_table.hpp",
+   "symbol": "index_",
+   "scope": [
+    "namespace pineforge :: detail #1",
+    "template < class Value > class DiskResidentIdTable #1",
+    "std :: pair < iterator , bool > place ( const std :: string & key , Value value ) #1"
+   ],
+   "cls": "grow",
+   "spelling": "index_ . insert ( . . . )",
+   "ordinal": 1
+  },
+  {
+   "file": "include/pineforge/detail/disk_resident_id_table.hpp",
+   "symbol": "row",
+   "scope": [
+    "namespace pineforge :: detail #1",
+    "template < class Value > class DiskResidentIdTable #1",
+    "std :: pair < iterator , bool > place ( const std :: string & key , Value value ) #1"
+   ],
+   "cls": "alloc",
+   "spelling": "make_shared",
+   "ordinal": 1
+  },
   {
    "file": "include/pineforge/detail/disk_resident_id_table.hpp",
    "symbol": "index_",
@@ -31113,17 +31137,6 @@
    "spelling": "result = . . .",
    "ordinal": 1
   },
-  {
-   "file": "src/source/pine_adapter.cpp",
-   "symbol": "cohorts_by_id_",
-   "scope": [
-    ":: pineforge :: source :: detail :: PositionTailDiagnostics :: pineforge :: source :: detail :: PositionTailNoDiagnostics namespace pineforge :: source #1",
-    "native_order :: CohortHandle PineExecutionAdapter :: cohort_for ( const SourceId & id ) #1"
-   ],
-   "cls": "grow",
-   "spelling": "cohorts_by_id_ . emplace ( . . . )",
-   "ordinal": 1
-  },
   {
    "file": "src/source/pine_adapter.cpp",
    "symbol": "adopted_reissues",

docs/design/arena-bound.md

@@ -340,8 +340,8 @@ indexPeak(k) = 24 * slots(k)                       # k > 32
 | --- | --- | --- |
 | Equity writer / selected hold-back / SpillCurve | 24*E | `pf_equity_point_t` size assertion `include/pineforge/metrics_stream.hpp:16`; `src/report_scratch.cpp:238` |
 | Position tail | (30+L)*T | `src/source/pine_position_tail.hpp:20`; footer packing at `src/source/pine_position_tail.cpp:94` |
-| Cohort AVL and five-word summary | (96+L)*D + 1[D>0] | `Links` at `include/pineforge/detail/disk_id_table.hpp:165-182`, `end_` at `:348`; `CohortFacts::Record` at `include/pineforge/source/pine_adapter.hpp:2059-2062` |
-| Entry-rank AVL | (60+L)*J + 1[J>0] | `exit_rank_keys_` at `src/source/pine_adapter.cpp:6800-6801`; `Links` at `include/pineforge/detail/disk_id_table.hpp:165-182`, `end_` at `:348` |
+| Cohort AVL and five-word summary | (96+L)*D + 1[D>0] | `Links` at `include/pineforge/detail/disk_id_table.hpp:165-182`, `end_` at `:349`; `CohortFacts::Record` at `include/pineforge/source/pine_adapter.hpp:2059-2062` |
+| Entry-rank AVL | (60+L)*J + 1[J>0] | `exit_rank_keys_` at `src/source/pine_adapter.cpp:6800-6801`; `Links` at `include/pineforge/detail/disk_id_table.hpp:165-182`, `end_` at `:349` |
 | Erased minima data | (40+L)*D + (40+2*L)*Q | `erased_minima_` at `include/pineforge/source/pine_adapter.hpp:960`; 40-byte header at `include/pineforge/detail/id_pair_store.hpp:69` |
 | Erased minima index | indexPeak(D+Q) | 16-byte slot at `include/pineforge/detail/id_pair_store.hpp:69`; replacement at `:642` |
 | Consumed partial-cycle data | (41+2*L)*Q | `include/pineforge/source/pine_adapter.hpp:2838`; joined key at `src/source/pine_adapter.cpp:25506` |

include/pineforge/detail/disk_id_table.hpp

@@ -61,6 +61,7 @@ class DiskIdTable {
         return file_.scratch_owner() && file_.scratch_resource() == ScratchResource::cohort;
     }
     void set_directory(const std::string& directory) { file_.set_directory(directory); }
+    void open_for_insert() { file_.open("pineforge-ids"); }
     void bind_scratch(const std::shared_ptr<ScratchSlab>& slab, ScratchResource resource) {
         if (size_ != 0)
             throw coded<std::logic_error>(RunFailureCode::engine_invariant, {},

include/pineforge/detail/disk_resident_id_table.hpp

@@ -440,17 +440,30 @@ class DiskResidentIdTable {
         const auto row = resident_.find(at->second);
         return {this, row->first, row->second.owner};
     }
-    std::pair<iterator, bool> emplace(const std::string& key, Value value) {
+    void admit(const std::string& key, bool spillable) {
         // A refused new logical row must not first overwrite/evict an eligible
         // resident row. This search reads index bytes only, without restoring.
         index_.require_insert(key);
         std::size_t extra = 0;
-        if (index_.bounded_cohorts() && !value.spillable() && !index_.find(key)) {
+        if (index_.bounded_cohorts() && !spillable && !index_.find(key)) {
             extra = 1; // An existing logical row ignores the incoming value.
             require_live_growth(extra);
         }
         spill_before_admission(extra);
         spill_implicitly();
+        index_.open_for_insert();
+    }
+    std::pair<iterator, bool> place(const std::string& key, Value value) {
+        const auto placed = index_.insert(key, value.retired_record());
+        if (!placed.second)
+            throw coded<std::logic_error>(RunFailureCode::engine_invariant, {},
+                "resident id placement found duplicate after admission");
+        auto row = std::make_shared<Row>(key, std::move(value));
+        insert_resident(placed.first, row);
+        return {{this, placed.first, std::move(row)}, true};
+    }
+    std::pair<iterator, bool> emplace(const std::string& key, Value value) {
+        admit(key, value.spillable());
         const auto placed = index_.insert(key, value.retired_record());
         if (!placed.second) return {{this, placed.first, load(placed.first)}, false};
         auto row = std::make_shared<Row>(key, std::move(value));

scripts/test_cr150_source_gate.py

@@ -1216,12 +1216,12 @@ def test_real_tree_scans_files_and_sites_and_is_not_a_pass_by_construction(self)
         self.assertEqual(c['findings'], 0)
         self.assertEqual(c['PASS(bounded)'], 0)
         # Exact manifest union: 1,344 original unreviewed groups, ten time-format
-        # groups and two map groups, and 47 groups of the integrated tree (51
+        # groups and two map groups, and 46 groups of the integrated tree (50
         # emitted, 4 stale entries removed), one group of the local list of
         # materialize_relative_exits and three groups of
         # SeriesPercentileNearestRank (its ring, the growth of the ring and its
         # window); all 184 pending-flat-memory groups remain.
-        self.assertEqual(c['OPEN'], 1356 + 51 + 184)
+        self.assertEqual(c['OPEN'], 1356 + 50 + 184)
         # Every matched entry is exactly one of PASS(bounded) / OPEN.
         self.assertEqual(c['PASS(bounded)'] + c['OPEN'], c['entries_matched'])
         strict = run_gate(ROOT, '--strict')

src/source/pine_adapter.cpp

@@ -3217,16 +3217,16 @@ native_order::CohortHandle PineExecutionAdapter::cohort_for(const SourceId& id)
     auto found = cohorts_by_id_.find(id);
     if (found != cohorts_by_id_.end()) return found->second.handle;
     CohortFacts facts;
+    cohorts_by_id_.admit(id, facts.spillable());
     facts.handle = require_host().cohort_open();
     if (facts.handle.value == 0) {
         throw coded<std::logic_error>(RunFailureCode::engine_invariant, {},
                                       "native cohort allocation refused");
     }
     const auto result = facts.handle;
-    cohorts_by_id_.emplace(id, std::move(facts));
+    cohorts_by_id_.place(id, std::move(facts));
     return result;
 }
-
 PineSizingSnapshot PineExecutionAdapter::sizing_snapshot() const {
     PineSizingSnapshot snapshot;
     const auto& host = require_host();

tests/cr150.d/cr365-cohort-for-admission.cmake

@@ -0,0 +1,8 @@
+if(PINEFORGE_BUILD_SOURCE_LAYER)
+    list(APPEND TEST_SOURCES test_cohort_for_live_limit_new_id)
+    pineforge_add_unit_test(test_cohort_for_live_limit_new_id)
+    list(APPEND TEST_SOURCES test_cohort_for_existing_id_cadence)
+    pineforge_add_unit_test(test_cohort_for_existing_id_cadence)
+    list(APPEND TEST_SOURCES test_cohort_for_admission_io_refusal)
+    pineforge_add_unit_test(test_cohort_for_admission_io_refusal)
+endif()

tests/test_cohort_for_admission_io_refusal.cpp

@@ -0,0 +1,125 @@
+// CR-365 R2-A2 (extra to the brief): an admission refusal that the read-only
+// live-limit ticket cannot see must still precede the host cohort handle.
+//
+// cohort_for opens the table's backing file during admission. When that open is
+// refused (here: the scratch directory is a regular file) the new id must be
+// refused before require_host().cohort_open(), so no kernel cohort handle or
+// receipt is consumed. The witness is dynamic: it reads the kernel's next
+// handle before and after the refused call, and it first proves, through the
+// table's own emplace and without touching the kernel, that the injected fault
+// really refuses placement. Both facts are read from the real objects.
+#include <pineforge/source/pine_strategy_host.hpp>
+
+#include <cstddef>
+#include <cstdint>
+#include <cstdio>
+#include <stdexcept>
+#include <string>
+#include <system_error>
+#include <type_traits>
+#include <vector>
+
+namespace {
+using Adapter = pineforge::source::PineExecutionAdapter;
+
+// The same explicit-instantiation access seam as test_id_adapter_storage.cpp:
+// no changed class definition or production friend.
+template<class Tag, auto Member> struct Access {
+    friend auto access(Tag) { return Member; }
+};
+struct Cohorts { friend auto access(Cohorts); };
+struct CohortFor { friend auto access(CohortFor); };
+template struct Access<Cohorts, &Adapter::cohorts_by_id_>;
+template struct Access<CohortFor, &Adapter::cohort_for>;
+
+class IoRefusalHost final : public pineforge::source::PineStrategyHost {
+public:
+    IoRefusalHost() {
+        pineforge::source::PineStrategyConfig config;
+        config.initial_capital = 1e12;
+        config.margin_long = config.margin_short = 0.0;
+        config.pyramiding = 5000;
+        configure_pine_strategy(config);
+        set_syminfo_mintick(0.25);
+        set_syminfo_timezone("UTC");
+        set_syminfo_session("24x7");
+    }
+
+    void on_source_bar(const pineforge::Bar&) override {
+        if (pine_bar_index() == 0) probe();
+    }
+
+    bool probed = false;
+    std::string verdict;
+
+private:
+    void fail(const std::string& why) {
+        if (verdict.empty()) verdict = why;
+    }
+
+    void probe() {
+        probed = true;
+        auto& table = adapter_.*access(Cohorts{});
+        // A regular file is not a directory: a scratch child of it cannot be created.
+        table.set_directory("/dev/null/pineforge-cr365-r2-refused");
+        // Setup proof, independent of cohort_for and of the kernel: the table
+        // itself refuses to place a row, and stays empty.
+        using Facts = std::decay_t<decltype(table.find(std::string())->second)>;
+        bool setup_refused = false;
+        try {
+            table.emplace("cr365-setup-probe", Facts{});
+        } catch (const std::system_error&) {
+            setup_refused = true;
+        }
+        const auto logical = table.size();
+        const auto first = cohort_open();
+        const auto hash_before = broker_state_hash_projection();
+        bool refused = false;
+        std::string what;
+        try {
+            (void)(adapter_.*access(CohortFor{}))("cr365-admission-io-refused");
+        } catch (const std::system_error& error) {
+            refused = true;
+            what = error.what();
+        }
+        const auto hash_after = broker_state_hash_projection();
+        const auto second = cohort_open();
+        std::printf("R2-A2 setup_refused=%d logical=%llu first_handle=%llu second_handle=%llu "
+                    "refused=%d same_hash=%d text=\"%s\"\n",
+                    setup_refused ? 1 : 0, static_cast<unsigned long long>(logical),
+                    static_cast<unsigned long long>(first.value),
+                    static_cast<unsigned long long>(second.value), refused ? 1 : 0,
+                    hash_after == hash_before ? 1 : 0, what.c_str());
+        if (!setup_refused || logical != 0)
+            return fail("SETUP: the injected scratch fault did not refuse a table placement");
+        if (!refused)
+            return fail("BEHAVIOR: cohort_for accepted a new id although the table refuses placement");
+        if (table.size() != logical)
+            return fail("BEHAVIOR: the refused cohort_for left a source row behind");
+        if (second.value != first.value + 1)
+            return fail("BEHAVIOR: the refused cohort_for consumed a host cohort handle (next="
+                        + std::to_string(second.value) + " expected="
+                        + std::to_string(first.value + 1) + ")");
+        if (hash_after != hash_before)
+            return fail("BEHAVIOR: the refused cohort_for changed the broker state");
+    }
+};
+} // namespace
+
+int main() {
+    try {
+        std::vector<pineforge::Bar> bars(4);
+        for (std::size_t i = 0; i < bars.size(); ++i)
+            bars[i] = {100.0, 101.0, 99.0, 100.0, 1000.0,
+                       1736121600000LL + static_cast<std::int64_t>(i) * 60000};
+        IoRefusalHost host;
+        host.run(bars.data(), static_cast<int>(bars.size()), "1", "1");
+        if (!host.probed) throw std::runtime_error("SETUP: the probe bar was never reached");
+        if (!host.verdict.empty()) throw std::runtime_error(host.verdict);
+        std::puts("cohort_for admission io refusal: OK");
+        return 0;
+    } catch (const std::exception& error) {
+        std::fprintf(stderr, "FAIL: %s\n", error.what());
+        return 1;
+    }
+}