code wiki / (root) / nx_store_put_load_file_t196.nx
`nx_store_put_load_file_t196.nx` source

↩ module page · 858 lines · 48879 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"
29import "nx_fs_create_lib.nx"
30const NSP_MAGIC_4096: i64 = 4096
31
32// 2026-08-08 RAISED 1048576 -> 33554432. THE FIFTH HAND-COPY OF THIS CAP, AND THE ONE NOBODY RAISED:
33// nx_debt DB_CAP, nx_debtlive DL_CAP, nx_debtcluster and nx_debt_view were all moved to 33554432 on
34// 2026-08-06 in lockstep; nx_dora's DCAP carries the measurement and the reason -- "knowledge/store/debt-
35// loads to 4,396,892B. At the old 1 MiB this organ silently lost 76% of the board, and because AN
36// APPEND-ONLY PLANE PAST A PREFIX CAP LOSES ITS NEWEST ROWS FIRST it was reporting on the OLDEST
37// quarter -- degrading exactly as new work arrived." This organ was missed. Measured today: that plane
38// is 4,974,565B / 3,785 rows.
39// TWO LIVE CONSEQUENCES, both from this one constant:
40//  (1) the `load` verb returned a silent 1 MiB PREFIX -- that is what fed nx_debtmine a 21.3% read it
41//      reported as complete coverage (debt 1786235467).
42//  (2) put/close/setcol were EFFECTIVELY DEAD on any plane over 1 MiB. They rebuild the whole store from
43//      the rows they loaded, so on a 4.9 MB plane `neu` holds ~1 MiB of rows and nsp_would_shrink
44//      correctly REFUSES every commit. NO DATA WAS EVER AT RISK -- that guard is the reason -- but the
45//      unified write verb could not write to the estate's largest planes at all.
46// ***THE GUARD IS WHY THIS WAS AN OUTAGE AND NOT A CATASTROPHE.*** nsp_would_shrink turns a truncated
47// read into a refusal instead of a whole-store rewrite that silently drops the newest 2,900 rows. Do not
48// weaken it when raising this cap: a larger buffer moves the cliff, the guard removes the precipice.
49// Re-verify the sizing with the estate's own ruler: nx_planefit knowledge/store/debt- 33554432
50const NSP_CAP: i64 = 33554432
51const NSP_NL: i64 = 10
52const NSP_TAB: i64 = 9
53const NSP_PIPE: i64 = 124
54const NSP_DASH: i64 = 45
55const NSP_NCOL: i64 = 7
56const NSP_PAIR: i64 = 2
57const NSP_SP_BYTES: i64 = 128
58const NSP_STDERR: i64 = 2
59const NSP_PFXCAP: i64 = 256
60const NSP_MSGCAP: i64 = 256
61const NSP_ARGC_PUT_MIN: i64 = 6  // prog prefix put actor id + at least 1 more field
62const NSP_ARGC_PUT_MAX: i64 = 20 // id + up to 15 fields (16-col plane cap)
63const NSP_ARGC_CLOSE: i64 = 6    // prog prefix close actor id note
64const NSP_ARGC_SETCOL: i64 = 7   // prog prefix setcol actor id colidx value
65const NSP_ARGC_MIN: i64 = 3
66const NSP_MAXCOL: i64 = 16       // max columns a plane row may carry
67const NSP_SP2_BYTES: i64 = 256   // 16 cols * 2 slots * 8B
68const NSP_EXIT_USAGE: i64 = 2
69const NSP_EXIT_IO: i64 = 1
70const NSP_EXIT_REFUSED: i64 = 4
71// putmany (2026-08-18): file-fed batch. NSP_BATCH_MAX bounds the per-call row count so the range
72// arrays are sized once; a records file with more lines is REFUSED (never silently truncated). The
73// records file is read whole via sys_read_file (lseek-END sized, cannot short-read), so it carries no
74// cap of its own. 4096 rows is ~46x the 88-contract comparewatch queue that motivated it -- headroom
75// named, not a guess, and the refusal makes an over-large batch a loud error rather than lost rows.
76const NSP_HASH: i64 = 35         // '#': comment lines in a records file are skipped, like a staging buffer
77const NSP_BATCH_MAX: i64 = 4096
78const NSP_F_RECFILE: i64 = 4   // putmany argv: prog prefix putmany actor recfile
79const NSP_ARGC_PUTMANY: i64 = 5
80// put field argv indices:
81const NSP_F_ACTOR: i64 = 3
82const NSP_F_ID: i64 = 4
83const NSP_F_LAST: i64 = 10
84// close argv indices:
85const NSP_C_ID: i64 = 4
86const NSP_C_NOTE: i64 = 5
87// status column index in the 7-col row:
88const NSP_COL_STATUS: i64 = 3
89const NSP_COL_NOTE: i64 = 6
90
91func nsp_slen(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
92func nsp_werr(s: *u8) -> i64 { sys_write(NSP_STDERR, s, nsp_slen(s)); return 0 }
93func 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
98}
99// slice [a,b) of q equals C-string s
100// two byte-slices equal? (putmany matches an input-record id slice against a loaded-plane id slice)
101func 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
106}
107func 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
113}
114// split line [ls,le) into up to NSP_MAXCOL (start,end) pairs (callers pass NSP_SP2_BYTES scratch)
115func 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
129}
130func 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
135}
136// copy slice with inner tabs -> '|' (hist rows stay 6-col clean)
137func 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
148}
149// derive the hist prefix: "knowledge/store/debt-" -> "knowledge/store/debthist-"
150func 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
158}
159// append one hist revision row: epoch actor verb id old new
160func 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
199}
200// span-atoi for the q:n value slice (ss_get hands back ptr+len, not a NUL string)
201func 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
206}
207// BATCHED HIST COMMIT (2026-08-19, the putmany residual). putmany collapsed the MAIN plane to one
208// commit and then appended its audit rows ONE AT A TIME: N x sts_append_fast_locked = N x (lock +
209// q:n read + segment commit + manifest rename + syncdir) on the hist plane -- MEASURED as the
210// surviving slow leg: a 163-row batch spent ~15 min in btrfs_log_inode_parent doing exactly this
211// while holding the MAIN plane lock, which is the whole convoy shrunk by one layer, not cured.
212// This is sts_append_fast generalized to k rows: ONE lock, ONE q:n read, k adds, ONE commit.
213// Same CAS discipline (gen0 before the read, refuse-on-stale under the lock); same bootstrap
214// (no q:n yet -> full seed of just these rows); lock order stays main-then-hist.
215// rows: concatenated row TEXTS (no trailing newlines); offs/lens: per-row slices; returns 0 or -1.
216func 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
259}
260func 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
269}
270
271func 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
284}
285
286// SHARED ANTI-CLOBBER TOOTH (2026-08-02, debts 1785713047/1785713373) -- ONE guard, FOUR callers.
287// EVERY write verb here rebuilds the WHOLE plane from what sts_load could reach, then re-seeds it. If that
288// load came back PARTIAL, every row it could not see is silently DELETED. put/putn ADD rows; close/setcol
289// MODIFY one row in place -- so NONE of them can ever legitimately REDUCE the row count. A count below what
290// the plane already declares therefore means the load was short, and committing it would destroy data
291// (commontask- 37->10, dedupq- 21->3, both recovered 2026-08-02). Refuse instead of committing the loss.
292// FAIL-OPEN when q:n is absent -- that call is BOOTSTRAPPING the plane, which is legitimate.
293// ★Deliberately NOT in the shared sts_seed primitive: there it would refuse nx_plane_append's ROLLBACK and
294// nx_plane_repair, both of which shrink BY DEFINITION (1785713047). The guard belongs to the WRITER.
295// Proven both directions by nx_store_put_shrink_gate 4/4: it BITES a manufactured over-declared plane
296// (rc=4, store untouched) and SPARES a healthy one (rc=0, q:n 3->4).
297func 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
305}
306func 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
311}
312
313
314// Retained full-plane export: data bypasses the bounded MCP stdout channel.
315// Exclusive publication and identical-byte retries preserve existing artifacts.
316func nsp_export_equal(path:*u8,b:*u8,n:i64)->i64{
let fd:i64=sys_openat_rd(path);if fd<0{return -1}
let size:i64=sys_lseek(fd,0,2);if size!=n{sys_close(fd);return 0}
if sys_lseek(fd,0,0)<0{sys_close(fd);return -1}
let chunk:*u8=sts_mm(FSX_DENT_BUF);var off:i64=0
while off<n{var take:i64=n-off;if take>FSX_DENT_BUF{take=FSX_DENT_BUF}
let got:i64=sys_read(fd,chunk,take);if got<=0{sys_close(fd);sys_munmap(chunk,FSX_DENT_BUF);return -1}
var i:i64=0;while i<got{if chunk[i]!=b[off+i]{sys_close(fd);sys_munmap(chunk,FSX_DENT_BUF);return 0};i=i+1};off=off+got
};sys_close(fd);sys_munmap(chunk,FSX_DENT_BUF);return 1
325}
326func nsp_export_sync_parent(path:*u8)->i64{
let n:i64=nsp_slen(path);var last:i64=-1;var i:i64=0
while i<n{if path[i]==(47 as u8){last=i};i=i+1}
let parent:*u8=sts_mm(n+2);if last<0{parent[0]=46 as u8;parent[1]=0 as u8}else{if last==0{parent[0]=47 as u8;parent[1]=0 as u8}else{i=0;while i<last{parent[i]=path[i];i=i+1};parent[last]=0 as u8}}
let fd:i64=sys_openat_rd(parent);sys_munmap(parent,n+2);if fd<0{return -1};let rc:i64=sys_fsync(fd);sys_close(fd);return rc
331}
332func nsp_export_coverage(prefix:*u8)->i64{
let sp:*i64=sts_mm(16) as *i64;let count:i64=ss_manifest_dyn(prefix,sp);if count<1{return -1}
let names:*i64=sp[0] as *i64;let pn:i64=nsp_slen(prefix);var i:i64=0
while i<count{let name:*u8=names[i] as *u8;let nn:i64=nsp_slen(name);let path:*u8=sts_mm(pn+nn+6)
var p:i64=ss_cat(path,0,prefix);p=ss_cat(path,p,name);p=ss_cat(path,p,".docs");path[p]=0 as u8
let fd:i64=sys_openat_rd(path);sys_munmap(path,pn+nn+6);if fd<0{return -1};let size:i64=sys_lseek(fd,0,2);sys_close(fd);if size<0{return -1};i=i+1
};return count
339}
340func nsp_load_file(prefix:*u8,path:*u8)->i64{
if fsx_denied(prefix)==1||fsx_write_denied(path)==1{nsp_werr("LOAD-FILE REFUSED reason=path-policy\n");return NSP_EXIT_REFUSED}
let segments:i64=nsp_export_coverage(prefix);if segments<1{nsp_werr("LOAD-FILE REFUSED reason=segment-coverage-unproven\n");return NSP_EXIT_IO}
let ln:*i64=sts_mm(16) as *i64;let b:*u8=sts_load_fit(prefix,ln);let n:i64=ln[0]
if (b as i64)==0||n<1{nsp_werr("LOAD-FILE REFUSED reason=empty-or-incomplete-row-plane\n");return NSP_EXIT_IO}
let honest:*u8=sts_mm(n+1);let flags:*i64=sts_mm(24) as *i64;let hn:i64=sts_load_honest(prefix,honest,n+1,flags)
if hn!=n{nsp_werr("LOAD-FILE REFUSED reason=initialized-span-disagreement\n");return NSP_EXIT_REFUSED}
var rows:i64=0;var i:i64=0;var same:i64=1;while i<n{if b[i]==(10 as u8){rows=rows+1};if honest[i]!=b[i]{same=0};i=i+1}
if hn!=n||same!=1||flags[0]!=rows||flags[1]!=rows||flags[2]!=0{nsp_werr("LOAD-FILE REFUSED reason=row-coverage-disagreement\n");return NSP_EXIT_REFUSED}
let rc:i64=fxc_create(path,b,n);var changed:i64=1
if rc==FXC_EXISTS{changed=0}else{if rc!=n{nsp_werr("LOAD-FILE REFUSED reason=exclusive-write-or-resource-failure\n");return NSP_EXIT_IO}}
if nsp_export_equal(path,b,n)!=1{nsp_werr("LOAD-FILE REFUSED reason=existing-artifact-differs-or-readback-failed\n");return NSP_EXIT_REFUSED}
if nsp_export_sync_parent(path)<0{nsp_werr("LOAD-FILE INCOMPLETE reason=parent-fsync-failed artifact-may-exist=1\n");return NSP_EXIT_IO}
fsx_puts("LOAD-FILE OK bytes=");fsx_putn(n);fsx_puts(" rows=");fsx_putn(rows);fsx_puts(" declared_rows=");fsx_putn(flags[0]);fsx_puts(" loaded_rows=");fsx_putn(flags[1]);fsx_puts(" beyond_declared_observed=");fsx_putn(flags[2]);fsx_puts(" segments_readable=");fsx_putn(segments);fsx_puts(" changed=");fsx_putn(changed);fsx_puts(" coverage=declared-live-manifest archive_scanned=0 beyond_probe_window=");fsx_putn(STS_PROBE_WINDOW);fsx_puts(" beyond_probe_miss_run=");fsx_putn(STS_PROBE_MISS_RUN);fsx_puts(" path=");fsx_puts(path);fsx_puts("\n");return 0
354}
355
356func 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
360
  // 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.
  if nsp_eqs("load-file",verb)==1 { if argc!=4{return NSP_EXIT_USAGE};let lock:i64=sts_lock(prefix);if lock<0{return NSP_EXIT_IO};return nsp_load_file(prefix,argv[3] as *u8) }
  sts_lock(prefix)
375
  // ---- 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
  }
437
  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
  }
511
  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
  }
589
  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
  }
662
  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
  }
733
  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
  }
854
  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
858}
nx_store_put_load_file_t196.nx source

`nx_store_put_load_file_t196.nx` source
