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

↩ module page · 488 lines · 33308 B
1// nx_search_ship.nx -- THE SEARCH SHIP LOOP AS ONE ORGAN (operator 2026-09-14: "make sure our primitives get
2// shifted into workflows get shifted to agents so we are 1 2 3 step"). On 2026-09-14 the seat shipped the docportal
3// search daemon four times and every ship was ten hand-pecked calls: build the daemon, build the judged-set referee,
4// run the promote ruler, run the promoted AND the staged referee and diff their per-query lines, run the sticky gate,
5// bank the live binary by hash, deploy through the health-checked /api/deploy with expect_sha256, wait out the
6// warm-before-listen window, probe cold two-common-term queries through the edge and read phase_us, then journal.
7// This organ is that loop as ONE call. It refuses at the first RED, and it composes the estate's rulers by FORK
8// (nx_mgmt_call, nx_contentdiff, nx_qrels_bench, nx_filehash, nx_filecopy, nx_https_get_cli2) -- it re-implements
9// none of them, so there is still exactly one of each. The pure parsers live in nx_search_ship_lib (gate-driven).
10// TWO MODES (the Codex method: a worker ends at a PRIVATE CANDIDATE + RECEIPT, only the root adopts):
11//   nx_search_ship dry     -- build + ruler + referee A/B + receipt; NOTHING is deployed (the candidate)
12//   nx_search_ship deploy  -- the same, then bank + deploy + warm-wait + cold probes + journal (the adoption)
13// OPTIONS: allow_loss=<n> (named lost runs the ruler may report; each is journalled) | bar=<ms> (cold-probe bar,
14// default the search-query row of knowledge/uat_journeys.conf, else 3000) | probes=<path> (one query per line,
15// default knowledge/search_probes.conf, else the journey's own query family photos).
16// REFEREE RULE: identical per-query lines = GREEN; lines moved with mean ndcg AND mrr not lower = PROCEEDS ANNOUNCED
17// with the first differing line printed and journalled (a ranking change the caller adjudicates); a lower mean
18// REFUSES. EXIT: 0 GREEN | 2 usage | 12 build refused | 11 wrong src_path | 31 ruler RED | 41 referee worse |
19// 53 bank failed | 51 deploy did not land | 52 deploy_status not GREEN | 61 daemon never answered | 71 a probe over
20// the bar. The LAST line is SEARCH-SHIP verdict=... for positional readers.
21// license_tier: ORIGINAL  expect_exit: 0
22import "nx_syscalls.nx"
23import "nx_estate_path.nx"
24import "nx_tool_run.nx"
25import "nx_search_ship_lib.nx"
26
27const SS_CAP: i64 = 262144
28const SS_SMALL: i64 = 8192
29const SS_TMO_MS: i64 = 900000
30const SS_MGMT: *u8 = "./nx_mgmt_call.elf" as *u8
31const SS_CONTENTDIFF: *u8 = "./nx_contentdiff.elf" as *u8
32const SS_FILEHASH: *u8 = "./nx_filehash.elf" as *u8
33const SS_FILECOPY: *u8 = "./nx_filecopy.elf" as *u8
34const SS_HTTPS: *u8 = "./nx_https_get_cli2.elf" as *u8
35const SS_BENCH_LIVE: *u8 = "./nx_qrels_bench.elf" as *u8
36const SS_BENCH_STAGED: *u8 = "./nx_qrels_bench.sov.elf.new" as *u8
37const SS_DAEMON: *u8 = "nx_docportal_admin_daemon" as *u8
38const SS_DAEMON_ELF: *u8 = "nx_docportal_admin_daemon.elf" as *u8
39const SS_DAEMON_NEW: *u8 = "nx_docportal_admin_daemon.sov.elf.new" as *u8
40const SS_DAEMON_SRC: *u8 = "/nx_docportal_admin_daemon.nx" as *u8
41const SS_BENCH: *u8 = "nx_qrels_bench" as *u8
42const SS_BENCH_SRC: *u8 = "/nx_qrels_bench.nx" as *u8
43const SS_DEPLOY_BODY: *u8 = "target=docportal&confirm=yes&expect_sha256=" as *u8
44const SS_EDGE: *u8 = "127.0.0.1:8443" as *u8
45const SS_URL_HEAD: *u8 = "https://nishifamily.com/api/search?q=" as *u8
46const SS_URL_TAIL: *u8 = "&scope=web" as *u8
47const SS_JOURNEYS: *u8 = "knowledge/uat_journeys.conf" as *u8
48const SS_JOURNEY_ID: *u8 = "search-query" as *u8
49const SS_PROBES: *u8 = "knowledge/search_probes.conf" as *u8
50const SS_PROBE_DEFAULT: *u8 = "family photos" as *u8
51const SS_PLAN: *u8 = "buildroot/knowledge/compare/search.plan" as *u8
52const SS_JRNL: *u8 = "knowledge/status/search_ship.jrnl" as *u8
53const SS_BANK_DIR: *u8 = "knowledge/status/bank/" as *u8
54const SS_DEFAULT_BAR_MS: i64 = 3000
55const SS_WARM_TRIES: i64 = 30
56const SS_WARM_STEP_MS: i64 = 2000
57const SS_SETTLE_MS: i64 = 1500
58const SS_MAX_PROBES: i64 = 16
59const SS_PROBE_W: i64 = 96
60const SS_LINE: i64 = 4096
61const SS_MODE_0644: i64 = 420
62const SS_QPFX: *u8 = "  q=" as *u8
63const SS_RUNG_DEFAULT: *u8 = "L3" as *u8   // the rung a ship journals under when the caller names none; rung=<id> overrides it
64static sx_rung_g: *u8
65
66func sx_puts(s: *u8) -> i64 { sys_write(1, s, sx_slen(s)); return 0 }
67func sx_pn(v: i64) -> i64 { let b: *u8 = sys_mmap(32); sx_catn(b, 0, v); sx_puts(b); sys_munmap(b, 32); return 0 }
68// print the last `lines` lines of a capture (a receipt's verdict is its tail)
69func sx_tail(buf: *u8, n: i64, lines: i64) -> i64 {
  var i: i64 = n
  var seen: i64 = 0
  var go: i64 = 1
  while go == 1 {
      if i <= 0 { go = 0 } else {
          if buf[i - 1] == (SX_LF as u8) { seen = seen + 1; if seen > lines { go = 0 } else { i = i - 1 } } else { i = i - 1 }
      }
  }
  sys_write(1, (buf as i64 + i) as *u8, n - i)
  if n > 0 { if buf[n - 1] != (SX_LF as u8) { sx_puts("\n" as *u8) } }
  return 0
81}
82// fork with up to three args (a2/a3 may be 0), capturing stdout; returns the child's rc
83func sx_fork(path: *u8, a1: *u8, a2: *u8, a3: *u8, out: *u8, cap: i64, olen: *i64) -> i64 {
  let av: *i64 = sys_mmap(8 * 6) as *i64
  av[0] = path as i64
  av[1] = a1 as i64
  var k: i64 = 2
  if (a2 as i64) != 0 { av[k] = a2 as i64; k = k + 1 }
  if (a3 as i64) != 0 { av[k] = a3 as i64; k = k + 1 }
  av[k] = 0
  olen[0] = 0
  let rc: i64 = tr_run_capture_to(path, av, out, cap - 1, olen, SS_TMO_MS)
  sys_munmap(av as *u8, 8 * 6)
  return rc
95}
96func sx_mgmt(method: *u8, path: *u8, body: *u8, out: *u8, olen: *i64) -> i64 { return sx_fork(SS_MGMT, method, path, body, out, SS_CAP, olen) }
97// the sha256 of a file by the ONE hash organ, read BY KEY from its JSON (never the first hex run)
98func sx_sha_of(path: *u8, dst: *u8) -> i64 {
  let out: *u8 = sys_mmap(SS_SMALL)
  let olen: *i64 = sys_mmap(16) as *i64
  sx_fork(SS_FILEHASH, path, 0 as *u8, 0 as *u8, out, SS_SMALL, olen)
  let r: i64 = sx_json_str(out, olen[0], "\"sha256\":" as *u8, dst, 80)
  sys_munmap(out, SS_SMALL)
  sys_munmap(olen as *u8, 16)
  return r
106}
107func sx_append(path: *u8, line: *u8, n: i64) -> i64 {
  let fd: i64 = sys_openat_append(path, SS_MODE_0644)
  if fd < 0 { return 0 - 1 }
  sys_write(fd, line, n)
  sys_close(fd)
  return 0
113}
114// one journal frame: epoch TAB nx_search_ship TAB stage TAB verdict TAB detail (a line never splits: detail is cut)
115func sx_jrnl(stage: *u8, verdict: *u8, detail: *u8) -> i64 {
  let ln: *u8 = sys_mmap(SS_LINE)
  var o: i64 = sx_catn(ln, 0, sys_now_realtime_sec())
  ln[o] = 9 as u8; o = o + 1
  o = sx_cat(ln, o, "nx_search_ship" as *u8); ln[o] = 9 as u8; o = o + 1
  o = sx_cat(ln, o, stage); ln[o] = 9 as u8; o = o + 1
  o = sx_cat(ln, o, verdict); ln[o] = 9 as u8; o = o + 1
  var i: i64 = 0
  while detail[i] != (0 as u8) { if o < SS_LINE - 2 { var c: i64 = detail[i] as i64; if c == SX_LF { c = SX_SPACE } ln[o] = c as u8; o = o + 1 } i = i + 1 }
  ln[o] = SX_LF as u8; o = o + 1
  sx_append(SS_JRNL, ln, o)
  sys_munmap(ln, SS_LINE)
  return 0
128}
129// the board is the ESTATE's ledger: one log row per ship on search.plan, pipes and newlines folded so the row parses
130func sx_plan_log(kind: *u8, text: *u8) -> i64 {
  let ln: *u8 = sys_mmap(SS_LINE)
  var o: i64 = sx_cat(ln, 0, "log|" as *u8)
  o = sx_catn(ln, o, sys_now_realtime_sec())
  o = sx_cat(ln, o, "|" as *u8)
  o = sx_cat(ln, o, sx_rung_g)
  o = sx_cat(ln, o, "|" as *u8)
  o = sx_cat(ln, o, kind)
  o = sx_cat(ln, o, "|" as *u8)
  var i: i64 = 0
  while text[i] != (0 as u8) { if o < SS_LINE - 2 { var c: i64 = text[i] as i64; if c == SX_PIPE { c = 47 } if c == SX_LF { c = SX_SPACE } ln[o] = c as u8; o = o + 1 } i = i + 1 }
  ln[o] = SX_LF as u8; o = o + 1
  sx_append(SS_PLAN, ln, o)
  sys_munmap(ln, SS_LINE)
  return 0
145}
146// bank the live daemon by hash under a tag; 1 = the copier re-read and verified byte-for-byte
147// pathout (0 = not wanted) receives the bank path, so the agent mode can restore exactly what it banked
148func sx_bank(tag: *u8, sha: *u8, pathout: *u8) -> i64 {
  let dst: *u8 = sys_mmap(512)
  var o: i64 = sx_cat(dst, 0, SS_BANK_DIR)
  o = sx_cat(dst, o, SS_DAEMON_ELF)
  o = sx_cat(dst, o, "." as *u8)
  o = sx_cat(dst, o, tag)
  o = sx_cat(dst, o, "-" as *u8)
  var k: i64 = 0
  while k < 8 { if sha[k] != (0 as u8) { dst[o] = sha[k]; o = o + 1 } k = k + 1 }
  dst[o] = 0 as u8
  o = sx_cat(dst, o, "-" as *u8)
  o = sx_catn(dst, o, sys_now_realtime_sec())
  dst[o] = 0 as u8
  if (pathout as i64) != 0 { sx_cat(pathout, 0, dst) }
  let out: *u8 = sys_mmap(SS_SMALL)
  let olen: *i64 = sys_mmap(16) as *i64
  sx_fork(SS_FILECOPY, SS_DAEMON_ELF, dst, 0 as *u8, out, SS_SMALL, olen)
  sx_puts("    BANK " as *u8); sx_puts(dst); sx_puts(" " as *u8); sx_tail(out, olen[0], 1)
  let ok: i64 = sx_find(out, olen[0], "verified=1" as *u8)
  sys_munmap(out, SS_SMALL)
  sys_munmap(olen as *u8, 16)
  sys_munmap(dst, 512)
  if ok >= 0 { return 1 }
  return 0
172}
173// build one target through the mgmt API; fills sha; 0 ok | 12 refused | 11 src_path names another target
174func sx_build(target: *u8, srcsuffix: *u8, stage: *u8, sha: *u8, out: *u8, olen: *i64) -> i64 {
  let body: *u8 = sys_mmap(512)
  var bo: i64 = sx_cat(body, 0, "target=" as *u8)
  bo = sx_cat(body, bo, target)
  bo = sx_cat(body, bo, "&bust=r" as *u8)
  bo = sx_catn(body, bo, sys_now_realtime_sec())
  sx_mgmt("POST" as *u8, "/api/build" as *u8, body, out, olen)
  sys_munmap(body, 512)
  sx_puts("  [" as *u8); sx_puts(stage); sx_puts("] BUILD " as *u8); sx_puts(target); sx_puts(" " as *u8)
  if sx_find(out, olen[0], "\"BUILT\"" as *u8) < 0 { sx_puts("REFUSED by /api/build (exit 12) -- the API's own reason:\n" as *u8); sx_tail(out, olen[0], 6); sx_jrnl("BUILD" as *u8, "REFUSED" as *u8, target); return 12 }
  sx_json_str(out, olen[0], "\"sha256\":" as *u8, sha, 80)
  let srcp: *u8 = sys_mmap(512)
  sx_json_str(out, olen[0], "\"src_path\":" as *u8, srcp, 512)
  if sx_find(srcp, sx_slen(srcp), srcsuffix) < 0 { sx_puts("REFUSED: src_path does not name the target (exit 11): " as *u8); sx_puts(srcp); sx_puts("\n" as *u8); sx_jrnl("BUILD" as *u8, "WRONG-SRC" as *u8, srcp); return 11 }
  sx_puts("BUILT sha=" as *u8); sx_puts(sha); sx_puts(" src_path=" as *u8); sx_puts(srcp); sx_puts("\n" as *u8)
  sx_jrnl("BUILD" as *u8, target, sha)
  sys_munmap(srcp, 512)
  return 0
192}
193
194// The DEPLOY watchdog's up budget, read from the SAME conf nx_mgmt_data md_probe_conf reads (deploy_probe.conf in the nishihost
195// root: `up_tries N` / `up_tick_ms N`, '#' lines are comments, a zero or unparsable value keeps the default), so the ship waits
196// exactly as long as the watchdog may. Defaults mirror the watchdog's compiled-in 10 x 3000 ms.
197const SS_PROBE_CONF: *u8 = "deploy_probe.conf" as *u8
198const SS_PROBE_UP_TRIES_DFLT: i64 = 10
199const SS_PROBE_UP_TICK_MS_DFLT: i64 = 3000
200func sx_conf_key_int(buf: *u8, n: i64, key: *u8, dflt: i64) -> i64 {
  var i: i64 = 0
  var v: i64 = dflt
  while i < n {
      // end of this line: the LF index or n (a flag, never a sentinel written into the cursor)
      var e: i64 = i
      var hit: i64 = 0
      while hit == 0 { if e >= n { hit = 1 } else { if buf[e] == (SX_LF as u8) { hit = 1 } else { e = e + 1 } } }
      if e > i { if (buf[i] as i64) != SX_HASH {
          if sx_starts((buf as i64 + i) as *u8, key) == 1 {
              let got: i64 = sx_int_after((buf as i64 + i) as *u8, e - i, key)
              if got > 0 { v = got }
          }
      } }
      i = e + 1
  }
  return v
217}
218func sx_probe_budget_ms() -> i64 {
  let szp: *i64 = sys_mmap(16) as *i64
  szp[0] = 0
  let buf: *u8 = sys_read_file(SS_PROBE_CONF, szp)
  var tries: i64 = SS_PROBE_UP_TRIES_DFLT
  var tick: i64 = SS_PROBE_UP_TICK_MS_DFLT
  if (buf as i64) != 0 { if szp[0] > 0 {
      tries = sx_conf_key_int(buf, szp[0], "up_tries " as *u8, tries)
      tick = sx_conf_key_int(buf, szp[0], "up_tick_ms " as *u8, tick)
  } }
  return tries * tick
229}
230func main(argc: i64, argv: *i64) -> i64 {
  ep_anchor()
  if argc < 2 { sx_puts("usage: nx_search_ship dry|deploy|agent [allow_loss=<n>] [bar=<ms>] [probes=<path>] [rung=<id>]\n" as *u8); return 2 }
  let mode: *u8 = argv[1] as *u8
  // do_deploy: 0 = dry (private candidate + receipt), 1 = deploy (the root adopts), 2 = agent (W2: adopts on its
  // own ONLY a candidate the ruler and referee prove is a pure speed change; holds a moved ranking for the root)
  var do_deploy: i64 = 0
  if sx_streq(mode, "deploy" as *u8) == 1 { do_deploy = 1 } else { if sx_streq(mode, "agent" as *u8) == 1 { do_deploy = 2 } else { if sx_streq(mode, "dry" as *u8) == 0 { sx_puts("usage: nx_search_ship dry|deploy|agent [allow_loss=<n>] [bar=<ms>] [probes=<path>] [rung=<id>]\n" as *u8); return 2 } } }
  var allow_loss: i64 = 0
  var bar: i64 = 0
  var probes_path: *u8 = SS_PROBES
  sx_rung_g = SS_RUNG_DEFAULT
  var ai: i64 = 2
  while ai < argc {
      let a: *u8 = argv[ai] as *u8
      if sx_starts(a, "allow_loss=" as *u8) == 1 { allow_loss = sx_int_after(a, sx_slen(a), "allow_loss=" as *u8) }
      else { if sx_starts(a, "bar=" as *u8) == 1 { bar = sx_int_after(a, sx_slen(a), "bar=" as *u8) }
      else { if sx_starts(a, "probes=" as *u8) == 1 { probes_path = (a as i64 + 7) as *u8 }
      else { if sx_starts(a, "rung=" as *u8) == 1 { sx_rung_g = (a as i64 + sx_slen("rung=" as *u8)) as *u8 } } } }
      ai = ai + 1
  }
  if bar <= 0 {
      let jb: *i64 = sys_mmap(16) as *i64
      let jcf: *u8 = sys_read_file(SS_JOURNEYS, jb)
      bar = SS_DEFAULT_BAR_MS
      if (jcf as i64) != 0 { bar = sx_journey_bar(jcf, jb[0], SS_JOURNEY_ID, SS_DEFAULT_BAR_MS) }
  }
  sx_puts("=== NX-SEARCH-SHIP mode=" as *u8); sx_puts(mode); sx_puts(" bar_ms=" as *u8); sx_pn(bar); sx_puts(" (one loop: build daemon > build referee > ruler > referee A/B > [bank > deploy > warm > cold probes > journal]; refuses at the first RED)\n" as *u8)
  let out: *u8 = sys_mmap(SS_CAP)
  let olen: *i64 = sys_mmap(16) as *i64
260
  // [1/8] + [2/8] BUILD the daemon and the referee on the same tree
  let sha: *u8 = sys_mmap(80)
  let rc1: i64 = sx_build(SS_DAEMON, SS_DAEMON_SRC, "1/8" as *u8, sha, out, olen)
  if rc1 != 0 { return rc1 }
  let shab: *u8 = sys_mmap(80)
  let rc2: i64 = sx_build(SS_BENCH, SS_BENCH_SRC, "2/8" as *u8, shab, out, olen)
  if rc2 != 0 { return rc2 }
268
  // [3/8] RULER: the staged daemon must keep every printable run the live one has (or every loss is NAMED and allowed)
  sx_fork(SS_CONTENTDIFF, SS_DAEMON_ELF, SS_DAEMON_NEW, 0 as *u8, out, SS_CAP, olen)
  let lost: i64 = sx_int_after(out, olen[0], "lost_from_live=" as *u8)
  var named: i64 = 0
  var li: i64 = 0
  while li < olen[0] { let p: i64 = sx_find((out as i64 + li) as *u8, olen[0] - li, "LOST:" as *u8); if p < 0 { li = olen[0] } else { named = named + 1; li = li + p + 5 } }
  let green: i64 = sx_find(out, olen[0], "verdict=GREEN" as *u8)
  sx_puts("  [3/8] RULER lost_from_live=" as *u8); sx_pn(lost); sx_puts(" named=" as *u8); sx_pn(named); sx_puts(" " as *u8); sx_tail(out, olen[0], 1)
  if green < 0 {
      var bar_lost: i64 = named
      if lost > bar_lost { bar_lost = lost }
      if allow_loss > 0 { if named > 0 { if bar_lost <= allow_loss {
          sx_puts("    ruler RED on " as *u8); sx_pn(bar_lost); sx_puts(" named lost run(s), caller allowed " as *u8); sx_pn(allow_loss); sx_puts(" -- proceeding, every named run journalled:\n" as *u8); sx_tail(out, olen[0], 12)
          sx_jrnl("RULER" as *u8, "ALLOWED" as *u8, sha)
      } else { sx_puts("    REFUSED (exit 31): lost runs exceed allow_loss\n" as *u8); sx_tail(out, olen[0], 12); sx_jrnl("RULER" as *u8, "RED" as *u8, sha); return 31 } }
      else { sx_puts("    REFUSED (exit 31): the ruler is RED and named no lost run -- an allowance cannot cover a ruler that did not speak\n" as *u8); sx_tail(out, olen[0], 12); sx_jrnl("RULER" as *u8, "RED" as *u8, sha); return 31 } }
      else { sx_puts("    REFUSED (exit 31): the staged daemon loses printable content the live one has. Ruler tail:\n" as *u8); sx_tail(out, olen[0], 12); sx_jrnl("RULER" as *u8, "RED" as *u8, sha); return 31 }
  } else { sx_jrnl("RULER" as *u8, "GREEN" as *u8, sha) }
287
  // [4/8] REFEREE A/B: the promoted referee and the staged one on the same index, per-query lines compared byte-for-byte
  let outa: *u8 = sys_mmap(SS_CAP)
  let alen: *i64 = sys_mmap(16) as *i64
  let outb: *u8 = sys_mmap(SS_CAP)
  let blen: *i64 = sys_mmap(16) as *i64
  sx_fork(SS_BENCH_LIVE, "ab-live" as *u8, 0 as *u8, 0 as *u8, outa, SS_CAP, alen)
  sx_fork(SS_BENCH_STAGED, "ab-staged" as *u8, 0 as *u8, 0 as *u8, outb, SS_CAP, blen)
  let qa: *u8 = sys_mmap(SS_SMALL)
  let qb: *u8 = sys_mmap(SS_SMALL)
  let qal: *i64 = sys_mmap(16) as *i64
  let qbl: *i64 = sys_mmap(16) as *i64
  let na: i64 = sx_lines_with(outa, alen[0], SS_QPFX, qa, SS_SMALL, qal)
  let nb: i64 = sx_lines_with(outb, blen[0], SS_QPFX, qb, SS_SMALL, qbl)
  let ca: i64 = sx_int_after(outa, alen[0], "corpus_docs=" as *u8)
  let cb: i64 = sx_int_after(outb, blen[0], "corpus_docs=" as *u8)
  let ndcg_a: i64 = sx_int_after(outa, alen[0], "diag_mean_ndcg_permil_UNPUBLISHABLE=" as *u8)
  let ndcg_b: i64 = sx_int_after(outb, blen[0], "diag_mean_ndcg_permil_UNPUBLISHABLE=" as *u8)
  let mrr_a: i64 = sx_int_after(outa, alen[0], "diag_mean_mrr_permil_UNPUBLISHABLE=" as *u8)
  let mrr_b: i64 = sx_int_after(outb, blen[0], "diag_mean_mrr_permil_UNPUBLISHABLE=" as *u8)
  let same: i64 = sx_bytes_eq(qa, qal[0], qb, qbl[0])
  var referee: *u8 = "IDENTICAL" as *u8
  sx_puts("  [4/8] REFEREE live=" as *u8); sx_pn(na); sx_puts(" staged=" as *u8); sx_pn(nb); sx_puts(" queries, corpus_docs " as *u8); sx_pn(ca); sx_puts("/" as *u8); sx_pn(cb); sx_puts(" ndcg " as *u8); sx_pn(ndcg_a); sx_puts("->" as *u8); sx_pn(ndcg_b); sx_puts(" mrr " as *u8); sx_pn(mrr_a); sx_puts("->" as *u8); sx_pn(mrr_b)
  if na == 0 { sx_puts(" REFUSED (exit 41): the live referee scored NO query -- read its tail:\n" as *u8); sx_tail(outa, alen[0], 6); sx_jrnl("REFEREE" as *u8, "NO-LIVE-ARM" as *u8, sha); return 41 }
  if nb == 0 { sx_puts(" REFUSED (exit 41): the staged referee scored NO query -- read its tail:\n" as *u8); sx_tail(outb, blen[0], 6); sx_jrnl("REFEREE" as *u8, "NO-STAGED-ARM" as *u8, sha); return 41 }
  if ca != cb { sx_puts(" WARNING index moved between the arms (a crawler ship): compare with that in mind" as *u8) }
  if same == 1 { sx_puts(" IDENTICAL on every per-query line\n" as *u8); sx_jrnl("REFEREE" as *u8, "IDENTICAL" as *u8, sha) }
  else {
      let dl: i64 = sx_first_diff_line(qa, qal[0], qb, qbl[0])
      sx_puts(" MOVED first differing line=" as *u8); sx_pn(dl); sx_puts("\n    live:\n" as *u8); sys_write(1, qa, qal[0]); sx_puts("    staged:\n" as *u8); sys_write(1, qb, qbl[0])
      if ndcg_b < ndcg_a { sx_puts("    REFUSED (exit 41): mean ndcg fell -- a ranking regression, not a speed change\n" as *u8); sx_jrnl("REFEREE" as *u8, "WORSE-NDCG" as *u8, sha); return 41 }
      if mrr_b < mrr_a { sx_puts("    REFUSED (exit 41): mean mrr fell -- a ranking regression, not a speed change\n" as *u8); sx_jrnl("REFEREE" as *u8, "WORSE-MRR" as *u8, sha); return 41 }
      sx_puts("    PROCEEDING ANNOUNCED: the ranking moved and neither mean fell -- this is a ranking change the caller adjudicates against its pre-declared accept rule; the differing lines are the receipt\n" as *u8)
      sx_jrnl("REFEREE" as *u8, "MOVED-NOT-WORSE" as *u8, sha)
      referee = "MOVED-NOT-WORSE" as *u8
  }
323
  // AGENT MODE (W2, 2026-09-14): the worker adopts on its own ONLY the class whose safety is mechanical -- the
  // candidate differs from live, the ruler lost nothing and the referee is byte-identical on every judged query
  // (a pure speed change). A MOVED ranking is HELD as a private candidate for the root, never adopted by an
  // agent: that is the RACI, not caution theatre. The decision is the lib's pure function under the gate's T8.
  if do_deploy == 2 {
      let live0: *u8 = sys_mmap(80)
      sx_sha_of(SS_DAEMON_ELF, live0)
      let dec: i64 = sx_agent_decide(sx_streq(live0, sha), same, lost)
      if dec == SX_AGENT_NOCHANGE { sx_jrnl("AGENT" as *u8, "NOCHANGE" as *u8, sha); sx_puts("SEARCH-SHIP verdict=GREEN mode=agent NO-CHANGE: the candidate is byte-identical to live " as *u8); sx_puts(sha); sx_puts(" -- nothing to adopt\n" as *u8); return 0 }
      if dec == SX_AGENT_HOLD { sx_jrnl("AGENT" as *u8, "HOLD" as *u8, sha); sx_puts("SEARCH-SHIP verdict=GREEN mode=agent HELD: a PRIVATE CANDIDATE " as *u8); sx_puts(sha); sx_puts(" whose ranking moved (referee=" as *u8); sx_puts(referee); sx_puts(") waits for the root: nx_search_ship deploy\n" as *u8); return 0 }
      sx_jrnl("AGENT" as *u8, "ADOPT" as *u8, sha)
      sx_puts("  AGENT ADOPT: the candidate differs from live, the ruler lost 0 and the referee is IDENTICAL -- a pure speed change, adopting through the same stages as deploy\n" as *u8)
      allow_loss = 0
  }
  if do_deploy == 0 {
      sx_jrnl("CANDIDATE" as *u8, "READY" as *u8, sha)
      sx_puts("SEARCH-SHIP verdict=GREEN mode=dry daemon_staged=" as *u8); sx_puts(sha); sx_puts(" referee=" as *u8); sx_puts(referee); sx_puts(" -- a PRIVATE CANDIDATE with its receipt; nothing deployed. Adopt with: nx_search_ship deploy\n" as *u8)
      return 0
  }
343
  // [5/8] BANK the live daemon by hash BEFORE anything replaces it (the rollback slot holds the previous deploy, not the last known good)
  let livesha: *u8 = sys_mmap(80)
  sx_sha_of(SS_DAEMON_ELF, livesha)
  let prebank: *u8 = sys_mmap(512)
  sx_puts("  [5/8] BANK live=" as *u8); sx_puts(livesha); sx_puts("\n" as *u8)
  if sx_bank("pre" as *u8, livesha, prebank) == 0 { sx_puts("    REFUSED (exit 53): the pre-change binary could not be banked and verified -- never deploy without a way back\n" as *u8); sx_jrnl("BANK" as *u8, "FAILED" as *u8, livesha); return 53 }
  sx_jrnl("BANK" as *u8, "PRE" as *u8, livesha)
351
  // [6/8] DEPLOY through the health-checked lane with expect_sha256, then READ THE ARTIFACT
  let db: *u8 = sys_mmap(512)
  var dbo: i64 = sx_cat(db, 0, SS_DEPLOY_BODY)
  dbo = sx_cat(db, dbo, sha)
  if allow_loss > 0 { dbo = sx_cat(db, dbo, "&allow_capability_loss=yes" as *u8) }
  sx_mgmt("POST" as *u8, "/api/deploy" as *u8, db, out, olen)
  sx_puts("  [6/8] DEPLOY " as *u8); sx_tail(out, olen[0], 1)
  if sx_find(out, olen[0], "\"PROMOTED\"" as *u8) < 0 { sx_puts("    REFUSED (exit 51): /api/deploy did not answer PROMOTED -- its reason is above; LIVE IS UNTOUCHED\n" as *u8); sx_jrnl("DEPLOY" as *u8, "REFUSED" as *u8, sha); return 51 }
  let nowsha: *u8 = sys_mmap(80)
  sx_sha_of(SS_DAEMON_ELF, nowsha)
  if sx_streq(nowsha, sha) == 0 { sys_sleep_ms(SS_SETTLE_MS); sx_sha_of(SS_DAEMON_ELF, nowsha) }
  if sx_streq(nowsha, sha) == 0 { sx_puts("    REFUSED (exit 51): live artifact != staged sha after deploy -- the receipt is not the artifact. live=" as *u8); sx_puts(nowsha); sx_puts("\n" as *u8); sx_jrnl("DEPLOY" as *u8, "NOT-LANDED" as *u8, nowsha); return 51 }
  // [6b] WAIT FOR THE WATCHDOG'S TERMINAL STATE (2026-09-16, measured): /api/deploy_status reads RUNNING for as long as the
  // watchdog's up budget lets the daemon warm BEFORE it listens (deploy_probe.conf, else the compiled-in 10 x 3000 ms; the
  // docportal guard itself tolerates 600 s). Two reads 4.5 s after the deploy took a referee-IDENTICAL build for a failure
  // (exit 52) while the watchdog was still counting -- a level read of a transient, the deploy-time twin of the health-probe
  // law. Poll until the status leaves RUNNING, bounded by the SAME budget the watchdog reads plus one settle, so the ship and
  // the watchdog share one bound instead of two typed constants; the wait is announced.
  let bud_ms: i64 = sx_probe_budget_ms() + SS_SETTLE_MS * 2
  var waited_ms: i64 = 0
  var dst_done: i64 = 0
  sys_sleep_ms(SS_SETTLE_MS)
  while dst_done == 0 {
      sx_mgmt("GET" as *u8, "/api/deploy_status" as *u8, 0 as *u8, out, olen)
      if sx_find(out, olen[0], "RUNNING" as *u8) < 0 { dst_done = 1 } else {
          if waited_ms >= bud_ms { dst_done = 1 } else { sys_sleep_ms(SS_WARM_STEP_MS); waited_ms = waited_ms + SS_WARM_STEP_MS }
      }
  }
  sx_puts("    watchdog_wait_ms=" as *u8); sx_pn(waited_ms); sx_puts(" budget_ms=" as *u8); sx_pn(bud_ms); sx_puts("\n" as *u8)
  sx_puts("    deploy_status " as *u8); sx_tail(out, olen[0], 1)
  if sx_find(out, olen[0], "DEPLOYED-GREEN" as *u8) < 0 { sx_puts("    REFUSED (exit 52): the watchdog did not report DEPLOYED-GREEN -- read its status; the auto-rollback may be acting\n" as *u8); sx_jrnl("DEPLOY" as *u8, "NOT-GREEN" as *u8, sha); return 52 }
  sx_jrnl("DEPLOY" as *u8, "GREEN" as *u8, sha)
  sx_bank("ship" as *u8, sha, 0 as *u8)
385
  // [7/8] WARM-WAIT then COLD PROBES: the daemon opens its port only after the shard is warm, and the restart empties
  // the result memo, so the first answer of every probe is a true cold read through the edge with phase_us
  let probes: *u8 = sys_mmap(SS_PROBE_W * SS_MAX_PROBES)
  var np: i64 = 0
  let pb: *i64 = sys_mmap(16) as *i64
  let pcf: *u8 = sys_read_file(probes_path, pb)
  if (pcf as i64) != 0 { np = sx_probes(pcf, pb[0], probes, SS_PROBE_W, SS_MAX_PROBES) }
  if np == 0 { sx_cat(probes, 0, SS_PROBE_DEFAULT); np = 1 }
  let url: *u8 = sys_mmap(1024)
  let ph: *i64 = sys_mmap(8 * SX_PHASE_SLOTS) as *i64
  var over: i64 = 0
  var maxrecv: i64 = 0
  var maxq: *u8 = probes
  var ready: i64 = 0
  let summary: *u8 = sys_mmap(SS_LINE)
  var so: i64 = sx_cat(summary, 0, "nx_search_ship " as *u8)
  so = sx_cat(summary, so, mode)
  so = sx_cat(summary, so, ": daemon " as *u8)
  so = sx_cat(summary, so, sha)
  so = sx_cat(summary, so, " live (ruler GREEN, referee " as *u8)
  so = sx_cat(summary, so, referee)
  so = sx_cat(summary, so, ", pre-change " as *u8)
  so = sx_cat(summary, so, livesha)
  so = sx_cat(summary, so, " banked). COLD PROBES through the edge, recv ms / phase_us total,prep,s1,l0dec,auth: " as *u8)
  var pi: i64 = 0
  while pi < np {
      let q: *u8 = (probes as i64 + pi * SS_PROBE_W) as *u8
      var uo: i64 = sx_cat(url, 0, SS_URL_HEAD)
      uo = sx_urlenc(url, uo, q)
      uo = sx_cat(url, uo, SS_URL_TAIL)
      var tries: i64 = 0
      var got: i64 = 0
      while got == 0 {
          sx_fork(SS_HTTPS, url, SS_EDGE, 0 as *u8, out, SS_CAP, olen)
          if sx_find(out, olen[0], "HTTP/1.1 200" as *u8) >= 0 { got = 1; ready = 1 } else {
              tries = tries + 1
              if ready == 1 { got = 2 } else { if tries >= SS_WARM_TRIES { got = 2 } else { sys_sleep_ms(SS_WARM_STEP_MS) } }
          }
      }
      if got == 2 { if ready == 0 { sx_puts("  [7/8] PROBE the daemon never answered 200 inside the warm window (exit 61). Last answer:\n" as *u8); sx_tail(out, olen[0], 3); sx_jrnl("PROBE" as *u8, "NEVER-ANSWERED" as *u8, q); return 61 } }
      sx_phase(out, olen[0], ph)
      sx_puts("  [7/8] PROBE q=" as *u8); sx_puts(q); sx_puts(" recv_ms=" as *u8); sx_pn(ph[0]); sx_puts(" total_us=" as *u8); sx_pn(ph[6]); sx_puts(" prep=" as *u8); sx_pn(ph[1]); sx_puts(" s1=" as *u8); sx_pn(ph[2]); sx_puts(" l0dec=" as *u8); sx_pn(ph[3]); sx_puts(" auth=" as *u8); sx_pn(ph[4]); sx_puts(" cand=" as *u8); sx_pn(ph[7])
      if got == 2 { sx_puts(" NO-200" as *u8); over = over + 1 }
      if ph[0] > bar { sx_puts(" OVER-BAR" as *u8); over = over + 1 }
      if ph[0] > maxrecv { maxrecv = ph[0]; maxq = q }
      sx_puts("\n" as *u8)
      so = sx_cat(summary, so, q); so = sx_cat(summary, so, " " as *u8); so = sx_catn(summary, so, ph[0]); so = sx_cat(summary, so, "/" as *u8); so = sx_catn(summary, so, ph[6]); so = sx_cat(summary, so, "," as *u8); so = sx_catn(summary, so, ph[1]); so = sx_cat(summary, so, "," as *u8); so = sx_catn(summary, so, ph[2]); so = sx_cat(summary, so, "," as *u8); so = sx_catn(summary, so, ph[3]); so = sx_cat(summary, so, "," as *u8); so = sx_catn(summary, so, ph[4]); so = sx_cat(summary, so, "; " as *u8)
      pi = pi + 1
  }
  // [7c/8] HTML SERP WITNESS: the consumer journey's own URL and must_contain, through the same edge. The JSON
  // probes above exercise only the /api/search branch of the daemon's router; on 2026-09-14 a daemon whose JSON
  // probes all read GREEN had broken the HTML /search branch through the buffered edge, and no stage saw it.
  // A failed witness counts as an over-bar probe so the agent rollback and the exit code both fire on it.
  let hjb: *i64 = sys_mmap(16) as *i64
  let hjcf: *u8 = sys_read_file(SS_JOURNEYS, hjb)
  let hu: *u8 = sys_mmap(SS_LINE)
  let hm: *u8 = sys_mmap(SS_LINE)
  var hun: i64 = 0
  var hmn: i64 = 0
  if (hjcf as i64) != 0 { hun = sx_journey_field(hjcf, hjb[0], SS_JOURNEY_ID, SX_JF_URL, hu, SS_LINE); hmn = sx_journey_field(hjcf, hjb[0], SS_JOURNEY_ID, SX_JF_MUST, hm, SS_LINE) }
  var hokw: *u8 = "FAIL" as *u8
  if hun == 0 || hmn == 0 { sx_puts("  [7c/8] HTML SERP WITNESS UNAVAILABLE: journey " as *u8); sx_puts(SS_JOURNEY_ID); sx_puts(" has no url or no must_contain in " as *u8); sx_puts(SS_JOURNEYS); sx_puts(" -- counted as a failed probe\n" as *u8); over = over + 1; sx_jrnl("PROBE" as *u8, "HTML-NO-JOURNEY" as *u8, SS_JOURNEY_ID) }
  else {
      sx_fork(SS_HTTPS, hu, SS_EDGE, 0 as *u8, out, SS_CAP, olen)
      let hok: i64 = sx_html_ok(out, olen[0], hm)
      sx_puts("  [7c/8] HTML SERP through the edge url=" as *u8); sx_puts(hu); sx_puts(" must_contain=" as *u8); sx_puts(hm); sx_puts(" bytes=" as *u8); sx_pn(olen[0])
      if hok == 1 { hokw = "OK" as *u8; sx_puts(" OK\n" as *u8) }
      else { sx_puts(" FAIL -- the 200 status line or the marker is missing; last lines:\n" as *u8); sx_tail(out, olen[0], 3); over = over + 1; sx_jrnl("PROBE" as *u8, "HTML-FAIL" as *u8, hu) }
  }
  so = sx_cat(summary, so, "html_serp=" as *u8); so = sx_cat(summary, so, hokw); so = sx_cat(summary, so, " " as *u8)
  so = sx_cat(summary, so, "bar_ms=" as *u8); so = sx_catn(summary, so, bar); so = sx_cat(summary, so, " over_bar=" as *u8); so = sx_catn(summary, so, over)
457
  // AGENT ROLLBACK (W2): an adopted candidate that answers over the bar is not left live for a seat to find in the
  // morning. The pre-change binary banked in [5/8] is staged back and deployed through the same health-checked
  // lane, verified by hash, and the outcome is journalled either way; the run still exits 71 because the
  // candidate failed its bar -- a rollback is a remedy, not a pass.
  if do_deploy == 2 { if over > 0 {
      sx_puts("  [7b/8] AGENT ROLLBACK: " as *u8); sx_pn(over); sx_puts(" probe(s) over the bar -- restoring the banked pre-change binary " as *u8); sx_puts(livesha); sx_puts(" from " as *u8); sx_puts(prebank); sx_puts("\n" as *u8)
      sx_fork(SS_FILECOPY, prebank, SS_DAEMON_NEW, "overwrite" as *u8, out, SS_CAP, olen)
      if sx_find(out, olen[0], "verified=1" as *u8) < 0 { sx_puts("    ROLLBACK FAILED: the bank could not be staged -- read the copier's tail:\n" as *u8); sx_tail(out, olen[0], 2); sx_jrnl("ROLLBACK" as *u8, "STAGE-FAILED" as *u8, livesha) } else {
          var rbo: i64 = sx_cat(db, 0, SS_DEPLOY_BODY)
          rbo = sx_cat(db, rbo, livesha)
          sx_mgmt("POST" as *u8, "/api/deploy" as *u8, db, out, olen)
          sx_puts("    rollback deploy " as *u8); sx_tail(out, olen[0], 1)
          sys_sleep_ms(SS_SETTLE_MS)
          let rbsha: *u8 = sys_mmap(80)
          sx_sha_of(SS_DAEMON_ELF, rbsha)
          if sx_streq(rbsha, livesha) == 1 { sx_puts("    ROLLBACK LANDED: live is the pre-change binary again\n" as *u8); sx_jrnl("ROLLBACK" as *u8, "LANDED" as *u8, livesha) }
          else { sx_puts("    ROLLBACK NOT LANDED: live=" as *u8); sx_puts(rbsha); sx_puts(" -- the watchdog's own rollback may be acting; read /api/deploy_status\n" as *u8); sx_jrnl("ROLLBACK" as *u8, "NOT-LANDED" as *u8, rbsha) }
      }
      so = sx_cat(summary, so, " AGENT-ROLLBACK-ATTEMPTED" as *u8)
  } }
478
  // [8/8] JOURNAL: the estate's ledger (search.plan) and this organ's own frame, then the positional verdict line
  var verdict: *u8 = "GREEN" as *u8
  if over > 0 { verdict = "RED" as *u8 }
  sx_plan_log("land" as *u8, summary)
  sx_jrnl("SHIPPED" as *u8, verdict, summary)
  sx_puts("  [8/8] JOURNAL search.plan log row appended + " as *u8); sx_puts(SS_JRNL); sx_puts("\n" as *u8)
  sx_puts("SEARCH-SHIP verdict=" as *u8); sx_puts(verdict); sx_puts(" daemon=" as *u8); sx_puts(sha); sx_puts(" referee=" as *u8); sx_puts(referee); sx_puts(" probes=" as *u8); sx_pn(np); sx_puts(" over_bar=" as *u8); sx_pn(over); sx_puts(" html_serp=" as *u8); sx_puts(hokw); sx_puts(" max_recv_ms=" as *u8); sx_pn(maxrecv); sx_puts(" (" as *u8); sx_puts(maxq); sx_puts(") bar_ms=" as *u8); sx_pn(bar); sx_puts(" -- every stage above carries its evidence line\n" as *u8)
  if over > 0 { return 71 }
  return 0
488}
nx_search_ship.nx source

`nx_search_ship.nx` source
