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

↩ module page · 609 lines · 33931 B
1// nx_ws_kickoff_sync.nx -- CR-R1b MECHANICAL WORKSTREAM CAPTURE: the "sync from kickoff, survive
2// interruptions" rung of the orchestration north-star (Pillar 7). A workstream KICKS OFF, BEATS its
3// state continuously, DONEs on completion; the append-only journal IS the resume surface -- a crash
4// costs nothing but reading it. Each frame is ONE O_APPEND write (atomic for <4KB lines on one host)
5// so PARALLEL sessions converge without clobber (the coindex append-and-derive property).
6// license_tier: ORIGINAL   expect_exit: 0
7//   kickoff  <journal> <ws> <actor> <note>
8//   beat     <journal> <ws> <actor> <note>
9//   done     <journal> <ws> <actor> <note>
10//   resume   <journal>                 -> in-flight (KICKOFF w/o DONE) + last checkpoint = resume list
11//   board    <journal> [window_sec]    -> per-ws latest state + liveness (ACTIVE/STALE/DONE)
12//   selftest <journal>                 -> gate T1..T6 (caller pre-cleans the path); exit 0 iff all pass
13import "nx_syscalls.nx"
14import "nx_itoa_lib.nx"   // shared MSB-first emitter (zero-alloc)
15import "nx_logtail.nx"    // shared HONEST TAIL READER -- this organ is where it was proven, and it is
16const K_MAGIC_2000: i64 = 2000
17const K_MAGIC_3000: i64 = 3000
                        // now LIFTED so the 2nd..Nth consumer inherits the proof (debt 1786054115)
