Census sample: the sampling rule as one .dag fold, checked against the pin at the pin's revision

src/v2/compiler/reference_conservation_census.dag

@@ -40,7 +40,11 @@ import v2.compiler.reference_conservation_admission {
   conservation_read_of
 }
 import v2.extdeps.languages.dag { dag_language_model }
-import v2.std.algebra { fold_list }
+import v2.std.algebra { fold_list, length }
+import v2.std.collection { empty_map, map_insert, map_lookup }
+import v2.std.optional { Absent, Present }
+import std.bytes { bytes_octets, utf8_encode_bytes }
+import gunbc.namespace_step0_subject_collector { Step0SubjectCollected, Step0SubjectCollectionRefused, collect_step0_subject_vector_at }
 import v2.std.integer { integer_int_to_decimal_string }
 import std.algebra { Cons, Empty, FreeMonoid, list_snoc_item }
 import v2.std.compilers.lexing { symbol_lexeme }
@@ -287,20 +291,25 @@ fn reference_conservation_changed_scope_observation_exit(paths_csv: String) -> P
 }
 
 
-// THE PINNED STRATIFIED SAMPLE (XL-2; first pinned at main fec339d561, REPINNED at main 8fcd8e77b8).
-// The whole-corpus census is about 55 CPU-hours on the interpreter, so the drop shapes are first
-// read off this sample. The selection is a rule, not a hand pick: every `.dag` file under dag/ and
-// src/v2/ is put in a stratum (its first two path segments under dag/, its first three under
-// src/v2/); each stratum takes k = max(6, round(300 * the stratum's share of ALL `.dag` files))
-// files -- at most as many as it has -- from its files of at most 8,000 bytes sorted by path, at
-// indices floor(i * n / k) for i in 0..k-1 over those n files. Applying this rule at fec339d561
-// reproduces the original 315-path list exactly, which is what fixed the one ambiguity in the
-// earlier wording (the share is over all files, not over the small ones). The 8,000-byte cap bounds
-// the run and biases the sample toward small files; a shape that only appears in large files is
-// not in it, and the full census is what closes that gap. The list is kept here so the sample
-// re-runs byte-for-byte. A listed path the tree no longer holds is reported SamplePathMissing (its
-// absence established by a listing, not inferred from a failed read), so a stale pin is visible in
-// the census rather than a smaller population; repinning re-applies the rule at a named revision.
+// THE PINNED STRATIFIED SAMPLE (XL-2; first pinned at main fec339d561, repinned at the revision
+// reference_conservation_stratified_sample_revision names). The whole-corpus census is about 55
+// CPU-hours on the interpreter, so the drop shapes are first read off this sample. The selection is
+// a rule, not a hand pick, and the rule has one authority: sample_paths_by_rule below. The list is
+// the rule's output at that revision, kept literally so the census re-runs byte-for-byte without
+// reading the object store. reference_conservation_sample_by_rule_at_pin_exit re-derives it at the
+// pinned revision and refuses if the two differ -- but ONLY WHEN THAT INSTRUMENT IS RUN (`gunbc run
+// --function reference_conservation_sample_by_rule_at_pin_exit`): it reads every `.dag` blob at the
+// revision out of the object store, which no hermetic floor claim can do, so no gate executes it.
+// The floor claims in v2.test.claim.namespace_xl0.reference_conservation_census pin the rule's
+// arithmetic on supplied inputs; the list-to-revision pairing is established by the instrument's run
+// recorded with the change that repins, and a repin must run it again. The rule is applied to the PINNED revision, never to the checked-out tree: the sample is
+// a snapshot, and comparing it to a tree that has since grown would only detect change. The
+// 8,000-byte cap bounds the run and biases the sample toward small files; a shape that only appears
+// in large files is not in it, and the full census is what closes that gap. A listed path the tree
+// no longer holds is reported SamplePathMissing (its absence established by a listing, not inferred
+// from a failed read), so a stale pin is visible in the census rather than a smaller population.
+data reference_conservation_stratified_sample_revision: String = "8fcd8e77b898f8188a281be07119fc5e97d09bef"
+
 data reference_conservation_stratified_sample_paths: List<String> = [
   "dag/examples/blackjack/cards.dag",
   "dag/examples/blackjack/hand.dag",
@@ -627,3 +636,181 @@ fn reference_conservation_stratified_sample_census() -> String {
     cons: fn(acc, path) { acc + reference_conservation_census_for_path(path: path) }
   )
 }
