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

↩ module page · 347 lines · 15518 B
1// nx_worklog_ingest_lib.nx -- the PRESENT-axis workhorse: ingest the session transcript (.jsonl the
2// harness writes out-of-band as work happens) into the append-only work-journal. A byte-offset
3// CURSOR makes it idempotent: each run appends one framed line per NEW tool_use since the cursor,
4// then advances the cursor to the last complete line. Idempotence is the key property -- the SAME
5// organ can fire on PostToolUse (real-time / mid-turn crash durability), on Stop (turn end), and on a
6// Cron timer (fully out-of-band redundancy) with ZERO duplicate lines, because the cursor gates them
7// all. Reading the harness's OWN record means capture can never miss an action.
8//
9// REUSE: getdents64 + fstatat(mtime@88) directory walk is the proven nx_ws_index_lib/nx_sov_tree_audit
10// idiom; line-boundary cursor is the standard JSONL tail-follow; atomic cursor write = tmp+renameat.
11// Sovereign: nx_syscalls + nx_worklog_lib + nx_json_lib only. license_tier: ORIGINAL
12import "nx_syscalls.nx"
13import "nx_worklog_lib.nx"
14import "nx_json_lib.nx"
15const WLI_MAGIC_1024: i64 = 1024
16const WLI_MAGIC_65536: i64 = 65536
17const WLI_MAGIC_2048: i64 = 2048
18const WLI_MAGIC_131072: i64 = 131072
19
20const WLI_READCAP: i64 = 8388608     // 8MB per ingest pass (process up to last newline within it)
21const WLI_TOOLCAP: i64 = 64
22const WLI_TGTCAP:  i64 = 256
23const WLI_TGTWIN:  i64 = 4096        // bound the first-string-value search inside an input object
24
25func wli_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
30}
31func wli_catn(dst: *u8, off: i64, v: i64) -> i64 {
  var o: i64 = off
  var m: i64 = v
  if m == 0 { dst[o] = 48 as u8; return o + 1 }
  if m < 0 { dst[o] = 45 as u8; o = o + 1; m = 0 - m }
  let t: *u8 = sys_mmap(32)
  var k: i64 = 0
  while m > 0 { t[k] = ((48 + (m % 10)) as u8); m = m / 10; k = k + 1 }
  while k > 0 { dst[o] = t[k - 1]; o = o + 1; k = k - 1 }
  return o
41}
42
43// read a decimal byte-offset cursor; absent/empty/garbage -> 0 (start from the top, safe).
44func wli_read_cursor(cursor_path: *u8) -> i64 {
  let fd: i64 = sys_openat_rd(cursor_path)
  if fd < 0 { return 0 }
  let buf: *u8 = sys_mmap(64)
  let nrd: i64 = sys_read(fd, buf, 63)
  sys_close(fd)
  if nrd <= 0 { return 0 }
  var v: i64 = 0
  var i: i64 = 0
  while i < nrd {
      let c: i64 = buf[i] as i64
      if c >= 48 { if c <= 57 { v = v * 10 + (c - 48) } }
      i = i + 1
  }
  return v
59}
60
61// write the cursor atomically (tmp + renameat) so a crash mid-write never yields a torn offset.
62func wli_write_cursor(cursor_path: *u8, v: i64) -> i64 {
  let buf: *u8 = sys_mmap(64)
  var o: i64 = wli_catn(buf, 0, v)
  buf[o] = 10 as u8; o = o + 1
  let tmp: *u8 = sys_mmap(WLI_MAGIC_1024)
  var t: i64 = wli_cat(tmp, 0, cursor_path)
  let suf: *u8 = ".tmp" as *u8
  t = wli_cat(tmp, t, suf)
  tmp[t] = 0 as u8
  let fd: i64 = sys_openat_wr(tmp, 420)
  if fd < 0 { return 0 - 1 }
  sys_write(fd, buf, o)
  sys_close(fd)
  sys_renameat(tmp, cursor_path)
  return 0
77}
78
79// per-transcript cursor path: <WL_DIR>/cursors/<basename(transcript)>.cur -- so distinct sessions
80// never share an offset (which would corrupt capture across a restart). returns length.
81func wli_default_cursor(transcript: *u8, out: *u8, outcap: i64) -> i64 {
  let tl: i64 = jx_len(transcript)
  var bstart: i64 = 0
  var i: i64 = 0
  while i < tl { if transcript[i] == (47 as u8) { bstart = i + 1 } i = i + 1 }
  var o: i64 = wli_cat(out, 0, WL_DIR)               // const passed as arg = OK
  let mid: *u8 = "/cursors/" as *u8
  o = wli_cat(out, o, mid)
  var j: i64 = bstart
  while j < tl { out[o] = transcript[j]; o = o + 1; j = j + 1 }
  let suf: *u8 = ".cur" as *u8
  o = wli_cat(out, o, suf)
  out[o] = 0 as u8
  return o
95}
96
97// THE INGEST: append one "ingest" work-journal line per NEW tool_use since the cursor; advance cursor.
98// returns the number of newly-captured actions (0 if none / no transcript yet -- always graceful).
99func wli_ingest(transcript: *u8, cursor_path: *u8, worklog: *u8) -> i64 {
  let start: i64 = wli_read_cursor(cursor_path)
  let fd: i64 = sys_openat_rd(transcript)
  if fd < 0 { return 0 }                              // no transcript yet -> nothing, no error
  sys_lseek(fd, start, 0)                             // SEEK_SET
  let buf: *u8 = sys_mmap(WLI_READCAP)
  var total: i64 = 0
  var nrd: i64 = sys_read(fd, buf, WLI_READCAP)
  while nrd > 0 {
      total = total + nrd
      if total >= WLI_READCAP { nrd = 0 } else { nrd = sys_read(fd, ((buf as i64) + total) as *u8, WLI_READCAP - total) }
  }
  sys_close(fd)
  if total <= 0 { return 0 }
113
  // only process COMPLETE lines: trim to the last newline so no tool_use ever spans the cursor.
  var lastnl: i64 = 0 - 1
  var p: i64 = total - 1
  while p >= 0 { if buf[p] == (10 as u8) { lastnl = p; p = 0 - 1 } else { p = p - 1 } }
  if lastnl < 0 { return 0 }
  let limit: i64 = lastnl + 1
120
  let anchor:   *u8 = "\"type\":\"tool_use\"" as *u8
  let namekey:  *u8 = "name" as *u8
  let inputpat: *u8 = "\"input\":{" as *u8
  let colq:     *u8 = ":\"" as *u8
  let inlen: i64 = jx_len(inputpat)
126
  var count: i64 = 0
  var at: i64 = jx_find(buf, limit, 0, anchor)
  while at >= 0 {
      let tool: *u8 = sys_mmap(WLI_TOOLCAP)
      tool[0] = 45 as u8; tool[1] = 0 as u8           // default "-"
      let tgt: *u8 = sys_mmap(WLI_TGTCAP)
      tgt[0] = 45 as u8; tgt[1] = 0 as u8             // default "-"
      let after_name: i64 = jx_get_str(buf, limit, at, namekey, tool, WLI_TOOLCAP)
      var nexti: i64 = at + 16
      if after_name >= 0 {
          nexti = after_name
          let inpos: i64 = jx_find(buf, limit, after_name, inputpat)
          if inpos >= 0 {
              var win: i64 = inpos + WLI_TGTWIN
              if win > limit { win = limit }
              let vq: i64 = jx_find(buf, win, inpos + inlen, colq)
              if vq >= 0 { jx_copy_str(buf, limit, vq + 2, tgt, WLI_TGTCAP) }
          }
          wl_append3(worklog, "ingest" as *u8, tool, tgt)
          count = count + 1
      }
      at = jx_find(buf, limit, nexti, anchor)
  }
  wli_write_cursor(cursor_path, start + limit)
  return count
152}
153
154// translate a Windows path ("C:\Users\...") to its WSL form ("/mnt/c/Users/..."). Hook stdin reports
155// Windows paths but the organ runs under WSL. Already-POSIX (leading '/') paths copy through. The
156// JSON de-escape already collapsed "\\" -> "\", so here we only fold drive-letter + backslashes.
157func wli_winpath_to_wsl(src: *u8, out: *u8) -> i64 {
  let n: i64 = jx_len(src)
  if n == 0 { out[0] = 0 as u8; return 0 }
  if src[0] == (47 as u8) {                          // already POSIX
      var i: i64 = 0
      while src[i] != (0 as u8) { out[i] = src[i]; i = i + 1 }
      out[i] = 0 as u8
      return i
  }
  if n >= 2 { if src[1] == (58 as u8) {              // "<drive>:" form
      let mnt: *u8 = "/mnt/" as *u8
      var o: i64 = wli_cat(out, 0, mnt)
      var dl: i64 = src[0] as i64
      if dl >= 65 { if dl <= 90 { dl = dl + 32 } }   // drive letter upper -> lower
      out[o] = dl as u8; o = o + 1
      var i: i64 = 2
      while i < n {
          var c: i64 = src[i] as i64
          if c == 92 { c = 47 }                       // backslash -> slash
          out[o] = c as u8; o = o + 1
          i = i + 1
      }
      out[o] = 0 as u8
      return o
  } }
  var j: i64 = 0                                      // unknown shape: copy verbatim
  while src[j] != (0 as u8) { out[j] = src[j]; j = j + 1 }
  out[j] = 0 as u8
  return j
186}
187
188// ".jsonl" suffix test (for the cron newest-transcript fallback).
189func wli_is_jsonl(name: *u8) -> i64 {
  let n: i64 = jx_len(name)
  if n < 7 { return 0 }
  let suf: *u8 = ".jsonl" as *u8
  var k: i64 = 0
  while k < 6 { if name[n - 6 + k] != suf[k] { return 0 } k = k + 1 }
  return 1
196}
197
198// newest *.jsonl in `dir` (highest st_mtim.tv_sec) -> out. returns 0 if found, -1 if none/unopenable.
199// lets a Cron timer run with no args and still target the live session. getdents64 + fstatat idiom.
200func wli_newest_transcript(dir: *u8, out: *u8, outcap: i64) -> i64 {
  let fd: i64 = sys_openat_rd(dir)
  if fd < 0 { return 0 - 1 }
  let gbuf: *u8 = sys_mmap(WLI_MAGIC_65536)
  let stbuf: *u8 = sys_mmap(256)
  let child: *u8 = sys_mmap(WLI_MAGIC_2048)
  var best_mt: i64 = 0 - 1
  var found: i64 = 0
  var nread: i64 = __syscall(217, fd, gbuf, WLI_MAGIC_65536, 0, 0, 0)
  while nread > 0 {
      var off: i64 = 0
      while off < nread {
          let reclen: i64 = (gbuf[off + 16] as i64) | ((gbuf[off + 17] as i64) << 8)
          if reclen <= 0 { off = nread } else {
              let name: *u8 = ((gbuf as i64) + off + 19) as *u8
              if wli_is_jsonl(name) == 1 {
                  var co: i64 = wli_cat(child, 0, dir)
                  child[co] = 47 as u8; co = co + 1
                  var ci: i64 = 0
                  while name[ci] != (0 as u8) { child[co] = name[ci]; co = co + 1; ci = ci + 1 }
                  child[co] = 0 as u8
                  var mt: i64 = 0
                  if sys_fstatat(child, stbuf) == 0 {
                      mt = (stbuf[88] as i64) | ((stbuf[89] as i64) << 8) | ((stbuf[90] as i64) << 16) | ((stbuf[91] as i64) << 24) | ((stbuf[92] as i64) << 32) | ((stbuf[93] as i64) << 40)
                  }
                  if mt > best_mt {
                      best_mt = mt
                      var oo: i64 = 0
                      while child[oo] != (0 as u8) { out[oo] = child[oo]; oo = oo + 1 }
                      out[oo] = 0 as u8
                      found = 1
                  }
              }
              off = off + reclen
          }
      }
      nread = __syscall(217, fd, gbuf, WLI_MAGIC_65536, 0, 0, 0)
  }
  sys_close(fd)
  if found == 1 { return 0 }
  return 0 - 1
241}
242
243// ONE capture entry point shared by every runner/hook (DRY): acquire the transcript by precedence
244//   explicit path (argpath != 0)  >  hook stdin "transcript_path"  >  newest *.jsonl in projdir
245// (Windows paths translated to WSL), compute the per-transcript cursor, ingest. returns lines added.
246// Graceful: no transcript anywhere -> 0, never errors. Caller must NOT also read stdin separately.
247// Placed last so every callee is already defined above (definition-before-use).
248func wli_capture(argpath: *u8, projdir: *u8, cursors_dir: *u8) -> i64 {
  wl_ensure_dir(WL_DIR)
  sys_mkdir(cursors_dir, 511)
  let transcript: *u8 = sys_mmap(WLI_MAGIC_2048)
  var have: i64 = 0
  if (argpath as i64) != 0 {
      wli_winpath_to_wsl(argpath, transcript)
      have = 1
  }
  if have == 0 {
      let sbuf: *u8 = sys_mmap(WLI_MAGIC_131072)
      var sn: i64 = 0
      var r: i64 = sys_read(0, sbuf, WLI_MAGIC_131072)
      while r > 0 {
          sn = sn + r
          if sn >= WLI_MAGIC_131072 { r = 0 } else { r = sys_read(0, ((sbuf as i64) + sn) as *u8, WLI_MAGIC_131072 - sn) }
      }
      if sn > 0 {
          let raw: *u8 = sys_mmap(WLI_MAGIC_2048)
          let key: *u8 = "transcript_path" as *u8
          if jx_get_str(sbuf, sn, 0, key, raw, WLI_MAGIC_2048) >= 0 {
              wli_winpath_to_wsl(raw, transcript)
              have = 1
          }
      }
  }
  if have == 0 {
      if wli_newest_transcript(projdir, transcript, WLI_MAGIC_2048) == 0 { have = 1 }
  }
  if have == 0 { return 0 }
  let cursor: *u8 = sys_mmap(WLI_MAGIC_2048)
  wli_default_cursor(transcript, cursor, WLI_MAGIC_2048)
  return wli_ingest(transcript, cursor, WL_PATH)
281}
282
283// ".cur" suffix test (cursor files are named "<transcript_basename>.cur").
284func wli_is_cur(name: *u8) -> i64 {
  let n: i64 = jx_len(name)
  if n < 5 { return 0 }
  let suf: *u8 = ".cur" as *u8
  var k: i64 = 0
  while k < 4 { if name[n - 4 + k] != suf[k] { return 0 } k = k + 1 }
  return 1
291}
292
293// CRASH-HEAL SWEEP -- the missing crash-durability piece. Re-ingest the stranded tail of EVERY session
294// whose cursor sits BEHIND its transcript: that gap is the exact signature of a session that WAS being
295// captured but never got a final Stop hook (i.e. it CRASHED). Driven by the CURSORS dir -- one .cur per
296// session the worklog already owns -- NOT the transcript dir, so (a) pre-worklog transcripts are never
297// touched and (b) cost is one stat per known session. Idempotent: a cleanly-finished session has
298// cursor==size, so its ingest is a no-op (0 new lines, zero duplicates); that same idempotence makes a
299// transient re-visit during the dir walk harmless. After ANY crash, the next session's first Stop calls
300// this and the crashed tail is captured -- closing the hole BY CONSTRUCTION. returns lines healed.
301func wli_sweep_behind_to(projdir: *u8, cursors_dir: *u8, worklog: *u8) -> i64 {
  let fd: i64 = sys_openat_rd(cursors_dir)
  if fd < 0 { return 0 }                              // no cursors yet -> nothing, graceful
  let gbuf: *u8 = sys_mmap(WLI_MAGIC_65536)
  let stbuf: *u8 = sys_mmap(256)
  let cpath: *u8 = sys_mmap(WLI_MAGIC_2048)
  let tpath: *u8 = sys_mmap(WLI_MAGIC_2048)
  var recovered: i64 = 0
  var nread: i64 = __syscall(217, fd, gbuf, WLI_MAGIC_65536, 0, 0, 0)
  while nread > 0 {
      var off: i64 = 0
      while off < nread {
          let reclen: i64 = (gbuf[off + 16] as i64) | ((gbuf[off + 17] as i64) << 8)
          if reclen <= 0 { off = nread } else {
              let name: *u8 = ((gbuf as i64) + off + 19) as *u8
              if wli_is_cur(name) == 1 {
                  var co: i64 = wli_cat(cpath, 0, cursors_dir)    // cursor path = cursors_dir/name
                  cpath[co] = 47 as u8; co = co + 1
                  var ci: i64 = 0
                  while name[ci] != (0 as u8) { cpath[co] = name[ci]; co = co + 1; ci = ci + 1 }
                  cpath[co] = 0 as u8
                  var to: i64 = wli_cat(tpath, 0, projdir)        // transcript = projdir/(name minus ".cur")
                  tpath[to] = 47 as u8; to = to + 1
                  let blen: i64 = ci - 4                          // ci = strlen(name); strip ".cur"
                  var ti: i64 = 0
                  while ti < blen { tpath[to] = name[ti]; to = to + 1; ti = ti + 1 }
                  tpath[to] = 0 as u8
                  var sz: i64 = 0 - 1                             // size > cursor => un-ingested tail (a crash)
                  if sys_fstatat(tpath, stbuf) == 0 {
                      sz = (stbuf[48] as i64) | ((stbuf[49] as i64) << 8) | ((stbuf[50] as i64) << 16) | ((stbuf[51] as i64) << 24) | ((stbuf[52] as i64) << 32) | ((stbuf[53] as i64) << 40)
                  }
                  let cur: i64 = wli_read_cursor(cpath)
                  if sz > cur { recovered = recovered + wli_ingest(tpath, cpath, worklog) }
              }
              off = off + reclen
          }
      }
      nread = __syscall(217, fd, gbuf, WLI_MAGIC_65536, 0, 0, 0)
  }
  sys_close(fd)
  return recovered
342}
343
344// convenience: heal into the real work-journal (WL_PATH). Called by the Stop-hook / cron runner.
345func wli_sweep_behind(projdir: *u8, cursors_dir: *u8) -> i64 {
  return wli_sweep_behind_to(projdir, cursors_dir, WL_PATH)
347}
nx_worklog_ingest_lib.nx source

`nx_worklog_ingest_lib.nx` source