19const K_MAGIC_4096: i64 = 4096
20const K_MAGIC_262144: i64 = 262144
21const K_MAGIC_262140: i64 = 262140
22const K_MAGIC_3600: i64 = 3600
23
24func ks_puts(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 }
25// MIGRATED to the shared emitter (debt 1785563586). The old body mmapped a scratch buffer
26// per call and never freed it. At PAGE granularity that is 4096B leaked PER CALL -- the
27// defect that took 28.5GB of a 36GB host in nx_ts_lumadiff (2MB input, ~3.66M calls).
28// nxi_* is MSB-first, allocates NOTHING, and emits identical bytes including the sign.
29func ks_putn(v: i64) -> i64 { nxi_out(v); return 0 }
30func ks_cat(d: *u8, o: i64, s: *u8) -> i64 { var i: i64=0; while s[i]!=(0 as u8){ d[o]=s[i]; o=o+1; i=i+1 } return o }
31func ks_catn(d: *u8, o: i64, v: i64) -> i64 { let t: *u8=sys_mmap(28); var m: i64=v; if m<0{d[o]=45 as u8;o=o+1;m=0-m} var k: i64=0; if m==0{t[0]=48 as u8;k=1} while m>0{t[k]=(48+(m%10)) as u8;m=m/10;k=k+1} var i: i64=0; while i<k{d[o]=t[k-1-i];o=o+1;i=i+1} return o }
32func ks_vlen(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} return n }
33func ks_atoi_z(s: *u8) -> i64 { var v: i64=0; var i: i64=0; while s[i]!=(0 as u8){ let c: i64=s[i] as i64; if c>=48 { if c<=57 { v=v*10+(c-48) } } i=i+1 } return v }
34
35// append one frame: <ts>\t<verb>\t<ws>\t<actor>\t<note>\n  (ts<0 => real clock)
36func ks_append(journal: *u8, ts: i64, verb: *u8, ws: *u8, actor: *u8, note: *u8) -> i64 {
  var t: i64 = ts
  if t < 0 { t = sys_now_realtime_sec() }
  let ln: *u8 = sys_mmap(K_MAGIC_4096)
  var o: i64 = 0
  o = ks_catn(ln, o, t)
  ln[o]=9 as u8; o=o+1
  o = ks_cat(ln, o, verb)
  ln[o]=9 as u8; o=o+1
  o = ks_cat(ln, o, ws)
  ln[o]=9 as u8; o=o+1
  o = ks_cat(ln, o, actor)
  ln[o]=9 as u8; o=o+1
  o = ks_cat(ln, o, note)
  ln[o]=10 as u8; o=o+1
  let fd: i64 = sys_openat_append(journal, 0x1a4)
  if fd < 0 { return -1 }
  sys_write(fd, ln, o)
  sys_close(fd)
  return 0
56}
57func ks_read(path: *u8, buf: *u8, cap: i64) -> i64 {
  let fd: i64 = sys_openat_rd(path)
  if fd < 0 { return 0 }
  var n: i64 = 0; var go: i64 = 1
  while go == 1 { let r: i64 = sys_read(fd, ((buf as i64)+n) as *u8, cap-n); if r <= 0 { go = 0 } else { n = n + r } if n >= cap { go = 0 } }
  sys_close(fd)
  return n
64}
65// ---------------------------------------------------------------------------------------------
66// HONEST TAIL READ (2026-08-06, ws=resume-surface-rearm). THE DEFECT THIS REPLACES: every reader
67// below did sys_mmap(K_MAGIC_262144) + ks_read(...,K_MAGIC_262140) -- a FORWARD read from offset 0
68// into a 256KiB window, with n NEVER compared against the file size. On a journal larger than the
69// window that shows the OLDEST 19% and reports it as the whole truth, SILENTLY: no flag, no count.
70// MEASURED 2026-08-06 on the live surface: ws_sync.jrnl was ~1.363MB = 5.2x the window, so `board`
71// reported its newest frame as ts 1784596191 (Jul 20) while the file already carried Aug-06 frames,
72// and four kickoffs appended minutes earlier were invisible.
73// THE HARM IS NOT ONLY BLINDNESS -- ks_closed() and ks_last_ts() test DONE/ORPHAN-REAPED and
74// last-activity WITHIN THE BUFFER, so truncation FABRICATES: (a) any ws DONE past the cap reads
75// IN-FLIGHT FOREVER, (b) every "@ts" is the last IN-WINDOW ts = a false staleness age that reads as
76// an abandoned lane. A reader that silently drops the tail of a CRASH-RESUME surface is the banned
77// partial-as-complete class (law L011), same shape as the MEMORY.md cliff.
78// KNOWN GOOD MATCHED: nx_fs declares scanned=/scan_cap=/FILE-EXCEEDS-SCAN-WINDOW on every read.
79// This matches that contract then exceeds it -- it STREAMS THE WHOLE FILE so file_bytes is EXACT
80// (nx_fs stops at its cap and cannot report true size), RETAINS THE NEWEST bytes (a resume surface
81// must never prefer the oldest), and starts on a frame boundary so a partial line is never parsed.
82// env[0]=file_bytes(exact) env[1]=scanned(kept) env[2]=dropped(oldest) env[3]=truncated(0|1)
83const K_JRNL_BUF: i64 = 8388608
84const K_JRNL_RD: i64 = 8388600
85
86// THE READER THAT LIVED HERE IS NOW runtime/nx_logtail.nx (lt_read_tail). It was born in this file and
87// proven here, so this organ is the natural FIRST consumer of the lifted version -- and the migration is
88// its own equivalence test: if the extraction were unfaithful, the 12 teeth below would not still pass.
89// A SHARED LIB WHOSE ONLY CALLER IS ITS OWN GATE IS THE ADOPTION GAP, NOT A CAPABILITY.
90func ks_env_print(env: *i64) -> i64 {
  ks_puts(" ENVELOPE file_bytes=" as *u8); ks_putn(env[0])
  ks_puts(" scanned=" as *u8); ks_putn(env[1])
  ks_puts(" dropped_oldest=" as *u8); ks_putn(env[2])
  ks_puts(" window_bytes=" as *u8); ks_putn(K_JRNL_RD)
  if env[3]==0 { ks_puts(" coverage_complete=1\n" as *u8) }
  else { ks_puts(" coverage_complete=0 TRUNCATED-OLDEST-DROPPED-NEWEST-KEPT\n" as *u8) }
  return 0
98}
99// span [out0,out1) of column c in line [ls,le); 1 if found
100func ks_col(q: *u8, ls: i64, le: i64, c: i64, out: *i64) -> i64 {
  var col: i64 = 0
  var p: i64 = ls
  while col < c {
      var s: i64 = 1
      while s == 1 { if p >= le { return 0 } if q[p]==(9 as u8) { s = 0 } else { p = p+1 } }
      p = p + 1
      col = col + 1
  }
  var e: i64 = p
  var s2: i64 = 1
  while s2 == 1 { if e >= le { s2 = 0 } else { if q[e]==(9 as u8) { s2 = 0 } else { e = e+1 } } }
  out[0] = p; out[1] = e
  return 1
114}
115func ks_lit_eq(q: *u8, s: i64, e: i64, lit: *u8) -> i64 {
  var i: i64 = 0
  while s+i < e { if lit[i]==(0 as u8) { return 0 } if q[s+i]!=lit[i] { return 0 } i=i+1 }
  if lit[i]!=(0 as u8) { return 0 }
  return 1
120}
121func ks_span_eq(q: *u8, s1: i64, e1: i64, s2: i64, e2: i64) -> i64 {
  if e1-s1 != e2-s2 { return 0 }
  var i: i64 = 0
  while s1+i < e1 { if q[s1+i]!=q[s2+i] { return 0 } i=i+1 }
  return 1
126}
127func ks_atoi(q: *u8, s: i64, e: i64) -> i64 {
  var v: i64 = 0; var i: i64 = s
  while i < e { let c: i64 = q[i] as i64; if c>=48 { if c<=57 { v = v*10 + (c-48) } } i=i+1 }
  return v
131}
132// line end from i
133func ks_le(q: *u8, i: i64, n: i64) -> i64 {
  var le: i64 = i; var s: i64 = 1
  while s==1 { if le>=n { s=0 } else { if q[le]==(10 as u8){s=0} else {le=le+1} } }
  return le
137}
138// any frame verb==VERB (literal) and col2 span == [ws_s,ws_e) ?
139func ks_has(q: *u8, n: i64, verb: *u8, ws_s: i64, ws_e: i64) -> i64 {
  let cv: *i64 = sys_mmap(16) as *i64
  let cw: *i64 = sys_mmap(16) as *i64
  var i: i64 = 0
  while i < n {
      let le: i64 = ks_le(q,i,n)
      if ks_col(q,i,le,1,cv)==1 { if ks_lit_eq(q,cv[0],cv[1],verb)==1 {
          if ks_col(q,i,le,2,cw)==1 { if ks_span_eq(q,cw[0],cw[1],ws_s,ws_e)==1 { return 1 } }
      } }
      i = le + 1
  }
  return 0
151}
152// TERMINAL-CLOSURE (2026-08-06). A row leaves the board when it is DONE **or** ORPHAN-REAPED.
153// The F214 reaper (nx_seat reap -> nx_claims reap) already releases provably-expired claims and
154// appends ORPHAN-REAPED frames HERE, but every reader below tested the literal "DONE" only -- so a
155// reaped row stayed IN-FLIGHT FOREVER. MEASURED 2026-08-06: memory-rail and game-save were reaped
156// in the claims plane ("ORPHAN-REAPED by=claude-swarm-d-f214 expired age=100794s ttl=5400s") and
157// STILL read IN-FLIGHT on the resume surface. The reap ran; the board never learned.
158// A REAP IS TERMINAL BUT IT IS NOT A COMPLETION -- kept a DISTINCT verb on purpose so nx_seat shift
159// keeps counting only real DONE closures as closed rungs; laundering reaps into DONE would inflate
160// tokens-per-closed-rung with abandoned work.
161// last ts of frames whose verb==VERB for this ws span; -1 if none.
162func ks_last_ts_verb(q: *u8, n: i64, verb: *u8, ws_s: i64, ws_e: i64) -> i64 {
  let cw: *i64 = sys_mmap(16) as *i64
  let cv: *i64 = sys_mmap(16) as *i64
  let c0: *i64 = sys_mmap(16) as *i64
  var last: i64 = -1
  var i: i64 = 0
  while i < n {
      let le: i64 = ks_le(q,i,n)
      if ks_col(q,i,le,2,cw)==1 { if ks_span_eq(q,cw[0],cw[1],ws_s,ws_e)==1 {
          if ks_col(q,i,le,1,cv)==1 { if ks_lit_eq(q,cv[0],cv[1],verb)==1 {
              if ks_col(q,i,le,0,c0)==1 { last = ks_atoi(q,c0[0],c0[1]) }
          } }
      } }
      i = le + 1
  }
  return last
178}
179// A REAP IS REVOCABLE; A DONE IS NOT. DONE is the seat's own statement that the work finished, so it
180// is final anywhere in the file. ORPHAN-REAPED is a THIRD PARTY's inference that the seat died -- and
181// an inference can be WRONG. If we treated it as final-anywhere (the first cut of this function did),
182// one false reap would hide a LIVE workstream from the board FOREVER, and the louder that seat beat
183// the more invisible it would stay. So a reap only closes the row while nothing has spoken since:
184// any BEAT or KICKOFF newer than the last reap REOPENS it. That makes a mistaken reap self-healing on
185// the seat's very next beat, which is what lets the sweep run unattended without needing to be right
186// every time. Fail-safe direction chosen deliberately: false-reopen (visible, costs a board row) is
187// cheap; false-close (invisible, loses a live lane) is the expensive one.
188func ks_closed(q: *u8, n: i64, ws_s: i64, ws_e: i64) -> i64 {
  if ks_has(q,n,"DONE" as *u8,ws_s,ws_e)==1 { return 1 }
  let rt: i64 = ks_last_ts_verb(q,n,"ORPHAN-REAPED" as *u8,ws_s,ws_e)
  if rt < 0 { return 0 }
  let bt: i64 = ks_last_ts_verb(q,n,"BEAT" as *u8,ws_s,ws_e)
  if bt > rt { return 0 }
  let kt: i64 = ks_last_ts_verb(q,n,"KICKOFF" as *u8,ws_s,ws_e)
  if kt > rt { return 0 }
  // CHECKPOINT counts too. `nx_seat brief` writes CHECKPOINT frames as the lossless-flush protocol,
  // so a seat that briefed after being reaped is demonstrably alive and its row must reopen. I had
  // omitted this verb; the sibling working nx_seat's mirror of this predicate caught it. Any verb a
  // WORKING seat emits belongs in this list -- missing one silently keeps a live lane buried.
  let ct: i64 = ks_last_ts_verb(q,n,"CHECKPOINT" as *u8,ws_s,ws_e)
  if ct > rt { return 0 }
  return 1
203}
204// any frame verb==VERB (literal) and col2 == ws (literal) ?  (for selftest)
205func ks_has_lit(q: *u8, n: i64, verb: *u8, ws: *u8) -> i64 {
  let cv: *i64 = sys_mmap(16) as *i64
  let cw: *i64 = sys_mmap(16) as *i64
  var i: i64 = 0
  while i < n {
      let le: i64 = ks_le(q,i,n)
      if ks_col(q,i,le,1,cv)==1 { if ks_lit_eq(q,cv[0],cv[1],verb)==1 {
          if ks_col(q,i,le,2,cw)==1 { if ks_lit_eq(q,cw[0],cw[1],ws)==1 { return 1 } }
      } }
      i = le + 1
  }
  return 0
217}
218// first KICKOFF for ws span at line-start upto? (no earlier KICKOFF same ws in [0,upto))
219func ks_first_kick(q: *u8, upto: i64, ws_s: i64, ws_e: i64) -> i64 {
  let cv: *i64 = sys_mmap(16) as *i64
  let cw: *i64 = sys_mmap(16) as *i64
  var i: i64 = 0
  while i < upto {
      let le: i64 = ks_le(q,i,upto)
      if ks_col(q,i,le,1,cv)==1 { if ks_lit_eq(q,cv[0],cv[1],"KICKOFF" as *u8)==1 {
          if ks_col(q,i,le,2,cw)==1 { if ks_span_eq(q,cw[0],cw[1],ws_s,ws_e)==1 { return 0 } }
      } }
      i = le + 1
  }
  return 1
231}
232// ts of the LAST (file-order) frame for ws span; -1 if none
233func ks_last_ts(q: *u8, n: i64, ws_s: i64, ws_e: i64) -> i64 {
  let cw: *i64 = sys_mmap(16) as *i64
  let c0: *i64 = sys_mmap(16) as *i64
  var last: i64 = -1
  var i: i64 = 0
  while i < n {
      let le: i64 = ks_le(q,i,n)
      if ks_col(q,i,le,2,cw)==1 { if ks_span_eq(q,cw[0],cw[1],ws_s,ws_e)==1 {
          if ks_col(q,i,le,0,c0)==1 { last = ks_atoi(q,c0[0],c0[1]) }
      } }
      i = le + 1
  }
  return last
246}
247// print the last checkpoint (note col4) of the last frame for ws span
248func ks_print_last_note(q: *u8, n: i64, ws_s: i64, ws_e: i64) -> i64 {
  let cw: *i64 = sys_mmap(16) as *i64
  let cn: *i64 = sys_mmap(16) as *i64
  var ns: i64 = -1; var ne: i64 = -1
  var i: i64 = 0
  while i < n {
      let le: i64 = ks_le(q,i,n)
      if ks_col(q,i,le,2,cw)==1 { if ks_span_eq(q,cw[0],cw[1],ws_s,ws_e)==1 {
          if ks_col(q,i,le,4,cn)==1 { ns=cn[0]; ne=cn[1] }
      } }
      i = le + 1
  }
  if ns>=0 { sys_write(1,(q as i64 + ns) as *u8, ne-ns) }
  return 0
262}
263// ------------------------------------------------------------------------------------------------
264// SINGLE-PASS PER-WS INDEX (2026-08-06, ws=resume-surface-rearm). Retires debt 1786054039, which I filed
265// against my OWN work an hour earlier.
266// THE COST BEING REMOVED WAS MEASURED FIRST, WITH A CONTROL: `resume` on the live 1364805-byte journal ran
267// 2024/2002/2012 ms (median 2012, spread +/-11ms) against a 70ms transport-only floor measured on a
268// 14-byte file -- so ~1942ms was pure scanning, and the tight spread proves it was compute-bound, not
269// network noise. CAUSE: ks_resume/ks_board walk every line and, PER CANDIDATE ROW, call ks_closed (which
270// itself rescans up to FIVE times), ks_last_ts, and ks_print_last_note -- each restarting from offset 0.
271// That is O(rows x bytes), and it had been INVISIBLE because the 256KiB truncation bug capped the buffer.
272// ***FIXING THE CORRECTNESS DEFECT REMOVED AN ACCIDENTAL BRAKE AND EXPOSED THE QUADRATIC UNDERNEATH IT.***
273// THE INDEX aggregates everything the emitters need in ONE pass: last ts, last note span, DONE, last reap
274// ts, and the last ts of any WORKING verb. Cost becomes O(bytes) + O(rows x distinct_ws) for name lookup,
275// and memory is bounded by distinct_ws instead of by journal size.
276// THE SLOW SCANNERS ARE KEPT ON PURPOSE AND ARE NOT DEAD CODE. Tooth T12 runs the index and the reference
277// over the SAME journal and asserts they agree. A rewrite that merely still passes the old teeth has only
278// shown it did not break them; a DIFFERENTIAL check shows it computes the same answer. Keeping the slow
279// oracle in-tree is the cheapest permanent guard against a clever index quietly drifting from the truth.
280const K_IX_MAX: i64 = 4096
281const K_IX_STRIDE: i64 = 9
282// slot layout: 0 ws_s | 1 ws_e | 2 last_ts | 3 note_s | 4 note_e | 5 done | 6 reap_ts | 7 live_ts | 8 first_kick_off
283func ks_ix_find(ix: *i64, cnt: i64, q: *u8, s: i64, e: i64) -> i64 {
  var k: i64 = 0
  while k < cnt {
      let b: i64 = k * K_IX_STRIDE
      if ks_span_eq(q, ix[b], ix[b+1], s, e) == 1 { return k }
      k = k + 1
  }
  return 0 - 1
291}
292func ks_ix_build(q: *u8, n: i64, ix: *i64) -> i64 {
  let c0: *i64 = sys_mmap(16) as *i64
  let cv: *i64 = sys_mmap(16) as *i64
  let cw: *i64 = sys_mmap(16) as *i64
  let cn: *i64 = sys_mmap(16) as *i64
  var cnt: i64 = 0
  var i: i64 = 0
  while i < n {
      let le: i64 = ks_le(q, i, n)
      if ks_col(q,i,le,2,cw) == 1 {
          var k: i64 = ks_ix_find(ix, cnt, q, cw[0], cw[1])
          if k < 0 { if cnt < K_IX_MAX {
              k = cnt
              let b0: i64 = k * K_IX_STRIDE
              ix[b0] = cw[0]; ix[b0+1] = cw[1]
              ix[b0+2] = 0 - 1; ix[b0+3] = 0 - 1; ix[b0+4] = 0 - 1
              ix[b0+5] = 0; ix[b0+6] = 0 - 1; ix[b0+7] = 0 - 1; ix[b0+8] = 0 - 1
              cnt = cnt + 1
          } }
          if k >= 0 {
              let b: i64 = k * K_IX_STRIDE
              var ts: i64 = 0 - 1
              if ks_col(q,i,le,0,c0) == 1 { ts = ks_atoi(q, c0[0], c0[1]) }
              ix[b+2] = ts
              if ks_col(q,i,le,4,cn) == 1 { ix[b+3] = cn[0]; ix[b+4] = cn[1] }
              if ks_col(q,i,le,1,cv) == 1 {
                  if ks_lit_eq(q,cv[0],cv[1],"DONE" as *u8) == 1 { ix[b+5] = 1 }
                  if ks_lit_eq(q,cv[0],cv[1],"ORPHAN-REAPED" as *u8) == 1 { if ts > ix[b+6] { ix[b+6] = ts } }
                  if ks_lit_eq(q,cv[0],cv[1],"BEAT" as *u8) == 1 { if ts > ix[b+7] { ix[b+7] = ts } }
                  if ks_lit_eq(q,cv[0],cv[1],"CHECKPOINT" as *u8) == 1 { if ts > ix[b+7] { ix[b+7] = ts } }
                  if ks_lit_eq(q,cv[0],cv[1],"KICKOFF" as *u8) == 1 {
                      if ts > ix[b+7] { ix[b+7] = ts }
                      if ix[b+8] < 0 { ix[b+8] = i }
                  }
              }
          }
      }
      i = le + 1
  }
  return cnt
332}
333// MIRRORS ks_closed EXACTLY -- same DONE-first order, same reap-then-revival logic, same verb set
334// {BEAT,KICKOFF,CHECKPOINT}. If you change one, change both; T12 is what catches you if you do not.
335func ks_ix_closed(ix: *i64, k: i64) -> i64 {
  let b: i64 = k * K_IX_STRIDE
  if ix[b+5] == 1 { return 1 }
  if ix[b+6] < 0 { return 0 }
  if ix[b+7] > ix[b+6] { return 0 }
  return 1
341}
342// RESUME: print in-flight workstreams (KICKOFF w/o DONE) + last checkpoint; returns count
343func ks_resume(q: *u8, n: i64) -> i64 {
  let cv: *i64 = sys_mmap(16) as *i64
  let cw: *i64 = sys_mmap(16) as *i64
  let ix: *i64 = sys_mmap(K_IX_MAX * K_IX_STRIDE * 8) as *i64
  let cnt: i64 = ks_ix_build(q, n, ix)
  var inflight: i64 = 0
  var i: i64 = 0
  while i < n {
      let le: i64 = ks_le(q,i,n)
      if ks_col(q,i,le,1,cv)==1 { if ks_lit_eq(q,cv[0],cv[1],"KICKOFF" as *u8)==1 {
          if ks_col(q,i,le,2,cw)==1 {
              let k: i64 = ks_ix_find(ix, cnt, q, cw[0], cw[1])
              // EMIT ORDER PRESERVED DELIBERATELY. This second pass walks lines and fires only on the line
              // the index recorded as the FIRST KICKOFF -- exactly ks_first_kick's semantics, but O(1)
              // instead of a rescan. Emitting straight out of the index would order rows by first
              // APPEARANCE of the ws under ANY verb, a subtly different sequence, and a performance fix
              // that silently reorders a human-read surface is not a pure win.
              if k >= 0 { if ix[k*K_IX_STRIDE+8] == i {
                  if ks_ix_closed(ix, k) == 0 {
                      inflight = inflight + 1
                      ks_puts("  IN-FLIGHT ws=" as *u8); sys_write(1,(q as i64 + cw[0]) as *u8, cw[1]-cw[0])
                      ks_puts(" @" as *u8); ks_putn(ix[k*K_IX_STRIDE+2])
                      ks_puts(" last=" as *u8)
                      let ns: i64 = ix[k*K_IX_STRIDE+3]
                      let ne: i64 = ix[k*K_IX_STRIDE+4]
                      if ns >= 0 { sys_write(1,(q as i64 + ns) as *u8, ne-ns) }
                      ks_puts("\n" as *u8)
                  }
              } }
          }
      } }
      i = le + 1
  }
  return inflight
377}
378// BOARD: distinct ws -> state + liveness
379func ks_board(q: *u8, n: i64, window: i64) -> i64 {
  let cv: *i64 = sys_mmap(16) as *i64
  let cw: *i64 = sys_mmap(16) as *i64
  let now: i64 = sys_now_realtime_sec()
  let ix: *i64 = sys_mmap(K_IX_MAX * K_IX_STRIDE * 8) as *i64
  let cnt: i64 = ks_ix_build(q, n, ix)
  var i: i64 = 0
  while i < n {
      let le: i64 = ks_le(q,i,n)
      if ks_col(q,i,le,1,cv)==1 { if ks_lit_eq(q,cv[0],cv[1],"KICKOFF" as *u8)==1 {
          if ks_col(q,i,le,2,cw)==1 {
              let k: i64 = ks_ix_find(ix, cnt, q, cw[0], cw[1])
              if k >= 0 { if ix[k*K_IX_STRIDE+8] == i {
              let b: i64 = k * K_IX_STRIDE
              ks_puts("  ws=" as *u8); sys_write(1,(q as i64 + cw[0]) as *u8, cw[1]-cw[0])
              if ix[b+5]==1 { ks_puts(" DONE" as *u8) }
              else {
                // REAPED is terminal but is NOT a completion -- shown distinctly so the board never
                // launders an abandoned seat into a finished one (and nx_seat shift keeps counting
                // only real DONE frames as closed rungs). DONE is tested FIRST because ks_ix_closed
                // is true for both, exactly as the scanning version ordered its two tests.
                if ks_ix_closed(ix,k)==1 { ks_puts(" REAPED" as *u8) }
                else {
                  let lt: i64 = ix[b+2]
                  if now - lt <= window { ks_puts(" ACTIVE" as *u8) } else { ks_puts(" STALE" as *u8) }
                  ks_puts(" last=" as *u8)
                  let ns: i64 = ix[b+3]
                  let ne: i64 = ix[b+4]
                  if ns >= 0 { sys_write(1,(q as i64 + ns) as *u8, ne-ns) }
                }
              }
              ks_puts("\n" as *u8)
              } }
          }
      } }
      i = le + 1
  }
  return 0
417}
418func ks_assert(name: *u8, got: i64, want: i64, fails: *i64) -> i64 {
  if got==want { ks_puts("  PASS " as *u8) } else { ks_puts("  FAIL " as *u8); fails[0]=fails[0]+1 }
  ks_puts(name); ks_puts(" got=" as *u8); ks_putn(got); ks_puts(" want=" as *u8); ks_putn(want); ks_puts("\n" as *u8)
  return 0
422}
423// inflight-count via a fresh read
424// REFERENCE IMPLEMENTATION -- the O(rows x bytes) scanner, KEPT ON PURPOSE as the differential oracle for
425// the indexed path (tooth T12). It is slow by design and must stay behaviourally frozen. If you need to
426// change what "in flight" MEANS, change ks_closed and let BOTH paths inherit it -- do not edit this body to
427// match a new index, or the oracle starts agreeing with the thing it is supposed to check.
428func ks_inflight_ref(journal: *u8) -> i64 {
  let ev: *i64 = sys_mmap(64) as *i64
  let q: *u8 = sys_mmap(K_JRNL_BUF)
  let n: i64 = lt_read_tail(journal, q, K_JRNL_RD, ev)
  let cv: *i64 = sys_mmap(16) as *i64
  let cw: *i64 = sys_mmap(16) as *i64
  var c: i64 = 0
  var i: i64 = 0
  while i < n {
      let le: i64 = ks_le(q,i,n)
      if ks_col(q,i,le,1,cv)==1 { if ks_lit_eq(q,cv[0],cv[1],"KICKOFF" as *u8)==1 {
          if ks_col(q,i,le,2,cw)==1 { if ks_first_kick(q,i,cw[0],cw[1])==1 {
              if ks_closed(q,n,cw[0],cw[1])==0 { c=c+1 }
          } }
      } }
      i = le + 1
  }
  return c
446}
447// INDEXED -- the production path. One index build, then a count straight off the aggregates. Order does not
448// matter for a count, so this does not need the line-walk that ks_resume/ks_board use to preserve emit order.
449func ks_inflight(journal: *u8) -> i64 {
  let ev: *i64 = sys_mmap(64) as *i64
  let q: *u8 = sys_mmap(K_JRNL_BUF)
  let n: i64 = lt_read_tail(journal, q, K_JRNL_RD, ev)
  let ix: *i64 = sys_mmap(K_IX_MAX * K_IX_STRIDE * 8) as *i64
  let cnt: i64 = ks_ix_build(q, n, ix)
  var c: i64 = 0
  var k: i64 = 0
  while k < cnt {
      if ix[k*K_IX_STRIDE+8] >= 0 { if ks_ix_closed(ix,k)==0 { c = c + 1 } }
      k = k + 1
  }
  return c
462}
463func ks_selftest(journal: *u8) -> i64 {
  let fails: *i64 = sys_mmap(16) as *i64
  fails[0]=0
  ks_puts("== nx_ws_kickoff_sync selftest ==\n" as *u8)
  // T6 neg: empty journal (caller pre-cleaned) -> 0 in-flight, no fabrication
  ks_assert("T6-empty-no-fabrication" as *u8, ks_inflight(journal), 0, fails)
  // T1 kickoff A -> in-flight 1
  ks_append(journal, -1, "KICKOFF" as *u8, "A" as *u8, "s1" as *u8, "start-A" as *u8)
  ks_assert("T1-kickoff-tracked" as *u8, ks_inflight(journal), 1, fails)
  // T2 beat A -> still 1, last note advances (checked via has)
  ks_append(journal, -1, "BEAT" as *u8, "A" as *u8, "s1" as *u8, "checkpoint-A1" as *u8)
  ks_assert("T2-beat-keeps-inflight" as *u8, ks_inflight(journal), 1, fails)
  // T5 kickoff B -> in-flight 2 (two concurrent, no clobber)
  ks_append(journal, -1, "KICKOFF" as *u8, "B" as *u8, "s2" as *u8, "start-B" as *u8)
  ks_assert("T5-two-concurrent" as *u8, ks_inflight(journal), 2, fails)
  // T3/T4 done A -> in-flight 1 (B remains = the CRASH-RESUME property: B kicked-off, never done, still resumable)
  ks_append(journal, -1, "DONE" as *u8, "A" as *u8, "s1" as *u8, "finished-A" as *u8)
  ks_assert("T3-done-clears" as *u8, ks_inflight(journal), 1, fails)
  // load-bearing: A must be gone, B must remain
  let ev2: *i64 = sys_mmap(64) as *i64
  let q: *u8 = sys_mmap(K_JRNL_BUF)
  let n: i64 = lt_read_tail(journal, q, K_JRNL_RD, ev2)
  var bfail: i64 = 0
  if ks_has_lit(q,n,"DONE" as *u8,"A" as *u8)==0 { bfail=1 }
  if ks_has_lit(q,n,"KICKOFF" as *u8,"B" as *u8)==0 { bfail=1 }
  if ks_has_lit(q,n,"DONE" as *u8,"B" as *u8)==1 { bfail=1 }
  ks_assert("T4-crash-resume-B-survives" as *u8, bfail, 0, fails)
  ks_puts("== RESUME view ==\n" as *u8)
  ks_resume(q, n)
  // T7 (2026-08-06) THE REAP-TERMINAL TOOTH. B is still in-flight. The F214 reaper appends exactly
  // this frame (nx_seat.nx:1073) once nx_claims has released a provably-expired claim, so B must
  // then LEAVE the in-flight set. Without this tooth the entire reap path runs GREEN while the
  // board never changes -- which is precisely what was MEASURED in production on 2026-08-06:
  // memory-rail and game-save were ORPHAN-REAPED in the claims plane and still read IN-FLIGHT here.
  ks_append(journal, -1, "ORPHAN-REAPED" as *u8, "B" as *u8, "reaper" as *u8, "expired-ttl" as *u8)
  ks_assert("T7-reap-closes-row" as *u8, ks_inflight(journal), 0, fails)
  // T8 (2026-08-06) ENVELOPE HONEST ON A COMPLETE READ. A journal that FITS must declare
  // coverage_complete=1 with dropped=0 and file_bytes==scanned==n. If a full read can ever be
  // reported as partial (or a partial one as complete) the instrument is lying about its own
  // coverage, which is the entire defect class being closed here.
  let e8: *i64 = sys_mmap(64) as *i64
  let b8: *u8 = sys_mmap(K_JRNL_BUF)
  let n8: i64 = lt_read_tail(journal, b8, K_JRNL_RD, e8)
  var f8: i64 = 0
  if n8 <= 0 { f8=1 }
  if e8[3]!=0 { f8=1 }
  if e8[2]!=0 { f8=1 }
  if e8[0]!=n8 { f8=1 }
  if e8[1]!=n8 { f8=1 }
  ks_assert("T8-envelope-complete-honest" as *u8, f8, 0, fails)
  // T9 THE NEG-CONTROL THAT WOULD HAVE CAUGHT THE PRODUCTION DEFECT. Read the SAME journal through
  // a DELIBERATELY TINY window, so truncation is FORCED without authoring a multi-megabyte fixture
  // (a tooth that needs an 8MB file is a tooth nobody runs). The OLD reader kept the OLDEST bytes
  // and said nothing; this one must keep the NEWEST, declare the drop, close the arithmetic, and
  // start on a frame boundary. BOTH DIRECTIONS ARE ASSERTED ON PURPOSE: the newest ws must be
  // PRESENT and the oldest MUST BE ABSENT -- asserting only "newest present" would also pass for a
  // reader that quietly returned the whole file, i.e. it would not discriminate.
  let tinywin: i64 = 100
  let e9: *i64 = sys_mmap(64) as *i64
  let b9: *u8 = sys_mmap(K_MAGIC_4096)
  let n9: i64 = lt_read_tail(journal, b9, tinywin, e9)
  var f9: i64 = 0
  if e9[3]!=1 { f9=1 }
  if e9[0]<=n9 { f9=1 }
  if e9[1]!=n9 { f9=1 }
  if e9[2]!=e9[0]-e9[1] { f9=1 }
  if b9[0]==(10 as u8) { f9=1 }
  if ks_has_lit(b9,n9,"ORPHAN-REAPED" as *u8,"B" as *u8)==0 { f9=1 }
  if ks_has_lit(b9,n9,"KICKOFF" as *u8,"A" as *u8)==1 { f9=1 }
  ks_assert("T9-tail-keeps-newest-drops-oldest-declares" as *u8, f9, 0, fails)
  // T10a/T10/T11 (2026-08-06) THE REVOCATION TEETH. Deliberately appended AFTER T9's assertion has
  // already run: T9 asserts about the journal's NEWEST bytes, so adding frames before it would push
  // B's reap out of its 100-byte window and break a sibling's tooth. Execution order is the
  // contract here, not final file state.
  // These run on a FRESH ws C with EXPLICIT timestamps. Every frame above uses -1 (real clock) and
  // the whole selftest completes inside ONE second, so -1 frames are indistinguishable in ts order --
  // the ordering rule these teeth exist to test would pass or fail on scheduling luck.
  ks_append(journal, 1000, "KICKOFF" as *u8, "C" as *u8, "s3" as *u8, "start-C" as *u8)
  // NOTE the expected counts are 1 -> 0 -> 1, NOT 2 -> 1 -> 2: B was already closed by T7 above, so
  // it never contributes here. First cut of these teeth asserted 2/1/2 and went RED on all three --
  // the gate caught the author's arithmetic, which is the whole point of running it.
  ks_assert("T10a-kickoff-C-inflight" as *u8, ks_inflight(journal), 1, fails)
  ks_append(journal, K_MAGIC_2000, "ORPHAN-REAPED" as *u8, "C" as *u8, "reaper" as *u8, "expired-ttl" as *u8)
  ks_assert("T10-reap-closes-row" as *u8, ks_inflight(journal), 0, fails)
  // T11: the reaped seat speaks again. The reap was an INFERENCE that the seat died; this BEAT is
  // direct evidence it did not, so C must RETURN to the in-flight set. Without this, one false reap
  // buries a live workstream forever and beating harder never rescues it.
  ks_append(journal, K_MAGIC_3000, "BEAT" as *u8, "C" as *u8, "s3" as *u8, "alive-after-all" as *u8)
  ks_assert("T11-beat-revokes-reap" as *u8, ks_inflight(journal), 1, fails)
  // T12 (2026-08-06) THE DIFFERENTIAL TOOTH. The indexed path and the O(rows x bytes) reference scanner
  // must return the SAME count on the SAME journal, with every terminal/revival case above already in it.
  // This is what makes the single-pass rewrite PROVABLE rather than merely plausible: T1-T11 show only
  // that the new code did not break the old assertions, which a subtly-wrong index could also achieve
  // (drop one verb from the revival set and most of them still pass). TWO INDEPENDENT IMPLEMENTATIONS
  // AGREEING IS EVIDENCE; ONE IMPLEMENTATION PASSING ITS OWN TESTS IS NOT.
  ks_assert("T12-index-agrees-with-reference-scanner" as *u8, ks_inflight(journal), ks_inflight_ref(journal), fails)
  if fails[0]==0 { ks_puts("VERDICT=GREEN (13/13)\n" as *u8); return 0 }
  ks_puts("VERDICT=RED fails=" as *u8); ks_putn(fails[0]); ks_puts("\n" as *u8)
  return 1
562}
563func main(argc: i64, argv: *i64) -> i64 {
  if argc < 3 { ks_puts("usage: nx_ws_kickoff_sync {kickoff|beat|done <journal> <ws> <actor> <note> | resume|board <journal> [window] | selftest <journal>}\n" as *u8); sys_exit(2); return 2 }
  let verb: *u8 = argv[1] as *u8
  let journal: *u8 = argv[2] as *u8
  let vv: *u8 = verb
  if ks_lit_eq(vv,0,ks_vlen(vv),"selftest" as *u8)==1 { sys_exit(ks_selftest(journal)); return 0 }
  if ks_lit_eq(vv,0,ks_vlen(vv),"resume" as *u8)==1 {
      let ev: *i64 = sys_mmap(64) as *i64
      let q: *u8 = sys_mmap(K_JRNL_BUF); let n: i64 = lt_read_tail(journal,q,K_JRNL_RD,ev)
      let c: i64 = ks_resume(q,n); ks_puts(" RESUME in-flight=" as *u8); ks_putn(c); ks_puts("\n" as *u8); ks_env_print(ev); sys_exit(0); return 0
  }
  if ks_lit_eq(vv,0,ks_vlen(vv),"xcheck" as *u8)==1 {
      // LIVE DIFFERENTIAL, added 2026-08-06 in the middle of an investigation it would have ended in one
      // call. Tooth T12 compares the indexed path against the reference scanner on the SELFTEST fixture --
      // three workstreams, short notes, no revivals. Production has ~300 workstreams, reap-then-revive
      // sequences, and single notes over 18KB. When the indexed count on the live journal came back 9
      // against a 26 banked forty minutes earlier, I could not tell from OUTSIDE whether the index was
      // wrong or the DATA had moved, and I nearly rolled back a correct build on a guess.
      // ***A FIXTURE-ONLY DIFFERENTIAL PROVES THE FAST PATH ON DATA THAT CANNOT BREAK IT.*** This verb runs
      // both implementations over any real journal and exits 3 on disagreement, so the question is
      // answerable in one call by anyone, forever, instead of being re-reasoned from scratch each time.
      let ixc: i64 = ks_inflight(journal)
      let refc: i64 = ks_inflight_ref(journal)
      ks_puts(" XCHECK indexed=" as *u8); ks_putn(ixc)
      ks_puts(" reference=" as *u8); ks_putn(refc)
      if ixc == refc { ks_puts(" AGREE\n" as *u8); sys_exit(0); return 0 }
      ks_puts(" DISAGREE\n" as *u8); sys_exit(3); return 3
  }
  if ks_lit_eq(vv,0,ks_vlen(vv),"board" as *u8)==1 {
      var w: i64 = K_MAGIC_3600
      if argc>=4 { w = ks_atoi_z(argv[3] as *u8) }
      let ev: *i64 = sys_mmap(64) as *i64
      let q: *u8 = sys_mmap(K_JRNL_BUF); let n: i64 = lt_read_tail(journal,q,K_JRNL_RD,ev)
      ks_board(q,n,w); ks_env_print(ev); sys_exit(0); return 0
  }
  // append verbs
  if argc < 6 { ks_puts("kickoff|beat|done need <journal> <ws> <actor> <note>\n" as *u8); sys_exit(2); return 2 }
  let ws: *u8 = argv[3] as *u8
  let actor: *u8 = argv[4] as *u8
  let note: *u8 = argv[5] as *u8
  var vtag: *u8 = "BEAT" as *u8
  if ks_lit_eq(vv,0,ks_vlen(vv),"kickoff" as *u8)==1 { vtag = "KICKOFF" as *u8 }
  if ks_lit_eq(vv,0,ks_vlen(vv),"done" as *u8)==1 { vtag = "DONE" as *u8 }
  let rc: i64 = ks_append(journal, -1, vtag, ws, actor, note)
  if rc==0 { ks_puts(" SYNC " as *u8); ks_puts(vtag); ks_puts(" ws=" as *u8); ks_puts(ws); ks_puts("\n" as *u8); sys_exit(0); return 0 }
  ks_puts(" append FAILED\n" as *u8); sys_exit(1); return 1
609}
nx_ws_kickoff_sync.nx source

`nx_ws_kickoff_sync.nx` source