+
+// ---- the sampling rule, the one authority for the pinned list ------------------------------------
+
+// One `.dag` file the rule may select: its path, and whether it is at most the 8,000-byte cap.
+type SampleFile {
+  path: String
+  within_cap: Bool
+}
+
+data sample_rule_byte_cap: Int = 8000
+data sample_rule_target: Int = 300
+data sample_rule_stratum_floor: Int = 6
+
+// UTF-8 never takes fewer bytes than code points, so text of more code points than the cap is over
+// it without being encoded; only text at or under the cap in code points has its bytes counted.
+fn sample_file_within_cap(content: String) -> Bool {
+  if string_length(s: content) > sample_rule_byte_cap {
+    false
+  } else {
+    length(xs: bytes_octets(b: utf8_encode_bytes(s: content))) <= sample_rule_byte_cap
+  }
+}
+
+// A file's stratum: its first two path segments under dag/, its first three under src/v2/.
+fn sample_stratum(path: String) -> String {
+  let segments = split(s: path, delimiter: "/")
+  let depth = if starts_with(s: path, prefix: "dag/") { 2 } else { 3 }
+  fold_list(xs: segments, empty: SampleStratumAcc { taken: 0, text: "" }, cons: fn(acc, seg) {
+    if acc.taken >= depth {
+      acc
+    } else {
+      if acc.taken == 0 {
+        SampleStratumAcc { taken: 1, text: seg }
+      } else {
+        SampleStratumAcc { taken: acc.taken + 1, text: acc.text + "/" + seg }
+      }
+    }
+  }).text
+}
+
+type SampleStratumAcc {
+  taken: Int
+  text: String
+}
+
+// round(q / d) with ties to even, in integers (d > 0, q >= 0).
+fn sample_round_half_even(q: Int, d: Int) -> Int {
+  let f = q / d
+  let twice_rem = 2 * (q - f * d)
+  if twice_rem > d {
+    f + 1
+  } else {
+    if twice_rem == d {
+      if f % 2 == 1 { f + 1 } else { f }
+    } else {
+      f
+    }
+  }
+}
+
+// A stratum's quota: max(6, round(300 * share)), at most the files it can take from.
+fn sample_stratum_quota(stratum_count: Int, total: Int, small_count: Int) -> Int {
+  let rounded = sample_round_half_even(q: sample_rule_target * stratum_count, d: total)
+  let wanted = if rounded < sample_rule_stratum_floor { sample_rule_stratum_floor } else { rounded }
+  if wanted > small_count { small_count } else { wanted }
+}
+
+type SampleTakeAcc {
+  index: Int
+  next: Int
+  taken: FreeMonoid<String>
+}
+
+// The small files at indices floor(i * n / k), i in 0..k-1, in one ordered walk: the targets are
+// strictly increasing, so each element is compared with the next target only.
+fn sample_stratum_take(small: FreeMonoid<String>, quota: Int) -> FreeMonoid<String> {
+  let n = length(xs: small)
+  fold_list(xs: small, empty: SampleTakeAcc { index: 0, next: 0, taken: Empty }, cons: fn(acc, path) {
+    if acc.next < quota && acc.index == (acc.next * n) / quota {
+      SampleTakeAcc { index: acc.index + 1, next: acc.next + 1, taken: list_snoc_item(xs: acc.taken, item: path) }
+    } else {
+      SampleTakeAcc { index: acc.index + 1, next: acc.next, taken: acc.taken }
+    }
+  }).taken
+}
+
+type SampleRuleAcc {
+  stratum: String
+  count: Int
+  small: FreeMonoid<String>
+  selected: FreeMonoid<String>
+}
+
+fn sample_rule_flush(acc: SampleRuleAcc, total: Int) -> FreeMonoid<String> {
+  if acc.count == 0 {
+    acc.selected
+  } else {
+    fold_list(
+      xs: sample_stratum_take(small: acc.small, quota: sample_stratum_quota(stratum_count: acc.count, total: total, small_count: length(xs: acc.small))),
+      empty: acc.selected,
+      cons: fn(out, path) { list_snoc_item(xs: out, item: path) }
+    )
+  }
+}
+
+// The `.dag` suffix is read with the builtin substring: the bare name ends_with also names a corpus
+// function (gunbc.rust_item_scan), so the floor's import hygiene refuses it unimported here.
+fn sample_file_in_rule_roots(path: String) -> Bool {
+  let n = string_length(s: path)
+  n > 4 && substring(s: path, start: n - 4, end: n) == ".dag" && (starts_with(s: path, prefix: "dag/") || starts_with(s: path, prefix: "src/v2/"))
+}
+
+// THE RULE. Sorting by path makes every stratum one contiguous run -- a stratum's paths share a
+// prefix ending at a segment boundary, so no other path sorts between two of them -- and one pass
+// then groups them. The output is in path order.
+fn sample_paths_by_rule(files: FreeMonoid<SampleFile>) -> FreeMonoid<String> {
+  let in_roots = sort_by(filter(files, f => sample_file_in_rule_roots(path: f.path)), f => f.path)
+  let total = length(xs: in_roots)
+  let last = fold_list(xs: in_roots, empty: SampleRuleAcc { stratum: "", count: 0, small: Empty, selected: Empty }, cons: fn(acc, f) {
+    let st = sample_stratum(path: f.path)
+    let current = if acc.count > 0 && st == acc.stratum {
+      acc
+    } else {
+      SampleRuleAcc { stratum: st, count: 0, small: Empty, selected: sample_rule_flush(acc: acc, total: total) }
+    }
+    SampleRuleAcc {
+      stratum: current.stratum,
+      count: current.count + 1,
+      small: if f.within_cap { list_snoc_item(xs: current.small, item: f.path) } else { current.small },
+      selected: current.selected
+    }
+  })
+  sample_rule_flush(acc: last, total: total)
+}
+
+// THE PAIRING CHECK, run as an instrument (`gunbc run --function
+// reference_conservation_sample_by_rule_at_pin_exit`): read the tree at the pinned revision out of
+// the object store, apply the rule, and refuse unless the result is the pinned list exactly.
+fn reference_conservation_sample_by_rule_at(ref: String) -> SampleRuleAtRevision {
+  match collect_step0_subject_vector_at(ref: ref) {
+    Step0SubjectCollectionRefused { cause: _ } => SampleRuleRevisionUnreadable { ref: ref }
+    Step0SubjectCollected { vector: vector } =>
+      SampleRuleApplied {
+        selected: sample_paths_by_rule(files: fold(vector.members, init: Empty, f: fn(acc, m) {
+          list_snoc_item(xs: acc, item: SampleFile { path: m.path as String, within_cap: sample_file_within_cap(content: m.content) })
+        }))
+      }
+  }
+}
+
+type SampleRuleAtRevision
+  = SampleRuleApplied { selected: FreeMonoid<String> }
+  | SampleRuleRevisionUnreadable { ref: String }
+
+fn sample_paths_not_in(xs: FreeMonoid<String>, ys: FreeMonoid<String>) -> FreeMonoid<String> {
+  let present = fold_list(xs: ys, empty: empty_map(), cons: fn(m, y) { map_insert(m: m, key: y, value: true) })
+  fold_list(xs: xs, empty: Empty, cons: fn(acc, x) {
+    match map_lookup(m: present, key: x) {
+      Present { value: _ } => acc
+      Absent => list_snoc_item(xs: acc, item: x)
+    }
+  })
+}
+
+fn reference_conservation_sample_by_rule_at_pin_exit() -> ProcessExit {
+  match reference_conservation_sample_by_rule_at(ref: reference_conservation_stratified_sample_revision) {
+    SampleRuleRevisionUnreadable { ref: r } => exit_failure(reason: "the pinned revision " + r + " could not be read from the object store")
+    SampleRuleApplied { selected: derived } =>
+      let pinned = sort_by(reference_conservation_stratified_sample_paths, p => p)
+      let missing = sample_paths_not_in(xs: derived, ys: pinned)
+      let extra = sample_paths_not_in(xs: pinned, ys: derived)
+      if length(xs: missing) == 0 && length(xs: extra) == 0 && length(xs: derived) == length(xs: pinned) {
+        ExitSuccess
+      } else {
+        exit_failure(reason: "the rule at " + reference_conservation_stratified_sample_revision + " selects " + integer_int_to_decimal_string(value: length(xs: derived)) + " paths; the pin lists " + integer_int_to_decimal_string(value: length(xs: pinned)) + "; rule-only " + integer_int_to_decimal_string(value: length(xs: missing)) + fold_list(xs: missing, empty: "", cons: fn(a, p) { a + " " + p }) + "; pin-only " + integer_int_to_decimal_string(value: length(xs: extra)) + fold_list(xs: extra, empty: "", cons: fn(a, p) { a + " " + p }))
+      }
+  }
+}

