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

↩ module page · 1109 lines · 56340 B
1// nx_seat_meter_beat.nx -- ES1 of /compare/engineshift: THE SCOREBOARD'S PRODUCER (laptop-local, 2026-09-02).
2// nx_seat shift (F831) grades the engine shift from knowledge/status/token_efficiency.log, and every row that log
3// has ever held was appended BY HAND: 16 rows, the last dated 2026-08-05, no hook, clock row or organ names
4// `nx_seat meter`. A gauge whose rows are hand-appended is a diary, not a gauge -- it stops the day the hand
5// stops and reads as "no movement" forever (the local column read ZERO for 44 days and was quoted twice as a
6// finding about the estate). This organ is the producer. It rides the SessionStart hook beside nx_crashwin.
7//
8// WHAT IT MEASURES. Every Claude Code transcript (<sessions>/*.jsonl) carries one line per assistant content
9// block, and EVERY block of one message repeats that message's usage object (measured 2026-09-02: 367 assistant
10// lines for 58 distinct message ids in one transcript). So usage is counted ONCE PER MESSAGE ID, never per line;
11// a naive sum overcounts about six-fold. Per complete UTC day and model family it sums output_tokens,
12// cache_read_input_tokens and cache_creation_input_tokens, counts messages, and counts the transcripts active
13// that day (the row's `sessions` column). A line is an assistant line when it carries a raw "type":"assistant"
14// (tool results are escaped strings and cannot); the message's OWN model, id and usage are the DEPTH-1 children of
15// "message":{ -- a tool_use input is raw JSON and may carry model and id fields of its own, so position on the line
16// decides nothing (neg-controls in the gate). Harness API-error lines (model "<synthetic>") are their own bucket.
17//
18//   nx_seat_meter_beat scan <sessions-dir> [--now <epoch>] [--days N]
19//       prints one DAY row per (day, family) inside the window and a METERBEAT summary; writes nothing.
20//   nx_seat_meter_beat beat <sessions-dir> <state-dir> <mcp-elf> <base-url> <cap-file> [--now <epoch>] [--days N]
21//                           [--dry] [--json]
22//       for every COMPLETE day (strictly before today, UTC) in the window whose (day, family) is not yet in
23//       <state-dir>/meterbeat.days, forks <mcp-elf> (nx_mcp_call) to run `nx_seat meter` on the estate and, on
24//       the reply "METERED row appended", records the marker -- idempotent by day marker: a second run appends
25//       nothing. --dry plans and prints, forks nothing, marks nothing. --json wraps the digest as a SessionStart
26//       hook envelope and exits 0 (the hook framework parses stdout only on exit 0).
27// Today is NEVER pushed: an incomplete day would be re-summed tomorrow and the marker would make it final.
28// Window: --days N (default MB_DAYS_DEFAULT) complete days; transcripts whose mtime predates the window start
29// are not read (they cannot carry a newer day), so a session-start run touches only recent files.
30// A transcript whose last line has no newline is counted PARTIAL: its complete lines count, the tail does not,
31// and the summary says so -- a truncated file is never read as zero.
32// Exit: 0 GREEN | 1 RED (a push was refused or answered without the METERED receipt) | 2 usage |
33//       3 UNOBSERVABLE (the sessions dir cannot be opened). verdict= is the LAST token of the last line.
34// license_tier: ORIGINAL. Writes only <state-dir>/meterbeat.days. No hw writes (Rule 26).
35import "nx_syscalls.nx"
36import "nx_memplane_lib.nx"
37import "nx_tool_run.nx"
38
39const MB_DAYS_DEFAULT: i64 = 3            // a hook run owes yesterday and the day before; older days were owed earlier
40const MB_DAYS_MAX: i64 = 400              // distinct days one run can hold; reaching it announces days_floor=1
41const MB_FAM_MAX: i64 = 8                 // model families per run (fable, opus, sonnet, haiku, ...)
42const MB_FAM_CAP: i64 = 32                // one family token cell
43const MB_SLOT: i64 = 256                  // one transcript name cell (names measured 42 B)
44const MB_FILES_MAX: i64 = 4096            // 220 transcripts on 2026-09-02; reaching the cap announces files_floor=1
45const MB_PATH: i64 = 4096                 // PATH_MAX
46const MB_DENTBUF: i64 = 262144            // getdents64 batch buffer, LOOPED to completion
47const MB_STATBUF: i64 = 256               // struct stat is 144 B on x86_64
48const MB_STAT_MTIME_SLOT: i64 = 11        // st_mtim.tv_sec at byte 88 = i64[11]
49const MB_DOT: i64 = 46
50const MB_QUOTE: i64 = 34
51const MB_NL: i64 = 10
52const MB_DASH: i64 = 45
53const MB_ID_CAP: i64 = 64                 // msg_ ids measured 30 B
54const MB_ID_RING: i64 = 8                 // blocks of one message are consecutive; a ring of 8 absorbs interleaving
55const MB_OUT: i64 = 262144                // report buffer (400 days x 8 families x ~120 B fits 4x over)
56const MB_DAY_LEN: i64 = 10                // YYYY-MM-DD
57const MB_SEC_PER_DAY: i64 = 86400
58const MB_EXIT_OK: i64 = 0
59const MB_EXIT_RED: i64 = 1
60const MB_EXIT_USAGE: i64 = 2
61const MB_EXIT_UNOBS: i64 = 3
62const MB_FORK_TIMEOUT_MS: i64 = 60000     // the estate appends one row; 60 s covers a saturated box
63const MB_RESP_CAP: i64 = 65536
64const MB_READ_SLACK: i64 = 16             // sys_read_file maps size+16; released with the same arithmetic
65const MB_MARK_CAP: i64 = 1048576          // marker file window; 400 days x 8 families x ~64 B is 200 KB
66const MB_ARGV_MAX: i64 = 8
67const MB_FRAG_CAP: i64 = 1024
68const MB_NOTE_CAP: i64 = 256
69const MB_DIGITS_MAX: i64 = 18             // an i64 holds 18 full decimal digits; longer runs are malformed
70const MB_HINNANT_OFFSET: i64 = 719468     // days_from_civil: 1970-01-01 is day 719468 of the proleptic era count
71const MB_ERA_DAYS: i64 = 146097
72const MB_ERA_YEARS: i64 = 400
73const MB_DOY_MUL: i64 = 153
74const MB_DOY_ADD: i64 = 2
75const MB_DOY_DIV: i64 = 5
76const MB_YEAR_DAYS: i64 = 365
77const MB_KEY_YEAR_MUL: i64 = 10000        // day key = y*10000 + m*100 + d, printable back with dashes
78const MB_KEY_MON_MUL: i64 = 100
79const MB_CELLS: i64 = 3200                // MB_DAYS_MAX * MB_FAM_MAX
80const MB_I64: i64 = 8
81
82// ---- small helpers --------------------------------------------------------------------------------------
83func mb_slot(base: *u8, k: i64) -> *u8 { return ((base as i64) + k * MB_SLOT) as *u8 }
84func mb_famslot(base: *u8, k: i64) -> *u8 { return ((base as i64) + k * MB_FAM_CAP) as *u8 }
85
86func mb_copy(dst: *u8, src: *u8, cap: i64) -> i64 {
  var i: i64 = 0
  while i < cap - 1 {
      if src[i] == (0 as u8) { dst[i] = 0 as u8; return i }
      dst[i] = src[i]
      i = i + 1
  }
  dst[i] = 0 as u8
  return i
95}
96
97func mb_ends_with(name: *u8, ext: *u8) -> i64 {
  let nl: i64 = mp_len(name)
  let el: i64 = mp_len(ext)
  if nl < el { return 0 }
  var i: i64 = 0
  while i < el { if name[nl - el + i] != ext[i] { return 0 } i = i + 1 }
  return 1
104}
105
106func mb_atoi(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
115}
116
117// find literal `lit` in buf[s..e) scanning FORWARD; returns the offset of its first byte or -1
118func mb_find(buf: *u8, s: i64, e: i64, lit: *u8) -> i64 {
  let ll: i64 = mp_len(lit)
  if ll == 0 { return 0 - 1 }
  var i: i64 = s
  while i + ll <= e {
      var j: i64 = 0
      var same: i64 = 1
      var scan: i64 = 1
      while scan == 1 {
          if j >= ll { scan = 0 } else {
              if buf[i + j] != lit[j] { same = 0; scan = 0 } else { j = j + 1 }
          }
      }
      if same == 1 { return i }
      i = i + 1
  }
  return 0 - 1
135}
136
137// find the LAST occurrence of `lit` in buf[s..e); returns its offset or -1
138func mb_rfind(buf: *u8, s: i64, e: i64, lit: *u8) -> i64 {
  let ll: i64 = mp_len(lit)
  if ll == 0 { return 0 - 1 }
  var i: i64 = e - ll
  while i >= s {
      var j: i64 = 0
      var same: i64 = 1
      var scan: i64 = 1
      while scan == 1 {
          if j >= ll { scan = 0 } else {
              if buf[i + j] != lit[j] { same = 0; scan = 0 } else { j = j + 1 }
          }
      }
      if same == 1 { return i }
      i = i - 1
  }
  return 0 - 1
155}
156
157// parse the decimal integer that starts at buf[i] and ends before e or a non-digit; -1 when no digit is there
158func mb_int_at(buf: *u8, i0: i64, e: i64) -> i64 {
  var i: i64 = i0
  var v: i64 = 0
  var digits: i64 = 0
  var scan: i64 = 1
  while scan == 1 {
      if i >= e { scan = 0 } else {
          let c: i64 = buf[i] as i64
          if c < 48 { scan = 0 } else { if c > 57 { scan = 0 } else {
              if digits >= MB_DIGITS_MAX { return 0 - 1 }
              v = v * 10 + (c - 48); digits = digits + 1; i = i + 1
          } }
      }
  }
  if digits == 0 { return 0 - 1 }
  return v
174}
175
176// the integer after `key` within buf[s..e), FIRST occurrence; -1 when absent or unparseable
177func mb_int_after(buf: *u8, s: i64, e: i64, key: *u8) -> i64 {
  let p: i64 = mb_find(buf, s, e, key)
  if p < 0 { return 0 - 1 }
  return mb_int_at(buf, p + mp_len(key), e)
181}
182
183// Hinnant days_from_civil, valid for every civil date after 0000-03-01; integer arithmetic only
184func mb_days_from_civil(y0: i64, m: i64, d: i64) -> i64 {
  var y: i64 = y0
  if m <= 2 { y = y - 1 }
  let era: i64 = y / MB_ERA_YEARS
  let yoe: i64 = y - era * MB_ERA_YEARS
  var mm: i64 = m + 9
  if m > 2 { mm = m - 3 }
  let doy: i64 = (MB_DOY_MUL * mm + MB_DOY_ADD) / MB_DOY_DIV + d - 1
  let doe: i64 = yoe * MB_YEAR_DAYS + yoe / 4 - yoe / 100 + doy
  return era * MB_ERA_DAYS + doe - MB_HINNANT_OFFSET
194}
195
196// YYYY-MM-DD at buf[i..i+10) -> day key y*10000+m*100+d, or -1 when the shape is wrong
197func mb_daykey_at(buf: *u8, i: i64, e: i64) -> i64 {
  if i + MB_DAY_LEN > e { return 0 - 1 }
  if (buf[i + 4] as i64) != MB_DASH { return 0 - 1 }
  if (buf[i + 7] as i64) != MB_DASH { return 0 - 1 }
  let y: i64 = mb_int_at(buf, i, i + 4)
  let m: i64 = mb_int_at(buf, i + 5, i + 7)
  let d: i64 = mb_int_at(buf, i + 8, i + 10)
  if y < 0 { return 0 - 1 }
  if m < 1 { return 0 - 1 }
  if m > 12 { return 0 - 1 }
  if d < 1 { return 0 - 1 }
  if d > 31 { return 0 - 1 }
  return y * MB_KEY_YEAR_MUL + m * MB_KEY_MON_MUL + d
210}
211
212func mb_key_daynum(key: i64) -> i64 {
  let y: i64 = key / MB_KEY_YEAR_MUL
  let m: i64 = (key / MB_KEY_MON_MUL) % MB_KEY_MON_MUL
  let d: i64 = key % MB_KEY_MON_MUL
  return mb_days_from_civil(y, m, d)
217}
218
219// print a day key as YYYY-MM-DD into out at off
220func mb_cat_key(out: *u8, off: i64, key: i64) -> i64 {
  var o: i64 = off
  o = mp_catn(out, o, key / MB_KEY_YEAR_MUL)
  out[o] = MB_DASH as u8; o = o + 1
  let m: i64 = (key / MB_KEY_MON_MUL) % MB_KEY_MON_MUL
  if m < 10 { out[o] = 48 as u8; o = o + 1 }
  o = mp_catn(out, o, m)
  out[o] = MB_DASH as u8; o = o + 1
  let d: i64 = key % MB_KEY_MON_MUL
  if d < 10 { out[o] = 48 as u8; o = o + 1 }
  o = mp_catn(out, o, d)
  return o
232}
233
234func mb_mtime(path: *u8) -> i64 {
  let sb: *u8 = sys_mmap(MB_STATBUF)
  let r: i64 = sys_fstatat(path, sb)
  if r < 0 { sys_munmap(sb, MB_STATBUF); return 0 - 1 }
  let sw: *i64 = sb as *i64
  let mt: i64 = sw[MB_STAT_MTIME_SLOT]
  sys_munmap(sb, MB_STATBUF)
  return mt
242}
243
244// ---- the per-run tables ---------------------------------------------------------------------------------
245// keys[d] day key; fams[f] family token; cell c = d*MB_FAM_MAX + f; per cell: msgs, out, cread, ccre, files.
246// ctr slots: 0 files_seen 1 files_read 2 files_skipped_mtime 3 lines 4 assistant 5 msgs 6 dup_blocks
247//            7 malformed 8 partial_files 9 days 10 fams 11 days_floor 12 files_floor 13 statfail 14 outside_window
248const MB_C_FILES: i64 = 0
249const MB_C_READ: i64 = 1
250const MB_C_SKIPPED: i64 = 2
251const MB_C_LINES: i64 = 3
252const MB_C_ASSIST: i64 = 4
253const MB_C_MSGS: i64 = 5
254const MB_C_DUP: i64 = 6
255const MB_C_MALFORMED: i64 = 7
256const MB_C_PARTIAL: i64 = 8
257const MB_C_DAYS: i64 = 9
258const MB_C_FAMS: i64 = 10
259const MB_C_DAYSFLOOR: i64 = 11
260const MB_C_FILESFLOOR: i64 = 12
261const MB_C_STATFAIL: i64 = 13
262const MB_C_OUTSIDE: i64 = 14
263const MB_C_SYNTH: i64 = 15                // harness-written API-error lines: model "<synthetic>", no msg_ id, nothing to meter
264const MB_C_RECOVERED: i64 = 16            // rows the estate log already held that the local marker had lost (a dropped receipt)
265const MB_C_DEFERRED: i64 = 17             // rows NOT pushed because the estate log could not be read first
266const MB_C_SLOTS: i64 = 24
267const MB_LOG_CAP: i64 = 262144            // one meterlog fetch: 400 rows x ~120 B escaped fits 5x over
268const MB_LOG_ROWS: *u8 = "400"            // tail rows to read back; 40 days x 2 families x 2 retries is 160
269const MB_ROW_BACK: i64 = 200              // a log row is epoch TAB seat TAB 4 numbers TAB note: the seat sits within 200 B before its note
270
271func mb_day_index(keys: *i64, ctr: *i64, key: i64) -> i64 {
  var i: i64 = 0
  let n: i64 = ctr[MB_C_DAYS]
  while i < n { if keys[i] == key { return i } i = i + 1 }
  if n >= MB_DAYS_MAX { ctr[MB_C_DAYSFLOOR] = 1; return 0 - 1 }
  keys[n] = key
  ctr[MB_C_DAYS] = n + 1
  return n
279}
280
281// find `key` where it starts a DEPTH-1 field of the object whose body begins at s (just past its '{'). Strings
282// are skipped with their escapes and nested objects and arrays are skipped whole, so a tool_use input that
283// carries its own "model" or "id" (raw JSON inside content) can never be mistaken for the message's field.
284// Returns the key's offset, or -1 when the object closes first or the key is absent.
285func mb_depth1_find(buf: *u8, s: i64, e: i64, key: *u8) -> i64 {
  let kl: i64 = mp_len(key)
  var depth: i64 = 1
  var instr: i64 = 0
  var i: i64 = s
  while i < e {
      let c: i64 = buf[i] as i64
      if instr == 1 {
          if c == 92 { i = i + 1 } else { if c == MB_QUOTE { instr = 0 } }
      } else {
          if c == MB_QUOTE {
              if depth == 1 { if i + kl <= e { if mb_find(buf, i, i + kl, key) == i { return i } } }
              instr = 1
          } else {
              if c == 123 { depth = depth + 1 } else { if c == 91 { depth = depth + 1 } else {
                  if c == 125 { depth = depth - 1 } else { if c == 93 { depth = depth - 1 } }
              } }
              if depth <= 0 { return 0 - 1 }
          }
      }
      i = i + 1
  }
  return 0 - 1
308}
309
310// family token = the run between "claude-" and the next '-' in the model id at offset p (the message's OWN
311// model field, found at depth 1 by the caller); p < 0 files as "other"
312func mb_fam_index(fams: *u8, ctr: *i64, buf: *u8, p: i64, e: i64) -> i64 {
  let tok: *u8 = sys_mmap(MB_FAM_CAP)
  var tl: i64 = 0
  if p < 0 { tl = mp_cat(tok, 0, "other" as *u8) } else {
      var i: i64 = p + mp_len("\"model\":\"claude-" as *u8)
      var scan: i64 = 1
      while scan == 1 {
          if i >= e { scan = 0 } else {
              let c: i64 = buf[i] as i64
              if c == MB_DASH { scan = 0 } else { if c == MB_QUOTE { scan = 0 } else {
                  if tl < MB_FAM_CAP - 1 { tok[tl] = c as u8; tl = tl + 1 }
                  i = i + 1
              } }
          }
      }
      if tl == 0 { tl = mp_cat(tok, 0, "other" as *u8) }
  }
  tok[tl] = 0 as u8
  var f: i64 = 0
  let n: i64 = ctr[MB_C_FAMS]
  while f < n { if mp_streq(mb_famslot(fams, f), tok) == 1 { sys_munmap(tok, MB_FAM_CAP); return f } f = f + 1 }
  if n >= MB_FAM_MAX { sys_munmap(tok, MB_FAM_CAP); return 0 - 1 }
  mb_copy(mb_famslot(fams, n), tok, MB_FAM_CAP)
  ctr[MB_C_FAMS] = n + 1
  sys_munmap(tok, MB_FAM_CAP)
  return n
338}
339
340// scan ONE transcript into the tables. ring/ringn: the last MB_ID_RING message ids seen in this file (per-file dedupe).
341// fileday[f]: the last day key this file was counted as active for family f (per-file session counting).
342func mb_scan_file(path: *u8, keys: *i64, fams: *u8, msgs: *i64, outt: *i64, cread: *i64, ccre: *i64, files: *i64,
                ctr: *i64, lo_day: i64, hi_day: i64) -> i64 {
  let lenp: *i64 = sys_mmap(MB_I64) as *i64
  let buf: *u8 = sys_read_file(path, lenp)
  let n: i64 = lenp[0]
  if (buf as i64) == 0 { sys_munmap(lenp as *u8, MB_I64); return 0 - 1 }
  let ring: *u8 = sys_mmap(MB_ID_RING * MB_ID_CAP)
  var ringn: i64 = 0
  var ringw: i64 = 0
  let fileday: *i64 = sys_mmap(MB_FAM_MAX * MB_I64) as *i64
  var f0: i64 = 0
  while f0 < MB_FAM_MAX { fileday[f0] = 0 - 1; f0 = f0 + 1 }
  var i: i64 = 0
  var partial: i64 = 0
  while i < n {
      var e: i64 = i
      var scan: i64 = 1
      while scan == 1 { if e >= n { scan = 0 } else { if (buf[e] as i64) == MB_NL { scan = 0 } else { e = e + 1 } } }
      if e >= n { partial = 1 } else {
          ctr[MB_C_LINES] = ctr[MB_C_LINES] + 1
          // a raw "type":"assistant" is written only by the harness at top level or inside an assistant
          // tool_use input (tool results are escaped strings), so its presence classifies the line
          var isasst: i64 = 0
          if mb_find(buf, i, e, "\"type\":\"assistant\"" as *u8) >= 0 { isasst = 1 }
          if isasst == 1 {
              ctr[MB_C_ASSIST] = ctr[MB_C_ASSIST] + 1
              // the message's OWN fields are the depth-1 children of "message":{ -- a tool_use input inside
              // content is raw JSON too and may carry "model":"..." or "id":"..." of its own (MEASURED: two
              // messages on 2026-09-01 carried five raw model fields each, and position on the line filed
              // them under the wrong family), so depth decides, never position
              let mp: i64 = mb_find(buf, i, e, "\"message\":{" as *u8)
              var ms: i64 = 0 - 1
              if mp >= 0 { ms = mp + mp_len("\"message\":{" as *u8) }
              var up: i64 = 0 - 1
              var idp: i64 = 0 - 1
              var modp: i64 = 0 - 1
              if ms >= 0 {
                  up = mb_depth1_find(buf, ms, e, "\"usage\":{" as *u8)
                  idp = mb_depth1_find(buf, ms, e, "\"id\":\"msg_" as *u8)
                  modp = mb_depth1_find(buf, ms, e, "\"model\":\"claude-" as *u8)
              }
              let tsp: i64 = mb_rfind(buf, i, e, "\"timestamp\":\"" as *u8)
              var ok: i64 = 1
              if up < 0 { ok = 0 }
              if tsp < 0 { ok = 0 }
              if idp < 0 { ok = 0 }
              var key: i64 = 0 - 1
              var outv: i64 = 0 - 1
              var crv: i64 = 0 - 1
              var ccv: i64 = 0 - 1
              if ok == 1 {
                  key = mb_daykey_at(buf, tsp + mp_len("\"timestamp\":\"" as *u8), e)
                  outv = mb_int_after(buf, up, e, "\"output_tokens\":" as *u8)
                  crv = mb_int_after(buf, up, e, "\"cache_read_input_tokens\":" as *u8)
                  ccv = mb_int_after(buf, up, e, "\"cache_creation_input_tokens\":" as *u8)
                  if key < 0 { ok = 0 }
                  if outv < 0 { ok = 0 }
                  if crv < 0 { ok = 0 }
                  if ccv < 0 { ok = 0 }
              }
              // an API-error line is written by the harness with model "<synthetic>" and no msg_ id (MEASURED
              // 2026-09-02: 15 of them in two days); it is a named bucket, never MALFORMED and never a message
              if ok == 0 { if mb_find(buf, i, e, "\"model\":\"<synthetic>\"" as *u8) >= 0 { ctr[MB_C_SYNTH] = ctr[MB_C_SYNTH] + 1 } else { ctr[MB_C_MALFORMED] = ctr[MB_C_MALFORMED] + 1 } } else {
                  // per-file dedupe: the id span "msg_..." up to the closing quote
                  let ids: i64 = idp + mp_len("\"id\":\"" as *u8)
                  var ide: i64 = ids
                  var s2: i64 = 1
                  while s2 == 1 { if ide >= e { s2 = 0 } else { if (buf[ide] as i64) == MB_QUOTE { s2 = 0 } else { ide = ide + 1 } } }
                  var idl: i64 = ide - ids
                  if idl > MB_ID_CAP - 1 { idl = MB_ID_CAP - 1 }
                  var dup: i64 = 0
                  var r: i64 = 0
                  while r < ringn {
                      let rs: *u8 = ((ring as i64) + r * MB_ID_CAP) as *u8
                      var same: i64 = 1
                      var k: i64 = 0
                      while k < idl { if rs[k] != buf[ids + k] { same = 0 } k = k + 1 }
                      if rs[idl] != (0 as u8) { same = 0 }
                      if same == 1 { dup = 1 }
                      r = r + 1
                  }
                  if dup == 1 { ctr[MB_C_DUP] = ctr[MB_C_DUP] + 1 } else {
                      let ws: *u8 = ((ring as i64) + ringw * MB_ID_CAP) as *u8
                      var k2: i64 = 0
                      while k2 < idl { ws[k2] = buf[ids + k2]; k2 = k2 + 1 }
                      ws[idl] = 0 as u8
                      ringw = (ringw + 1) % MB_ID_RING
                      if ringn < MB_ID_RING { ringn = ringn + 1 }
                      let dn: i64 = mb_key_daynum(key)
                      var inwin: i64 = 1
                      if dn < lo_day { inwin = 0 }
                      if dn > hi_day { inwin = 0 }
                      if inwin == 0 { ctr[MB_C_OUTSIDE] = ctr[MB_C_OUTSIDE] + 1 } else {
                          let d: i64 = mb_day_index(keys, ctr, key)
                          let f: i64 = mb_fam_index(fams, ctr, buf, modp, e)
                          if d < 0 { ctr[MB_C_MALFORMED] = ctr[MB_C_MALFORMED] + 1 } else { if f < 0 { ctr[MB_C_MALFORMED] = ctr[MB_C_MALFORMED] + 1 } else {
                              let c: i64 = d * MB_FAM_MAX + f
                              msgs[c] = msgs[c] + 1
                              outt[c] = outt[c] + outv
                              cread[c] = cread[c] + crv
                              ccre[c] = ccre[c] + ccv
                              ctr[MB_C_MSGS] = ctr[MB_C_MSGS] + 1
                              if fileday[f] != key { files[c] = files[c] + 1; fileday[f] = key }
                          } }
                      }
                  }
              }
          }
      }
      i = e + 1
  }
  if partial == 1 { ctr[MB_C_PARTIAL] = ctr[MB_C_PARTIAL] + 1 }
  sys_munmap(ring, MB_ID_RING * MB_ID_CAP)
  sys_munmap(fileday as *u8, MB_FAM_MAX * MB_I64)
  sys_munmap(buf, n + MB_READ_SLACK)
  sys_munmap(lenp as *u8, MB_I64)
  return 0
459}
460
461// ---- RESIDUE (ES11 sm_residue): which bytes of a session were REASONING and which were MECHANICAL --------------
462// The 2026-07-21 hand measurement (92.9 percent mechanical, 7.1 percent reasoning over 222 transcripts) as a standing
463// organ. Four buckets partition every content byte a line carries, measured as stored (escaped JSON string bytes):
464//   reasoning   assistant text and thinking blocks      -- the residue only a frontier seat produces today
465//   authored    tool_use input objects                   -- code and commands the seat wrote (the local maker's target)
466//   mechanical  tool_result content on user lines        -- what tools returned and the context re-reads every turn
467//   prompt      plain user text                          -- the operator
468// Per transcript in the window one RESIDUE row (a session's own permil), then the window total. A byte that fits no
469// bucket is counted as OTHER so the partition still sums.
470const MB_R_REASON: i64 = 0
471const MB_R_AUTHORED: i64 = 1
472const MB_R_MECH: i64 = 2
473const MB_R_PROMPT: i64 = 3
474const MB_R_OTHER: i64 = 4
475const MB_R_LINES: i64 = 5
476const MB_R_SLOTS: i64 = 8
477const MB_LBRACE: i64 = 123
478const MB_RBRACE: i64 = 125
479const MB_LBRACK: i64 = 91
480const MB_RBRACK: i64 = 93
481const MB_COMMA: i64 = 44
482const MB_PERMIL: i64 = 1000
483
484// index of the closing unescaped quote of the string whose first byte is buf[s]; e when unterminated
485func mb_str_end(buf: *u8, s: i64, e: i64) -> i64 {
  var i: i64 = s
  while i < e {
      let c: i64 = buf[i] as i64
      if c == 92 { i = i + 2 } else { if c == MB_QUOTE { return i } else { i = i + 1 } }
  }
  return e
492}
493// index just past the value that starts at buf[s] (object, array, string, or a bare scalar up to , ] })
494func mb_skip_value(buf: *u8, s: i64, e: i64) -> i64 {
  if s >= e { return e }
  let c0: i64 = buf[s] as i64
  if c0 == MB_QUOTE { let q: i64 = mb_str_end(buf, s + 1, e); if q >= e { return e } return q + 1 }
  if c0 == MB_LBRACE { } else { if c0 == MB_LBRACK { } else {
      var j: i64 = s
      var go: i64 = 1
      while go == 1 { if j >= e { go = 0 } else { let c: i64 = buf[j] as i64; if c == MB_COMMA { go = 0 } else { if c == MB_RBRACE { go = 0 } else { if c == MB_RBRACK { go = 0 } else { j = j + 1 } } } } }
      return j
  } }
  var depth: i64 = 0
  var instr: i64 = 0
  var i: i64 = s
  while i < e {
      let c: i64 = buf[i] as i64
      if instr == 1 {
          if c == 92 { i = i + 1 } else { if c == MB_QUOTE { instr = 0 } }
      } else {
          if c == MB_QUOTE { instr = 1 } else {
              if c == MB_LBRACE { depth = depth + 1 } else { if c == MB_LBRACK { depth = depth + 1 } else {
                  if c == MB_RBRACE { depth = depth - 1 } else { if c == MB_RBRACK { depth = depth - 1 } }
              } }
              if depth == 0 { return i + 1 }
          }
      }
      i = i + 1
  }
  return e
522}
523// byte length of the string value of depth-1 key `key` inside the object body starting at s; 0 when absent
524func mb_strlen_of(buf: *u8, s: i64, e: i64, key: *u8) -> i64 {
  let p: i64 = mb_depth1_find(buf, s, e, key)
  if p < 0 { return 0 }
  let vs: i64 = p + mp_len(key)
  return mb_str_end(buf, vs, e) - vs
529}
530// byte length of the object value of depth-1 key `key` (key literal ends with :{ ), 0 when absent
531func mb_objlen_of(buf: *u8, s: i64, e: i64, key: *u8) -> i64 {
  let p: i64 = mb_depth1_find(buf, s, e, key)
  if p < 0 { return 0 }
  let os: i64 = p + mp_len(key) - 1
  return mb_skip_value(buf, os, e) - os
536}
537// walk the content ARRAY whose '[' is at buf[a]; per element object, add its bytes to the right bucket by "type"
538func mb_residue_content(buf: *u8, a: i64, e: i64, isasst: i64, acc: *i64) -> i64 {
  var i: i64 = a + 1
  var go: i64 = 1
  while go == 1 {
      if i >= e { go = 0 } else {
          let c: i64 = buf[i] as i64
          if c == MB_RBRACK { go = 0 } else { if c == MB_LBRACE {
              let es: i64 = i + 1
              let ee: i64 = mb_skip_value(buf, i, e)
              let tp: i64 = mb_depth1_find(buf, es, ee, "\"type\":\"" as *u8)
              var kind: i64 = MB_R_OTHER
              var bytes: i64 = ee - i
              if tp >= 0 {
                  let ts: i64 = tp + mp_len("\"type\":\"" as *u8)
                  if mb_find(buf, ts, ee, "text\"" as *u8) == ts { if isasst == 1 { kind = MB_R_REASON } else { kind = MB_R_PROMPT } bytes = mb_strlen_of(buf, es, ee, "\"text\":\"" as *u8) }
                  if mb_find(buf, ts, ee, "thinking\"" as *u8) == ts { kind = MB_R_REASON; bytes = mb_strlen_of(buf, es, ee, "\"thinking\":\"" as *u8) }
                  if mb_find(buf, ts, ee, "tool_use\"" as *u8) == ts { kind = MB_R_AUTHORED; bytes = mb_objlen_of(buf, es, ee, "\"input\":{" as *u8) }
                  if mb_find(buf, ts, ee, "tool_result\"" as *u8) == ts {
                      kind = MB_R_MECH
                      let cp: i64 = mb_depth1_find(buf, es, ee, "\"content\":" as *u8)
                      if cp < 0 { bytes = 0 } else {
                          let vs: i64 = cp + mp_len("\"content\":" as *u8)
                          if (buf[vs] as i64) == MB_QUOTE { bytes = mb_str_end(buf, vs + 1, ee) - (vs + 1) } else {
                              bytes = mb_skip_value(buf, vs, ee) - vs
                          }
                      }
                  }
              }
              acc[kind] = acc[kind] + bytes
              i = ee
          } else { i = i + 1 } }
      }
  }
  return 0
572}
573// one line: classify its message content into the buckets (only lines with a "message":{ object carry content)
574func mb_residue_line(buf: *u8, i: i64, e: i64, acc: *i64) -> i64 {
  acc[MB_R_LINES] = acc[MB_R_LINES] + 1
  let mp: i64 = mb_find(buf, i, e, "\"message\":{" as *u8)
  if mp < 0 { return 0 }
  let ms: i64 = mp + mp_len("\"message\":{" as *u8)
  var isasst: i64 = 0
  if mb_find(buf, i, e, "\"type\":\"assistant\"" as *u8) >= 0 { isasst = 1 }
  let cp: i64 = mb_depth1_find(buf, ms, e, "\"content\":" as *u8)
  if cp < 0 { return 0 }
  let vs: i64 = cp + mp_len("\"content\":" as *u8)
  let c0: i64 = buf[vs] as i64
  if c0 == MB_QUOTE {
      let n: i64 = mb_str_end(buf, vs + 1, e) - (vs + 1)
      if isasst == 1 { acc[MB_R_REASON] = acc[MB_R_REASON] + n } else { acc[MB_R_PROMPT] = acc[MB_R_PROMPT] + n }
      return 0
  }
  if c0 == MB_LBRACK { mb_residue_content(buf, vs, e, isasst, acc) }
  return 0
592}
593// per transcript: sum the buckets; write ONE RESIDUE row; add to the window accumulator
594func mb_residue_file(path: *u8, name: *u8, ob: *u8, op: *i64, win: *i64) -> i64 {
  let lenp: *i64 = sys_mmap(MB_I64) as *i64
  let buf: *u8 = sys_read_file(path, lenp)
  let n: i64 = lenp[0]
  if (buf as i64) == 0 { sys_munmap(lenp as *u8, MB_I64); return 0 - 1 }
  let acc: *i64 = sys_mmap(MB_R_SLOTS * MB_I64) as *i64
  var k: i64 = 0
  while k < MB_R_SLOTS { acc[k] = 0; k = k + 1 }
  var i: i64 = 0
  while i < n {
      var e: i64 = i
      var scan: i64 = 1
      while scan == 1 { if e >= n { scan = 0 } else { if (buf[e] as i64) == MB_NL { scan = 0 } else { e = e + 1 } } }
      if e < n { mb_residue_line(buf, i, e, acc) }
      i = e + 1
  }
  let total: i64 = acc[MB_R_REASON] + acc[MB_R_AUTHORED] + acc[MB_R_MECH] + acc[MB_R_PROMPT] + acc[MB_R_OTHER]
  var o: i64 = op[0]
  o = mp_cat(ob, o, "RESIDUE file=" as *u8); o = mp_cat(ob, o, name)
  o = mp_cat(ob, o, " lines=" as *u8); o = mp_catn(ob, o, acc[MB_R_LINES])
  o = mp_cat(ob, o, " reasoning=" as *u8); o = mp_catn(ob, o, acc[MB_R_REASON])
  o = mp_cat(ob, o, " authored=" as *u8); o = mp_catn(ob, o, acc[MB_R_AUTHORED])
  o = mp_cat(ob, o, " mechanical=" as *u8); o = mp_catn(ob, o, acc[MB_R_MECH])
  o = mp_cat(ob, o, " prompt=" as *u8); o = mp_catn(ob, o, acc[MB_R_PROMPT])
  o = mp_cat(ob, o, " other=" as *u8); o = mp_catn(ob, o, acc[MB_R_OTHER])
  o = mp_cat(ob, o, " total=" as *u8); o = mp_catn(ob, o, total)
  o = mp_cat(ob, o, " reasoning_permil=" as *u8)
  if total > 0 { o = mp_catn(ob, o, (acc[MB_R_REASON] * MB_PERMIL) / total) } else { o = mp_cat(ob, o, "UNMEASURED" as *u8) }
  o = mp_cat(ob, o, "\n" as *u8)
  op[0] = o
  k = 0
  while k < MB_R_SLOTS { win[k] = win[k] + acc[k]; k = k + 1 }
  sys_munmap(acc as *u8, MB_R_SLOTS * MB_I64)
  sys_munmap(buf, n + MB_READ_SLACK)
  sys_munmap(lenp as *u8, MB_I64)
  return 0
630}
631
632// walk <dir> for *.jsonl newer than min_mtime and scan each. Returns 0, or -1 when the dir cannot be opened.
633// mode 0 = the meter tables; mode 1 = RESIDUE rows into ob/op with the window accumulator win.
634func mb_scan_dir(dir: *u8, min_mtime: i64, keys: *i64, fams: *u8, msgs: *i64, outt: *i64, cread: *i64, ccre: *i64,
               files: *i64, ctr: *i64, lo_day: i64, hi_day: i64, mode: i64, ob: *u8, op: *i64, win: *i64) -> i64 {
  let fd: i64 = sys_openat_rd(dir)
  if fd < 0 { return 0 - 1 }
  let dbuf: *u8 = sys_mmap(MB_DENTBUF)
  let path: *u8 = sys_mmap(MB_PATH)
  var run: i64 = 1
  while run == 1 {
      let got: i64 = sys_getdents64(fd, dbuf, MB_DENTBUF)
      if got <= 0 { run = 0 } else {
          var off: i64 = 0
          var inner: i64 = 1
          while inner == 1 {
              if off >= got { inner = 0 } else {
                  let rec: *u8 = ((dbuf as i64) + off) as *u8
                  let reclen: i64 = dirent_reclen(rec)
                  if reclen <= 0 { inner = 0 } else {
                      let name: *u8 = dirent_name(rec)
                      var take: i64 = 1
                      if (name[0] as i64) == MB_DOT { take = 0 }
                      if take == 1 { if mb_ends_with(name, ".jsonl" as *u8) == 0 { take = 0 } }
                      if take == 1 {
                          ctr[MB_C_FILES] = ctr[MB_C_FILES] + 1
                          if ctr[MB_C_FILES] > MB_FILES_MAX { ctr[MB_C_FILESFLOOR] = 1 } else {
                              mp_join(path, dir, name)
                              let mt: i64 = mb_mtime(path)
                              if mt < 0 { ctr[MB_C_STATFAIL] = ctr[MB_C_STATFAIL] + 1 } else {
                                  if mt < min_mtime { ctr[MB_C_SKIPPED] = ctr[MB_C_SKIPPED] + 1 } else {
                                      var r: i64 = 0
                                      if mode == 1 { r = mb_residue_file(path, name, ob, op, win) } else { r = mb_scan_file(path, keys, fams, msgs, outt, cread, ccre, files, ctr, lo_day, hi_day) }
                                      if r < 0 { ctr[MB_C_STATFAIL] = ctr[MB_C_STATFAIL] + 1 } else { ctr[MB_C_READ] = ctr[MB_C_READ] + 1 }
                                  }
                              }
                          }
                      }
                      off = off + reclen
                  }
              }
          }
      }
  }
  sys_close(fd)
  sys_munmap(dbuf, MB_DENTBUF)
  sys_munmap(path, MB_PATH)
  return 0
679}
680
681// ---- marker file: one line per pushed (day, family): YYYY-MM-DD\tfamily\tepoch\tsessions\tout\tcread\tccreate ----
682func mb_marked(mark: *u8, mn: i64, key: i64, fam: *u8) -> i64 {
  let probe: *u8 = sys_mmap(MB_NOTE_CAP)
  var o: i64 = mb_cat_key(probe, 0, key)
  probe[o] = 9 as u8; o = o + 1
  o = mp_cat(probe, o, fam)
  probe[o] = 9 as u8; o = o + 1
  probe[o] = 0 as u8
  var found: i64 = 0
  // a marker line starts the file or follows a newline -- an anchored match, never a substring of a note
  var i: i64 = 0
  while i < mn {
      var at: i64 = 0
      if i == 0 { at = 1 } else { if (mark[i - 1] as i64) == MB_NL { at = 1 } }
      if at == 1 { if mb_find(mark, i, mn, probe) == i { found = 1 } }
      i = i + 1
  }
  sys_munmap(probe, MB_NOTE_CAP)
  return found
700}
701
702func mb_mark_append(markpath: *u8, key: i64, fam: *u8, now: i64, sess: i64, ov: i64, cr: i64, cc: i64) -> i64 {
  let ln: *u8 = sys_mmap(MB_NOTE_CAP)
  var o: i64 = mb_cat_key(ln, 0, key)
  ln[o] = 9 as u8; o = o + 1
  o = mp_cat(ln, o, fam)
  ln[o] = 9 as u8; o = o + 1
  o = mp_catn(ln, o, now)
  ln[o] = 9 as u8; o = o + 1
  o = mp_catn(ln, o, sess)
  ln[o] = 9 as u8; o = o + 1
  o = mp_catn(ln, o, ov)
  ln[o] = 9 as u8; o = o + 1
  o = mp_catn(ln, o, cr)
  ln[o] = 9 as u8; o = o + 1
  o = mp_catn(ln, o, cc)
  ln[o] = MB_NL as u8; o = o + 1
  let fd: i64 = sys_openat_append(markpath, 420)
  if fd < 0 { sys_munmap(ln, MB_NOTE_CAP); return 0 - 1 }
  let w: i64 = sys_write(fd, ln, o)
  sys_close(fd)
  sys_munmap(ln, MB_NOTE_CAP)
  if w != o { return 0 - 1 }
  return 0
725}
726
727// push ONE row through the sovereign MCP client: nx_mcp_call <base_url> nx_seat <cap_file> "argv":[meter ...]
728func mb_push(mcp: *u8, base: *u8, cap: *u8, key: i64, fam: *u8, sess: i64, ov: i64, cr: i64, cc: i64, msgs: i64, resp: *u8) -> i64 {
  let frag: *u8 = sys_mmap(MB_FRAG_CAP)
  var o: i64 = mp_cat(frag, 0, "\"argv\":[\"meter\",\"claude-" as *u8)
  o = mp_cat(frag, o, fam)
  o = mp_cat(frag, o, "-daily\",\"" as *u8)
  o = mp_catn(frag, o, sess)
  o = mp_cat(frag, o, "\",\"" as *u8)
  o = mp_catn(frag, o, ov)
  o = mp_cat(frag, o, "\",\"" as *u8)
  o = mp_catn(frag, o, cr)
  o = mp_cat(frag, o, "\",\"" as *u8)
  o = mp_catn(frag, o, cc)
  o = mp_cat(frag, o, "\",\"beat-" as *u8)
  o = mb_cat_key(frag, o, key)
  o = mp_cat(frag, o, "-msgs=" as *u8)
  o = mp_catn(frag, o, msgs)
  o = mp_cat(frag, o, "-files=" as *u8)
  o = mp_catn(frag, o, sess)
  o = mp_cat(frag, o, "-src=nx_seat_meter_beat\"]" as *u8)
  frag[o] = 0 as u8
  let av: *i64 = sys_mmap(MB_ARGV_MAX * MB_I64) as *i64
  av[0] = mcp as i64; av[1] = base as i64; av[2] = "nx_seat" as i64; av[3] = cap as i64; av[4] = frag as i64; av[5] = 0
  let rl: *i64 = sys_mmap(MB_I64) as *i64
  let rc: i64 = tr_run_capture_to(mcp, av, resp, MB_RESP_CAP, rl, MB_FORK_TIMEOUT_MS)
  let n: i64 = rl[0]
  if n >= 0 { if n < MB_RESP_CAP { resp[n] = 0 as u8 } else { resp[MB_RESP_CAP - 1] = 0 as u8 } } else { resp[0] = 0 as u8 }
  var ok: i64 = 0
  if n > 0 { if mb_find(resp, 0, n, "METERED row appended" as *u8) >= 0 { ok = 1 } }
  sys_munmap(frag, MB_FRAG_CAP)
  sys_munmap(av as *u8, MB_ARGV_MAX * MB_I64)
  sys_munmap(rl as *u8, MB_I64)
  if rc < 0 { return 0 - 1 }
  if ok == 0 { return 0 - 2 }
  return 0
762}
763
764// read the estate's meter log tail ONCE per run: nx_mcp_call <base_url> nx_seat <cap_file> "argv":["meterlog","400"].
765// Returns the captured byte count, or -1 when the transport failed or the reply carries no METERLOG header.
766func mb_fetch_meterlog(mcp: *u8, base: *u8, capf: *u8, log: *u8) -> i64 {
  let frag: *u8 = sys_mmap(MB_FRAG_CAP)
  var o: i64 = mp_cat(frag, 0, "\"argv\":[\"meterlog\",\"" as *u8)
  o = mp_cat(frag, o, MB_LOG_ROWS)
  o = mp_cat(frag, o, "\"]" as *u8)
  frag[o] = 0 as u8
  let av: *i64 = sys_mmap(MB_ARGV_MAX * MB_I64) as *i64
  av[0] = mcp as i64; av[1] = base as i64; av[2] = "nx_seat" as i64; av[3] = capf as i64; av[4] = frag as i64; av[5] = 0
  let rl: *i64 = sys_mmap(MB_I64) as *i64
  let rc: i64 = tr_run_capture_to(mcp, av, log, MB_LOG_CAP, rl, MB_FORK_TIMEOUT_MS)
  var n: i64 = rl[0]
  if n < 0 { n = 0 }
  if n >= MB_LOG_CAP { n = MB_LOG_CAP - 1 }
  log[n] = 0 as u8
  sys_munmap(frag, MB_FRAG_CAP)
  sys_munmap(av as *u8, MB_ARGV_MAX * MB_I64)
  sys_munmap(rl as *u8, MB_I64)
  if rc < 0 { return 0 - 1 }
  if mb_find(log, 0, n, "METERLOG file_bytes=" as *u8) < 0 { return 0 - 1 }
  return n
786}
787
788// does the fetched log (JSON-escaped, so TAB reads as backslash-t) already carry this beat's row for (day, family)?
789// A row is  epoch TAB claude-<fam>-daily TAB ... TAB beat-<day>-msgs=...  so the seat token must sit within
790// MB_ROW_BACK bytes BEFORE the note; every occurrence of the note is checked, never only the first.
791func mb_remote_has(log: *u8, n: i64, key: i64, fam: *u8) -> i64 {
  let note: *u8 = sys_mmap(MB_NOTE_CAP)
  var o: i64 = mp_cat(note, 0, "beat-" as *u8)
  o = mb_cat_key(note, o, key)
  o = mp_cat(note, o, "-msgs=" as *u8)
  note[o] = 0 as u8
  let seat: *u8 = sys_mmap(MB_NOTE_CAP)
  var so: i64 = mp_cat(seat, 0, "claude-" as *u8)
  so = mp_cat(seat, so, fam)
  so = mp_cat(seat, so, "-daily" as *u8)
  seat[so] = 0 as u8
  var found: i64 = 0
  var from: i64 = 0
  var scan: i64 = 1
  while scan == 1 {
      let p: i64 = mb_find(log, from, n, note)
      if p < 0 { scan = 0 } else {
          // the seat token must sit in the SAME row: a row starts after the previous separator, which in the
          // escaped reply is the two bytes backslash-n and in a raw capture is byte 10 -- take the nearer one.
          // Without this anchor the previous row's seat (families alternate day by day) reads as this row's.
          var lo: i64 = p - MB_ROW_BACK
          if lo < 0 { lo = 0 }
          var rs: i64 = lo
          let esc: i64 = mb_rfind(log, lo, p, "\\n" as *u8)
          if esc >= 0 { rs = esc + 2 }
          var q: i64 = p - 1
          var back: i64 = 1
          while back == 1 { if q < rs { back = 0 } else { if (log[q] as i64) == MB_NL { rs = q + 1; back = 0 } else { q = q - 1 } } }
          if mb_find(log, rs, p, seat) >= 0 { found = 1; scan = 0 } else { from = p + 1 }
      }
  }
  sys_munmap(note, MB_NOTE_CAP)
  sys_munmap(seat, MB_NOTE_CAP)
  return found
825}
826
827const SM_PUSHED: i64 = 0
828const SM_ALREADY: i64 = 1
829const SM_WOULD: i64 = 2
830const SM_TRANSPORT: i64 = 3
831const SM_REFUSED: i64 = 4
832const SM_UNMARKED: i64 = 5
833const SM_RECOVERED: i64 = 6
834const SM_DEFERRED: i64 = 7
835// THE BEAT DECISION for one (day, family) row -- the ES1 contract sm_beat: the local marker first, then the estate's
836// own log (a dropped receipt is a row that LANDED without a marker -- MEASURED on this organ's first live run, where
837// one push was reported failed and the estate log held it), then the push through the sovereign MCP client, then the
838// marker append. With the estate log unreadable NOTHING is pushed: an insert is not idempotent and a blind retry
839// double-applies. Every outcome is a NAMED state, never a bare bit.
840func sm_beat(mcp: *u8, base: *u8, capf: *u8, markpath: *u8, mark: *u8, mn: i64, log: *u8, ln: i64, key: i64, fam: *u8, sess: i64, ov: i64, cr: i64, cc: i64, nmsgs: i64, dry: i64, now: i64, resp: *u8) -> i64 {
  if mb_marked(mark, mn, key, fam) == 1 { return SM_ALREADY }
  if dry == 1 { return SM_WOULD }
  if ln < 0 { return SM_DEFERRED }
  if mb_remote_has(log, ln, key, fam) == 1 {
      mb_mark_append(markpath, key, fam, now, sess, ov, cr, cc)
      return SM_RECOVERED
  }
  let pr: i64 = mb_push(mcp, base, capf, key, fam, sess, ov, cr, cc, nmsgs, resp)
  if pr == 0 - 1 { return SM_TRANSPORT }
  if pr != 0 { return SM_REFUSED }
  if mb_mark_append(markpath, key, fam, now, sess, ov, cr, cc) != 0 { return SM_UNMARKED }
  return SM_PUSHED
853}
854
855func mb_usage() -> i64 {
  let m: *u8 = "usage: nx_seat_meter_beat scan <sessions-dir> [--now <epoch>] [--days N] | beat <sessions-dir> <state-dir> <mcp-elf> <base-url> <cap-file> [--now <epoch>] [--days N] [--dry] [--json] | remotecheck <meterlog-capture> <YYYY-MM-DD> <family> | residue <sessions-dir> [--now <epoch>] [--days N]\n" as *u8
  mp_write_all(2, m, mp_len(m))
  return MB_EXIT_USAGE
859}
860
861// THE RESIDUE VERB -- ES11 contract sm_residue: every transcript touched inside the window gets one RESIDUE row and
862// the window total closes with the partition printed; exit 0 GREEN, 3 UNOBSERVABLE (dir unreadable or zero bytes)
863func sm_residue(dir: *u8, now: i64, days: i64) -> i64 {
  let today: i64 = now / MB_SEC_PER_DAY
  let lo_day: i64 = today - days
  let min_mtime: i64 = lo_day * MB_SEC_PER_DAY
  let ctr: *i64 = sys_mmap(MB_C_SLOTS * MB_I64) as *i64
  let ob: *u8 = sys_mmap(MB_OUT)
  let op: *i64 = sys_mmap(MB_I64) as *i64
  op[0] = 0
  let win: *i64 = sys_mmap(MB_R_SLOTS * MB_I64) as *i64
  var k: i64 = 0
  while k < MB_R_SLOTS { win[k] = 0; k = k + 1 }
  let sr: i64 = mb_scan_dir(dir, min_mtime, 0 as *i64, 0 as *u8, 0 as *i64, 0 as *i64, 0 as *i64, 0 as *i64, 0 as *i64, ctr, lo_day, today, 1, ob, op, win)
  var o: i64 = op[0]
  if sr < 0 {
      o = mp_cat(ob, o, "RESIDUE-TOTAL sessions_dir_unreadable=1 verdict=UNOBSERVABLE\n" as *u8)
      mp_write_all(1, ob, o)
      return MB_EXIT_UNOBS
  }
  let total: i64 = win[MB_R_REASON] + win[MB_R_AUTHORED] + win[MB_R_MECH] + win[MB_R_PROMPT] + win[MB_R_OTHER]
  o = mp_cat(ob, o, "RESIDUE-TOTAL window_days=" as *u8); o = mp_catn(ob, o, days)
  o = mp_cat(ob, o, " files_seen=" as *u8); o = mp_catn(ob, o, ctr[MB_C_FILES])
  o = mp_cat(ob, o, " files_read=" as *u8); o = mp_catn(ob, o, ctr[MB_C_READ])
  o = mp_cat(ob, o, " files_skipped_mtime=" as *u8); o = mp_catn(ob, o, ctr[MB_C_SKIPPED])
  o = mp_cat(ob, o, " lines=" as *u8); o = mp_catn(ob, o, win[MB_R_LINES])
  o = mp_cat(ob, o, " reasoning=" as *u8); o = mp_catn(ob, o, win[MB_R_REASON])
  o = mp_cat(ob, o, " authored=" as *u8); o = mp_catn(ob, o, win[MB_R_AUTHORED])
  o = mp_cat(ob, o, " mechanical=" as *u8); o = mp_catn(ob, o, win[MB_R_MECH])
  o = mp_cat(ob, o, " prompt=" as *u8); o = mp_catn(ob, o, win[MB_R_PROMPT])
  o = mp_cat(ob, o, " other=" as *u8); o = mp_catn(ob, o, win[MB_R_OTHER])
  o = mp_cat(ob, o, " total=" as *u8); o = mp_catn(ob, o, total)
  o = mp_cat(ob, o, " partition=" as *u8); o = mp_catn(ob, o, win[MB_R_REASON] + win[MB_R_AUTHORED] + win[MB_R_MECH] + win[MB_R_PROMPT] + win[MB_R_OTHER])
  if total > 0 {
      o = mp_cat(ob, o, " reasoning_permil=" as *u8); o = mp_catn(ob, o, (win[MB_R_REASON] * MB_PERMIL) / total)
      o = mp_cat(ob, o, " authored_permil=" as *u8); o = mp_catn(ob, o, (win[MB_R_AUTHORED] * MB_PERMIL) / total)
      o = mp_cat(ob, o, " mechanical_permil=" as *u8); o = mp_catn(ob, o, (win[MB_R_MECH] * MB_PERMIL) / total)
      o = mp_cat(ob, o, " verdict=GREEN\n" as *u8)
      mp_write_all(1, ob, o)
      return MB_EXIT_OK
  }
  o = mp_cat(ob, o, " reasoning_permil=UNMEASURED verdict=UNOBSERVABLE\n" as *u8)
  mp_write_all(1, ob, o)
  return MB_EXIT_UNOBS
905}
906
907func main(argc: i64, argv: *i64) -> i64 {
  if argc < 3 { sys_exit(mb_usage()); return MB_EXIT_USAGE }
  let verb: *u8 = argv[1] as *u8
  // remotecheck: the read-back ruler on a captured meterlog reply, exposed so the gate can drive it
  if mp_streq(verb, "remotecheck" as *u8) == 1 {
      if argc < 5 { sys_exit(mb_usage()); return MB_EXIT_USAGE }
      let lp: *i64 = sys_mmap(MB_I64) as *i64
      let lg: *u8 = sys_read_file(argv[2] as *u8, lp)
      let ln0: i64 = lp[0]
      if (lg as i64) == 0 { let um: *u8 = "REMOTECHECK verdict=UNOBSERVABLE (capture unreadable)\n" as *u8; mp_write_all(1, um, mp_len(um)); sys_exit(MB_EXIT_UNOBS); return MB_EXIT_UNOBS }
      let dk: i64 = mb_daykey_at(argv[3] as *u8, 0, MB_DAY_LEN)
      if dk < 0 { sys_exit(mb_usage()); return MB_EXIT_USAGE }
      let hit: i64 = mb_remote_has(lg, ln0, dk, argv[4] as *u8)
      if hit == 1 { let ym: *u8 = "REMOTECHECK verdict=REMOTE-MARKED\n" as *u8; mp_write_all(1, ym, mp_len(ym)) } else { let nm: *u8 = "REMOTECHECK verdict=REMOTE-ABSENT\n" as *u8; mp_write_all(1, nm, mp_len(nm)) }
      sys_exit(MB_EXIT_OK)
      return MB_EXIT_OK
  }
  // residue <sessions-dir> [--now <epoch>] [--days N]: the ES11 contract sm_residue, one row per transcript
  if mp_streq(verb, "residue" as *u8) == 1 {
      let rdir: *u8 = argv[2] as *u8
      var rnow: i64 = sys_now_realtime_sec()
      var rdays: i64 = MB_DAYS_DEFAULT
      var ri: i64 = 3
      while ri < argc {
          let ra: *u8 = argv[ri] as *u8
          var rused: i64 = 0
          if mp_streq(ra, "--now" as *u8) == 1 { if ri + 1 < argc { rnow = mb_atoi(argv[ri + 1] as *u8); ri = ri + 1 } rused = 1 }
          if mp_streq(ra, "--days" as *u8) == 1 { if ri + 1 < argc { rdays = mb_atoi(argv[ri + 1] as *u8); ri = ri + 1 } rused = 1 }
          if rused == 0 { sys_exit(mb_usage()); return MB_EXIT_USAGE }
          ri = ri + 1
      }
      if rdays < 1 { rdays = 1 }
      sys_exit(sm_residue(rdir, rnow, rdays))
      return 0
  }
  var isbeat: i64 = 0
  if mp_streq(verb, "beat" as *u8) == 1 { isbeat = 1 } else { if mp_streq(verb, "scan" as *u8) == 0 { sys_exit(mb_usage()); return MB_EXIT_USAGE } }
  if isbeat == 1 { if argc < 7 { sys_exit(mb_usage()); return MB_EXIT_USAGE } }
  let sessdir: *u8 = argv[2] as *u8
  var statedir: *u8 = "" as *u8
  var mcp: *u8 = "" as *u8
  var base: *u8 = "" as *u8
  var capf: *u8 = "" as *u8
  var ai: i64 = 3
  if isbeat == 1 { statedir = argv[3] as *u8; mcp = argv[4] as *u8; base = argv[5] as *u8; capf = argv[6] as *u8; ai = 7 }
  var now: i64 = sys_now_realtime_sec()
  var days: i64 = MB_DAYS_DEFAULT
  var dry: i64 = 0
  var jsonf: i64 = 0
  while ai < argc {
      let a: *u8 = argv[ai] as *u8
      var used: i64 = 0
      if mp_streq(a, "--dry" as *u8) == 1 { dry = 1; used = 1 }
      if mp_streq(a, "--json" as *u8) == 1 { jsonf = 1; used = 1 }
      if mp_streq(a, "--now" as *u8) == 1 { if ai + 1 < argc { now = mb_atoi(argv[ai + 1] as *u8); ai = ai + 1 } used = 1 }
      if mp_streq(a, "--days" as *u8) == 1 { if ai + 1 < argc { days = mb_atoi(argv[ai + 1] as *u8); ai = ai + 1 } used = 1 }
      if used == 0 { sys_exit(mb_usage()); return MB_EXIT_USAGE }
      ai = ai + 1
  }
  if days < 1 { days = 1 }
  if days > MB_DAYS_MAX - 1 { days = MB_DAYS_MAX - 1 }
  let today: i64 = now / MB_SEC_PER_DAY
  let hi_day: i64 = today - 1                       // the newest COMPLETE day
  let lo_day: i64 = today - days
  let min_mtime: i64 = (lo_day) * MB_SEC_PER_DAY     // a file last written before the window start carries no in-window day
972
  let keys: *i64 = sys_mmap(MB_DAYS_MAX * MB_I64) as *i64
  let fams: *u8 = sys_mmap(MB_FAM_MAX * MB_FAM_CAP)
  let msgs: *i64 = sys_mmap(MB_CELLS * MB_I64) as *i64
  let outt: *i64 = sys_mmap(MB_CELLS * MB_I64) as *i64
  let cread: *i64 = sys_mmap(MB_CELLS * MB_I64) as *i64
  let ccre: *i64 = sys_mmap(MB_CELLS * MB_I64) as *i64
  let files: *i64 = sys_mmap(MB_CELLS * MB_I64) as *i64
  let ctr: *i64 = sys_mmap(MB_C_SLOTS * MB_I64) as *i64
  let ob: *u8 = sys_mmap(MB_OUT)
  var o: i64 = 0
983
  let sr: i64 = mb_scan_dir(sessdir, min_mtime, keys, fams, msgs, outt, cread, ccre, files, ctr, lo_day, hi_day, 0, ob, 0 as *i64, 0 as *i64)
  if sr < 0 {
      o = mp_cat(ob, o, "METERBEAT sessions_dir_unreadable=1 dir=" as *u8)
      o = mp_cat(ob, o, sessdir)
      o = mp_cat(ob, o, " verdict=UNOBSERVABLE\n" as *u8)
      if jsonf == 1 { mp_hook_json("SessionStart" as *u8, ob, o); sys_exit(MB_EXIT_OK); return MB_EXIT_OK }
      mp_write_all(1, ob, o)
      sys_exit(MB_EXIT_UNOBS)
      return MB_EXIT_UNOBS
  }
994
  // marker file (beat only)
  let markpath: *u8 = sys_mmap(MB_PATH)
  var mark: *u8 = 0 as *u8
  var mn: i64 = 0
  if isbeat == 1 {
      mp_join(markpath, statedir, "meterbeat.days" as *u8)
      let ml: *i64 = sys_mmap(MB_I64) as *i64
      mark = sys_read_file(markpath, ml)
      mn = ml[0]
      if (mark as i64) == 0 { mn = 0 }
      sys_munmap(ml as *u8, MB_I64)
  }
  // the estate's own log, read ONCE before any push (dry runs fork nothing)
  let log: *u8 = sys_mmap(MB_LOG_CAP)
  var ln: i64 = 0 - 1
  if isbeat == 1 { if dry == 0 { ln = mb_fetch_meterlog(mcp, base, capf, log) } }
1011
  // rows, oldest day first (the table is in first-seen order; sort by key with a simple selection order)
  var pushed: i64 = 0
  var planned: i64 = 0
  var already: i64 = 0
  var failed: i64 = 0
  var sumout: i64 = 0
  var sumcr: i64 = 0
  let resp: *u8 = sys_mmap(MB_RESP_CAP)
  let nd: i64 = ctr[MB_C_DAYS]
  let nf: i64 = ctr[MB_C_FAMS]
  let done: *i64 = sys_mmap(MB_DAYS_MAX * MB_I64) as *i64
  var emitted: i64 = 0
  while emitted < nd {
      // pick the smallest not-yet-emitted key
      var best: i64 = 0 - 1
      var di: i64 = 0
      while di < nd { if done[di] == 0 { if best < 0 { best = di } else { if keys[di] < keys[best] { best = di } } } di = di + 1 }
      done[best] = 1
      emitted = emitted + 1
      var f: i64 = 0
      while f < nf {
          let c: i64 = best * MB_FAM_MAX + f
          if msgs[c] > 0 {
              let fam: *u8 = mb_famslot(fams, f)
              o = mp_cat(ob, o, "DAY " as *u8)
              o = mb_cat_key(ob, o, keys[best])
              o = mp_cat(ob, o, " fam=" as *u8); o = mp_cat(ob, o, fam)
              o = mp_cat(ob, o, " sessions=" as *u8); o = mp_catn(ob, o, files[c])
              o = mp_cat(ob, o, " msgs=" as *u8); o = mp_catn(ob, o, msgs[c])
              o = mp_cat(ob, o, " out=" as *u8); o = mp_catn(ob, o, outt[c])
              o = mp_cat(ob, o, " cread=" as *u8); o = mp_catn(ob, o, cread[c])
              o = mp_cat(ob, o, " ccreate=" as *u8); o = mp_catn(ob, o, ccre[c])
              sumout = sumout + outt[c]
              sumcr = sumcr + cread[c]
              if isbeat == 1 {
                  let st: i64 = sm_beat(mcp, base, capf, markpath, mark, mn, log, ln, keys[best], fam, files[c], outt[c], cread[c], ccre[c], msgs[c], dry, now, resp)
                  if st == SM_ALREADY { already = already + 1; o = mp_cat(ob, o, " status=ALREADY-MARKED" as *u8) } else { planned = planned + 1 }
                  if st == SM_WOULD { o = mp_cat(ob, o, " status=WOULD-PUSH" as *u8) }
                  if st == SM_RECOVERED { ctr[MB_C_RECOVERED] = ctr[MB_C_RECOVERED] + 1; o = mp_cat(ob, o, " status=RECOVERED-estate-log-already-held-it" as *u8) }
                  if st == SM_DEFERRED { ctr[MB_C_DEFERRED] = ctr[MB_C_DEFERRED] + 1; o = mp_cat(ob, o, " status=DEFERRED-estate-log-unreadable" as *u8) }
                  if st == SM_PUSHED { pushed = pushed + 1; o = mp_cat(ob, o, " status=PUSHED" as *u8) }
                  if st == SM_UNMARKED { failed = failed + 1; o = mp_cat(ob, o, " status=PUSHED-BUT-UNMARKED" as *u8) }
                  if st == SM_TRANSPORT { failed = failed + 1; o = mp_cat(ob, o, " status=PUSH-FAILED-transport" as *u8) }
                  if st == SM_REFUSED { failed = failed + 1; o = mp_cat(ob, o, " status=PUSH-REFUSED-no-METERED-receipt" as *u8) }
              }
              o = mp_cat(ob, o, "\n" as *u8)
          }
          f = f + 1
      }
  }
  o = mp_cat(ob, o, "METERBEAT window_days=" as *u8); o = mp_catn(ob, o, days)
  o = mp_cat(ob, o, " today_daynum=" as *u8); o = mp_catn(ob, o, today)
  o = mp_cat(ob, o, " files_seen=" as *u8); o = mp_catn(ob, o, ctr[MB_C_FILES])
  o = mp_cat(ob, o, " files_read=" as *u8); o = mp_catn(ob, o, ctr[MB_C_READ])
  o = mp_cat(ob, o, " files_skipped_mtime=" as *u8); o = mp_catn(ob, o, ctr[MB_C_SKIPPED])
  o = mp_cat(ob, o, " statfail=" as *u8); o = mp_catn(ob, o, ctr[MB_C_STATFAIL])
  o = mp_cat(ob, o, " lines=" as *u8); o = mp_catn(ob, o, ctr[MB_C_LINES])
  o = mp_cat(ob, o, " assistant_lines=" as *u8); o = mp_catn(ob, o, ctr[MB_C_ASSIST])
  o = mp_cat(ob, o, " msgs=" as *u8); o = mp_catn(ob, o, ctr[MB_C_MSGS])
  o = mp_cat(ob, o, " dup_blocks=" as *u8); o = mp_catn(ob, o, ctr[MB_C_DUP])
  o = mp_cat(ob, o, " outside_window=" as *u8); o = mp_catn(ob, o, ctr[MB_C_OUTSIDE])
  o = mp_cat(ob, o, " malformed=" as *u8); o = mp_catn(ob, o, ctr[MB_C_MALFORMED])
  o = mp_cat(ob, o, " synthetic=" as *u8); o = mp_catn(ob, o, ctr[MB_C_SYNTH])
  o = mp_cat(ob, o, " partial_files=" as *u8); o = mp_catn(ob, o, ctr[MB_C_PARTIAL])
  o = mp_cat(ob, o, " days=" as *u8); o = mp_catn(ob, o, nd)
  o = mp_cat(ob, o, " fams=" as *u8); o = mp_catn(ob, o, nf)
  o = mp_cat(ob, o, " days_floor=" as *u8); o = mp_catn(ob, o, ctr[MB_C_DAYSFLOOR])
  o = mp_cat(ob, o, " files_floor=" as *u8); o = mp_catn(ob, o, ctr[MB_C_FILESFLOOR])
  o = mp_cat(ob, o, " window_out=" as *u8); o = mp_catn(ob, o, sumout)
  o = mp_cat(ob, o, " window_cread=" as *u8); o = mp_catn(ob, o, sumcr)
  // partition: assistant_lines = msgs + dup_blocks + outside_window + malformed + synthetic (printed so a reader can check it)
  o = mp_cat(ob, o, " partition=" as *u8); o = mp_catn(ob, o, ctr[MB_C_MSGS] + ctr[MB_C_DUP] + ctr[MB_C_OUTSIDE] + ctr[MB_C_MALFORMED] + ctr[MB_C_SYNTH])
  if isbeat == 1 {
      o = mp_cat(ob, o, " planned=" as *u8); o = mp_catn(ob, o, planned)
      o = mp_cat(ob, o, " pushed=" as *u8); o = mp_catn(ob, o, pushed)
      o = mp_cat(ob, o, " already=" as *u8); o = mp_catn(ob, o, already)
      o = mp_cat(ob, o, " failed=" as *u8); o = mp_catn(ob, o, failed)
      o = mp_cat(ob, o, " recovered=" as *u8); o = mp_catn(ob, o, ctr[MB_C_RECOVERED])
      o = mp_cat(ob, o, " deferred=" as *u8); o = mp_catn(ob, o, ctr[MB_C_DEFERRED])
      if dry == 1 { o = mp_cat(ob, o, " remote_log=NOT-READ-dry" as *u8) } else { if ln < 0 { o = mp_cat(ob, o, " remote_log=UNREADABLE" as *u8) } else { o = mp_cat(ob, o, " remote_log=READ" as *u8) } }
      o = mp_cat(ob, o, " dry=" as *u8); o = mp_catn(ob, o, dry)
  }
  var rc: i64 = MB_EXIT_OK
  if failed > 0 { rc = MB_EXIT_RED }
  if rc == MB_EXIT_OK { o = mp_cat(ob, o, " verdict=GREEN\n" as *u8) } else { o = mp_cat(ob, o, " verdict=RED\n" as *u8) }
  if jsonf == 1 {
      // the hook envelope carries ONE line: the summary; the DAY rows stay in plain mode
      let sp: i64 = mb_rfind(ob, 0, o, "METERBEAT " as *u8)
      var s0: i64 = 0
      if sp >= 0 { s0 = sp }
      mp_hook_json("SessionStart" as *u8, ((ob as i64) + s0) as *u8, o - s0)
      sys_exit(MB_EXIT_OK)
      return MB_EXIT_OK
  }
  mp_write_all(1, ob, o)
  sys_exit(rc)
  return rc
1109}
nx_seat_meter_beat.nx source

`nx_seat_meter_beat.nx` source
