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

↩ module page · 990 lines · 47176 B
1// nx_mvault_walk.nx -- the LIVE reindex DRIVER, VW02 (MV-1 bulk-migration engine).
2// VW01 was a single-level, full-read walk: it SKIPPED >512MiB files (seq119 OOM
3// guard) -- i.e. most cam recordings -- and its one-reg_put-per-item store path
4// would have both crawled (a new segment per put) and corrupted the ids index
5// past ~15k items (fixed 1MiB index buffer). VW02 is the measured rebuild:
6//   * RECURSIVE: iterative BFS dir queue, ONE shared getdents buffer.
7//   * STREAMING CID: every file hashes in chunks (the head window doubles as
8//     the sniff bytes) -- nothing is skipped for size, RAM is O(1) per file.
9//   * IN-MEMORY DEDUP SETS seeded once from mv:ids / p:ids -- zero per-file
10//     store reads; BATCHED commits (one segment per ~24MiB of records, not one
11//     per item) -- the seg-store quadratic + index-rewrite blowup is dodged.
12//   * PATH-RESUME PLANE knowledge/store/mvaultpath-: key p:<cid of "path|size">,
13//     val = content CID. Re-runs skip already-hashed unchanged paths (daily DVR
14//     ingest = incremental by design). Size-keyed, not mtime-keyed: recordings
15//     are append-once; an in-place same-size rewrite would be missed (declared).
16//   * BG MODE (`bg <logfile>`): forks a detached worker; fds 0/1/2 close so the
17//     tools-daemon capture pipe EOFs immediately; progress + final JSON append
18//     to <logfile>. TB-scale hashing runs for hours -- a synchronous MCP call
19//     drops its response at ~15s.
20//   * POLICY FILTERS (counted, never silent): dot-names skipped (.trickplay
21//     sidecar dirs), *-poster.jpg sidecars skipped, 0-byte files skipped
22//     (hard-crash torn artifacts must not become vault records).
23//   * selftest: KAT-anchored teeth, emits verdict=GREEN|RED (D001 anchor).
24// RECORD-ONLY = NON-MUTATING BY CONSTRUCTION (unchanged): never moves a byte;
25// only the seg-store index planes are written (and only on commit).
26// ENVELOPE (declared): set ~950k cids / queue 16MiB dirs / ids blob 64MiB --
27// overflow sets fatal=1 in the output, never a silent cap.
28//
29//   nx_mvault_walk <dir> <gen|real> <source> [commit|dry] [bg <logfile>]
30//   nx_mvault_walk selftest
31// license_tier: ORIGINAL   No hw writes (Rule 26).
32import "nx_syscalls.nx"
33import "nx_sha256.nx"
34import "nx_canon_cid.nx"
35import "_hdl_build/nx_proc_ctl.nx"   // proc_count_by_name -- the single-walker guard
36import "nx_media_pool.nx"
37import "nx_media_sniff.nx"
38import "nx_mvault_ingest.nx"
39import "nx_mvault_context.nx"
40import "nx_mvault_record.nx"
41import "nx_mvault_layout.nx"
42import "nx_mvault_reclaim.nx"
43import "nx_mvault_reindex.nx"
44import "nx_seg_store.nx"
45const MW_MAGIC_4096: i64 = 4096
46const MW_MAGIC_8192: i64 = 8192
47const MW_MAGIC_8191: i64 = 8191
48const MW_MAGIC_300000: i64 = 300000
49const MW_MAGIC_1024: i64 = 1024
50
51const MW_DIRBUF: i64 = 1048576      // 1 MiB getdents buffer (shared; BFS reuses it)
52const MW_PATH: i64 = 4096
53const MW_LARGE: i64 = 104857600     // 100 MiB -> "large file surfaced for value review"
54const MW_HEAD: i64 = 1048576        // head window: sniff bytes + first hash read
55const MW_CHUNK: i64 = 4194304       // 4 MiB streaming read chunk
56const MW_QCAP: i64 = 16777216       // dir-queue arena (~280k dirs)
57const MW_SETSLOTS: i64 = 1048576    // dedup hash slots (pow2; fills at 75%)
58const MW_SETARENA: i64 = 67108864   // 64 MiB cid arena (~950k cids)
59const MW_IDSCAP: i64 = 67108864     // ids-index blob cap (rewritten per flush)
60const MW_WCAP: i64 = 33554432       // 32 MiB batch writer buffer
61const MW_FLUSH: i64 = 25165824      // flush a batch past 24 MiB of records
62const MW_PROG: i64 = 256            // progress line every N hashed files (bg)
63const MW_OLDMAX: i64 = 536870912    // VW01's skip threshold -- now just a counter
64
65func mw_seq(a: *u8, b: *u8) -> i64 {
  var i: i64 = 0
  while 1 == 1 {
      if a[i] != b[i] { return 0 }
      if a[i] == (0 as u8) { return 1 }
      i = i + 1
  }
  return 0
73}
74
75// build "<dir>/<name>\0" into dst
76func mw_join(dst: *u8, dir: *u8, name: *u8) -> i64 {
  var o: i64 = 0
  var i: i64 = 0
  while dir[i] != (0 as u8) { dst[o] = dir[i]; o = o + 1; i = i + 1 }
  dst[o] = 47 as u8
  o = o + 1
  i = 0
  while name[i] != (0 as u8) { dst[o] = name[i]; o = o + 1; i = i + 1 }
  dst[o] = 0 as u8
  return o
86}
87
88// extension after the LAST '.'; "bin" if none
89func mw_ext(name: *u8, ext: *u8) -> i64 {
  var last: i64 = 0 - 1
  var i: i64 = 0
  while name[i] != (0 as u8) {
      if name[i] == (46 as u8) { last = i }
      i = i + 1
  }
  if last < 0 {
      ext[0] = 98 as u8
      ext[1] = 105 as u8
      ext[2] = 110 as u8
      ext[3] = 0 as u8
      return 3
  }
  var j: i64 = last + 1
  var o: i64 = 0
  while name[j] != (0 as u8) { ext[o] = name[j]; o = o + 1; j = j + 1 }
  ext[o] = 0 as u8
  return o
108}
109
110func mw_ch(dst: *u8, off: i64, c: i64) -> i64 { dst[off] = c as u8; return off + 1 }
111func mw_cat(dst: *u8, off: i64, s: *u8) -> i64 {
  var o: i64 = off
  var i: i64 = 0
  while s[i] != (0 as u8) { dst[o] = s[i]; o = o + 1; i = i + 1 }
  return o
116}
117// zero-alloc decimal (VW01's mmap'd a page per call -- per-file callers leak GBs)
118func mw_udec(dst: *u8, off: i64, v: i64) -> i64 {
  if v == 0 { dst[off] = 48 as u8; return off + 1 }
  var div: i64 = 1
  var m: i64 = v
  while m >= 10 { div = div * 10; m = m / 10 }
  var o: i64 = off
  var rest: i64 = v
  while div > 0 {
      let d: i64 = rest / div
      dst[o] = (48 + d) as u8
      rest = rest - d * div
      div = div / 10
      o = o + 1
  }
  return o
133}
134
135// name ends with "-poster.jpg" (cam-capture sidecar, not source media)
136func mw_is_poster(name: *u8) -> i64 {
  var n: i64 = 0
  while name[n] != (0 as u8) { n = n + 1 }
  if n < 11 { return 0 }
  let suf: *u8 = "-poster.jpg" as *u8
  var i: i64 = 0
  while i < 11 {
      if name[n - 11 + i] != suf[i] { return 0 }
      i = i + 1
  }
  return 1
147}
148
149// ---- open-addressing cid set: h[0]=table (slot=arena off+1), h[1]=slots,
150// h[2]=items, h[3]=arena, h[4]=arena used, h[5]=arena cap ----
151func mws_new() -> *i64 {
  let h: *i64 = sys_mmap(64) as *i64
  h[0] = (sys_mmap(8 * MW_SETSLOTS)) as i64
  h[1] = MW_SETSLOTS
  h[2] = 0
  h[3] = (sys_mmap(MW_SETARENA)) as i64
  h[4] = 0
  h[5] = MW_SETARENA
  return h
160}
161// first 16 hex chars after "nxc1-" -> i64 (sha256 is uniform)
162func mws_hash(cid: *u8) -> i64 {
  var v: i64 = 0
  var i: i64 = 5
  while i < 21 {
      let c: i64 = cid[i]
      var d: i64 = c - 48
      if c >= 97 { d = c - 87 }
      v = (v << 4) | (d & 15)
      i = i + 1
  }
  return v
173}
174func mws_eq(a: *u8, b: *u8) -> i64 {
  var i: i64 = 0
  while i < 69 { if a[i] != b[i] { return 0 } i = i + 1 }
  return 1
178}
179func mws_has(h: *i64, cid: *u8) -> i64 {
  let tab: *i64 = h[0] as *i64
  let mask: i64 = h[1] - 1
  var s: i64 = mws_hash(cid) & mask
  while 1 == 1 {
      let v: i64 = tab[s]
      if v == 0 { return 0 }
      let ap: *u8 = (h[3] + (v - 1)) as *u8
      if mws_eq(ap, cid) == 1 { return 1 }
      s = (s + 1) & mask
  }
  return 0
191}
192// 0 added, 1 was-present, -1 table/arena full (caller FAILS LOUD)
193func mws_add(h: *i64, cid: *u8) -> i64 {
  if mws_has(h, cid) == 1 { return 1 }
  if h[2] * 4 >= h[1] * 3 { return 0 - 1 }
  if h[4] + 70 > h[5] { return 0 - 1 }
  let tab: *i64 = h[0] as *i64
  let mask: i64 = h[1] - 1
  var s: i64 = mws_hash(cid) & mask
  var done: i64 = 0
  while done == 0 {
      if tab[s] == 0 { done = 1 } else { s = (s + 1) & mask }
  }
  let off: i64 = h[4]
  let ap: *u8 = (h[3] + off) as *u8
  var i: i64 = 0
  while i < 69 { ap[i] = cid[i]; i = i + 1 }
  ap[69] = 0 as u8
  h[4] = off + 70
  tab[s] = off + 1
  h[2] = h[2] + 1
  return 0
213}
214
215// seed a set (and optionally the run-level ids blob) from <prefix><idxkey>;
216// returns cid count, -1 on set overflow.
217func mw_load_ids(prefix: *u8, idxkey: *u8, h: *i64, blob: *u8, bloblen: *i64) -> i64 {
  let ipo: *i64 = sys_mmap(16) as *i64
  let ilo: *i64 = sys_mmap(16) as *i64
  var n: i64 = 0
  if ss_get(prefix, idxkey, ipo, ilo) >= 0 {
      let src: *u8 = ipo[0] as *u8
      let sl: i64 = ilo[0]
      if (blob as i64) != 0 {
          var c: i64 = 0
          while c < sl { blob[c] = src[c]; c = c + 1 }
          bloblen[0] = sl
      }
      var ls: i64 = 0
      var i: i64 = 0
      while i <= sl {
          var eol: i64 = 0
          if i == sl { eol = 1 } else { if src[i] == (10 as u8) { eol = 1 } }
          if eol == 1 {
              if i - ls == 69 {
                  if mws_add(h, (src as i64 + ls) as *u8) < 0 { return 0 - 1 }
                  n = n + 1
              }
              ls = i + 1
          }
          i = i + 1
      }
  }
  return n
245}
246
247// free the per-hash scratch sha256_init mmaps (per-file callers leak GBs otherwise)
248func mw_sha_free(c: *Sha256) -> i64 {
  sys_munmap(c.bufptr as *u8, 64)
  sys_munmap(c.kptr as *u8, 512)
  sys_munmap(c.wptr as *u8, 512)
  sys_munmap(c.k32ptr as *u8, 256)
  sys_munmap(c.st8ptr as *u8, 32)
  return 0
255}
256
257// stream-hash one file in ONE pass: head lands in headbuf (the sniff window),
258// the remainder streams through chunkbuf. O(1) memory, scratch freed.
259// out: cid, szbox[0]=total bytes, hnbox[0]=head bytes. 0 ok, -1 error.
260func mw_hash_file(path: *u8, ctx: *Sha256, headbuf: *u8, chunkbuf: *u8,
                dg: *u8, cid: *u8, szbox: *i64, hnbox: *i64) -> i64 {
  let fd: i64 = sys_openat_rd(path)
  if fd < 0 { return 0 - 1 }
  sha256_init(ctx)
  var bad: i64 = 0
  var hn: i64 = 0
  var done: i64 = 0
  while done == 0 {
      let n: i64 = sys_read(fd, (headbuf as i64 + hn) as *u8, MW_HEAD - hn)
      if n < 0 { bad = 1; done = 1 }
      if n == 0 { done = 1 }
      if n > 0 {
          hn = hn + n
          if hn >= MW_HEAD { done = 1 }
      }
  }
  var total: i64 = hn
  if bad == 0 {
      if hn > 0 { sha256_update(ctx, headbuf, hn) }
      if hn >= MW_HEAD {
          var more: i64 = 1
          while more == 1 {
              let n2: i64 = sys_read(fd, chunkbuf, MW_CHUNK)
              if n2 < 0 { bad = 1; more = 0 }
              if n2 == 0 { more = 0 }
              if n2 > 0 { sha256_update(ctx, chunkbuf, n2); total = total + n2 }
          }
      }
  }
  sys_close(fd)
  var r: i64 = 0 - 1
  if bad == 0 {
      sha256_final(ctx, dg)
      cid_from_digest(dg, cid)
      szbox[0] = total
      hnbox[0] = hn
      r = 0
  }
  mw_sha_free(ctx)
  return r
301}
302
303// cid of the resume key "<path>|<size>" (a hash key, not content). zero-alloc.
304func mw_pathkey(path: *u8, sz: i64, ctx: *Sha256, pkbuf: *u8, dg: *u8, pkcid: *u8) -> i64 {
  var o: i64 = 0
  var i: i64 = 0
  while path[i] != (0 as u8) { pkbuf[o] = path[i]; o = o + 1; i = i + 1 }
  pkbuf[o] = 124 as u8
  o = o + 1
  o = mw_udec(pkbuf, o, sz)
  sha256_init(ctx)
  sha256_update(ctx, pkbuf, o)
  sha256_final(ctx, dg)
  cid_from_digest(dg, pkcid)
  mw_sha_free(ctx)
  return o
317}
318
319// append one id line to a run-level ids blob
320func mw_ids_append(idsbuf: *u8, lenbox: *i64, cid: *u8) -> i64 {
  var o: i64 = lenbox[0]
  if o + 70 > MW_IDSCAP { return 0 - 1 }
  var i: i64 = 0
  while i < 69 { idsbuf[o] = cid[i]; o = o + 1; i = i + 1 }
  idsbuf[o] = 10 as u8
  lenbox[0] = o + 1
  return 0
328}
329
330// add a "<kp><cid>" record to the batch writer
331func mw_add_rec(wbox: *i64, kp: *u8, cid: *u8, val: *u8, vlen: i64, kbuf: *u8) -> i64 {
  var o: i64 = mw_cat(kbuf, 0, kp)
  o = mw_cat(kbuf, o, cid)
  kbuf[o] = 0 as u8
  let w: *i64 = wbox[0] as *i64
  return ss_add(w, 1, kbuf, val, vlen)
337}
338
339// commit the current batch (records + the complete ids index in the SAME
340// segment -- readers stay atomic) and start a fresh writer. 1=committed,
341// 0=nothing pending, -1=error.
342func mw_flush_plane(prefix: *u8, wbox: *i64, idxkey: *u8, idsbuf: *u8, idslen: i64) -> i64 {
  let w: *i64 = wbox[0] as *i64
  if w[1] == 0 { return 0 }
  if ss_add(w, 1, idxkey, idsbuf, idslen) < 0 { return 0 - 1 }
  // CANONICAL writer-side id (uncapped max+1): count-as-segid re-clobbers
  // after compaction or past any cap (the proven toolreg seg-256 class)
  let segid: i64 = ss_next_segid(prefix)
  if segid < 0 { return 0 - 1 }
  let rc: i64 = ss_commit(prefix, w, segid)
  sys_munmap(w[0] as *u8, w[2])
  sys_munmap(w as *u8, 32)
  wbox[0] = (ss_begin_cap(MW_WCAP)) as i64
  if rc < 0 { return 0 - 1 }
  return 1
356}
357
358func mw_prog(logfd: i64, st: *i64, buf: *u8) -> i64 {
  if logfd < 0 { return 0 }
  var o: i64 = 0
  o = mw_cat(buf, o, "PROG files=" as *u8); o = mw_udec(buf, o, st[0])
  o = mw_cat(buf, o, " seen=" as *u8); o = mw_udec(buf, o, st[1])
  o = mw_cat(buf, o, " already=" as *u8); o = mw_udec(buf, o, st[2])
  o = mw_cat(buf, o, " dirs=" as *u8); o = mw_udec(buf, o, st[5])
  o = mw_cat(buf, o, " commits=" as *u8); o = mw_udec(buf, o, st[7])
  o = mw_cat(buf, o, " rerr=" as *u8); o = mw_udec(buf, o, st[4])
  o = mw_ch(buf, o, 10)
  sys_write(logfd, buf, o)
  return 0
370}
371
372func mw_fail(out: *u8, msg: *u8) -> i64 {
  var o: i64 = 0
  o = mw_cat(out, o, "{\"error\":\"" as *u8)
  o = mw_cat(out, o, msg)
  o = mw_cat(out, o, "\"}" as *u8)
  o = mw_ch(out, o, 10)
  return o
379}
380
381func mw_emit(out: *u8, st: *i64, commit: i64) -> i64 {
  var o: i64 = 0
  o = mw_ch(out, o, 123)
  o = mw_cat(out, o, "\"envelope\":\"recursive-walk record-only non-mutating; streaming-cid; path-resume; batch-commit\"" as *u8)
  o = mw_ch(out, o, 44)
  o = mw_cat(out, o, "\"files\":" as *u8); o = mw_udec(out, o, st[0]); o = mw_ch(out, o, 44)
  o = mw_cat(out, o, "\"seen\":" as *u8); o = mw_udec(out, o, st[1]); o = mw_ch(out, o, 44)
  o = mw_cat(out, o, "\"unique\":" as *u8); o = mw_udec(out, o, st[11]); o = mw_ch(out, o, 44)
  o = mw_cat(out, o, "\"dup\":" as *u8); o = mw_udec(out, o, st[12]); o = mw_ch(out, o, 44)
  o = mw_cat(out, o, "\"reclaimable_bytes\":" as *u8); o = mw_udec(out, o, st[13]); o = mw_ch(out, o, 44)
  o = mw_cat(out, o, "\"already\":" as *u8); o = mw_udec(out, o, st[2]); o = mw_ch(out, o, 44)
  o = mw_cat(out, o, "\"large_streamed\":" as *u8); o = mw_udec(out, o, st[3]); o = mw_ch(out, o, 44)
  o = mw_cat(out, o, "\"read_errs\":" as *u8); o = mw_udec(out, o, st[4]); o = mw_ch(out, o, 44)
  o = mw_cat(out, o, "\"dirs\":" as *u8); o = mw_udec(out, o, st[5]); o = mw_ch(out, o, 44)
  o = mw_cat(out, o, "\"dirs_dropped\":" as *u8); o = mw_udec(out, o, st[6]); o = mw_ch(out, o, 44)
  o = mw_cat(out, o, "\"sidecars_skipped\":" as *u8); o = mw_udec(out, o, st[8]); o = mw_ch(out, o, 44)
  o = mw_cat(out, o, "\"empty_skipped\":" as *u8); o = mw_udec(out, o, st[9]); o = mw_ch(out, o, 44)
  o = mw_cat(out, o, "\"commits\":" as *u8); o = mw_udec(out, o, st[7]); o = mw_ch(out, o, 44)
  o = mw_cat(out, o, "\"skipped_large\":0" as *u8); o = mw_ch(out, o, 44)
  o = mw_cat(out, o, "\"fatal\":" as *u8); o = mw_udec(out, o, st[14]); o = mw_ch(out, o, 44)
  o = mw_cat(out, o, "\"commit\":" as *u8); o = mw_udec(out, o, commit)
  o = mw_ch(out, o, 125)
  o = mw_ch(out, o, 10)
  return o
405}
406
407// The full run: BFS from root, per-file streaming pipeline, batched commits.
408// Emits the result JSON into out (returns length); fills st (16 slots).
409// Wall-clock heartbeat interval. PROG ticks every MW_PROG NEWLY-PROCESSED files, which structurally CANNOT
410// distinguish "slow" from "stuck": a resumed run path-skips tens of thousands of entries emitting nothing, and
411// a single multi-GB recording stream-hashes for minutes emitting nothing. Both look identical to a dead worker.
412// A wall-clock tick proves LIVENESS independently of how much work has completed -- count seconds, not work.
413const MW_HB_SECS: i64 = 30
414
415func mw_run(root: *u8, prov: i64, source: *u8, commit: i64,
          mvprefix: *u8, ppprefix: *u8, dest_root: *u8,
          logfd: i64, out: *u8, st: *i64) -> i64 {
  let class_code: i64 = mv_class(prov)
  var hb_last: i64 = sys_now_realtime_sec()
  var i0: i64 = 0
  while i0 < 16 { st[i0] = 0; i0 = i0 + 1 }
  let ctx: *Sha256 = sys_mmap(256) as *Sha256
  let headbuf: *u8 = sys_mmap(MW_HEAD)
  let chunkbuf: *u8 = sys_mmap(MW_CHUNK)
  let dbuf: *u8 = sys_mmap(MW_DIRBUF)
  let dg: *u8 = sys_mmap(64)
  let ocid: *u8 = sys_mmap(96)
  let pkcid: *u8 = sys_mmap(96)
  let path: *u8 = sys_mmap(MW_PATH)
  let pkbuf: *u8 = sys_mmap(MW_PATH + 64)
  let ext: *u8 = sys_mmap(256)
  let opath: *u8 = sys_mmap(MW_PATH)
  let rec: *u8 = sys_mmap(MW_MAGIC_4096)
  let kbuf: *u8 = sys_mmap(512)
  let szbox: *i64 = sys_mmap(16) as *i64
  let hnbox: *i64 = sys_mmap(16) as *i64
  let progbuf: *u8 = sys_mmap(512)
  let rcl: *i64 = mv_reclaim_new()
  let cidset: *i64 = mws_new()
  let pathset: *i64 = mws_new()
  let ids: *u8 = sys_mmap(MW_IDSCAP)
  let idslen: *i64 = sys_mmap(16) as *i64
  idslen[0] = 0
  let pids: *u8 = sys_mmap(MW_IDSCAP)
  let pidslen: *i64 = sys_mmap(16) as *i64
  pidslen[0] = 0
  if mw_load_ids(mvprefix, "mv:ids" as *u8, cidset, ids, idslen) < 0 { return mw_fail(out, "cid-set overflow at load" as *u8) }
  if mw_load_ids(ppprefix, "p:ids" as *u8, pathset, pids, pidslen) < 0 { return mw_fail(out, "path-set overflow at load" as *u8) }
  let wm: *i64 = sys_mmap(16) as *i64
  wm[0] = (ss_begin_cap(MW_WCAP)) as i64
  let wp: *i64 = sys_mmap(16) as *i64
  wp[0] = (ss_begin_cap(MW_WCAP)) as i64
  let q: *u8 = sys_mmap(MW_QCAP)
  var qh: i64 = 0
  var qt: i64 = 0
  var rl0: i64 = 0
  while root[rl0] != (0 as u8) { q[qt] = root[rl0]; qt = qt + 1; rl0 = rl0 + 1 }
  q[qt] = 0 as u8
  qt = qt + 1
  var fatal: i64 = 0
  while qh < qt {
      let dir: *u8 = (q as i64 + qh) as *u8
      var dl: i64 = 0
      while dir[dl] != (0 as u8) { dl = dl + 1 }
      qh = qh + dl + 1
      st[5] = st[5] + 1
      let dfd: i64 = sys_openat_rd(dir)
      if dfd < 0 { st[4] = st[4] + 1 } else {
          var done: i64 = 0
          while done == 0 {
              if fatal == 1 { done = 1 } else {
                  let nread: i64 = sys_getdents64(dfd, dbuf, MW_DIRBUF)
                  if nread <= 0 { done = 1 } else {
                      var off: i64 = 0
                      while off < nread {
                          let drec: *u8 = (dbuf as i64 + off) as *u8
                          let reclen: i64 = dirent_reclen(drec)
                          let dtype: i64 = dirent_type(drec)
                          let name: *u8 = dirent_name(drec)
                          var skip: i64 = 0
                          if name[0] == (46 as u8) { skip = 1 }
                          if skip == 0 {
                              if dtype == DT_DIR {
                                  let need: i64 = mw_join(path, dir, name)
                                  if qt + need + 2 > MW_QCAP { st[6] = st[6] + 1 } else {
                                      var cq: i64 = 0
                                      while cq <= need { q[qt + cq] = path[cq]; cq = cq + 1 }
                                      qt = qt + need + 1
                                  }
                              }
                              if dtype == DT_REG {
                                  if mw_is_poster(name) == 1 { st[8] = st[8] + 1 } else {
                                      st[0] = st[0] + 1
                                      // LIVENESS TICK -- runs for EVERY directory entry (skipped or processed),
                                      // so a long path-resume run and a long single-file hash both keep proving
                                      // the worker is alive. Deliberately cheap: one clock read per entry.
                                      if logfd >= 0 {
                                          let hb_now: i64 = sys_now_realtime_sec()
                                          if hb_now - hb_last >= MW_HB_SECS {
                                              hb_last = hb_now
                                              var ho: i64 = mw_cat(progbuf, 0, "HB t=" as *u8)
                                              ho = mw_udec(progbuf, ho, hb_now)
                                              ho = mw_cat(progbuf, ho, " files=" as *u8); ho = mw_udec(progbuf, ho, st[0])
                                              ho = mw_cat(progbuf, ho, " seen=" as *u8); ho = mw_udec(progbuf, ho, st[1])
                                              ho = mw_cat(progbuf, ho, " already=" as *u8); ho = mw_udec(progbuf, ho, st[2])
                                              ho = mw_cat(progbuf, ho, " commits=" as *u8); ho = mw_udec(progbuf, ho, st[7])
                                              ho = mw_ch(progbuf, ho, 10)
                                              sys_write(logfd, progbuf, ho)
                                          }
                                      }
                                      mw_join(path, dir, name)
                                      let pfd: i64 = sys_openat_rd(path)
                                      var fsz: i64 = 0 - 1
                                      if pfd >= 0 {
                                          fsz = sys_lseek(pfd, 0, 2)
                                          sys_close(pfd)
                                      }
                                      if fsz < 0 { st[4] = st[4] + 1 } else {
                                          if fsz == 0 { st[9] = st[9] + 1 } else {
                                              mw_pathkey(path, fsz, ctx, pkbuf, dg, pkcid)
                                              if mws_has(pathset, pkcid) == 1 { st[2] = st[2] + 1 } else {
                                                  if mw_hash_file(path, ctx, headbuf, chunkbuf, dg, ocid, szbox, hnbox) < 0 { st[4] = st[4] + 1 } else {
                                                      if fsz > MW_OLDMAX { st[3] = st[3] + 1 }
                                                      mw_ext(name, ext)
                                                      let vtype: i64 = mv_classify_plan(headbuf, hnbox[0], class_code, MV_CTX_NONE, dest_root, source, ext, opath, ocid)
                                                      var isdup: i64 = 0
                                                      if mws_has(cidset, ocid) == 1 { isdup = 1 }
                                                      mv_reclaim_add(rcl, fsz, isdup, MW_LARGE)
                                                      st[1] = st[1] + 1
                                                      if commit == 1 {
                                                          if isdup == 0 {
                                                              let rl: i64 = mv_record_build(rec, class_code, vtype, MV_LVL_TS, source, opath)
                                                              if mw_add_rec(wm, "mv:" as *u8, ocid, rec, rl, kbuf) < 0 { fatal = 1 }
                                                              if mw_ids_append(ids, idslen, ocid) < 0 { fatal = 1 }
                                                          }
                                                          if mw_add_rec(wp, "p:" as *u8, pkcid, ocid, 70, kbuf) < 0 { fatal = 1 }
                                                          if mw_ids_append(pids, pidslen, pkcid) < 0 { fatal = 1 }
                                                          if mws_add(pathset, pkcid) < 0 { fatal = 1 }
                                                      }
                                                      if isdup == 0 { if mws_add(cidset, ocid) < 0 { fatal = 1 } }
                                                      if commit == 1 {
                                                          if fatal == 0 {
                                                              // seq625: also flush per 4096 files -- a multi-day
                                                              // run must advance the resume plane incrementally
                                                              st[15] = st[15] + 1
                                                              var doflush: i64 = 0
                                                              if st[15] >= MW_MAGIC_4096 { doflush = 1 }
                                                              let wmi: *i64 = wm[0] as *i64
                                                              if wmi[1] > MW_FLUSH { doflush = 1 }
                                                              let wpi: *i64 = wp[0] as *i64
                                                              if wpi[1] > MW_FLUSH { doflush = 1 }
                                                              if doflush == 1 {
                                                                  st[15] = 0
                                                                  let f1: i64 = mw_flush_plane(mvprefix, wm, "mv:ids" as *u8, ids, idslen[0])
                                                                  if f1 < 0 { fatal = 1 }
                                                                  if f1 == 1 { st[7] = st[7] + 1 }
                                                                  let f2: i64 = mw_flush_plane(ppprefix, wp, "p:ids" as *u8, pids, pidslen[0])
                                                                  if f2 < 0 { fatal = 1 }
                                                              }
                                                          }
                                                      }
                                                      st[10] = st[10] + 1
                                                      if st[10] >= MW_PROG { st[10] = 0; mw_prog(logfd, st, progbuf) }
                                                  }
                                              }
                                          }
                                      }
                                  }
                              }
                          }
                          if reclen <= 0 { off = nread } else { off = off + reclen }
                      }
                  }
              }
          }
          sys_close(dfd)
      }
      if fatal == 1 { qh = qt }
  }
  // 2026-07-30 DATA-LOSS FIX: the tail flush used to be guarded by
  // `fatal == 0`, so any run that set fatal DISCARDED up to MW_FLUSH (24MiB)
  // of ALREADY-COMPUTED records -- and still printed verdict=DONE. MEASURED
  // on /volume1/ai/images: files=127283 unique=127276 commits=30 fatal=1,
  // announced DONE, tail batch never written. Losing a completed batch is
  // strictly WORSE than a partial commit: those records are CORRECT; only the
  // announcement was wrong. Flush unconditionally; report honestly below.
  if commit == 1 {
      let f3: i64 = mw_flush_plane(mvprefix, wm, "mv:ids" as *u8, ids, idslen[0])
      if f3 < 0 { fatal = 1 }
      if f3 == 1 { st[7] = st[7] + 1 }
      let f4: i64 = mw_flush_plane(ppprefix, wp, "p:ids" as *u8, pids, pidslen[0])
      if f4 < 0 { fatal = 1 }
  }
  st[11] = mv_reclaim_unique(rcl)
  st[12] = mv_reclaim_dupcount(rcl)
  st[13] = mv_reclaim_reclaimable(rcl)
  st[14] = fatal
  return mw_emit(out, st, commit)
599}
600
601// loop-write a whole buffer (a single write may be partial)
602func mw_st_write(path: *u8, bytes: *u8, n: i64) -> i64 {
  let fd: i64 = sys_openat_wr(path, 420)
  if fd < 0 { return 0 - 1 }
  var off: i64 = 0
  var bad: i64 = 0
  while off < n {
      let w: i64 = sys_write(fd, (bytes as i64 + off) as *u8, n - off)
      if w <= 0 { bad = 1; off = n } else { off = off + w }
  }
  sys_close(fd)
  if bad == 1 { return 0 - 1 }
  return 0
614}
615
616func mw_st_tooth(name: *u8, ok: i64, passbox: *i64) -> i64 {
  let b: *u8 = sys_mmap(256)
  var o: i64 = mw_cat(b, 0, name)
  if ok == 1 { o = mw_cat(b, o, ": PASS" as *u8); passbox[0] = passbox[0] + 1 } else { o = mw_cat(b, o, ": FAIL" as *u8) }
  o = mw_ch(b, o, 10)
  sys_write(1, b, o)
  return 0
623}
624
625// VmSize (kB) from /proc/self/status -- the EXACT metric the seq905 finding used to prove the leak
626// ("all 9 organs' VmSize==VmPeak"). Reading our own address-space size is how a leak is measured from the
627// inside; RSS would understate it because leaked-but-untouched pages are not resident.
628const MW_VM_ITERS: i64 = 200
629// RATCHET (seq1015). Measured after the seq905 fixes: delta_kb=800 over MW_VM_ITERS=200 opens = exactly
630// 4096 bytes (ONE PAGE) per open, down from the whole store per open. 850 sits just above that, so the tooth
631// now LOCKS IN the gain: any regression worse than ~4.3KB/open fails immediately instead of hiding inside a
632// generous slack. Do NOT loosen this to make a failing build pass -- find the regression. When the last page
633// is eliminated, ratchet again toward 0.
634const MW_VM_SLACK_KB: i64 = 850
635func mw_vmsize() -> i64 {
  let fd: i64 = sys_openat_rd("/proc/self/status" as *u8)
  if fd < 0 { return 0 - 1 }
  let b: *u8 = sys_mmap(MW_MAGIC_8192)
  let n: i64 = sys_read(fd, b, MW_MAGIC_8191)
  sys_close(fd)
  if n <= 0 { sys_munmap(b, MW_MAGIC_8192); return 0 - 2 }
  b[n] = 0 as u8
  var i: i64 = 0
  var out: i64 = 0 - 1
  while i + 7 <= n {
      if out < 0 {
          if b[i] == (86 as u8) { if b[i+1] == (109 as u8) { if b[i+2] == (83 as u8) { if b[i+3] == (105 as u8) {
          if b[i+4] == (122 as u8) { if b[i+5] == (101 as u8) { if b[i+6] == (58 as u8) {
              var j: i64 = i + 7
              // /proc/self/status separates label and value with a TAB, not a space. Skipping only 0x20
              // left j parked on the tab, the digit scan hit a non-digit immediately and returned v=0 --
              // and 0 then failed the `vm0 > 0` guard, so T8 reported FAIL for four rounds while measuring
              // nothing (seq1010). Skip BOTH separators.
              var sk: i64 = 1
              while sk == 1 { sk = 0; if b[j] == (32 as u8) { j = j + 1; sk = 1 } else { if b[j] == (9 as u8) { j = j + 1; sk = 1 } } }
              var v: i64 = 0
              var run: i64 = 1
              while run == 1 {
                  let c: i64 = b[j] as i64
                  if c < 48 { run = 0 } else { if c > 57 { run = 0 } else { v = v * 10 + (c - 48); j = j + 1 } }
              }
              out = v
          } } } } } } }
      }
      i = i + 1
  }
  sys_munmap(b, MW_MAGIC_8192)
  // Distinct failure codes so a dead probe reports WHY instead of blending into a leak verdict (seq1010):
  // -1 open failed, -2 read failed, -3 parsed but no VmSize line matched.
  if out < 0 { return 0 - 3 }
  return out
672}
673
674func mw_selftest() -> i64 {
  let pass: *i64 = sys_mmap(16) as *i64
  pass[0] = 0
  sys_mkdir("tmp_mvwalk_st" as *u8, 511)
  sys_mkdir("tmp_mvwalk_st/tree" as *u8, 511)
  sys_mkdir("tmp_mvwalk_st/tree/sub" as *u8, 511)
  // deterministic re-runs: drop prior store artifacts (plain data files, not locks)
  sys_unlinkat("tmp_mvwalk_st/s-manifest.txt" as *u8)
  sys_unlinkat("tmp_mvwalk_st/p-manifest.txt" as *u8)
  let nb: *u8 = sys_mmap(128)
  var si: i64 = 0
  while si < 4 {
      var o1: i64 = mw_cat(nb, 0, "tmp_mvwalk_st/s-seg-" as *u8)
      o1 = mw_udec(nb, o1, si)
      o1 = mw_cat(nb, o1, ".docs" as *u8)
      nb[o1] = 0 as u8
      sys_unlinkat(nb)
      o1 = mw_cat(nb, 0, "tmp_mvwalk_st/s-seg-" as *u8)
      o1 = mw_udec(nb, o1, si)
      o1 = mw_cat(nb, o1, ".keys" as *u8)
      nb[o1] = 0 as u8
      sys_unlinkat(nb)
      o1 = mw_cat(nb, 0, "tmp_mvwalk_st/p-seg-" as *u8)
      o1 = mw_udec(nb, o1, si)
      o1 = mw_cat(nb, o1, ".docs" as *u8)
      nb[o1] = 0 as u8
      sys_unlinkat(nb)
      o1 = mw_cat(nb, 0, "tmp_mvwalk_st/p-seg-" as *u8)
      o1 = mw_udec(nb, o1, si)
      o1 = mw_cat(nb, o1, ".keys" as *u8)
      nb[o1] = 0 as u8
      sys_unlinkat(nb)
      si = si + 1
  }
  // T0 -- absolute anchor: FIPS KAT sha256("abc")
  let cid: *u8 = sys_mmap(96)
  cid_of("abc" as *u8, 3, cid)
  let kat: *u8 = "nxc1-ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad" as *u8
  mw_st_tooth("T0 cid_of sha256-KAT(abc)" as *u8, mws_eq(cid, kat), pass)
  // T1 -- streaming == one-shot; a tiny prime chunk forces block-boundary feeds
  let b100: *u8 = sys_mmap(128)
  var i: i64 = 0
  while i < 100 { b100[i] = ((i * 7 + 13) % 251) as u8; i = i + 1 }
  mw_st_write("tmp_mvwalk_st/tree/f1.bin" as *u8, b100, 100)
  let want: *u8 = sys_mmap(96)
  cid_of(b100, 100, want)
  let got: *u8 = sys_mmap(96)
  let sz: *i64 = sys_mmap(16) as *i64
  var t1: i64 = 0
  if cid_of_file_chunk("tmp_mvwalk_st/tree/f1.bin" as *u8, 7, sz, got) == 69 {
      if mws_eq(got, want) == 1 { if sz[0] == 100 { t1 = 1 } }
  }
  mw_st_tooth("T1 cid_of_file_chunk(7) == cid_of" as *u8, t1, pass)
  // T2 -- the unified hasher streams past the head window
  let big: i64 = MW_HEAD + MW_MAGIC_300000
  let bb: *u8 = sys_mmap(big + 16)
  i = 0
  while i < big { bb[i] = ((i % 253) + 1) as u8; i = i + 1 }
  mw_st_write("tmp_mvwalk_st/tree/f2.bin" as *u8, bb, big)
  let want2: *u8 = sys_mmap(96)
  cid_of(bb, big, want2)
  let ctx: *Sha256 = sys_mmap(256) as *Sha256
  let hb: *u8 = sys_mmap(MW_HEAD)
  let cb: *u8 = sys_mmap(MW_CHUNK)
  let dg: *u8 = sys_mmap(64)
  let hn: *i64 = sys_mmap(16) as *i64
  var t2: i64 = 0
  if mw_hash_file("tmp_mvwalk_st/tree/f2.bin" as *u8, ctx, hb, cb, dg, got, sz, hn) == 0 {
      if mws_eq(got, want2) == 1 { if sz[0] == big { if hn[0] == MW_HEAD { t2 = 1 } } }
  }
  mw_st_tooth("T2 mw_hash_file streams past head" as *u8, t2, pass)
  // T3 -- dedup set semantics
  let hset: *i64 = mws_new()
  var t3: i64 = 1
  if mws_add(hset, want) != 0 { t3 = 0 }
  if mws_has(hset, want) != 1 { t3 = 0 }
  if mws_add(hset, want) != 1 { t3 = 0 }
  if mws_has(hset, kat) == 1 { t3 = 0 }
  mw_st_tooth("T3 cid-set add/has/dup" as *u8, t3, pass)
  // dup content + a nested distinct file for the walk teeth
  mw_st_write("tmp_mvwalk_st/tree/f1dup.bin" as *u8, b100, 100)
  let b77: *u8 = sys_mmap(96)
  i = 0
  while i < 77 { b77[i] = (200 - i) as u8; i = i + 1 }
  mw_st_write("tmp_mvwalk_st/tree/sub/f3.bin" as *u8, b77, 77)
  let out: *u8 = sys_mmap(MW_MAGIC_4096)
  let st: *i64 = sys_mmap(16 * 8) as *i64
  let mvp: *u8 = "tmp_mvwalk_st/s-" as *u8
  let ppp: *u8 = "tmp_mvwalk_st/p-" as *u8
  let segs0: *i64 = sys_mmap(8 * 16) as *i64
  // T4 -- dry: counts + non-mutating
  mw_run("tmp_mvwalk_st/tree" as *u8, MV_PROV_REAL, "selftest" as *u8, 0, mvp, ppp, "/volume1/vault" as *u8, 0 - 1, out, st)
  var t4: i64 = 1
  if st[0] != 4 { t4 = 0 }
  if st[1] != 4 { t4 = 0 }
  if st[5] != 2 { t4 = 0 }
  if st[11] != 3 { t4 = 0 }
  if st[12] != 1 { t4 = 0 }
  if st[2] != 0 { t4 = 0 }
  if ss_manifest_cap(mvp, segs0, 16) > 0 { t4 = 0 }
  mw_st_tooth("T4 dry walk: counts + non-mutating" as *u8, t4, pass)
  // T5 -- commit: one batch segment, ids complete
  mw_run("tmp_mvwalk_st/tree" as *u8, MV_PROV_REAL, "selftest" as *u8, 1, mvp, ppp, "/volume1/vault" as *u8, 0 - 1, out, st)
  var t5: i64 = 1
  if st[1] != 4 { t5 = 0 }
  if st[7] != 1 { t5 = 0 }
  if st[14] != 0 { t5 = 0 }
  if ss_manifest_cap(mvp, segs0, 16) != 1 { t5 = 0 }
  let vh1: *i64 = mws_new()
  if mw_load_ids(mvp, "mv:ids" as *u8, vh1, 0 as *u8, 0 as *i64) != 3 { t5 = 0 }
  let vh2: *i64 = mws_new()
  if mw_load_ids(ppp, "p:ids" as *u8, vh2, 0 as *u8, 0 as *i64) != 4 { t5 = 0 }
  mw_st_tooth("T5 commit: 1 segment, 3 cids, 4 path keys" as *u8, t5, pass)
  // T6 -- resume: everything path-skipped, no new segment
  mw_run("tmp_mvwalk_st/tree" as *u8, MV_PROV_REAL, "selftest" as *u8, 1, mvp, ppp, "/volume1/vault" as *u8, 0 - 1, out, st)
  var t6: i64 = 1
  if st[2] != 4 { t6 = 0 }
  if st[1] != 0 { t6 = 0 }
  if st[7] != 0 { t6 = 0 }
  if ss_manifest_cap(mvp, segs0, 16) != 1 { t6 = 0 }
  mw_st_tooth("T6 resume: all already, no new commit" as *u8, t6, pass)
  // T7/T8 -- ss_close (the seq905 keystone). ss_open loads every live segment's blobs into memory and until
  // 2026-07-25 NOTHING ever released them (`func ss_close` grepped to 0 matches against ~180 ss_open call
  // sites), so every consumer leaked BY CONSTRUCTION -- measured as ~20.6 GB of 24.3 GB swap consumed, which
  // parks long walkers in uninterruptible sleep where they neither progress nor accept SIGKILL.
  var t7: i64 = 1
  if ss_close(0 as *i64) != 0 { t7 = 0 }
  mw_st_tooth("T7 ss_close(0) is a null-safe no-op" as *u8, t7, pass)
  // NON-VACUOUS BY SIZING: without ss_close, MW_VM_ITERS opens of even this tiny store grow VmSize by far
  // more than MW_VM_SLACK_KB, so this tooth FAILS on the pre-fix binary rather than passing trivially.
  let vm0: i64 = mw_vmsize()
  var itc: i64 = 0
  while itc < MW_VM_ITERS {
      let hx: *i64 = ss_open(mvp)
      if (hx as i64) != 0 { ss_close(hx) }
      itc = itc + 1
  }
  let vm1: i64 = mw_vmsize()
  var t8: i64 = 0
  if vm0 > 0 { if vm1 > 0 { if vm1 - vm0 <= MW_VM_SLACK_KB { t8 = 1 } } }
  // Report the MAGNITUDE, not just pass/fail. A binary FAIL says "something leaks"; the delta divided by
  // MW_VM_ITERS gives BYTES PER OPEN, which localises the culprit to a specific allocation size instead of
  // leaving the next session to guess. An instrument that only says NO is half an instrument.
  let d8: *u8 = sys_mmap(256)
  var do8: i64 = mw_cat(d8, 0, "    T8 delta_kb=" as *u8)
  do8 = mw_udec(d8, do8, vm1 - vm0)
  do8 = mw_cat(d8, do8, " vm0_kb=" as *u8); do8 = mw_udec(d8, do8, vm0)
  do8 = mw_cat(d8, do8, " vm1_kb=" as *u8); do8 = mw_udec(d8, do8, vm1)
  do8 = mw_cat(d8, do8, " iters=" as *u8); do8 = mw_udec(d8, do8, MW_VM_ITERS)
  do8 = mw_cat(d8, do8, " bytes_per_open=" as *u8); do8 = mw_udec(d8, do8, (vm1 - vm0) * MW_MAGIC_1024 / MW_VM_ITERS)
  if vm0 == 0 - 1 { do8 = mw_cat(d8, do8, " probe=OPENFAIL" as *u8) }
  if vm0 == 0 - 2 { do8 = mw_cat(d8, do8, " probe=READFAIL" as *u8) }
  if vm0 == 0 - 3 { do8 = mw_cat(d8, do8, " probe=PARSEFAIL" as *u8) }
  if vm0 > 0 { do8 = mw_cat(d8, do8, " probe=LIVE" as *u8) }
  do8 = mw_ch(d8, do8, 10)
  sys_write(1, d8, do8)
  sys_munmap(d8, 256)
  mw_st_tooth("T8 200x ss_open+ss_close: VmSize flat (seq905 leak closed)" as *u8, t8, pass)
  let vb: *u8 = sys_mmap(256)
  var vo: i64 = 0
  vo = mw_cat(vb, vo, "nx_mvault_walk selftest T=9 PASS=" as *u8)
  vo = mw_udec(vb, vo, pass[0])
  if pass[0] == 9 { vo = mw_cat(vb, vo, " verdict=GREEN" as *u8) } else { vo = mw_cat(vb, vo, " verdict=RED" as *u8) }
  vo = mw_ch(vb, vo, 10)
  sys_write(1, vb, vo)
  if pass[0] == 9 { return 0 }
  return 1
841}
842
843func main(argc: i64, argv: *i64) -> i64 {
  if argc >= 2 {
      if mw_seq((argv[1]) as *u8, "selftest" as *u8) == 1 { return mw_selftest() }
      // STOP verb (debt seq931). The single-walker guard could REFUSE a double-start but nothing could
      // TERMINATE a wedged worker -- the only recovery was ssh, which the api-first doctrine retires. A
      // long walker that stops committing is otherwise indistinguishable from a slow one AND unkillable,
      // so it holds the single-walker slot forever. Self-exclusion is mandatory: our own cmdline matches
      // the needle, so an unguarded kill-by-name SIGKILLs this process before it can report anything.
      if mw_seq((argv[1]) as *u8, "stop" as *u8) == 1 {
          // REPORT BEFORE KILLING (debt seq935). The first cut relied on __syscall(39) getpid to exclude
          // self; that did not hold, so the scanner SIGKILLed this process before it could write anything --
          // `stop` returned EMPTY while genuinely killing the worker, which reads as a no-op and invites the
          // operator to run it again. Ordering is the robust fix: the count comes from proc_count_by_name
          // (self-inclusive, so subtract 1) and the line is flushed to the capture pipe BEFORE any signal,
          // so even if we do kill ourselves the report has already left.
          let sb: *u8 = sys_mmap(256)
          var tgt: i64 = proc_count_by_name("nx_mvault_walk" as *u8) - 1
          if tgt < 0 { tgt = 0 }
          var sq: i64 = mw_cat(sb, 0, "{\"action\":\"STOPPING\",\"targets\":" as *u8)
          sq = mw_udec(sb, sq, tgt)
          sq = mw_cat(sb, sq, ",\"note\":\"SIGKILL by cmdline; re-run the walk to resume (path-resume plane skips completed paths)\"}" as *u8)
          sq = mw_ch(sb, sq, 10)
          sys_write(1, sb, sq)
          if tgt > 0 {
              let me: i64 = __syscall(39, 0, 0, 0, 0, 0, 0)
              proc_kill_by_name_except("nx_mvault_walk" as *u8, 9, me)
          }
          return 0
      }
  }
  if argc < 4 {
      let ub: *u8 = sys_mmap(256)
      var uo: i64 = mw_cat(ub, 0, "usage: nx_mvault_walk <dir> <gen|real> <source> [commit|dry] [bg <logfile>] | selftest" as *u8)
      uo = mw_ch(ub, uo, 10)
      sys_write(2, ub, uo)
      return 2
  }
  let dir: *u8 = (argv[1]) as *u8
  let prov_s: *u8 = (argv[2]) as *u8
  let source: *u8 = (argv[3]) as *u8
  var commit: i64 = 1
  if argc >= 5 { if mw_seq((argv[4]) as *u8, "dry" as *u8) == 1 { commit = 0 } }
  var prov: i64 = MV_PROV_REAL
  if mw_seq(prov_s, "gen" as *u8) == 1 { prov = MV_PROV_GEN }
  // CWD-robust: stores write CWD-relative knowledge/store/ (gate T6 lesson)
  // YIELD TO THE CONTROL PLANE BY CONSTRUCTION (2026-07-30). This walk is a
  // BULK job: on 2026-07-30 it saturated the NAS and every forked organ queued
  // behind its I/O, so EVERY agent MCP call 503'd for minutes. A manual
  // `renice 19` restored interactive service immediately -- so the walk now
  // does it to ITSELF at startup rather than waiting for an operator to notice.
  sys_setpriority(19)
  sys_mkdir("knowledge" as *u8, 511)
  sys_mkdir("knowledge/store" as *u8, 511)
  let mvprefix: *u8 = "knowledge/store/mvault-" as *u8
  let ppprefix: *u8 = "knowledge/store/mvaultpath-" as *u8
  let dest_root: *u8 = "/volume1/vault" as *u8
  let out: *u8 = sys_mmap(MW_MAGIC_4096)
  let st: *i64 = sys_mmap(16 * 8) as *i64
  var bg: i64 = 0
  if argc >= 7 { if mw_seq((argv[5]) as *u8, "bg" as *u8) == 1 { bg = 1 } }
  if bg == 1 {
      let logpath: *u8 = (argv[6]) as *u8
      // SINGLE-WALKER BY CONSTRUCTION (debt seq621). Two concurrent walkers both read-modify-write the
      // lock-free seg-store -- that is CORRUPTION of a 20k+ record plane, not merely duplicated work. The
      // old discipline was "the operator must remember not to double-start"; a tool that CANNOT do the
      // wrong thing beats a session that must remember. Our OWN cmdline matches too, so >=2 == another walker.
      if proc_count_by_name("nx_mvault_walk" as *u8) >= 2 {
          var ro: i64 = mw_cat(out, 0, "{\"action\":\"REFUSED\",\"verdict\":\"SINGLE-WALKER\",\"reason\":\"another nx_mvault_walk is already running; not starting a second (seg-store RMW would corrupt)\"}" as *u8)
          ro = mw_ch(out, ro, 10)
          sys_write(1, out, ro)
          return 3
      }
      let pid: i64 = sys_fork()
      if pid != 0 {
          var o: i64 = mw_cat(out, 0, "{\"action\":\"BG-STARTED\",\"log\":\"" as *u8)
          o = mw_cat(out, o, logpath)
          o = mw_cat(out, o, "\"}" as *u8)
          o = mw_ch(out, o, 10)
          sys_write(1, out, o)
          return 0
      }
      // child: detach from the capture pipe BEFORE the long run
      sys_close(0)
      sys_close(1)
      sys_close(2)
      // SESSION LEADER. Closing 0/1/2 detaches the capture pipe but NOT the session:
      // the worker stayed in the tools-daemon's session and died with it -- the measured
      // cause of the 2026-07-23 run stopping at 21,521 files with no verdict=DONE.
      // Every durable daemon here (nx_hostctl, gallery guard, pulse/conductor) does
      // fork + nx_setsid for exactly this reason; this one-shot worker had skipped it.
      nx_setsid()
      let logfd: i64 = sys_openat_append(logpath, 420)
      if logfd < 0 { sys_exit(3) }
      // START line BEFORE any hashing (debt seq618). The PROG tick fires every 256 NEWLY-PROCESSED files,
      // so on a RESUMED run (most paths path-skip instantly, then 256 multi-GB recordings must stream-hash)
      // the log stayed 0 bytes for a long time -- and "still hashing" was indistinguishable from "died on
      // launch". An immediate line makes worker liveness observable from byte one instead of inferred.
      var so: i64 = mw_cat(out, 0, "START nx_mvault_walk dir=" as *u8)
      so = mw_cat(out, so, dir)
      so = mw_cat(out, so, " prov=" as *u8)
      so = mw_cat(out, so, prov_s)
      so = mw_cat(out, so, " source=" as *u8)
      so = mw_cat(out, so, source)
      so = mw_cat(out, so, " commit=" as *u8)
      so = mw_udec(out, so, commit)
      so = mw_ch(out, so, 10)
      sys_write(logfd, out, so)
      let n: i64 = mw_run(dir, prov, source, commit, mvprefix, ppprefix, dest_root, logfd, out, st)
      var wo: i64 = 0
      var bad: i64 = 0
      while wo < n {
          let ww: i64 = sys_write(logfd, (out as i64 + wo) as *u8, n - wo)
          if ww <= 0 { bad = 1; wo = n } else { wo = wo + ww }
      }
      // 2026-07-30: the verdict must reflect the ERROR FLAG, not merely
      // reaching the end. This wrote DONE unconditionally, so a run with
      // fatal=1 announced success (measured: images root, 127,283 files,
      // fatal=1, verdict=DONE). A COMPLETION VERDICT THAT IGNORES ITS OWN
      // ERROR FLAG IS A FALSE GREEN. Scan the emitted JSON for "fatal":0.
      var okrun: i64 = 0
      var fi: i64 = 0
      while fi + 9 < n {
          if out[fi] == (102 as u8) {
              if out[fi + 1] == (97 as u8) {
                  if out[fi + 2] == (116 as u8) {
                      if out[fi + 3] == (97 as u8) {
                          if out[fi + 4] == (108 as u8) {
                              if out[fi + 7] == (48 as u8) { okrun = 1 }
                          }
                      }
                  }
              }
          }
          fi = fi + 1
      }
      if okrun == 1 {
          sys_write(logfd, "verdict=DONE\n" as *u8, 13)
      } else {
          sys_write(logfd, "verdict=INCOMPLETE fatal=1 -- some items may be missing; re-run this root\n" as *u8, 74)
      }
      sys_close(logfd)
      if bad == 1 { sys_exit(4) }
      sys_exit(0)
  }
  let n2: i64 = mw_run(dir, prov, source, commit, mvprefix, ppprefix, dest_root, 0 - 1, out, st)
  sys_write(1, out, n2)
  return 0
990}
nx_mvault_walk.nx source

`nx_mvault_walk.nx` source