src/v2/test/claim/namespace_xl0/reference_conservation_census_test.dag

@@ -2,6 +2,10 @@ module v2.test.claim.namespace_xl0.reference_conservation_census
 
 import extdeps.filesystem.filesystem_io { FilesystemReadRefused, filesystem_listing_observation }
 import v2.compiler.reference_conservation_census {
+  SampleFile,
+  sample_file_within_cap,
+  sample_paths_by_rule,
+  sample_stratum,
   SamplePathMissing,
   SamplePathSubjectRefused,
   SamplePathUnreadable,
@@ -11,6 +15,10 @@ import v2.compiler.reference_conservation_census {
 import v2.std.live_tree { LiveTreeDisposition, SubstrateInputsOnly }
 import v2.std.logic { Bool }
 import v2.std.text { String }
+import v2.std.algebra { fold_list, length }
+import v2.std.collection { List }
+import std.algebra { Empty, FreeMonoid, list_snoc_item }
+import v2.std.integer { Int }
 
 data live_tree_disposition: LiveTreeDisposition = SubstrateInputsOnly
 
@@ -80,3 +88,112 @@ test fn a_file_its_own_parent_does_not_list_is_missing_holds() -> Bool {
     _ => false
   }
 }
+
+// ---- the sampling rule's arithmetic, pinned on supplied file sets --------------------------------
+//
+// sample_paths_by_rule is the one authority for the pinned list; these controls pin each of its
+// parts on a small SUPPLIED file set whose answer is derived by hand, so a change to any part of the
+// rule reds here without reading a tree. Whether the pinned list IS the rule's output at its revision
+// is the instrument reference_conservation_sample_by_rule_at_pin_exit, run against the object store.
+
+data rule_digits: List<String> = ["0", "1", "2", "3", "4", "5", "6", "7", "8", "9"]
+
+type RuleNameAcc {
+  count: Int
+  names: FreeMonoid<String>
+}
+
+// The first n of "000".."999", in order (and in path order, being zero-padded).
+fn rule_names(n: Int) -> FreeMonoid<String> {
+  fold_list(xs: rule_digits, empty: RuleNameAcc { count: 0, names: Empty }, cons: fn(a1, d1) {
+    fold_list(xs: rule_digits, empty: a1, cons: fn(a2, d2) {
+      fold_list(xs: rule_digits, empty: a2, cons: fn(a3, d3) {
+        if a3.count < n { RuleNameAcc { count: a3.count + 1, names: list_snoc_item(xs: a3.names, item: d1 + d2 + d3) } } else { a3 }
+      })
+    })
+  }).names
+}
+
+fn rule_files(dir: String, n: Int, within_cap: Bool) -> FreeMonoid<SampleFile> {
+  fold_list(xs: rule_names(n: n), empty: Empty, cons: fn(acc, nm) {
+    list_snoc_item(xs: acc, item: SampleFile { path: dir + "/f" + nm + ".dag", within_cap: within_cap })
+  })
+}
+
+fn rule_join(a: FreeMonoid<SampleFile>, b: FreeMonoid<SampleFile>) -> FreeMonoid<SampleFile> {
+  fold_list(xs: b, empty: a, cons: fn(acc, f) { list_snoc_item(xs: acc, item: f) })
+}
+
+fn rule_selected_under(files: FreeMonoid<SampleFile>, dir: String) -> FreeMonoid<String> {
+  filter(sample_paths_by_rule(files: files), p => starts_with(s: p, prefix: dir + "/"))
+}
+
+// THE THREE LARGE FILE SETS ARE NULLARY AND ENROLLED WARM in v2.workflow.floor_pure_producer_share:
+// building them is not what the claims are about, so each claim pays only for the rule it runs.
+fn quota_fixture_files() -> FreeMonoid<SampleFile> {
+  rule_join(a: rule_files(dir: "dag/a", n: 10, within_cap: true), b: rule_files(dir: "dag/b", n: 490, within_cap: false))
+}
+
+fn under_floor_fixture_files() -> FreeMonoid<SampleFile> {
+  rule_join(a: rule_files(dir: "dag/a", n: 10, within_cap: true), b: rule_files(dir: "dag/b", n: 600, within_cap: false))
+}
+
+fn half_share_fixture_files() -> FreeMonoid<SampleFile> {
+  rule_join(a: rule_files(dir: "dag/a", n: 13, within_cap: true), b: rule_files(dir: "dag/b", n: 299, within_cap: false))
+}
+
+// THE floor(i*n/k) INDICES. dag/a holds 10 small files among 500 in all (the filler is 490 over-cap
+// files in dag/b), so its quota is 300 * 10 / 500 = 6, and floor(i * 10 / 6) for i in 0..5 is 0, 1,
+// 3, 5, 6, 8.
+test fn the_rule_takes_its_quota_at_the_floor_indices_holds() -> Bool {
+  rule_selected_under(files: quota_fixture_files(), dir: "dag/a") == ["dag/a/f000.dag", "dag/a/f001.dag", "dag/a/f003.dag", "dag/a/f005.dag", "dag/a/f006.dag", "dag/a/f008.dag"]
+}
+
+// A SHARE UNDER THE FLOOR IS RAISED TO 6: dag/a's 10 small files among 610 give round(4.92) = 5.
+test fn a_share_under_the_floor_takes_six_holds() -> Bool {
+  length(xs: rule_selected_under(files: under_floor_fixture_files(), dir: "dag/a")) == 6
+}
+
+// TIES ROUND TO EVEN: dag/a's 13 small files among 312 give 300 * 13 / 312 = 12.5, which rounds to 12
+// (half-up would take all 13).
+test fn a_half_share_rounds_to_even_holds() -> Bool {
+  length(xs: rule_selected_under(files: half_share_fixture_files(), dir: "dag/a")) == 12
+}
+
+// A STRATUM WITH FEWER SMALL FILES THAN ITS QUOTA TAKES THEM ALL: 3 small, quota 6.
+test fn a_small_stratum_takes_every_small_file_holds() -> Bool {
+  let files = rule_join(a: rule_files(dir: "dag/a", n: 3, within_cap: true), b: rule_files(dir: "dag/a", n: 0, within_cap: true))
+  rule_selected_under(files: files, dir: "dag/a") == ["dag/a/f000.dag", "dag/a/f001.dag", "dag/a/f002.dag"]
+}
+
+// ONLY dag/ AND src/v2/ ARE IN THE RULE'S ROOTS: a src/v1 file is never selected.
+test fn a_file_outside_the_rule_roots_is_never_selected_holds() -> Bool {
+  length(xs: sample_paths_by_rule(files: rule_files(dir: "src/v1/compiler", n: 3, within_cap: true))) == 0
+}
+
+test fn strata_are_two_segments_under_dag_and_three_under_src_v2_holds() -> Bool {
+  sample_stratum(path: "dag/std/access.dag") == "dag/std"
+    && sample_stratum(path: "dag/gunbc/recurring_failure_mode/x.dag") == "dag/gunbc"
+    && sample_stratum(path: "src/v2/compiler/self_host/a.dag") == "src/v2/compiler"
+}
+
+// THE CAP IS IN BYTES: code points bound bytes from below, so the boundary is exercised with a
+// two-byte character that makes 8,000 code points into 8,001 bytes.
+fn rule_repeat(unit: String, n: Int) -> String {
+  fold_list(xs: rule_names(n: n), empty: "", cons: fn(acc, _) { acc + unit })
+}
+
+fn rule_ascii(n100: Int) -> String {
+  rule_repeat(unit: rule_repeat(unit: "x", n: 100), n: n100)
+}
+
+test fn the_cap_admits_exactly_eight_thousand_bytes_holds() -> Bool {
+  sample_file_within_cap(content: rule_ascii(n100: 80))
+    && !sample_file_within_cap(content: rule_ascii(n100: 80) + "x")
+}
+
+test fn the_cap_counts_bytes_not_code_points_holds() -> Bool {
+  let short_by_two = substring(s: rule_ascii(n100: 80), start: 0, end: 7998)
+  !sample_file_within_cap(content: short_by_two + "x" + "\u{e9}")
+    && sample_file_within_cap(content: short_by_two + "\u{e9}")
+}

src/v2/workflow/floor_pure_producer_share.dag

@@ -907,6 +907,9 @@ data floor_cross_claim_pure_producers_warm: List<String> = [
   "v2.test.claim.namespace_xl0.reference_conservation.construct_tags_subject",
   "v2.test.claim.namespace_xl0.reference_conservation.dotted_spine_subject",
   "v2.test.claim.namespace_xl0.reference_conservation.normalization_refused_subject",
+  "v2.test.claim.namespace_xl0.reference_conservation_census.quota_fixture_files",
+  "v2.test.claim.namespace_xl0.reference_conservation_census.under_floor_fixture_files",
+  "v2.test.claim.namespace_xl0.reference_conservation_census.half_share_fixture_files",
   "v2.test.claim.occurrence_role.occurrence_role.roles_fixture_outcome",
   "v2.test.claim.namespace_xl0.reference_conservation.statement_let_binder_subject",
   "v2.test.claim.namespace_xl0.reference_conservation_accepted_drops.binary_operand_beside_match_subject",