code wiki / _hdl_build / nx_store_put.nx
`nx_store_put.nx` source

↩ module page · 813 lines · 44812 B
1// nx_store_put.nx -- THE UNIFIED INFORMATION-PLANE WRITE VERB (operator 2026-07-18: "it shouldnt
2// be on the tsv we are supposed to use a unified nishi information management plane system thats
3// better than git ... not just debt it should be new feature capable").
4// Row-level, provenanced, additive writes DIRECTLY against any 7-col seg-store plane
5// (id title sev status owner scope note) -- debt-, work- (features), or any sibling prefix.
6// Flat staging files demote to bootstrap/recovery; the LIVE write path is this verb.
7// BETTER-THAN-GIT properties, in-plane: every mutation appends a revision to <prefix minus '-'>hist-
8// as  epoch<TAB>actor<TAB>verb<TAB>id<TAB>old-row<TAB>new-row  (inner tabs -> '|'), so history +
9// provenance live in the SOVEREIGN plane itself -- no working tree, no index, no merge dance;
10// row-independent writes cannot conflict, and the full before/after of every row is queryable.
11//   nx_store_put <prefix> put    <actor> <id> <field>...            (N-col: id + 1..15 fields, ANY plane schema
12//                                                                    incl the 9-col frontier -- v2 uplevel 07-18,
13//                                                                    operator: "stop engaging with the tsv")
14//   nx_store_put <prefix> setcol <actor> <id> <colidx> <value>       (flip ONE cell, e.g. frontier status col5 T->D)
15//   nx_store_put <prefix> close  <actor> <id> <closing-note>         (7-col convention: col3 open->closed, note appended)
16//   nx_store_put <prefix> load                                       (dump the plane)
17//   nx_store_put <prefix> hist                                       (dump the revision plane)
18// put on an unseeded prefix BOOTSTRAPS the plane (first row creates it -- no staging file needed).
19// close/setcol are FAIL-CLOSED: unknown id (or colidx past the row) -> exit 4, store untouched.
20// Whole-store last-writer-wins per write; writers serialized socially by session claims (flock = next rung).
21// ROW-LEVEL COMMITS (2026-09-02, loadgov LV17): put/close/setcol commit ONE row -- sts_append_fast for a new id, sts_replace_fast
22// (overwrite q:<seq>, newest-wins, no reader change) for an existing one. The whole-plane sts_seed is now only the bootstrap of an
23// unseeded plane and the fallback when a row cannot be located by seq; the shrink guard stays on that fallback. MEASURED CAUSE:
24// five whole-plane reseeds in flight held the RAID in D-state and refused every build on the estate for hours (2026-09-02).
25// license_tier: ORIGINAL   No hw writes (Rule 26).   expect_exit: 0
26import "nx_store_seed_lib.nx"
27import "nx_seg_store.nx"
28import "nx_syscalls.nx"
29const NSP_MAGIC_4096: i64 = 4096
30
31// 2026-08-08 RAISED 1048576 -> 33554432. THE FIFTH HAND-COPY OF THIS CAP, AND THE ONE NOBODY RAISED:
32// nx_debt DB_CAP, nx_debtlive DL_CAP, nx_debtcluster and nx_debt_view were all moved to 33554432 on
33// 2026-08-06 in lockstep; nx_dora's DCAP carries the measurement and the reason -- "knowledge/store/debt-
34// loads to 4,396,892B. At the old 1 MiB this organ silently lost 76% of the board, and because AN
35// APPEND-ONLY PLANE PAST A PREFIX CAP LOSES ITS NEWEST ROWS FIRST it was reporting on the OLDEST
36// quarter -- degrading exactly as new work arrived." This organ was missed. Measured today: that plane
37// is 4,974,565B / 3,785 rows.
38// TWO LIVE CONSEQUENCES, both from this one constant:
39//  (1) the `load` verb returned a silent 1 MiB PREFIX -- that is what fed nx_debtmine a 21.3% read it
40//      reported as complete coverage (debt 1786235467).
41//  (2) put/close/setcol were EFFECTIVELY DEAD on any plane over 1 MiB. They rebuild the whole store from
42//      the rows they loaded, so on a 4.9 MB plane `neu` holds ~1 MiB of rows and nsp_would_shrink
43//      correctly REFUSES every commit. NO DATA WAS EVER AT RISK -- that guard is the reason -- but the
44//      unified write verb could not write to the estate's largest planes at all.
45// ***THE GUARD IS WHY THIS WAS AN OUTAGE AND NOT A CATASTROPHE.*** nsp_would_shrink turns a truncated
46// read into a refusal instead of a whole-store rewrite that silently drops the newest 2,900 rows. Do not
47// weaken it when raising this cap: a larger buffer moves the cliff, the guard removes the precipice.
48// Re-verify the sizing with the estate's own ruler: nx_planefit knowledge/store/debt- 33554432
49const NSP_CAP: i64 = 33554432
50const NSP_NL: i64 = 10
51const NSP_TAB: i64 = 9
52const NSP_PIPE: i64 = 124
53const NSP_DASH: i64 = 45
54const NSP_NCOL: i64 = 7
55const NSP_PAIR: i64 = 2
56const NSP_SP_BYTES: i64 = 128
57const NSP_STDERR: i64 = 2
58const NSP_PFXCAP: i64 = 256
59const NSP_MSGCAP: i64 = 256
60const NSP_ARGC_PUT_MIN: i64 = 6  // prog prefix put actor id + at least 1 more field
61const NSP_ARGC_PUT_MAX: i64 = 20 // id + up to 15 fields (16-col plane cap)
62const NSP_ARGC_CLOSE: i64 = 6    // prog prefix close actor id note
63const NSP_ARGC_SETCOL: i64 = 7   // prog prefix setcol actor id colidx value
64const NSP_ARGC_MIN: i64 = 3
65const NSP_MAXCOL: i64 = 16       // max columns a plane row may carry
66const NSP_SP2_BYTES: i64 = 256   // 16 cols * 2 slots * 8B
67const NSP_EXIT_USAGE: i64 = 2
68const NSP_EXIT_IO: i64 = 1
69const NSP_EXIT_REFUSED: i64 = 4
70// putmany (2026-08-18): file-fed batch. NSP_BATCH_MAX bounds the per-call row count so the range
71// arrays are sized once; a records file with more lines is REFUSED (never silently truncated). The
72// records file is read whole via sys_read_file (lseek-END sized, cannot short-read), so it carries no
73// cap of its own. 4096 rows is ~46x the 88-contract comparewatch queue that motivated it -- headroom
74// named, not a guess, and the refusal makes an over-large batch a loud error rather than lost rows.
75const NSP_HASH: i64 = 35         // '#': comment lines in a records file are skipped, like a staging buffer
76const NSP_BATCH_MAX: i64 = 4096
77const NSP_F_RECFILE: i64 = 4   // putmany argv: prog prefix putmany actor recfile
78const NSP_ARGC_PUTMANY: i64 = 5
79// put field argv indices:
80const NSP_F_ACTOR: i64 = 3
81const NSP_F_ID: i64 = 4
82const NSP_F_LAST: i64 = 10
83// close argv indices:
84const NSP_C_ID: i64 = 4
85const NSP_C_NOTE: i64 = 5
86// status column index in the 7-col row:
87const NSP_COL_STATUS: i64 = 3
88const NSP_COL_NOTE: i64 = 6
89
90func nsp_slen(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
91func nsp_werr(s: *u8) -> i64 { sys_write(NSP_STDERR, s, nsp_slen(s)); return 0 }
92func nsp_eqs(a: *u8, b: *u8) -> i64 {
  var i: i64 = 0
  while a[i] != (0 as u8) { if a[i] != b[i] { return 0 } i = i + 1 }
  if b[i] != (0 as u8) { return 0 }
  return 1
97}
98// slice [a,b) of q equals C-string s
99// two byte-slices equal? (putmany matches an input-record id slice against a loaded-plane id slice)
100func nsp_slice_eq_slice(qa: *u8, a1: i64, b1: i64, qb: *u8, a2: i64, b2: i64) -> i64 {
  if b1 - a1 != b2 - a2 { return 0 }
  var i: i64 = 0
  while a1 + i < b1 { if qa[a1 + i] != qb[a2 + i] { return 0 } i = i + 1 }
  return 1
105}
106func nsp_slice_eqs(q: *u8, a: i64, b: i64, s: *u8) -> i64 {
  let sn: i64 = nsp_slen(s)
  if b - a != sn { return 0 }
  var i: i64 = 0
  while i < sn { if q[a+i] != s[i] { return 0 } i = i + 1 }
  return 1
112}
113// split line [ls,le) into up to NSP_MAXCOL (start,end) pairs (callers pass NSP_SP2_BYTES scratch)
114func nsp_cols(q: *u8, ls: i64, le: i64, sp: *i64) -> i64 {
  var c: i64 = 0
  var p: i64 = ls
  while c < NSP_MAXCOL {
      var e: i64 = p
      var s: i64 = 1
      while s == 1 { if e >= le { s = 0 } else { if q[e] == (NSP_TAB as u8) { s = 0 } else { e = e + 1 } } }
      sp[c*NSP_PAIR] = p
      sp[c*NSP_PAIR+1] = e
      c = c + 1
      if e >= le { return c }
      p = e + 1
  }
  return c
128}
129func nsp_cat_slice(d: *u8, o: i64, q: *u8, a: i64, b: i64) -> i64 {
  var oo: i64 = o
  var i: i64 = a
  while i < b { d[oo] = q[i]; oo = oo + 1; i = i + 1 }
  return oo
134}
135// copy slice with inner tabs -> '|' (hist rows stay 6-col clean)
136func nsp_cat_slice_pipe(d: *u8, o: i64, q: *u8, a: i64, b: i64) -> i64 {
  var oo: i64 = o
  var i: i64 = a
  while i < b {
      var c: i64 = q[i]
      if c == NSP_TAB { c = NSP_PIPE }
      d[oo] = c as u8
      oo = oo + 1
      i = i + 1
  }
  return oo
147}
148// derive the hist prefix: "knowledge/store/debt-" -> "knowledge/store/debthist-"
149func nsp_hist_prefix(prefix: *u8, out: *u8) -> i64 {
  var n: i64 = nsp_slen(prefix)
  if n > 0 { if prefix[n-1] == (NSP_DASH as u8) { n = n - 1 } }
  var i: i64 = 0
  while i < n { out[i] = prefix[i]; i = i + 1 }
  var o: i64 = ss_cat(out, n, "hist-" as *u8)
  out[o] = 0 as u8
  return o
157}
158// append one hist revision row: epoch actor verb id old new
159func nsp_hist_append(prefix: *u8, actor: *u8, verb: *u8, id: *u8, q: *u8, olda: i64, oldb: i64, neu: *u8, na: i64, nb: i64) -> i64 {
  let hp: *u8 = sys_mmap(NSP_PFXCAP)
  nsp_hist_prefix(prefix, hp)
  let hb: *u8 = sys_mmap(NSP_CAP)
  // O(1) APPEND 2026-07-30: this used to sts_load the ENTIRE hist plane just to append ONE revision
  // row, then sts_seed the WHOLE plane back -- O(rows x segments) write amplification on EVERY put
  // through this chokepoint. The row is a PURE APPEND (no existing row is modified), so we now build
  // ONLY the new row at offset 0 and hand it to sts_append_fast.
  var o: i64 = 0
  o = ss_catn(hb, o, sys_now_realtime_sec())
  hb[o] = NSP_TAB as u8
  o = o + 1
  o = ss_cat(hb, o, actor)
  hb[o] = NSP_TAB as u8
  o = o + 1
  o = ss_cat(hb, o, verb)
  hb[o] = NSP_TAB as u8
  o = o + 1
  o = ss_cat(hb, o, id)
  hb[o] = NSP_TAB as u8
  o = o + 1
  if oldb > olda { o = nsp_cat_slice_pipe(hb, o, q, olda, oldb) } else { o = ss_cat(hb, o, "-" as *u8) }
  hb[o] = NSP_TAB as u8
  o = o + 1
  if nb > na { o = nsp_cat_slice_pipe(hb, o, neu, na, nb) } else { o = ss_cat(hb, o, "-" as *u8) }
  hb[o] = NSP_NL as u8
  o = o + 1
  // FAST PATH: O(1) append -- writes ONLY this row, not the whole plane. o-1 strips the trailing
  // newline because sts_append_fast takes the bare row. BOOTSTRAP: sts_append_fast requires an
  // existing q:n, so a hist plane that does not exist YET legitimately returns <0; in that case
  // (and only that case) fall back to a full seed to create it. Every later append is O(1).
  // LOCKED form: the hist plane is NOT the plane `main` locked, so this append is otherwise
  // unprotected -- and sts_append_fast is itself a read-modify-write on q:n (two appenders read the
  // same n, write the same q:<n>, and newest-wins silently OVERWRITES one row). Lock order is always
  // main-then-hist, so no ABBA deadlock is possible.
  var rc: i64 = sts_append_fast_locked(hp, hb, o - 1)
  if rc < 0 { rc = sts_seed(hp, hb, o) }
  if rc < 0 { return 0 - 1 }
  return 0
198}
199// span-atoi for the q:n value slice (ss_get hands back ptr+len, not a NUL string)
200func nsp_atoi_span(q: *u8, n: i64) -> i64 {
  var v: i64 = 0
  var i: i64 = 0
  while i < n { let c: i64 = q[i] as i64; if c >= 48 { if c <= 57 { v = v * 10 + (c - 48) } } i = i + 1 }
  return v
205}
206// BATCHED HIST COMMIT (2026-08-19, the putmany residual). putmany collapsed the MAIN plane to one
207// commit and then appended its audit rows ONE AT A TIME: N x sts_append_fast_locked = N x (lock +
208// q:n read + segment commit + manifest rename + syncdir) on the hist plane -- MEASURED as the
209// surviving slow leg: a 163-row batch spent ~15 min in btrfs_log_inode_parent doing exactly this
210// while holding the MAIN plane lock, which is the whole convoy shrunk by one layer, not cured.
211// This is sts_append_fast generalized to k rows: ONE lock, ONE q:n read, k adds, ONE commit.
212// Same CAS discipline (gen0 before the read, refuse-on-stale under the lock); same bootstrap
213// (no q:n yet -> full seed of just these rows); lock order stays main-then-hist.
214// rows: concatenated row TEXTS (no trailing newlines); offs/lens: per-row slices; returns 0 or -1.
215func nsp_hist_commit_rows(hp: *u8, rows: *u8, offs: *i64, lens: *i64, k: i64) -> i64 {
  if k <= 0 { return 0 }
  let lk: i64 = sts_lock(hp)
  let gen0: i64 = ss_max_segid(hp)
  let pq: *i64 = sys_mmap(16) as *i64
  let lq: *i64 = sys_mmap(16) as *i64
  if ss_get(hp, "q:n" as *u8, pq, lq) != 1 {
      // BOOTSTRAP: no q:n row yet -- seed a fresh plane holding exactly these rows.
      let nlj: *u8 = sys_mmap(NSP_CAP)
      var no2: i64 = 0
      var b: i64 = 0
      while b < k {
          var z: i64 = 0
          while z < lens[b] { nlj[no2] = rows[offs[b] + z]; no2 = no2 + 1; z = z + 1 }
          nlj[no2] = NSP_NL as u8; no2 = no2 + 1
          b = b + 1
      }
      let src: i64 = sts_seed(hp, nlj, no2)
      sts_unlock(lk)
      if src < 0 { return 0 - 1 }
      return 0
  }
  let n0: i64 = nsp_atoi_span(pq[0] as *u8, lq[0])
  var need: i64 = STS_WSLACK
  var t: i64 = 0
  while t < k { need = need + lens[t] + STS_KEYCAP + STS_ROWOVH; t = t + 1 }
  let w: *i64 = ss_begin_cap(need)
  let key: *u8 = sys_mmap(STS_KEYCAP)
  var i: i64 = 0
  var addfail: i64 = 0
  while i < k {
      sts_rowkey(n0 + i, key)
      if ss_add(w, STS_KIND_LIVE, key, ((rows as i64) + offs[i]) as *u8, lens[i]) < 0 { addfail = 1 }
      i = i + 1
  }
  let cb: *u8 = sys_mmap(STS_NUMCAP)
  let cl: i64 = ss_catn(cb, 0, n0 + k)
  if ss_add(w, STS_KIND_LIVE, "q:n" as *u8, cb, cl) < 0 { addfail = 1 }
  if addfail == 1 { sts_unlock(lk); return 0 - 1 }
  let crc: i64 = ss_commit_cas(hp, w, ss_next_segid(hp), gen0)
  sts_unlock(lk)
  if crc != 0 { return 0 - 1 }
  return 0
258}
259func nsp_atoi(s: *u8) -> i64 {
  var v: i64 = 0
  var i: i64 = 0
  while s[i] != (0 as u8) {
      let c: i64 = s[i]
      if c >= 48 { if c <= 57 { v = v * 10 + (c - 48) } }
      i = i + 1
  }
  return v
268}
269
270func nsp_report(verb: *u8, id: *u8, replaced: i64, rows: i64) -> i64 {
  let m: *u8 = sys_mmap(NSP_MSGCAP)
  var o: i64 = ss_cat(m, 0, verb)
  o = ss_cat(m, o, " " as *u8)
  o = ss_cat(m, o, id)
  o = ss_cat(m, o, " replaced=" as *u8)
  o = ss_catn(m, o, replaced)
  o = ss_cat(m, o, " rows=" as *u8)
  o = ss_catn(m, o, rows)
  m[o] = NSP_NL as u8
  o = o + 1
  sys_write(1, m, o)
  return 0
283}
284
285// SHARED ANTI-CLOBBER TOOTH (2026-08-02, debts 1785713047/1785713373) -- ONE guard, FOUR callers.
286// EVERY write verb here rebuilds the WHOLE plane from what sts_load could reach, then re-seeds it. If that
287// load came back PARTIAL, every row it could not see is silently DELETED. put/putn ADD rows; close/setcol
288// MODIFY one row in place -- so NONE of them can ever legitimately REDUCE the row count. A count below what
289// the plane already declares therefore means the load was short, and committing it would destroy data
290// (commontask- 37->10, dedupq- 21->3, both recovered 2026-08-02). Refuse instead of committing the loss.
291// FAIL-OPEN when q:n is absent -- that call is BOOTSTRAPPING the plane, which is legitimate.
292// ★Deliberately NOT in the shared sts_seed primitive: there it would refuse nx_plane_append's ROLLBACK and
293// nx_plane_repair, both of which shrink BY DEFINITION (1785713047). The guard belongs to the WRITER.
294// Proven both directions by nx_store_put_shrink_gate 4/4: it BITES a manufactured over-declared plane
295// (rc=4, store untouched) and SPARES a healthy one (rc=0, q:n 3->4).
296func nsp_would_shrink(prefix: *u8, rows: i64) -> i64 {
  let vq: *i64 = sys_mmap(NSP_MAGIC_4096) as *i64
  let vl: *i64 = sys_mmap(NSP_MAGIC_4096) as *i64
  if ss_get(prefix, "q:n" as *u8, vq, vl) == 1 {
      let declared: i64 = sts_atoi(vq[0] as *u8, vl[0])
      if rows < declared { return 1 }
  }
  return 0
304}
305func nsp_refuse_shrink(vname: *u8) -> i64 {
  nsp_werr("REFUSED: this " as *u8)
  nsp_werr(vname)
  nsp_werr(" would SHRINK the plane -- the load came back partial, so committing it would delete the rows it could not see (the commontask-/dedupq- clobber class). Store untouched; re-run, and if the plane is genuinely damaged use nx_plane_repair deliberately.\n" as *u8)
  return 0
310}
311
312func main(argc: i64, argv: *i64) -> i64 {
  if argc < NSP_ARGC_MIN { nsp_werr("usage: nx_store_put <prefix> {put <actor> <id> <title> <sev> <status> <owner> <scope> <note> | close <actor> <id> <note> | load | hist}\n" as *u8); sys_exit(NSP_EXIT_USAGE); return NSP_EXIT_USAGE }
  let prefix: *u8 = argv[1] as *u8
  let verb: *u8 = argv[2] as *u8
316
  // seq1559 MIGRATION 2026-07-30 (ws=sev-eater). Every mutating verb here is a READ-MODIFY-WRITE
  // over a shared plane: sts_load -> rewrite the whole buffer -> sts_seed. Unlocked, two concurrent
  // invocations interleave as A-loads / B-loads / A-commits / B-commits-without-A and one caller's
  // row is gone silently. That is not theoretical -- nx_sts_lock_gate measures 5 of 6 rows destroyed
  // under 6-way concurrency, and 0 lost once the lock is held.
  // Taken HERE, before the first load, so the critical section covers the entire operation rather
  // than just the commit -- locking only the write would still let two readers race the same
  // snapshot. Uses sts_lock, i.e. the SAME <prefix>plock nx_debt / nx_plane_append / sts_append_row
  // use; a private lock file would have been easier and would have excluded nobody.
  // NO UNLOCK IS NEEDED ON ANY PATH: this is a one-shot CLI organ and every exit -- success or a
  // fail-closed sys_exit() -- terminates the process, which releases the flock. There is therefore
  // no path that can leak the lock, which is why the lock is safe to take this early.
  sts_lock(prefix)
330
  // ---- ADOPTS sts_load_fit ON BOTH READ VERBS (2026-08-08) ---------------------------------
  // ***NSP_CAP IS NO LONGER IN THE CORRECTNESS PATH HERE.*** Raising it (which I did earlier today,
  // 1048576 -> 33554432) is the INFERIOR HALF of a migration four organs completed on 2026-08-06:
  // nx_dora:222, nx_pm_board:162/167/172/177, nx_sheriff:311 and nx_planefit:229 all moved to
  // sts_load_fit, and nx_debt's own DB_CAP comment had already named the requirement -- "THIS IS THE
  // THIRD RAISE, NOT A FIX ... A CAP THAT CAN BE CROSSED IN SILENCE WILL BE CROSSED AGAIN." This verb
  // was missed in BOTH halves, which is exactly why nx_debtmine forked it and received 1,048,576 of
  // 5,048,015 plane bytes -- 21.3% -- and reported it as complete coverage.
  // sts_load_fit needs no stat and no heuristic: sts_load truncates ONLY when its buffer fills, so a
  // return STRICTLY LESS THAN cap PROVES completeness; n == cap is the one ambiguous case and the only
  // one it grows on (8 MiB doubling to 1 GiB, munmapping each attempt), then it REFUSES with a named
  // diagnostic rather than hand back a partial board.
  if nsp_eqs("load" as *u8, verb) == 1 {
      let ln: *i64 = sys_mmap(16) as *i64
      let b: *u8 = sts_load_fit(prefix, ln)
      if (b as i64) == 0 { nsp_werr("load REFUSED: sts_load_fit exhausted its growth and will NOT return a partial board -- shard the plane or raise STS_FIT_MAXTRIES deliberately\n" as *u8); sys_exit(NSP_EXIT_IO); return NSP_EXIT_IO }
      let n: i64 = ln[0]
      // THREE STATES, NOT TWO (2026-08-15). This printed "plane EMPTY / unseeded" for EVERY zero-row
      // outcome, and that is A FALSE DIAGNOSIS WITH AN AUTHORITATIVE NAME whenever the prefix names a
      // store that HAS data. MEASURED THE SAME DAY, and it cost three successively-wrong debt rows and
      // one PUBLISHED falsehood before the control was finally run: knowledge/store/ecomat returns zero
      // rows here while its own manifest lists 35 segments and nx_ecosystem_maturity_rollup reads 26
      // domains out of that exact prefix in the same minute.
      // NOTHING IS BROKEN. This verb is a ROW-plane reader (id title sev status owner scope note);
      // ecomat is not a row plane, so it carries no q:n row index and a row reader correctly finds
      // nothing. The control that settled it was one call: the same verb on knowledge/store/roi-
      // returns every row, which exonerates sts_load_fit, sts_load, ss_open and ss_open2 at once.
      //   * A READER THAT REPORTS "EMPTY" WHEN IT MEANS "NOT MY FORMAT" INVITES THE NEXT READER TO
      //     RESEED A PLANE THAT IS ALREADY FULL.
      //   * ASK OF ANY VERDICT: IS THE SUBJECT THE THING THE MESSAGE NAMES? Here it was not.
      // The exit code is deliberately UNCHANGED (rule 19 -- an exit code is an API contract and no
      // caller asked for a new one); only the sentence a human reads is corrected.
      if n <= 0 {
          let hh: *i64 = ss_open(prefix)
          var segs: i64 = 0
          if (hh as i64) != 0 { segs = hh[0] }
          if segs > 0 {
              nsp_werr("load: NOT-A-ROW-PLANE -- this prefix resolves to a store that HAS segments but carries no q:n row index, so a ROW reader finds nothing. That is NOT the same fact as an empty plane: the records are present and addressed another way, and seeding here would DUPLICATE LIVE DATA. Read it with the organ that owns this plane.\n" as *u8)
              sys_exit(NSP_EXIT_IO)
              return NSP_EXIT_IO
          }
          nsp_werr("plane EMPTY / unseeded (no segments resolve under this prefix)\n" as *u8)
          sys_exit(NSP_EXIT_IO)
          return NSP_EXIT_IO
      }
      sys_write(1, b, n)
      sys_exit(0)
      return 0
  }
  if nsp_eqs("hist" as *u8, verb) == 1 {
      let hp: *u8 = sys_mmap(NSP_PFXCAP)
      nsp_hist_prefix(prefix, hp)
      let l2: *i64 = sys_mmap(16) as *i64
      let b2: *u8 = sts_load_fit(hp, l2)
      if (b2 as i64) == 0 { nsp_werr("hist REFUSED: sts_load_fit exhausted its growth and will NOT return a partial revision log\n" as *u8); sys_exit(NSP_EXIT_IO); return NSP_EXIT_IO }
      let n2: i64 = l2[0]
      if n2 <= 0 { nsp_werr("hist EMPTY (no revisions yet)\n" as *u8); sys_exit(NSP_EXIT_IO); return NSP_EXIT_IO }
      sys_write(1, b2, n2)
      sys_exit(0)
      return 0
  }
392
  if nsp_eqs("put" as *u8, verb) == 1 {
      if argc < NSP_ARGC_PUT_MIN { nsp_werr("put needs <actor> <id> <field>... (1..15 fields after id)\n" as *u8); sys_exit(NSP_EXIT_USAGE); return NSP_EXIT_USAGE }
      if argc > NSP_ARGC_PUT_MAX { nsp_werr("put: too many fields (16-col plane cap)\n" as *u8); sys_exit(NSP_EXIT_USAGE); return NSP_EXIT_USAGE }
      let actor: *u8 = argv[NSP_F_ACTOR] as *u8
      let id: *u8 = argv[NSP_F_ID] as *u8
      let cur: *u8 = sys_mmap(NSP_CAP)
      var cn: i64 = sts_load(prefix, cur, NSP_CAP - NSP_MAGIC_4096)
      if cn < 0 { cn = 0 }
      let neu: *u8 = sys_mmap(NSP_CAP)
      let sp: *i64 = sys_mmap(NSP_SP2_BYTES) as *i64
      // literal counters first
      var no: i64 = 0
      var replaced: i64 = 0
      var rows: i64 = 0
      var olda: i64 = 0
      var oldb: i64 = 0
      var i: i64 = 0
      while i < cn {
          var le: i64 = i
          var s: i64 = 1
          while s == 1 { if le >= cn { s = 0 } else { if cur[le] == (NSP_NL as u8) { s = 0 } else { le = le + 1 } } }
          if le > i {
              nsp_cols(cur, i, le, sp)
              if nsp_slice_eqs(cur, sp[0], sp[1], id) == 1 {
                  replaced = 1
                  olda = i
                  oldb = le
              } else {
                  no = nsp_cat_slice(neu, no, cur, i, le)
                  neu[no] = NSP_NL as u8
                  no = no + 1
                  rows = rows + 1
              }
          }
          i = le + 1
      }
      // append the new/updated row (ALL fields argv[4..argc-1] tab-joined -- N-col, schema-agnostic)
      let na: i64 = no
      let flast: i64 = argc - 1
      var f: i64 = NSP_F_ID
      while f <= flast {
          no = ss_cat(neu, no, argv[f] as *u8)
          if f < flast { neu[no] = NSP_TAB as u8; no = no + 1 }
          f = f + 1
      }
      let nb: i64 = no
      neu[no] = NSP_NL as u8
      no = no + 1
      rows = rows + 1
      // ROW-LEVEL COMMIT (LV17): commit ONE row, never the plane. An existing id is replaced at its q:<seq>; a new id is
      // appended as q:<n>. The whole-plane seed runs ONLY to bootstrap an unseeded plane or when the row cannot be located
      // by seq, and only there does the shrink guard still apply (it exists to stop a SEED from deleting unseen rows).
      var rc: i64 = 0 - 1
      var cpath_rc: *u8 = "replace" as *u8
      if replaced == 1 {
          let seq_rc: i64 = sts_find_seq(prefix, id)
          if seq_rc >= 0 { rc = sts_replace_fast(prefix, seq_rc, ((neu as i64) + na) as *u8, nb - na) }
      } else {
          rc = sts_append_fast(prefix, ((neu as i64) + na) as *u8, nb - na)
          cpath_rc = "append" as *u8
      }
      if rc < 0 {
          cpath_rc = "seed" as *u8
          if nsp_would_shrink(prefix, rows) == 1 { nsp_refuse_shrink("put" as *u8); sys_exit(NSP_EXIT_REFUSED); return NSP_EXIT_REFUSED }
          rc = sts_seed(prefix, neu, no)
      }
      nsp_werr("commit-path=" as *u8); nsp_werr(cpath_rc); nsp_werr("\n" as *u8)
      if rc < 0 { nsp_werr("plane commit error\n" as *u8); sys_exit(NSP_EXIT_IO); return NSP_EXIT_IO }
      if nsp_hist_append(prefix, actor, "put" as *u8, id, cur, olda, oldb, neu, na, nb) < 0 { nsp_werr("hist commit error\n" as *u8); sys_exit(NSP_EXIT_IO); return NSP_EXIT_IO }
      nsp_report("PUT" as *u8, id, replaced, rows)
      sys_exit(0)
      return 0
  }
466
  if nsp_eqs("close" as *u8, verb) == 1 {
      if argc < NSP_ARGC_CLOSE { nsp_werr("close needs <actor> <id> <note>\n" as *u8); sys_exit(NSP_EXIT_USAGE); return NSP_EXIT_USAGE }
      let actor2: *u8 = argv[NSP_F_ACTOR] as *u8
      let id2: *u8 = argv[NSP_C_ID] as *u8
      let cnote: *u8 = argv[NSP_C_NOTE] as *u8
      let cur2: *u8 = sys_mmap(NSP_CAP)
      let cn2: i64 = sts_load(prefix, cur2, NSP_CAP - NSP_MAGIC_4096)
      if cn2 <= 0 { nsp_werr("plane EMPTY: nothing to close\n" as *u8); sys_exit(NSP_EXIT_REFUSED); return NSP_EXIT_REFUSED }
      let neu2: *u8 = sys_mmap(NSP_CAP)
      let sp2: *i64 = sys_mmap(NSP_SP2_BYTES) as *i64
      var no2: i64 = 0
      var found: i64 = 0
      var rows2: i64 = 0
      var olda2: i64 = 0
      var oldb2: i64 = 0
      var na2: i64 = 0
      var nb2: i64 = 0
      var j: i64 = 0
      while j < cn2 {
          var le2: i64 = j
          var s2: i64 = 1
          while s2 == 1 { if le2 >= cn2 { s2 = 0 } else { if cur2[le2] == (NSP_NL as u8) { s2 = 0 } else { le2 = le2 + 1 } } }
          if le2 > j {
              let nc: i64 = nsp_cols(cur2, j, le2, sp2)
              if nsp_slice_eqs(cur2, sp2[0], sp2[1], id2) == 1 {
                  found = 1
                  olda2 = j
                  oldb2 = le2
                  na2 = no2
                  // rebuild: cols 0..2 as-is, col3 = closed, cols 4..5 as-is, col6 = old-note | close-note
                  var c: i64 = 0
                  while c < nc {
                      if c == NSP_COL_STATUS {
                          no2 = ss_cat(neu2, no2, "closed" as *u8)
                      } else {
                          no2 = nsp_cat_slice(neu2, no2, cur2, sp2[c*NSP_PAIR], sp2[c*NSP_PAIR+1])
                      }
                      if c == NSP_COL_NOTE {
                          no2 = ss_cat(neu2, no2, " | " as *u8)
                          no2 = ss_cat(neu2, no2, cnote)
                      }
                      if c < nc - 1 { neu2[no2] = NSP_TAB as u8; no2 = no2 + 1 }
                      c = c + 1
                  }
                  nb2 = no2
                  neu2[no2] = NSP_NL as u8
                  no2 = no2 + 1
                  rows2 = rows2 + 1
              } else {
                  no2 = nsp_cat_slice(neu2, no2, cur2, j, le2)
                  neu2[no2] = NSP_NL as u8
                  no2 = no2 + 1
                  rows2 = rows2 + 1
              }
          }
          j = le2 + 1
      }
      if found == 0 { nsp_werr("REFUSED: id not in plane (close is fail-closed; store untouched)\n" as *u8); sys_exit(NSP_EXIT_REFUSED); return NSP_EXIT_REFUSED }
      // ROW-LEVEL COMMIT (LV17): commit ONE row, never the plane. An existing id is replaced at its q:<seq>; a new id is
      // appended as q:<n>. The whole-plane seed runs ONLY to bootstrap an unseeded plane or when the row cannot be located
      // by seq, and only there does the shrink guard still apply (it exists to stop a SEED from deleting unseen rows).
      var rc2: i64 = 0 - 1
      var cpath_rc2: *u8 = "replace" as *u8
      let seq_rc2: i64 = sts_find_seq(prefix, id2)
      if seq_rc2 >= 0 { rc2 = sts_replace_fast(prefix, seq_rc2, ((neu2 as i64) + na2) as *u8, nb2 - na2) }
      if rc2 < 0 {
          cpath_rc2 = "seed" as *u8
          if nsp_would_shrink(prefix, rows2) == 1 { nsp_refuse_shrink("close" as *u8); sys_exit(NSP_EXIT_REFUSED); return NSP_EXIT_REFUSED }
          rc2 = sts_seed(prefix, neu2, no2)
      }
      nsp_werr("commit-path=" as *u8); nsp_werr(cpath_rc2); nsp_werr("\n" as *u8)
      if rc2 < 0 { nsp_werr("plane commit error\n" as *u8); sys_exit(NSP_EXIT_IO); return NSP_EXIT_IO }
      if nsp_hist_append(prefix, actor2, "close" as *u8, id2, cur2, olda2, oldb2, neu2, na2, nb2) < 0 { nsp_werr("hist commit error\n" as *u8); sys_exit(NSP_EXIT_IO); return NSP_EXIT_IO }
      nsp_report("CLOSED" as *u8, id2, found, rows2)
      sys_exit(0)
      return 0
  }
544
  if nsp_eqs("putn" as *u8, verb) == 1 {
      // BATCH put (W008 residual, 2026-07-18): N rows in ONE call -- one store commit, per-row hist.
      //   putn <actor> <ncols> <field>...   (groups of ncols fields, first field of each group = id)
      if argc < 7 { nsp_werr("putn needs <actor> <ncols> <field>... (groups of ncols, id first)\n" as *u8); sys_exit(NSP_EXIT_USAGE); return NSP_EXIT_USAGE }
      let actorn: *u8 = argv[NSP_F_ACTOR] as *u8
      let ncols: i64 = nsp_atoi(argv[4] as *u8)
      if ncols < 2 { nsp_werr("putn: ncols must be 2..15\n" as *u8); sys_exit(NSP_EXIT_USAGE); return NSP_EXIT_USAGE }
      if ncols > 15 { nsp_werr("putn: ncols must be 2..15\n" as *u8); sys_exit(NSP_EXIT_USAGE); return NSP_EXIT_USAGE }
      let nfields: i64 = argc - 5
      let ngroups: i64 = nfields / ncols
      if ngroups * ncols != nfields { nsp_werr("putn REFUSED: field count not a multiple of ncols (fail-closed, store untouched)\n" as *u8); sys_exit(NSP_EXIT_USAGE); return NSP_EXIT_USAGE }
      if ngroups < 1 { nsp_werr("putn: no groups\n" as *u8); sys_exit(NSP_EXIT_USAGE); return NSP_EXIT_USAGE }
      if ngroups > 32 { nsp_werr("putn: max 32 groups per call\n" as *u8); sys_exit(NSP_EXIT_USAGE); return NSP_EXIT_USAGE }
      var abuf: *u8 = sys_mmap(NSP_CAP)
      var bbuf: *u8 = sys_mmap(NSP_CAP)
      var an: i64 = sts_load(prefix, abuf, NSP_CAP - NSP_MAGIC_4096)
      if an < 0 { an = 0 }
      let spn: *i64 = sys_mmap(NSP_SP2_BYTES) as *i64
      var totrows: i64 = 0
      var g: i64 = 0
      while g < ngroups {
          let base: i64 = 5 + g * ncols
          let idn: *u8 = argv[base] as *u8
          var no: i64 = 0
          var olda: i64 = 0
          var oldb: i64 = 0
          var rows: i64 = 0
          var i2: i64 = 0
          while i2 < an {
              var le: i64 = i2
              var s: i64 = 1
              while s == 1 { if le >= an { s = 0 } else { if abuf[le] == (NSP_NL as u8) { s = 0 } else { le = le + 1 } } }
              if le > i2 {
                  nsp_cols(abuf, i2, le, spn)
                  if nsp_slice_eqs(abuf, spn[0], spn[1], idn) == 1 { olda = i2; oldb = le } else {
                      no = nsp_cat_slice(bbuf, no, abuf, i2, le)
                      bbuf[no] = NSP_NL as u8
                      no = no + 1
                      rows = rows + 1
                  }
              }
              i2 = le + 1
          }
          let na: i64 = no
          var f2: i64 = base
          let flast2: i64 = base + ncols - 1
          while f2 <= flast2 {
              no = ss_cat(bbuf, no, argv[f2] as *u8)
              if f2 < flast2 { bbuf[no] = NSP_TAB as u8; no = no + 1 }
              f2 = f2 + 1
          }
          let nb: i64 = no
          bbuf[no] = NSP_NL as u8
          no = no + 1
          rows = rows + 1
          if nsp_hist_append(prefix, actorn, "putn" as *u8, idn, abuf, olda, oldb, bbuf, na, nb) < 0 { nsp_werr("hist commit error\n" as *u8); sys_exit(NSP_EXIT_IO); return NSP_EXIT_IO }
          let tswap: *u8 = abuf
          abuf = bbuf
          bbuf = tswap
          an = no
          totrows = rows
          g = g + 1
      }
      // putn accumulates across groups, so totrows is the final count -- same invariant as put: a BATCH of
      // adds can only grow or hold steady. ⚠This one commits ONCE after N groups whose hist frames are
      // ALREADY written, so an unguarded partial load here loses rows the history says were just put.
      if nsp_would_shrink(prefix, totrows) == 1 { nsp_refuse_shrink("putn" as *u8); sys_exit(NSP_EXIT_REFUSED); return NSP_EXIT_REFUSED }
      if sts_seed(prefix, abuf, an) < 0 { nsp_werr("plane commit error\n" as *u8); sys_exit(NSP_EXIT_IO); return NSP_EXIT_IO }
      nsp_report("PUTN-GROUPS" as *u8, actorn, ngroups, totrows)
      sys_exit(0)
      return 0
  }
617
  if nsp_eqs("setcol" as *u8, verb) == 1 {
      if argc < NSP_ARGC_SETCOL { nsp_werr("setcol needs <actor> <id> <colidx> <value>\n" as *u8); sys_exit(NSP_EXIT_USAGE); return NSP_EXIT_USAGE }
      let actor3: *u8 = argv[NSP_F_ACTOR] as *u8
      let id3: *u8 = argv[NSP_C_ID] as *u8
      let colidx: i64 = nsp_atoi(argv[5] as *u8)
      let val: *u8 = argv[6] as *u8
      let cur3: *u8 = sys_mmap(NSP_CAP)
      let cn3: i64 = sts_load(prefix, cur3, NSP_CAP - NSP_MAGIC_4096)
      if cn3 <= 0 { nsp_werr("plane EMPTY: nothing to setcol\n" as *u8); sys_exit(NSP_EXIT_REFUSED); return NSP_EXIT_REFUSED }
      let neu3: *u8 = sys_mmap(NSP_CAP)
      let sp3: *i64 = sys_mmap(NSP_SP2_BYTES) as *i64
      var no3: i64 = 0
      var found3: i64 = 0
      var rows3: i64 = 0
      var olda3: i64 = 0
      var oldb3: i64 = 0
      var na3: i64 = 0
      var nb3: i64 = 0
      var j3: i64 = 0
      while j3 < cn3 {
          var le3: i64 = j3
          var s3: i64 = 1
          while s3 == 1 { if le3 >= cn3 { s3 = 0 } else { if cur3[le3] == (NSP_NL as u8) { s3 = 0 } else { le3 = le3 + 1 } } }
          if le3 > j3 {
              let nc3: i64 = nsp_cols(cur3, j3, le3, sp3)
              if nsp_slice_eqs(cur3, sp3[0], sp3[1], id3) == 1 {
                  if colidx >= nc3 { nsp_werr("REFUSED: colidx past the row (setcol is fail-closed; store untouched)\n" as *u8); sys_exit(NSP_EXIT_REFUSED); return NSP_EXIT_REFUSED }
                  found3 = 1
                  olda3 = j3
                  oldb3 = le3
                  na3 = no3
                  var c3: i64 = 0
                  while c3 < nc3 {
                      if c3 == colidx { no3 = ss_cat(neu3, no3, val) } else { no3 = nsp_cat_slice(neu3, no3, cur3, sp3[c3*NSP_PAIR], sp3[c3*NSP_PAIR+1]) }
                      if c3 < nc3 - 1 { neu3[no3] = NSP_TAB as u8; no3 = no3 + 1 }
                      c3 = c3 + 1
                  }
                  nb3 = no3
                  neu3[no3] = NSP_NL as u8
                  no3 = no3 + 1
                  rows3 = rows3 + 1
              } else {
                  no3 = nsp_cat_slice(neu3, no3, cur3, j3, le3)
                  neu3[no3] = NSP_NL as u8
                  no3 = no3 + 1
                  rows3 = rows3 + 1
              }
          }
          j3 = le3 + 1
      }
      if found3 == 0 { nsp_werr("REFUSED: id not in plane (setcol is fail-closed; store untouched)\n" as *u8); sys_exit(NSP_EXIT_REFUSED); return NSP_EXIT_REFUSED }
      // ROW-LEVEL COMMIT (LV17): commit ONE row, never the plane. An existing id is replaced at its q:<seq>; a new id is
      // appended as q:<n>. The whole-plane seed runs ONLY to bootstrap an unseeded plane or when the row cannot be located
      // by seq, and only there does the shrink guard still apply (it exists to stop a SEED from deleting unseen rows).
      var rc3: i64 = 0 - 1
      var cpath_rc3: *u8 = "replace" as *u8
      let seq_rc3: i64 = sts_find_seq(prefix, id3)
      if seq_rc3 >= 0 { rc3 = sts_replace_fast(prefix, seq_rc3, ((neu3 as i64) + na3) as *u8, nb3 - na3) }
      if rc3 < 0 {
          cpath_rc3 = "seed" as *u8
          if nsp_would_shrink(prefix, rows3) == 1 { nsp_refuse_shrink("setcol" as *u8); sys_exit(NSP_EXIT_REFUSED); return NSP_EXIT_REFUSED }
          rc3 = sts_seed(prefix, neu3, no3)
      }
      nsp_werr("commit-path=" as *u8); nsp_werr(cpath_rc3); nsp_werr("\n" as *u8)
      if rc3 < 0 { nsp_werr("plane commit error\n" as *u8); sys_exit(NSP_EXIT_IO); return NSP_EXIT_IO }
      if nsp_hist_append(prefix, actor3, "setcol" as *u8, id3, cur3, olda3, oldb3, neu3, na3, nb3) < 0 { nsp_werr("hist commit error\n" as *u8); sys_exit(NSP_EXIT_IO); return NSP_EXIT_IO }
      nsp_report("SETCOL" as *u8, id3, found3, rows3)
      sys_exit(0)
      return 0
  }
688
  if nsp_eqs("putmany" as *u8, verb) == 1 {
      // FILE-FED BATCH: one sts_load, one sts_seed, per-row hist. Collapses N forked full-plane
      // rewrites into ONE, killing the concurrent-writer D-state storm by construction (a single
      // sequential process holding the lock once). Records file = one row per line, TAB-separated
      // fields, first field = id; '#'/blank lines skipped -- byte-shape identical to a plane row.
      if argc < NSP_ARGC_PUTMANY { nsp_werr("putmany needs <actor> <recordsfile>\n" as *u8); sys_exit(NSP_EXIT_USAGE); return NSP_EXIT_USAGE }
      let actorm: *u8 = argv[NSP_F_ACTOR] as *u8
      let recpath: *u8 = argv[NSP_F_RECFILE] as *u8
      let rlen: *i64 = sys_mmap(16) as *i64
      let recs: *u8 = sys_read_file(recpath, rlen)
      if (recs as i64) == 0 { nsp_werr("putmany REFUSED: records file unreadable (store untouched)\n" as *u8); sys_exit(NSP_EXIT_IO); return NSP_EXIT_IO }
      let rn: i64 = rlen[0]
      // parse input records into (line, id) slice ranges. '#' and blank lines are skipped so a
      // records file can carry comments exactly like a staging buffer.
      let la: *i64 = sys_mmap(8 * NSP_BATCH_MAX) as *i64   // input line start
      let lb: *i64 = sys_mmap(8 * NSP_BATCH_MAX) as *i64   // input line end (exclusive)
      let ida: *i64 = sys_mmap(8 * NSP_BATCH_MAX) as *i64  // input id start
      let idb: *i64 = sys_mmap(8 * NSP_BATCH_MAX) as *i64  // input id end
      var nin: i64 = 0
      var rp: i64 = 0
      while rp < rn {
          var re: i64 = rp
          while re < rn { if recs[re] == (NSP_NL as u8) { break } re = re + 1 }
          if re > rp { if recs[rp] != (NSP_HASH as u8) {
              if nin >= NSP_BATCH_MAX { nsp_werr("putmany REFUSED: more than NSP_BATCH_MAX records (store untouched) -- split the batch or raise the const deliberately\n" as *u8); sys_exit(NSP_EXIT_REFUSED); return NSP_EXIT_REFUSED }
              // id = first TAB-delimited field
              var ie: i64 = rp
              while ie < re { if recs[ie] == (NSP_TAB as u8) { break } ie = ie + 1 }
              la[nin] = rp; lb[nin] = re; ida[nin] = rp; idb[nin] = ie
              nin = nin + 1
          } }
          rp = re + 1
      }
      if nin == 0 { nsp_report("PUTMANY" as *u8, actorm, 0, 0); sys_exit(0); return 0 }
      // load the plane ONCE
      let cur: *u8 = sys_mmap(NSP_CAP)
      var cn: i64 = sts_load(prefix, cur, NSP_CAP - NSP_MAGIC_4096)
      if cn < 0 { cn = 0 }
      let neu: *u8 = sys_mmap(NSP_CAP)
      let spm: *i64 = sys_mmap(NSP_SP2_BYTES) as *i64
      // per-input OLD range in cur (the row this input replaces, if any) for hist
      let olda: *i64 = sys_mmap(8 * NSP_BATCH_MAX) as *i64
      let oldb: *i64 = sys_mmap(8 * NSP_BATCH_MAX) as *i64
      var k0: i64 = 0
      while k0 < nin { olda[k0] = 0; oldb[k0] = 0; k0 = k0 + 1 }
      var no: i64 = 0
      var rows: i64 = 0
      // PASS 1: copy every existing row whose id is NOT in the input; record the OLD range for the
      // input that replaces it (matched-input hist gets a real before-image, new-input gets '-').
      var i: i64 = 0
      while i < cn {
          var le: i64 = i
          var s: i64 = 1
          while s == 1 { if le >= cn { s = 0 } else { if cur[le] == (NSP_NL as u8) { s = 0 } else { le = le + 1 } } }
          if le > i {
              nsp_cols(cur, i, le, spm)
              var matched: i64 = 0 - 1
              var q: i64 = 0
              while q < nin {
                  if nsp_slice_eq_slice(cur, spm[0], spm[1], recs, ida[q], idb[q]) == 1 { matched = q; q = nin } else { q = q + 1 }
              }
              if matched >= 0 { olda[matched] = i; oldb[matched] = le } else {
                  no = nsp_cat_slice(neu, no, cur, i, le)
                  neu[no] = NSP_NL as u8; no = no + 1
                  rows = rows + 1
              }
          }
          i = le + 1
      }
      // PASS 2: append every input row; record its NEW range for hist.
      let na: *i64 = sys_mmap(8 * NSP_BATCH_MAX) as *i64
      let nb: *i64 = sys_mmap(8 * NSP_BATCH_MAX) as *i64
      var k: i64 = 0
      while k < nin {
          na[k] = no
          no = nsp_cat_slice(neu, no, recs, la[k], lb[k])
          nb[k] = no
          neu[no] = NSP_NL as u8; no = no + 1
          rows = rows + 1
          k = k + 1
      }
      // the SAME anti-clobber guard the other writers use: a batch of replace-or-append can only grow
      // or hold the row count, so a lower count means the load came back partial -- refuse, never commit.
      if nsp_would_shrink(prefix, rows) == 1 { nsp_refuse_shrink("putmany" as *u8); sys_exit(NSP_EXIT_REFUSED); return NSP_EXIT_REFUSED }
      // ONE plane commit for the whole batch (this is the 88->1 fsync collapse).
      if sts_seed(prefix, neu, no) < 0 { nsp_werr("plane commit error\n" as *u8); sys_exit(NSP_EXIT_IO); return NSP_EXIT_IO }
      // per-row hist AFTER the single commit (matches putn's ordering). Each is an O(1) fast append;
      // a hist failure is reported per id but does not unwind the committed plane (same as putn).
      // ONE hist commit for the whole batch (2026-08-19): the per-row sts_append_fast loop was the
      // surviving fsync convoy (~15 min for 163 rows, measured in btrfs_log_inode_parent). Rows are
      // built with the SAME fields nsp_hist_append writes, then committed once.
      let hp2: *u8 = sys_mmap(NSP_PFXCAP)
      nsp_hist_prefix(prefix, hp2)
      let hrows: *u8 = sys_mmap(NSP_CAP)
      let hoffs: *i64 = sys_mmap(8 * NSP_BATCH_MAX) as *i64
      let hlens: *i64 = sys_mmap(8 * NSP_BATCH_MAX) as *i64
      let nowts: i64 = sys_now_realtime_sec()
      var ho: i64 = 0
      var hk: i64 = 0
      while hk < nin {
          hoffs[hk] = ho
          ho = ss_catn(hrows, ho, nowts)
          hrows[ho] = NSP_TAB as u8; ho = ho + 1
          ho = ss_cat(hrows, ho, actorm)
          hrows[ho] = NSP_TAB as u8; ho = ho + 1
          ho = ss_cat(hrows, ho, "putmany" as *u8)
          hrows[ho] = NSP_TAB as u8; ho = ho + 1
          ho = nsp_cat_slice(hrows, ho, recs, ida[hk], idb[hk])
          hrows[ho] = NSP_TAB as u8; ho = ho + 1
          if oldb[hk] > olda[hk] { ho = nsp_cat_slice_pipe(hrows, ho, cur, olda[hk], oldb[hk]) } else { ho = ss_cat(hrows, ho, "-" as *u8) }
          hrows[ho] = NSP_TAB as u8; ho = ho + 1
          if nb[hk] > na[hk] { ho = nsp_cat_slice_pipe(hrows, ho, neu, na[hk], nb[hk]) } else { ho = ss_cat(hrows, ho, "-" as *u8) }
          hlens[hk] = ho - hoffs[hk]
          hk = hk + 1
      }
      if nsp_hist_commit_rows(hp2, hrows, hoffs, hlens, nin) < 0 { nsp_werr("putmany: hist batch commit failed (plane committed; audit trail missing this batch)\n" as *u8) }
      nsp_report("PUTMANY" as *u8, actorm, nin, rows)
      sys_exit(0)
      return 0
  }
809
  nsp_werr("usage: nx_store_put <prefix> {put <actor> <id> <field>... | putmany <actor> <recordsfile> | setcol <actor> <id> <colidx> <value> | close <actor> <id> <note> | load | hist}\n" as *u8)
  sys_exit(NSP_EXIT_USAGE)
  return NSP_EXIT_USAGE
813}
nx_store_put.nx source

`nx_store_put.nx` source
