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

↩ module page · 1566 lines · 74612 B
1// nx_wflow_engine.nx -- R1 KEYSTONE of the Workflow Automation ladder (/compare/automation census row
2// "Unified multi-step durable workflow engine"). A WORKFLOW = DATA (law: data as data):
3//   flow header  = `flowid|event-kind|cond(k=v or -)`
4//   step row     = `flowid|idx|action|channel-or-dash|arg|max-attempts`   (idx 1-based, author ascending)
5// An EVENT = `kind~k=v~...~id=EN` (the nx_crm_flow contract). The engine upgrades nx_crm_flow from
6// single-shot in-memory rules to DURABLE EXECUTION (the Temporal model, sovereign):
7//   - every transition is ONE appended line in an append-only RUN LEDGER (flock + O_APPEND, torn-free):
8//     `WFRUN rid=<evid>.<flowid> flow=<f> step=<n> status=START|ATT|OK|FAILSTEP|FAILED|DONE att=<k>`
9//   - state is DERIVED BY REPLAY of the ledger (event-sourced; nothing mutates, history is sacred)
10//   - IDEMPOTENT by construction: a (event id, flow) pair with a START line never starts twice --
11//     the dedup survives crashes/restarts (stronger than the in-memory seen[] of nx_crm_flow)
12//   - per-step RETRY up to max-attempts; exhausted -> run FAILED at that step, later steps never run
13//   - wf_resume: any run with START but neither DONE nor FAILED (a crash mid-flight) is completed
14//     from its first non-OK step; already-OK steps are NOT re-executed (proven in the selftest)
15// ACTIONS v1 (in-process, composing the send plane like nx_crm_flow): create-task | send (any nx_send
16// channel, sr_valid-checked at load) | notify | update-field | probe-fail (DIAGNOSTIC: fails while
17// attempt < arg -- the deterministic retry witness; keep it out of production flows).
18// wf_load validates the WHOLE step set up front (unknown action / bad channel / attempts<1 = LOUD -1,
19// nothing half-loaded). Production ledger path suggestion: knowledge/status/wflow_runs.log.
20// R3 CONNECTOR LAYER (exec-organ): a step can run ANY blessed organ as a workflow step, resolved through a
21// GREEN fail-closed catalog (rows `name<TAB>elf<TAB>GREEN`, the tool_allowlist.conf shape; cx[3]=catalog path,
22// 0=exec-organ refused). Child exit 0 = step OK, nonzero = step fail (so per-step retry applies to REAL organ
23// execution). Production catalog: knowledge/wflow/connectors.conf.
24// PURE CORE (no main -- the house pure-core+gate idiom): gates = _hdl_build/nx_wflow_engine_gate (wf_selftest)
25// + _hdl_build/nx_wflow_connect_gate (connector battery). license_tier: ORIGINAL
26import "nx_send.nx"
27const WF_MAGIC_1089: i64 = 1089
28const WF_MAGIC_65536: i64 = 65536
29const WF_MAGIC_16777216: i64 = 16777216
30const WF_MAGIC_1024: i64 = 1024
31const WF_MAGIC_60000: i64 = 60000
32
33// ---- tilde-event helpers (nx_crm_flow idiom, local wf_ copies) ----
34func wf_has(hay: *u8, needle: *u8) -> i64 {
  let n: i64 = slen(hay)
  let m: i64 = slen(needle)
  if m == 0 { return 0 }
  var i: i64 = 0
  while i + m <= n {
      var k: i64 = 0
      var hit: i64 = 1
      while k < m { if hay[i+k] != needle[k] { hit = 0; k = m } else { k = k + 1 } }
      if hit == 1 { return 1 }
      i = i + 1
  }
  return 0
47}
48
49func wf_cat(dst: *u8, off: i64, s: *u8) -> i64 { var x: i64 = off; var i: i64 = 0; while s[i] != (0 as u8) { dst[x] = s[i]; x = x + 1; i = i + 1 } return x }
50func wf_catn(dst: *u8, off: i64, v: i64) -> i64 {
  var x: i64 = off
  var m: i64 = v
  if m < 0 { dst[x] = 45 as u8; x = x + 1; m = 0 - m }
  if m == 0 { dst[x] = 48 as u8; return x + 1 }
  let t: *u8 = sys_mmap(24)
  var k: i64 = 0
  while m > 0 { t[k] = (48 + (m - (m/10)*10)) as u8; m = m / 10; k = k + 1 }
  var j: i64 = 0
  while j < k { dst[x] = t[k-1-j]; x = x + 1; j = j + 1 }
  return x
61}
62func wf_atoi(s: *u8) -> i64 {
  var v: i64 = 0
  var i: i64 = 0
  var any: 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); any = 1 } }
      if c < 48 { return 0 - 1 }
      if c > 57 { return 0 - 1 }
      i = i + 1
  }
  if any == 0 { return 0 - 1 }
  return v
75}
76// raw __syscall(201) returns a broken constant on this backend (the getpid -25 class of quirk) --
77// use the proven clock_gettime wrapper so concurrent selftests never share a ledger path
78func wf_now() -> i64 { return sys_now_realtime_sec() }
79
80// ---- durable append-only ledger (flock + O_APPEND; one line per transition) ----
81// Required transition evidence: -4 must stop execution, never enter action retries.
82// fsync covers file contents; this does not claim directory-entry crash durability.
83const WF_EVIDENCE_ERROR: i64 = 0 - 4
84func wf_append(path: *u8, line: *u8) -> i64 {
  let fd: i64 = __syscall(257, 0 - 100, path as i64, WF_MAGIC_1089, 420, 0, 0)
  if fd < 0 { p("WFLOW evidence ERROR open failed\n"); return WF_EVIDENCE_ERROR }
  var locked: i64 = sys_flock(fd, SYS_LOCK_EX)
  while locked == (0 - 4) { locked = sys_flock(fd, SYS_LOCK_EX) }
  var rc: i64 = 0
  if locked != 0 { rc = WF_EVIDENCE_ERROR }
  let n: i64 = slen(line)
  var off: i64 = 0
  while off < n {
      if rc != 0 { break }
      let w: i64 = sys_write(fd, (line as i64 + off) as *u8, n - off)
      if w == (0 - 4) { continue }
      if w <= 0 { rc = WF_EVIDENCE_ERROR } else { off = off + w }
  }
  if rc == 0 {
      var sync: i64 = sys_fsync(fd)
      while sync == (0 - 4) { sync = sys_fsync(fd) }
      if sync != 0 { rc = WF_EVIDENCE_ERROR }
  }
  if locked == 0 { if sys_flock(fd, SYS_LOCK_UN) != 0 { rc = WF_EVIDENCE_ERROR } }
  if sys_close(fd) != 0 { rc = WF_EVIDENCE_ERROR }
  if rc != 0 { p("WFLOW evidence ERROR transition not confirmed; execution stops, reconcile ledger\n") }
  return rc
108}
109const WF_LEDCAP: i64 = 1048576
110import "nx_wflow_text_owned.nx"
111func wf_evid(ev:*u8,out:*u8,cap:i64)->i64 {
if cap<=0 { return 0 }
out[0]=0
let identity:*WfOwnedText=wf_id_event(ev)
if identity==(0 as *WfOwnedText) { return 0 }
if identity.len>=cap { wf_text_free(identity);return 0 }
var i:i64=0
while i<identity.len {
out[i]=identity.data[i]
i=i+1
}
out[i]=0
wf_text_free(identity)
return 1
125}
126
127// Compatibility reader retains its historical fixed allocation for existing callers.
128func wf_readall(path: *u8, szp: *i64) -> *u8 {
  let buf: *u8 = sys_mmap(WF_LEDCAP)
  szp[0] = 0
  let fd: i64 = __syscall(257, 0 - 100, path as i64, 0, 0, 0, 0)
  if fd < 0 { if fd == (0 - 2) { return buf } return 0 as *u8 }
  var off: i64 = 0
  var go: i64 = 1
  while go == 1 {
      let r: i64 = __syscall(0, fd, (buf as i64) + off, WF_LEDCAP - off, 0, 0, 0)
      if r == (0 - 4) { continue }
      if r < 0 { __syscall(3, fd, 0, 0, 0, 0, 0); return 0 as *u8 }
      if r == 0 { go = 0 }
      if go == 1 { off = off + r; if off >= WF_LEDCAP { __syscall(3, fd, 0, 0, 0, 0, 0); p("WFLOW ledger ERROR over cap -- fail loud\n" as *u8); return 0 as *u8 } }
  }
  __syscall(3, fd, 0, 0, 0, 0, 0)
  szp[0] = off
  return buf
145}
146func wf_read_snapshot(path:*u8,szp:*i64)->*u8 {
szp[0]=0
let fd:i64=sys_openat_rd(path)
if fd<0 { if fd==(0-2) { let empty:*u8=sys_mmap_try(1);if (empty as i64)<=0{return 0 as *u8};empty[0]=0;return empty };return 0 as *u8 }
let n:i64=sys_lseek(fd,0,2)
if n<0||n==9223372036854775807 {sys_close(fd);return 0 as *u8}
if sys_lseek(fd,0,0)!=0 {sys_close(fd);return 0 as *u8}
let buf:*u8=sys_mmap_try(n+1)
if (buf as i64)<=0 {sys_close(fd);return 0 as *u8}
var at:i64=0
while at<n {
let got:i64=sys_read(fd,((buf as i64)+at) as *u8,n-at)
if got!=(0-4) {if got<=0{sys_close(fd);sys_munmap_direct(buf,n+1);p("WFLOW read ERROR short or failed snapshot; no replay\n");return 0 as *u8};at=at+got}
}
var tail:i64=sys_read(fd,((buf as i64)+n) as *u8,1)
while tail==(0-4){tail=sys_read(fd,((buf as i64)+n) as *u8,1)}
let end:i64=sys_lseek(fd,0,2);sys_close(fd)
if tail!=0||end!=n {sys_munmap_direct(buf,n+1);p("WFLOW read ERROR changed snapshot; no replay\n");return 0 as *u8}
buf[n]=0;szp[0]=n;return buf
165}
166
167func wf_writeall(path: *u8, buf: *u8, n: i64) -> i64 {
  let fd: i64 = __syscall(257, 0 - 100, path as i64, 577, 420, 0, 0)
  if fd < 0 { p("WFLOW write ERROR cannot open out -- fail loud\n" as *u8); return 0 - 1 }
  let w: i64 = sys_write(fd, buf, n)
  __syscall(3, fd, 0, 0, 0, 0, 0)
  if w != n { p("WFLOW write ERROR short write -- fail loud\n" as *u8); return 0 - 1 }
  return 0
174}
175
176// ---- ledger replay queries ----
177func wf_has_rng(buf: *u8, ls: i64, le: i64, needle: *u8) -> i64 {
  let m: i64 = slen(needle)
  if m == 0 { return 0 }
  var i: i64 = ls
  while i + m <= le {
      var k: i64 = 0
      var hit: i64 = 1
      while k < m { if buf[i+k] != needle[k] { hit = 0; k = m } else { k = k + 1 } }
      if hit == 1 { return 1 }
      i = i + 1
  }
  return 0
189}
190// count ledger LINES containing BOTH substrings
191func wf_lines_with2(buf: *u8, n: i64, a: *u8, b: *u8) -> i64 {
  var c: i64 = 0
  var ls: i64 = 0
  while ls < n {
      var le: i64 = ls
      while le < n { if buf[le] == (10 as u8) { break } le = le + 1 }
      if wf_has_rng(buf, ls, le, a) == 1 { if wf_has_rng(buf, ls, le, b) == 1 { c = c + 1 } }
      ls = le + 1
  }
  return c
201}
202// parse the number after key within [ls,le); -1 if absent
203func wf_num_after(buf: *u8, ls: i64, le: i64, key: *u8) -> i64 {
  let m: i64 = slen(key)
  var i: i64 = ls
  while i + m <= le {
      var k: i64 = 0
      var hit: i64 = 1
      while k < m { if buf[i+k] != key[k] { hit = 0; k = m } else { k = k + 1 } }
      if hit == 1 {
          var q: i64 = i + m
          var v: i64 = 0
          var any: i64 = 0
          while q < le { let c: i64 = buf[q] as i64; if c >= 48 { if c <= 57 { v = v*10 + (c - 48); any = 1; q = q + 1 } else { q = le } } else { q = le } }
          if any == 1 { return v }
          return 0 - 1
      }
      i = i + 1
  }
  return 0 - 1
221}
222// copy the space-terminated value after key within [ls,le) into out; 1 if found
223func wf_val_after(buf: *u8, ls: i64, le: i64, key: *u8, out: *u8, cap: i64) -> i64 {
  let m: i64 = slen(key)
  var i: i64 = ls
  while i + m <= le {
      var k: i64 = 0
      var hit: i64 = 1
      while k < m { if buf[i+k] != key[k] { hit = 0; k = m } else { k = k + 1 } }
      if hit == 1 {
          var q: i64 = i + m
          var t: i64 = 0
          while q < le { if buf[q] == (32 as u8) { break } if buf[q] == (10 as u8) { break } if t < cap - 1 { out[t] = buf[q]; t = t + 1 } q = q + 1 }
          out[t] = 0 as u8
          return 1
      }
      i = i + 1
  }
  out[0] = 0 as u8
  return 0
241}
242// highest step idx with status=OK for this run token
243func wf_max_ok_step(buf: *u8, n: i64, ridtok: *u8) -> i64 {
  var mx: i64 = 0
  var ls: i64 = 0
  while ls < n {
      var le: i64 = ls
      while le < n { if buf[le] == (10 as u8) { break } le = le + 1 }
      if wf_has_rng(buf, ls, le, ridtok) == 1 { if wf_has_rng(buf, ls, le, "status=OK" as *u8) == 1 {
          let v: i64 = wf_num_after(buf, ls, le, " step=" as *u8)
          if v > mx { mx = v }
      } }
      ls = le + 1
  }
  return mx
256}
257
258// ---- definition validation (whole-set, loud, nothing half-loaded) ----
259func wf_action_ok(a: *u8) -> i64 {
  if seq(a, "create-task" as *u8) == 1 { return 1 }
  if seq(a, "send" as *u8) == 1 { return 1 }
  if seq(a, "notify" as *u8) == 1 { return 1 }
  if seq(a, "update-field" as *u8) == 1 { return 1 }
  if seq(a, "probe-fail" as *u8) == 1 { return 1 }
  if seq(a, "exec-organ" as *u8) == 1 { return 1 }
  if seq(a, "approve" as *u8) == 1 { return 1 }
  if seq(a, "set-var" as *u8) == 1 { return 1 }
  if seq(a, "for-each" as *u8) == 1 { return 1 }
  return 0
270}
271func wf_load(steps: *i64, n: i64) -> i64 {
  let a: *u8 = sys_mmap(64)
  let ch: *u8 = sys_mmap(64)
  let ix: *u8 = sys_mmap(32)
  let ma: *u8 = sys_mmap(32)
  var i: i64 = 0
  while i < n {
      let r: *u8 = steps[i] as *u8
      pipe_field(r, 1, ix, 32)
      if wf_atoi(ix) < 1 { p("WFLOW load ERROR bad step idx -- fail loud\n" as *u8); return 0 - 1 }
      pipe_field(r, 2, a, 64)
      if wf_action_ok(a) == 0 { p("WFLOW load ERROR unknown action=" as *u8); p(a); p(" -- fail loud\n" as *u8); return 0 - 1 }
      if seq(a, "send" as *u8) == 1 {
          pipe_field(r, 3, ch, 64)
          if sr_valid(ch) == 0 { p("WFLOW load ERROR bad send channel=" as *u8); p(ch); p(" -- fail loud\n" as *u8); return 0 - 1 }
      }
      if seq(a, "exec-organ" as *u8) == 1 {
          pipe_field(r, 3, ch, 64)
          var chok: i64 = 1
          if ch[0] == (0 as u8) { chok = 0 }
          if ch[0] == (45 as u8) { if ch[1] == (0 as u8) { chok = 0 } }
          if ch[0] == (62 as u8) { chok = 0 }
          var gsep: i64 = 0
          var gi2: i64 = 0
          while ch[gi2] != (0 as u8) { if ch[gi2] == (62 as u8) { gsep = gi2 } gi2 = gi2 + 1 }
          if gsep > 0 { if ch[gsep+1] == (0 as u8) { chok = 0 } }
          if chok == 0 { p("WFLOW load ERROR exec-organ needs connector or connector>var -- fail loud\n" as *u8); return 0 - 1 }
      }
      if seq(a, "approve" as *u8) == 1 {
          pipe_field(r, 3, ch, 64)
          var apok: i64 = 1
          if ch[0] == (0 as u8) { apok = 0 }
          if ch[0] == (45 as u8) { if ch[1] == (0 as u8) { apok = 0 } }
          if apok == 0 { p("WFLOW load ERROR approve needs an approver name -- fail loud\n" as *u8); return 0 - 1 }
      }
      if seq(a, "set-var" as *u8) == 1 {
          let ag: *u8 = sys_mmap(512)
          pipe_field(r, 4, ag, 512)
          var haseq: i64 = 0
          var gi: i64 = 0
          while ag[gi] != (0 as u8) { if ag[gi] == (61 as u8) { haseq = 1 } gi = gi + 1 }
          if haseq == 0 { p("WFLOW load ERROR set-var needs key=value -- fail loud\n" as *u8); return 0 - 1 }
      }
      if seq(a, "for-each" as *u8) == 1 {
          pipe_field(r, 3, ch, 64)
          var feok: i64 = 1
          if ch[0] == (0 as u8) { feok = 0 }
          if ch[0] == (45 as u8) { if ch[1] == (0 as u8) { feok = 0 } }
          if ch[0] == (62 as u8) { feok = 0 }
          let ag2: *u8 = sys_mmap(512)
          pipe_field(r, 4, ag2, 512)
          if ag2[0] == (0 as u8) { feok = 0 }
          if feok == 0 { p("WFLOW load ERROR for-each needs connector and a list arg -- fail loud\n" as *u8); return 0 - 1 }
      }
      pipe_field(r, 5, ma, 32)
      if wf_atoi(ma) < 1 { p("WFLOW load ERROR max-attempts under 1 -- fail loud\n" as *u8); return 0 - 1 }
      // R9: optional 7th field = per-step condition, must be k=v or dash
      let cnd: *u8 = sys_mmap(256)
      pipe_field(r, 6, cnd, 256)
      var cok: i64 = 1
      if cnd[0] != (0 as u8) {
          if cnd[0] == (45 as u8) { if cnd[1] == (0 as u8) { cok = 1 } else { cok = 0 } } else { cok = 0 }
          if cok == 0 {
              var hq: i64 = 0
              var ci: i64 = 0
              while cnd[ci] != (0 as u8) { if cnd[ci] == (61 as u8) { hq = 1 } ci = ci + 1 }
              if hq == 1 { cok = 1 }
          }
      }
      if cok == 0 { p("WFLOW load ERROR step condition must be k=v or dash -- fail loud\n" as *u8); return 0 - 1 }
      // R10: optional 8th field = on-fail target step idx (forward-only, > own idx)
      let onf: *u8 = sys_mmap(32)
      pipe_field(r, 7, onf, 32)
      var ofok: i64 = 1
      if onf[0] != (0 as u8) {
          var isdash: i64 = 0
          if onf[0] == (45 as u8) { if onf[1] == (0 as u8) { isdash = 1 } }
          if isdash == 0 {
              let tj: i64 = wf_atoi(onf)
              if tj < 1 { ofok = 0 }
              pipe_field(r, 1, ix, 32)
              if tj <= wf_atoi(ix) { ofok = 0 }
          }
      }
      if ofok == 0 { p("WFLOW load ERROR on-fail target must be a LATER step idx or dash -- fail loud\n" as *u8); return 0 - 1 }
      i = i + 1
  }
  return n
359}
360func wf_load_flows(flows: *i64, n: i64) -> i64 {
  let k: *u8 = sys_mmap(128)
  var i: i64 = 0
  while i < n {
      let r: *u8 = flows[i] as *u8
      pipe_field(r, 1, k, 128)
      if k[0] == (0 as u8) { p("WFLOW load ERROR flow without event-kind -- fail loud\n" as *u8); return 0 - 1 }
      i = i + 1
  }
  return n
370}
371// does flow header match event? (kind equal + cond k=v present, or cond '-')
372func wf_match(row:*u8,ev:*u8)->i64{
let fk:*WfOwnedText=wf_id_pipe(row,1);let cond:*WfOwnedText=wf_id_pipe(row,2);if fk==(0 as *WfOwnedText)||cond==(0 as *WfOwnedText){wf_text_free(fk);wf_text_free(cond);return 0}
var n:i64=0;while ev[n]!=0&&ev[n]!=(126 as u8){n=n+1};var result:i64=0
if fk.len==n&&wf_text_match(ev,0,n,fk.data,n)==1{if cond.len==0||seq(cond.data,"-")==1{result=1}else{let needle:*WfOwnedText=wf_id_join("~",cond.data,"");if needle!=(0 as *WfOwnedText){let total:i64=slen(ev);var i:i64=0;while i<total{if wf_text_match(ev,i,total,needle.data,needle.len)==1{let end:i64=i+needle.len;if end==total||ev[end]==(126 as u8){result=1}};i=i+1};wf_text_free(needle)}}}
wf_text_free(fk);wf_text_free(cond);return result
377}
378
379func wf_emit(led: *u8, rid: *u8, fid: *u8, step: i64, st: *u8, att: i64) -> i64 {
  let record:*WfOwnedText = wf_id_alloc(rid,fid,st,"",slen("WFRUN rid= flow= step= status= att=\n")+slen("9223372036854775807")*2+1)
  if record==(0 as *WfOwnedText){return WF_EVIDENCE_ERROR}
  let ln:*u8=record.data
  var o: i64 = 0
  o = wf_cat(ln, o, "WFRUN rid=" as *u8)
  o = wf_cat(ln, o, rid)
  o = wf_cat(ln, o, " flow=" as *u8)
  o = wf_cat(ln, o, fid)
  o = wf_cat(ln, o, " step=" as *u8)
  o = wf_catn(ln, o, step)
  o = wf_cat(ln, o, " status=" as *u8)
  o = wf_cat(ln, o, st)
  o = wf_cat(ln, o, " att=" as *u8)
  o = wf_catn(ln, o, att)
  ln[o] = 10 as u8
  ln[o+1] = 0 as u8
  let rc:i64=wf_append(led,ln)
  wf_text_free(record)
  return rc
399}
400
401// ---- R3 connector layer: GREEN fail-closed catalog, any blessed organ as a workflow step ----
402// catalog rows: name<TAB>elfpath<TAB>GREEN. No catalog / unknown name / flag not exactly GREEN = refuse.
403func wf_conn_resolve(cx: *i64, name: *u8, out: *u8) -> i64 {
  let conf: *u8 = cx[3] as *u8
  if (conf as i64) == 0 { p("WFLOW exec-organ ERROR no connector catalog loaded -- fail closed\n" as *u8); return 0 }
  let szp: *i64 = sys_mmap(8) as *i64
  let buf: *u8 = wf_read_snapshot(conf, szp)
  if (buf as i64) == 0 { return 0 }
  let n: i64 = szp[0]
  if n == 0 { p("WFLOW exec-organ ERROR connector catalog missing or empty -- fail closed\n" as *u8); return 0 }
  let nm: i64 = slen(name)
  var ls: i64 = 0
  while ls < n {
      var le: i64 = ls
      while le < n { if buf[le] == (10 as u8) { break } le = le + 1 }
      var t1: i64 = ls
      while t1 < le { if buf[t1] == (9 as u8) { break } t1 = t1 + 1 }
      var hit: i64 = 0
      if t1 - ls == nm {
          hit = 1
          var k: i64 = 0
          while k < nm { if buf[ls+k] != name[k] { hit = 0; k = nm } else { k = k + 1 } }
      }
      if hit == 1 {
          var t2: i64 = t1 + 1
          while t2 < le { if buf[t2] == (9 as u8) { break } t2 = t2 + 1 }
          let fl0: i64 = t2 + 1
          var okflag: i64 = 0
          if le - fl0 == 5 {
              okflag = 1
              let g: *u8 = "GREEN" as *u8
              var k2: i64 = 0
              while k2 < 5 { if buf[fl0+k2] != g[k2] { okflag = 0; k2 = 5 } else { k2 = k2 + 1 } }
          }
          if okflag == 0 { p("WFLOW exec-organ ERROR connector=" as *u8); p(name); p(" not GREEN -- fail closed\n" as *u8); return 0 }
          var q: i64 = t1 + 1
          var o: i64 = 0
          while q < t2 { if o < 500 { out[o] = buf[q]; o = o + 1 } q = q + 1 }
          out[o] = 0 as u8
          if o == 0 { return 0 }
          return 1
      }
      ls = le + 1
  }
  p("WFLOW exec-organ ERROR connector unknown=" as *u8); p(name); p(" -- fail closed\n" as *u8)
  return 0
447}
448func wf_rd32(b: *u8, off: i64) -> i64 {
  return (b[off] as i64) + ((b[off+1] as i64) * 256) + ((b[off+2] as i64) * WF_MAGIC_65536) + ((b[off+3] as i64) * WF_MAGIC_16777216)
450}
451// run the resolved organ as the step: argv = arg split on ~ ('-' or empty = no args); exit 0 = step OK.
452// R8b: channel `connector>varname` CAPTURES child stdout (bounded 480B, newlines to spaces, loud over-cap)
453// into the run variable plane -- step OUTPUT becomes later steps' INPUT.
454func wf_exec_organ(cx: *i64, name0: *u8, arg: *u8, rid: *u8) -> i64 {
  let name: *u8 = sys_mmap(128)
  let capv: *u8 = sys_mmap(128)
  var ni: i64 = 0
  var ci: i64 = 0
  var seen: i64 = 0
  var i0: i64 = 0
  while name0[i0] != (0 as u8) {
      if name0[i0] == (62 as u8) { seen = 1 } else {
          if seen == 0 { if ni < 126 { name[ni] = name0[i0]; ni = ni + 1 } } else { if ci < 126 { capv[ci] = name0[i0]; ci = ci + 1 } }
      }
      i0 = i0 + 1
  }
  name[ni] = 0 as u8
  capv[ci] = 0 as u8
  var docap: i64 = 0
  if ci > 0 { docap = 1 }
  let elf: *u8 = sys_mmap(512)
  if wf_conn_resolve(cx, name, elf) == 0 { return 0 }
  let av: *i64 = sys_mmap(8 * 16) as *i64
  av[0] = elf as i64
  var na: i64 = 1
  var skip: i64 = 0
  if arg[0] == (0 as u8) { skip = 1 }
  if arg[0] == (45 as u8) { if arg[1] == (0 as u8) { skip = 1 } }
  if skip == 0 {
      let ab: *u8 = sys_mmap(WF_MAGIC_1024)
      var i: i64 = 0
      var over: i64 = 0
      while arg[i] != (0 as u8) { if i < 1000 { ab[i] = arg[i] } else { over = 1 } i = i + 1 }
      if over == 1 { p("WFLOW exec-organ ERROR argv too long -- fail closed\n" as *u8); return 0 }
      ab[i] = 0 as u8
      let alen: i64 = i
      var st0: i64 = 0
      var j: i64 = 0
      while j <= alen {
          var cut: i64 = 0
          if j == alen { cut = 1 }
          if cut == 0 { if ab[j] == (126 as u8) { cut = 1 } }
          if cut == 1 {
              ab[j] = 0 as u8
              if na >= 14 { p("WFLOW exec-organ ERROR too many args -- fail closed\n" as *u8); return 0 }
              av[na] = (ab as i64) + st0
              na = na + 1
              st0 = j + 1
          }
          j = j + 1
      }
  }
  av[na] = 0
  let fb: *u8 = sys_mmap(16)
  if docap == 1 {
      if __syscall(293, fb as i64, 0, 0, 0, 0, 0) != 0 { p("WFLOW capture ERROR pipe failed -- fail closed\n" as *u8); return 0 }
  }
  let pid: i64 = sys_fork()
  if pid == 0 {
      if docap == 1 {
          let pw: i64 = wf_rd32(fb, 4)
          sys_dup3(pw, 1, 0)
          __syscall(3, wf_rd32(fb, 0), 0, 0, 0, 0, 0)
          __syscall(3, pw, 0, 0, 0, 0, 0)
      }
      let envp: *i64 = sys_mmap(16) as *i64
      envp[0] = "PATH=/usr/bin:/bin" as *u8 as i64
      envp[1] = 0
      sys_execve(elf, av, envp)
      sys_exit(127)
  }
  if pid < 0 { p("WFLOW exec-organ ERROR fork failed -- fail closed\n" as *u8); return 0 }
  let cbuf: *u8 = sys_mmap(WF_MAGIC_1024)
  var total: i64 = 0
  if docap == 1 {
      let pr: i64 = wf_rd32(fb, 0)
      __syscall(3, wf_rd32(fb, 4), 0, 0, 0, 0, 0)
      var go: i64 = 1
      while go == 1 {
          let rr: i64 = __syscall(0, pr, (cbuf as i64) + total, 1023 - total, 0, 0, 0)
          if rr <= 0 { go = 0 }
          if go == 1 { total = total + rr; if total >= 1023 { go = 0 } }
      }
      __syscall(3, pr, 0, 0, 0, 0, 0)
  }
  let st: *i64 = sys_mmap(16) as *i64
  var waited: i64 = sys_wait4(pid, st, 0)
  while waited == (0 - 4) { waited = sys_wait4(pid, st, 0) }
  if waited != pid { p("WFLOW exec-organ ERROR child wait failed -- fail closed\n" as *u8); return 0 }
  let ec: i64 = wait_status_rc(st[0])
  p("WFLOW-EXEC-ORGAN connector=" as *u8); p(name); p(" elf=" as *u8); p(elf); p(" exit=" as *u8); pn(ec); p("\n" as *u8)
  if ec != 0 { return 0 }
  if docap == 1 {
      if total > 480 { p("WFLOW capture ERROR output over 480 bytes -- fail loud\n" as *u8); return 0 }
      var t: i64 = 0
      while t < total { if cbuf[t] == (10 as u8) { cbuf[t] = 32 as u8 } if cbuf[t] == (13 as u8) { cbuf[t] = 32 as u8 } t = t + 1 }
      while total > 0 { if cbuf[total-1] == (32 as u8) { total = total - 1 } else { break } }
      cbuf[total] = 0 as u8
      let ledc: *u8 = cx[2] as *u8
      if wf_emit_var(ledc, rid, capv, cbuf) != 0 { return WF_EVIDENCE_ERROR }
      p("WFLOW-CAPTURE run=" as *u8); p(rid); p(" var=" as *u8); p(capv); p(" v=" as *u8); p(cbuf); p("\n" as *u8)
  }
  return 1
554}
555
556// ---- R11 loops: apply-to-each over the connector plane ----
557func wf_emit_iter(led: *u8, rid: *u8, n: i64, val: *u8) -> i64 {
let number:*WfOwnedText=wf_text_decimal(n)
if number==(0 as *WfOwnedText){return WF_EVIDENCE_ERROR}
let rn:i64=slen(rid);let vn:i64=slen(val)
var bytes:i64=slen("WFITER rid=")+slen(" n=")+slen(" item=")+2
if number.len>WF_TEXT_I64_MAX-bytes{wf_text_free(number);return WF_EVIDENCE_ERROR};bytes=bytes+number.len
if rn>WF_TEXT_I64_MAX-bytes{wf_text_free(number);return WF_EVIDENCE_ERROR};bytes=bytes+rn
if vn>WF_TEXT_I64_MAX-bytes{wf_text_free(number);return WF_EVIDENCE_ERROR};bytes=bytes+vn
let line:*u8=sys_mmap_try(bytes)
if (line as i64)<=0{wf_text_free(number);return WF_EVIDENCE_ERROR}
var at:i64=wf_cat(line,0,"WFITER rid=")
at=wf_cat(line,at,rid);at=wf_cat(line,at," n=");at=wf_cat(line,at,number.data)
at=wf_cat(line,at," item=");at=wf_cat(line,at,val);line[at]=10 as u8;line[at+1]=0
var rc:i64=wf_append(led,line)
if sys_munmap_direct(line,bytes)!=0{rc=WF_EVIDENCE_ERROR}
if wf_text_free(number)!=0{rc=WF_EVIDENCE_ERROR}
return rc
574}
575// for-each: arg = <listexpr>~<per-item template> (template default {item}). The list substitutes ONCE
576// (comma-split, empties skipped; input-sized storage); per item: bind {item} as a run var, emit WFITER,
577// substitute the template, fork the connector. Any failed item fails the STEP (retry/on-fail apply).
578// Empty list = zero iterations, step OK (apply-to-each on an empty collection is a no-op).
579func wf_for_each(cx: *i64, chspec: *u8, argraw: *u8, rid: *u8) -> i64 {
let led:*u8=cx[2] as *u8;let total:i64=slen(argraw)
var sep:i64=0;while sep<total{if argraw[sep]==(126 as u8){break};sep=sep+1}
let list:*WfOwnedText=wf_text_slice(argraw,0,sep)
if list==(0 as *WfOwnedText){return WF_EVIDENCE_ERROR}
var tpl:*WfOwnedText=0 as *WfOwnedText
if sep<total{tpl=wf_text_slice(argraw,sep+1,total-sep-1)}else{tpl=wf_text_slice("{item}",0,slen("{item}"))}
if tpl==(0 as *WfOwnedText){wf_text_free(list);return WF_EVIDENCE_ERROR}
let expanded:*WfOwnedText=wf_subst_owned(cx,rid,list.data)
wf_text_free(list)
if expanded==(0 as *WfOwnedText){wf_text_free(tpl);return 0}
var rc:i64=1;var start:i64=0;var cursor:i64=0;var item:i64=0
while cursor<=expanded.len{
var cut:i64=0
if cursor==expanded.len{cut=1}else{if expanded.data[cursor]==(44 as u8){cut=1}}
if cut==1{
 expanded.data[cursor]=0
 if cursor>start{
  let value:*u8=((expanded.data as i64)+start) as *u8;item=item+1
  if wf_emit_var(led,rid,"item",value)!=0{rc=WF_EVIDENCE_ERROR;break}
  if wf_emit_iter(led,rid,item,value)!=0{rc=WF_EVIDENCE_ERROR;break}
  let prepared:*WfOwnedText=wf_subst_owned(cx,rid,tpl.data)
  if prepared==(0 as *WfOwnedText){rc=0;break}
  let result:i64=wf_exec_organ(cx,chspec,prepared.data,rid)
  if wf_text_free(prepared)!=0{rc=WF_EVIDENCE_ERROR;break}
  if result!=1{rc=result;break}
 }
 start=cursor+1
}
cursor=cursor+1
}
if wf_text_free(expanded)!=0{rc=WF_EVIDENCE_ERROR}
if wf_text_free(tpl)!=0{rc=WF_EVIDENCE_ERROR}
return rc
613}
614
615// ---- action execution (in-process v1 + exec-organ; send-channel semantics from nx_send) ----
616func wf_exec(cx: *i64, action: *u8, ch: *u8, arg: *u8, att: i64, rid: *u8) -> i64 {
  p("WFLOW-ACTION action=" as *u8); p(action)
  if seq(action, "send" as *u8) == 1 { p(" channel=" as *u8); p(ch); p(" route=" as *u8); p(sr_route(ch)) }
  p(" arg=" as *u8); p(arg)
  p(" att=" as *u8); pn(att)
  p("\n" as *u8)
  if seq(action, "probe-fail" as *u8) == 1 {
      let k: i64 = wf_atoi(arg)
      if att < k { return 0 }
      return 1
  }
  if seq(action, "exec-organ" as *u8) == 1 { return wf_exec_organ(cx, ch, arg, rid) }
  if seq(action, "for-each" as *u8) == 1 { return wf_for_each(cx, ch, arg, rid) }
  return 1
630}
631
632// ---- R4 human-in-the-loop approvals: WFDEC decision records on the same ledger ----
633// WFDEC rid=<rid> step=<n> decision=APPROVED|DENIED by=<who>. DENY WINS if both exist (fail-safe,
634// deny-by-default). No decision = the approve step PARKS the run (status=WAIT, in-flight); wf_resume
635// re-examines it and completes once a decision lands.
636func wf_decision(cx: *i64, rid: *u8, idx: i64) -> i64 {
  let led: *u8 = cx[2] as *u8
  let szp: *i64 = sys_mmap_try(__size_of(i64)) as *i64
  if (szp as i64)<=0{return WF_EVIDENCE_ERROR}
  let buf: *u8 = wf_read_snapshot(led, szp)
  if (buf as i64) == 0 { sys_munmap_direct(szp as *u8,8); return 0 }
  let token:*WfOwnedText=wf_id_tok(rid)
  if token==(0 as *WfOwnedText){sys_munmap_direct(buf,szp[0]+1);sys_munmap_direct(szp as *u8,8);return WF_EVIDENCE_ERROR}
  let tok:*u8=token.data
  var q: i64 = 0
  q = wf_cat(tok, q, " rid=" as *u8)
  q = wf_cat(tok, q, rid)
  tok[q] = 32 as u8
  tok[q+1] = 0 as u8
  let nb: *u8 = sys_mmap_try(128)
  if (nb as i64)<=0{wf_text_free(token);sys_munmap_direct(buf,szp[0]+1);sys_munmap_direct(szp as *u8,__size_of(i64));return WF_EVIDENCE_ERROR}
  var o: i64 = 0
  o = wf_cat(nb, o, " step=" as *u8)
  o = wf_catn(nb, o, idx)
  o = wf_cat(nb, o, " decision=DENIED" as *u8)
  nb[o] = 0 as u8
  var result:i64=0
  if wf_lines_with2(buf, szp[0], tok, nb) > 0 { result=0-1 }
  var o2: i64 = 0
  o2 = wf_cat(nb, o2, " step=" as *u8)
  o2 = wf_catn(nb, o2, idx)
  o2 = wf_cat(nb, o2, " decision=APPROVED" as *u8)
  nb[o2] = 0 as u8
  if result==0&&wf_lines_with2(buf, szp[0], tok, nb) > 0 { result=1 }
  sys_munmap_direct(nb,128);wf_text_free(token);sys_munmap_direct(buf,szp[0]+1);sys_munmap_direct(szp as *u8,8)
  return result
667}
668// the operator surface: append a decision record (only APPROVED or DENIED accepted, loud else)
669func wf_decide(cx: *i64, rid: *u8, idx: i64, decision: *u8, who: *u8) -> i64 {
  var okd: i64 = 0
  if seq(decision, "APPROVED" as *u8) == 1 { okd = 1 }
  if seq(decision, "DENIED" as *u8) == 1 { okd = 1 }
  if okd == 0 { p("WFLOW decide ERROR decision must be APPROVED or DENIED -- fail loud\n" as *u8); return 0 - 1 }
  let led: *u8 = cx[2] as *u8
  let record:*WfOwnedText = wf_id_alloc(rid,decision,who,"",slen("WFDEC rid= step= decision= by=\n")+slen("9223372036854775807")+1)
  if record==(0 as *WfOwnedText){return WF_EVIDENCE_ERROR}
  let ln:*u8=record.data
  var o: i64 = 0
  o = wf_cat(ln, o, "WFDEC rid=" as *u8)
  o = wf_cat(ln, o, rid)
  o = wf_cat(ln, o, " step=" as *u8)
  o = wf_catn(ln, o, idx)
  o = wf_cat(ln, o, " decision=" as *u8)
  o = wf_cat(ln, o, decision)
  o = wf_cat(ln, o, " by=" as *u8)
  o = wf_cat(ln, o, who)
  ln[o] = 10 as u8
  ln[o+1] = 0 as u8
  let rc:i64=wf_append(led,ln)
  wf_text_free(record)
  return rc
692}
693
694// ---- R8 data passing: run-scoped variables on the SAME event-sourced ledger ----
695// WFVAR rid=<rid> k=<key> v=<value-to-end-of-line>. Trigger-event fields auto-bind at fire; a set-var
696// step writes derived vars; {key} placeholders in any step arg substitute at execution; LAST write wins;
697// unknown placeholder = LOUD step failure (the send_merge law). Values must not contain ~ or newline.
698func wf_emit_var(led: *u8, rid: *u8, k: *u8, v: *u8) -> i64 {
  // Allocate the complete serialized record, including newline and terminator.
  var bytes: i64 = slen("WFVAR rid=" as *u8)+slen(" k=" as *u8)+slen(" v=" as *u8)+2
  let rn: i64 = slen(rid);let kn: i64 = slen(k);let vn: i64 = slen(v)
  if rn>9223372036854775807-bytes{return 0-1};bytes=bytes+rn
  if kn>9223372036854775807-bytes{return 0-1};bytes=bytes+kn
  if vn>9223372036854775807-bytes{return 0-1};bytes=bytes+vn
  let ln: *u8 = sys_mmap_try(bytes)
  if (ln as i64)<=0{p("WFLOW variable ERROR allocation refused; record not appended\n" as *u8);return 0-1}
  var o: i64 = 0
  o = wf_cat(ln, o, "WFVAR rid=" as *u8)
  o = wf_cat(ln, o, rid)
  o = wf_cat(ln, o, " k=" as *u8)
  o = wf_cat(ln, o, k)
  o = wf_cat(ln, o, " v=" as *u8)
  o = wf_cat(ln, o, v)
  ln[o] = 10 as u8
  ln[o+1] = 0 as u8
  let rc:i64=wf_append(led,ln)
  let released:i64=sys_munmap_direct(ln,bytes)
  if released!=0{return 0-1}
  return rc
720}
721// copy value after key to END OF LINE (values may contain spaces)
722func wf_val_line_end(buf: *u8, ls: i64, le: i64, key: *u8, out: *u8, cap: i64) -> i64 {
  let m: i64 = slen(key)
  var i: i64 = ls
  while i + m <= le {
      var k: i64 = 0
      var hit: i64 = 1
      while k < m { if buf[i+k] != key[k] { hit = 0; k = m } else { k = k + 1 } }
      if hit == 1 {
          var q: i64 = i + m
          var t: i64 = 0
          while q < le { if t < cap - 1 { out[t] = buf[q]; t = t + 1 } q = q + 1 }
          out[t] = 0 as u8
          return 1
      }
      i = i + 1
  }
  out[0] = 0 as u8
  return 0
740}
741func wf_obs_tok_fwd(tok: *u8, rid: *u8) -> i64 {
  var q: i64 = 0
  q = wf_cat(tok, q, " rid=" as *u8)
  q = wf_cat(tok, q, rid)
  tok[q] = 32 as u8
  tok[q+1] = 0 as u8
  return q
748}
749// Latest WFVAR value: 1 found, 0 missing, -1 unresolved or insufficient capacity; no partial output.
750func wf_var_get(cx: *i64, rid: *u8, key: *u8, out: *u8, cap: i64) -> i64 {
if cap<=0{return 0-1};out[0]=0
let owned:*WfOwnedText=wf_var_owned(cx,rid,key)
if owned==(0 as *WfOwnedText){return 0-1}
var rc:i64=owned.status
if owned.status==1{
if owned.len>=cap{rc=0-1}else{var i:i64=0;while i<owned.len{out[i]=owned.data[i];i=i+1};out[owned.len]=0}
}
if wf_text_free(owned)!=0{out[0]=0;return 0-1}
return rc
760}
761// Substitute {key} from one owned ledger snapshot; -1 on unresolved input, allocation or insufficient output.
762func wf_subst(cx: *i64, rid: *u8, src: *u8, out: *u8, cap: i64) -> i64 {
if cap<=0{return 0-1};out[0]=0
let owned:*WfOwnedText=wf_subst_owned(cx,rid,src)
if owned==(0 as *WfOwnedText){return 0-1}
var rc:i64=owned.len
if owned.len>=cap{rc=0-1}else{var i:i64=0;while i<owned.len{out[i]=owned.data[i];i=i+1};out[owned.len]=0}
if wf_text_free(owned)!=0{out[0]=0;return 0-1}
return rc
770}
771// bind every k=v field of the trigger event as an initial run variable (segment 0 = kind, skipped)
772func wf_bind_event(cx: *i64, rid: *u8, ev: *u8) -> i64 {
  let n:i64=slen(ev)
  if n==9223372036854775807{return WF_EVIDENCE_ERROR}
  let bytes:i64=n+1
  let kb:*u8=sys_mmap_try(bytes)
  if (kb as i64)<=0{return WF_EVIDENCE_ERROR}
  let vb:*u8=sys_mmap_try(bytes)
  if (vb as i64)<=0{sys_munmap_direct(kb,bytes);return WF_EVIDENCE_ERROR}
  var i:i64=0
  var bound:i64=0
  var result:i64=0
  while i<n {
      if ev[i]!=(126 as u8){i=i+1}
      else {
          var j:i64=i+1
          var kn:i64=0
          var haseq:i64=0
          while j<n {
              if ev[j]==(126 as u8){break}
              if ev[j]==(61 as u8){haseq=1;j=j+1;break}
              kb[kn]=ev[j];kn=kn+1;j=j+1
          }
          kb[kn]=0 as u8
          if haseq==1 {
              var vn:i64=0
              while j<n {
                  if ev[j]==(126 as u8){break}
                  vb[vn]=ev[j];vn=vn+1;j=j+1
              }
              vb[vn]=0 as u8
              if kn>0 {
                  if wf_emit_var(cx[2] as *u8,rid,kb,vb)!=0{result=WF_EVIDENCE_ERROR;break}
                  bound=bound+1
              }
          }
          i=j
      }
  }
  let kr:i64=sys_munmap_direct(kb,bytes)
  let vr:i64=sys_munmap_direct(vb,bytes)
  if result!=0||kr!=0||vr!=0{return WF_EVIDENCE_ERROR}
  return bound
814}
815
816// ---- R12 versioning: @version N directive in the flows file; runs stamp WFVER; resume flags drift ----
817func wf_defs_version(path: *u8) -> i64 {
  let szp: *i64 = sys_mmap(8) as *i64
  let buf: *u8 = wf_read_snapshot(path, szp)
  if (buf as i64) == 0 { return 0 }
  let n: i64 = szp[0]
  var ls: i64 = 0
  while ls < n {
      var le: i64 = ls
      while le < n { if buf[le] == (10 as u8) { break } le = le + 1 }
      if wf_has_rng(buf, ls, le, "@version " as *u8) == 1 {
          let v: i64 = wf_num_after(buf, ls, le, "@version " as *u8)
          if v > 0 { return v }
      }
      ls = le + 1
  }
  return 0
833}
834func wf_emit_ver(led: *u8, rid: *u8, ver: i64) -> i64 {
  let record:*WfOwnedText = wf_id_alloc(rid,"","","",slen("WFVER rid= ver=\n")+slen("9223372036854775807")+1)
  if record==(0 as *WfOwnedText){return WF_EVIDENCE_ERROR}
  let ln:*u8=record.data
  var o: i64 = 0
  o = wf_cat(ln, o, "WFVER rid=" as *u8)
  o = wf_cat(ln, o, rid)
  o = wf_cat(ln, o, " ver=" as *u8)
  o = wf_catn(ln, o, ver)
  ln[o] = 10 as u8
  ln[o+1] = 0 as u8
  let rc:i64=wf_append(led,ln)
  wf_text_free(record)
  return rc
848}
849func wf_emit_drift(led: *u8, rid: *u8, ranv: i64, nowv: i64) -> i64 {
  let record:*WfOwnedText = wf_id_alloc(rid,"","","",slen("WFVERDRIFT rid= ran= now=\n")+slen("9223372036854775807")*2+1)
  if record==(0 as *WfOwnedText){return WF_EVIDENCE_ERROR}
  let ln:*u8=record.data
  var o: i64 = 0
  o = wf_cat(ln, o, "WFVERDRIFT rid=" as *u8)
  o = wf_cat(ln, o, rid)
  o = wf_cat(ln, o, " ran=" as *u8)
  o = wf_catn(ln, o, ranv)
  o = wf_cat(ln, o, " now=" as *u8)
  o = wf_catn(ln, o, nowv)
  ln[o] = 10 as u8
  ln[o+1] = 0 as u8
  let rc:i64=wf_append(led,ln)
  wf_text_free(record)
  return rc
865}
866// the version a run was STARTED under (last WFVER line for the rid; 0 = unversioned)
867func wf_run_ver(buf: *u8, n: i64, ridtok: *u8) -> i64 {
  var v: i64 = 0
  var ls: i64 = 0
  while ls < n {
      var le: i64 = ls
      while le < n { if buf[le] == (10 as u8) { break } le = le + 1 }
      if wf_has_rng(buf, ls, le, "WFVER " as *u8) == 1 { if wf_has_rng(buf, ls, le, ridtok) == 1 {
          let x: i64 = wf_num_after(buf, ls, le, " ver=" as *u8)
          if x > 0 { v = x }
      } }
      ls = le + 1
  }
  return v
880}
881
882// ---- R13 templates: gallery = <dir>/index (name|description rows) + <name>.flows/.steps pairs ----
883func wf_tpl_exists(path: *u8) -> i64 {
  let fd: i64 = __syscall(257, 0 - 100, path as i64, 0, 0, 0, 0)
  if fd < 0 { return 0 }
  __syscall(3, fd, 0, 0, 0, 0, 0)
  return 1
888}
889func wf_tpl_path(dir: *u8, name: *u8, ext: *u8, out: *u8) -> i64 {
  var o: i64 = 0
  o = wf_cat(out, o, dir)
  out[o] = 47 as u8
  o = o + 1
  o = wf_cat(out, o, name)
  o = wf_cat(out, o, ext)
  out[o] = 0 as u8
  return o
898}
899func wf_tpl_index_path(dir: *u8, out: *u8) -> i64 {
  var o: i64 = 0
  o = wf_cat(out, o, dir)
  out[o] = 47 as u8
  o = o + 1
  o = wf_cat(out, o, "index" as *u8)
  out[o] = 0 as u8
  return o
907}
908// list the gallery: every index row printed with file-existence + version; returns USABLE count, -1 loud
909func wf_tpl_list(dir: *u8) -> i64 {
  let ip: *u8 = sys_mmap(512)
  wf_tpl_index_path(dir, ip)
  let szp: *i64 = sys_mmap(8) as *i64
  let buf: *u8 = wf_read_snapshot(ip, szp)
  if (buf as i64) == 0 { return 0 - 1 }
  let n: i64 = szp[0]
  if n == 0 { p("WFLOW templates ERROR missing gallery index -- fail loud\n" as *u8); return 0 - 1 }
  let nm: *u8 = sys_mmap(128)
  let ds: *u8 = sys_mmap(512)
  let fp2: *u8 = sys_mmap(512)
  let sp2: *u8 = sys_mmap(512)
  var cnt: i64 = 0
  var ls: i64 = 0
  while ls < n {
      var le: i64 = ls
      while le < n { if buf[le] == (10 as u8) { break } le = le + 1 }
      var keep: i64 = 0
      if le > ls { keep = 1 }
      if keep == 1 { if buf[ls] == (35 as u8) { keep = 0 } }
      if keep == 1 {
          let row: *u8 = sys_mmap(le - ls + 2)
          var k: i64 = 0
          while ls + k < le { row[k] = buf[ls+k]; k = k + 1 }
          row[k] = 0 as u8
          pipe_field(row, 0, nm, 128)
          pipe_field(row, 1, ds, 512)
          wf_tpl_path(dir, nm, ".flows" as *u8, fp2)
          wf_tpl_path(dir, nm, ".steps" as *u8, sp2)
          let fe: i64 = wf_tpl_exists(fp2)
          let se: i64 = wf_tpl_exists(sp2)
          let v: i64 = wf_defs_version(fp2)
          p("WFTPL name=" as *u8); p(nm)
          p(" ver=" as *u8); pn(v)
          p(" flows=" as *u8); pn(fe)
          p(" steps=" as *u8); pn(se)
          p(" desc=" as *u8); p(ds)
          p("\n" as *u8)
          if fe == 1 { if se == 1 { cnt = cnt + 1 } }
      }
      ls = le + 1
  }
  p("WFTPL-TOTAL usable=" as *u8); pn(cnt); p("\n" as *u8)
  return cnt
953}
954func wf_tpl_indexed(dir: *u8, name: *u8) -> i64 {
  let ip: *u8 = sys_mmap(512)
  wf_tpl_index_path(dir, ip)
  let szp: *i64 = sys_mmap(8) as *i64
  let buf: *u8 = wf_read_snapshot(ip, szp)
  if (buf as i64) == 0 { return 0 - 1 }
  let n: i64 = szp[0]
  if n == 0 { return 0 - 1 }
  let nm: *u8 = sys_mmap(128)
  var ls: i64 = 0
  while ls < n {
      var le: i64 = ls
      while le < n { if buf[le] == (10 as u8) { break } le = le + 1 }
      if le > ls { if buf[ls] != (35 as u8) {
          let row: *u8 = sys_mmap(le - ls + 2)
          var k: i64 = 0
          while ls + k < le { row[k] = buf[ls+k]; k = k + 1 }
          row[k] = 0 as u8
          pipe_field(row, 0, nm, 128)
          if seq(nm, name) == 1 { return 1 }
      } }
      ls = le + 1
  }
  return 0
978}
979// instantiate: fail-closed to INDEXED templates; byte-copy both files; whole-set validate the COPY
980// (a broken template never lands silently). Returns the template version (0 = unversioned), -1 loud.
981func wf_tpl_instantiate(dir: *u8, name: *u8, df: *u8, ds: *u8) -> i64 {
  let ix: i64 = wf_tpl_indexed(dir, name)
  if ix != 1 { p("WFLOW instantiate ERROR template not in gallery index -- fail closed\n" as *u8); return 0 - 1 }
  let fp2: *u8 = sys_mmap(512)
  let sp2: *u8 = sys_mmap(512)
  wf_tpl_path(dir, name, ".flows" as *u8, fp2)
  wf_tpl_path(dir, name, ".steps" as *u8, sp2)
  let szp: *i64 = sys_mmap(8) as *i64
  let fb: *u8 = wf_read_snapshot(fp2, szp)
  if (fb as i64) == 0 { return 0 - 1 }
  if szp[0] == 0 { p("WFLOW instantiate ERROR template flows file missing -- fail loud\n" as *u8); return 0 - 1 }
  if wf_writeall(df, fb, szp[0]) != 0 { return 0 - 1 }
  let sb: *u8 = wf_read_snapshot(sp2, szp)
  if (sb as i64) == 0 { return 0 - 1 }
  if szp[0] == 0 { p("WFLOW instantiate ERROR template steps file missing -- fail loud\n" as *u8); return 0 - 1 }
  if wf_writeall(ds, sb, szp[0]) != 0 { return 0 - 1 }
  let fl: *i64 = sys_mmap(8 * 64) as *i64
  let st2: *i64 = sys_mmap(8 * 128) as *i64
  let nf: i64 = wf_lines_load(df, fl, 64)
  if nf < 1 { return 0 - 1 }
  let nst: i64 = wf_lines_load(ds, st2, 128)
  if nst < 1 { return 0 - 1 }
  if wf_load_flows(fl, nf) < 0 { return 0 - 1 }
  if wf_load(st2, nst) < 0 { return 0 - 1 }
  let v: i64 = wf_defs_version(df)
  p("WFLOW-INSTANTIATE template=" as *u8); p(name); p(" version=" as *u8); pn(v); p(" flows=" as *u8); pn(nf); p(" steps=" as *u8); pn(nst); p("\n" as *u8)
  return v
1008}
1009
1010// R9 branching: per-step condition k=v against the run variable plane. Missing variable or
1011// mismatch = 0 (the step SKIPs, recorded in the ledger); match = 1. Equality only, v1.
1012func wf_cond_ok(cx: *i64, rid: *u8, cond: *u8) -> i64 {
let n:i64=slen(cond);var split:i64=0
while split<n{if cond[split]==(61 as u8){break};split=split+1}
if split==0{return 0};if split==n{return 0}
let key:*WfOwnedText=wf_text_slice(cond,0,split)
if key==(0 as *WfOwnedText){return 0-1}
let value:*WfOwnedText=wf_var_owned(cx,rid,key.data)
wf_text_free(key)
if value==(0 as *WfOwnedText){return 0-1}
var result:i64=0
if value.status==1{result=seq(value.data,((cond as i64)+split+1) as *u8)}
if value.status<0{result=0-1}
if wf_text_free(value)!=0{return 0-1};return result
1025}
1026
1027// ---- the durable executor: run flow fid for run rid, starting AFTER step s0 ----
1028// cx bundle: cx[0]=steps ptr, cx[1]=nsteps, cx[2]=ledger path
1029func wf_run_from(cx: *i64, rid: *u8, fid: *u8, s0: i64) -> i64 {
let steps:*i64=cx[0] as *i64;var i:i64=0;var longest:i64=0
while i<cx[1]{let n:i64=slen(steps[i] as *u8);if n>longest{longest=n};i=i+1}
if longest==WF_TEXT_I64_MAX{return WF_EVIDENCE_ERROR}
let scratch:*WfRunScratch=wf_run_scratch_new(longest+1)
if scratch==(0 as *WfRunScratch){return WF_EVIDENCE_ERROR}
let rc:i64=wf_run_from_owned(cx,rid,fid,s0,scratch)
if wf_run_scratch_free(scratch)!=0{return WF_EVIDENCE_ERROR}
return rc
1038}
1039
1040func wf_run_from_owned(cx: *i64, rid: *u8, fid: *u8, s0: i64, scratch:*WfRunScratch) -> i64 {
  let steps: *i64 = cx[0] as *i64
  let ns: i64 = cx[1]
  let led: *u8 = cx[2] as *u8
  let f2: *u8 = scratch.fields.fid
  let ix: *u8 = scratch.fields.index
  let a: *u8 = scratch.fields.action
  let ch: *u8 = scratch.fields.channel
  let arg: *u8 = scratch.fields.arg
  let ma: *u8 = scratch.fields.retry
  var i: i64 = 0
  var done: i64 = 0
  var failed: i64 = 0
  var parked: i64 = 0
  var jumpto: i64 = 0
  while i < ns {
      if wf_text_free(scratch.prepared)!=0{return WF_EVIDENCE_ERROR};scratch.prepared=0 as *WfOwnedText
      let r: *u8 = steps[i] as *u8
      pipe_field(r, 0, f2, scratch.cap)
      var use: i64 = 0
      if seq(f2, fid) == 1 { use = 1 }
      var idx: i64 = 0
      if use == 1 { pipe_field(r, 1, ix, scratch.cap); idx = wf_atoi(ix); if idx <= s0 { use = 0 } }
      // R10: an active on-fail jump skips the remaining normal-path steps (recorded) until the target
      if use == 1 { if jumpto > 0 { if idx < jumpto { if wf_emit(led, rid, fid, idx, "SKIP" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }; use = 0 } else { jumpto = 0 } } }
      if use == 1 {
          pipe_field(r, 2, a, scratch.cap)
          pipe_field(r, 3, ch, scratch.cap)
          pipe_field(r, 4, arg, scratch.cap)
          pipe_field(r, 5, ma, scratch.cap)
          // R9: optional per-step condition (7th field, k=v) vs the variable plane -> SKIP on mismatch
          let cond: *u8 = scratch.fields.condition
          pipe_field(r, 6, cond, scratch.cap)
          var argx: *u8 = arg
          var mode: i64 = 0
          var docond: i64 = 0
          if cond[0] != (0 as u8) { docond = 1 }
          if cond[0] == (45 as u8) { if cond[1] == (0 as u8) { docond = 0 } }
          if docond == 1 { let condition:i64=wf_cond_ok(cx,rid,cond);if condition==0{mode=4};if condition<0{mode=3} }
          // R8: resolve {key} placeholders from the run variable plane; unknown = LOUD step failure.
          // for-each bypasses step-level subst ({item} binds inside the loop) -- the loop substitutes
          // the list once and the per-item template each iteration.
          if mode == 0 {
              var rawarg: i64 = 0
              if seq(a, "for-each" as *u8) == 1 { rawarg = 1 }
              if rawarg==0 {
                  scratch.prepared=wf_subst_owned(cx,rid,arg)
                  if scratch.prepared==(0 as *WfOwnedText){mode=3}else{argx=scratch.prepared.data}
              }
          }
          if mode == 4 {
              p("WFLOW-SKIP run=" as *u8); p(rid); p(" step=" as *u8); pn(idx); p(" cond=" as *u8); p(cond); p("\n" as *u8)
              if wf_emit(led, rid, fid, idx, "SKIP" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
          }
          if mode == 0 { if seq(a, "set-var" as *u8) == 1 { mode = 2 } }
          if mode == 0 { if seq(a, "approve" as *u8) == 1 { mode = 1 } }
          if mode == 3 {
              if wf_emit(led, rid, fid, idx, "FAILSTEP" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
              if wf_emit(led, rid, fid, idx, "FAILED" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
              failed = 1
              i = ns
          }
          if mode == 2 {
              var e:i64=0
              while argx[e]!=(0 as u8){if argx[e]==(61 as u8){break};e=e+1}
              if argx[e]==(61 as u8){if e>0{
                  argx[e]=0
                  let kb:*u8=argx
                  let vb:*u8=((argx as i64)+e+1) as *u8
                  if wf_emit_var(led, rid, kb, vb) != 0 { return WF_EVIDENCE_ERROR }
                  p("WFLOW-SETVAR run=" as *u8); p(rid); p(" k=" as *u8); p(kb); p(" v=" as *u8); p(vb); p("\n" as *u8)
                  if wf_emit(led, rid, fid, idx, "OK" as *u8, 1) != 0 { return WF_EVIDENCE_ERROR }
                  done = done + 1
              } else { mode = 3 } } else { mode = 3 }
              if mode == 3 {
                  p("WFLOW set-var ERROR needs key=value -- fail loud\n" as *u8)
                  if wf_emit(led, rid, fid, idx, "FAILSTEP" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
                  if wf_emit(led, rid, fid, idx, "FAILED" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
                  failed = 1
                  i = ns
              }
          }
          if mode == 1 {
              let dec: i64 = wf_decision(cx, rid, idx)
              if dec == 1 {
                  p("WFLOW-APPROVAL granted run=" as *u8); p(rid); p(" step=" as *u8); pn(idx); p("\n" as *u8)
                  if wf_emit(led, rid, fid, idx, "OK" as *u8, 1) != 0 { return WF_EVIDENCE_ERROR }
                  done = done + 1
              }
              if dec == (0 - 1) {
                  p("WFLOW-APPROVAL DENIED run=" as *u8); p(rid); p(" step=" as *u8); pn(idx); p(" -- run fails (deny wins)\n" as *u8)
                  if wf_emit(led, rid, fid, idx, "FAILSTEP" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
                  if wf_emit(led, rid, fid, idx, "FAILED" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
                  failed = 1
                  i = ns
              }
              if dec == 0 {
                  p("WFLOW-APPROVAL pending run=" as *u8); p(rid); p(" step=" as *u8); pn(idx)
                  p(" approver=" as *u8); p(ch); p(" what=" as *u8); p(argx); p(" -- run PARKED\n" as *u8)
                  if wf_emit(led, rid, fid, idx, "WAIT" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
                  parked = 1
                  i = ns
              }
          }
          if mode == 0 {
              let maxa: i64 = wf_atoi(ma)
              var att: i64 = 1
              var okd: i64 = 0
              while att <= maxa {
                  if wf_emit(led, rid, fid, idx, "ATT" as *u8, att) != 0 { return WF_EVIDENCE_ERROR }
                  let rc: i64 = wf_exec(cx, a, ch, argx, att, rid)
                  if rc == WF_EVIDENCE_ERROR { return rc }
                  if rc == 1 { if wf_emit(led, rid, fid, idx, "OK" as *u8, att) != 0 { return WF_EVIDENCE_ERROR }; okd = 1; att = maxa + 1 } else { att = att + 1 }
              }
              if okd == 0 {
                  if wf_emit(led, rid, fid, idx, "FAILSTEP" as *u8, maxa) != 0 { return WF_EVIDENCE_ERROR }
                  // R10: on-fail routing (8th field) -- exhausted ACTION failures only; the failure stays
                  // recorded (FAILSTEP + ONFAIL), wf_error vars land on the plane, execution jumps forward
                  let onf: *u8 = scratch.fields.onfail
                  pipe_field(r, 7, onf, scratch.cap)
                  var tj: i64 = 0
                  var haveof: i64 = 0
                  if onf[0] != (0 as u8) { haveof = 1 }
                  if onf[0] == (45 as u8) { if onf[1] == (0 as u8) { haveof = 0 } }
                  if haveof == 1 { tj = wf_atoi(onf) }
                  if tj > 0 {
                      if wf_emit(led, rid, fid, idx, "ONFAIL" as *u8, tj) != 0 { return WF_EVIDENCE_ERROR }
                      if wf_emit_var(led, rid, "wf_error" as *u8, "yes" as *u8) != 0 { return WF_EVIDENCE_ERROR }
                      let sv: *u8 = sys_mmap(32)
                      var so: i64 = 0
                      so = wf_cat(sv, so, "step" as *u8)
                      so = wf_catn(sv, so, idx)
                      sv[so] = 0 as u8
                      if wf_emit_var(led, rid, "wf_error_step" as *u8, sv) != 0 { return WF_EVIDENCE_ERROR }
                      p("WFLOW-ONFAIL run=" as *u8); p(rid); p(" step=" as *u8); pn(idx); p(" routes-to=" as *u8); pn(tj); p("\n" as *u8)
                      jumpto = tj
                  } else {
                      if wf_emit(led, rid, fid, idx, "FAILED" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
                      failed = 1
                      i = ns
                  }
              } else { done = done + 1 }
          }
      }
      i = i + 1
  }
  if failed == 1 { return 0 - 2 }
  if parked == 1 { return 0 - 3 }
  if wf_emit(led, rid, fid, 0, "DONE" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
  return done
1190}
1191
1192// ---- fire an event: durable ledger-deduped, one run per matched flow ----
1193func wf_fire(cx:*i64,flows:*i64,nf:i64,ev:*u8)->i64{
let event:*WfOwnedText=wf_id_event(ev);if event==(0 as *WfOwnedText){p("WFLOW identity ERROR invalid event; no dispatch\n");return 0-1}
let sz:*i64=sys_mmap_try(__size_of(i64)) as *i64;if (sz as i64)<=0{wf_text_free(event);return WF_EVIDENCE_ERROR};let buf:*u8=wf_read_snapshot(cx[2] as *u8,sz)
if buf==(0 as *u8){wf_text_free(event);sys_munmap_direct(sz as *u8,__size_of(i64));return WF_EVIDENCE_ERROR}
var started:i64=0;var matched:i64=0;var f:i64=0;var rc:i64=0
while f<nf{let row:*u8=flows[f] as *u8;if wf_match(row,ev)==1{matched=matched+1;let fid:*WfOwnedText=wf_id_pipe(row,0);if fid==(0 as *WfOwnedText){rc=WF_EVIDENCE_ERROR;break};let one:i64=wf_fire_pair(cx,buf,sz[0],event.data,ev,fid.data);wf_text_free(fid);if one<0{rc=one;break};started=started+one};f=f+1}
wf_text_free(event);sys_munmap_direct(buf,sz[0]+1);sys_munmap_direct(sz as *u8,__size_of(i64));if rc<0{return rc};if matched>0&&started==0{return 0-100};return started
1200}
1201
1202func wf_flow_known(cx:*i64,fid:*u8)->i64{
let steps:*i64=cx[0] as *i64;var i:i64=0;while i<cx[1]{let f:*WfOwnedText=wf_id_pipe(steps[i] as *u8,0);if f==(0 as *WfOwnedText){return 0-1};let same:i64=seq(f.data,fid);wf_text_free(f);if same==1{return 1};i=i+1};return 0
1204}
1205
1206func wf_resume(cx:*i64)->i64{
let sz:*i64=sys_mmap_try(__size_of(i64)) as *i64;if (sz as i64)<=0{return WF_EVIDENCE_ERROR};let buf:*u8=wf_read_snapshot(cx[2] as *u8,sz);if buf==(0 as *u8){sys_munmap_direct(sz as *u8,__size_of(i64));return WF_EVIDENCE_ERROR}
let n:i64=sz[0];var resumed:i64=0;var ls:i64=0;var rc:i64=0
while ls<n{var le:i64=ls;while le<n{if buf[le]==(10 as u8){break};le=le+1};if le==n{rc=WF_EVIDENCE_ERROR;break}
if wf_text_match(buf,ls,le,"WFRUN rid=",10)==1{
 let status:*WfOwnedText=wf_id_field(buf,ls,le," status=");if status==(0 as *WfOwnedText){rc=WF_EVIDENCE_ERROR;break}
 if seq(status.data,"START")==1{
  let rid:*WfOwnedText=wf_id_field(buf,ls,le," rid=");let fid:*WfOwnedText=wf_id_field(buf,ls,le," flow=")
  if rid==(0 as *WfOwnedText)||fid==(0 as *WfOwnedText){wf_text_free(rid);wf_text_free(fid);wf_text_free(status);rc=WF_EVIDENCE_ERROR;break}
  let token:*WfOwnedText=wf_id_tok(rid.data);if token==(0 as *WfOwnedText){wf_text_free(rid);wf_text_free(fid);wf_text_free(status);rc=WF_EVIDENCE_ERROR;break}
  if wf_lines_with2(buf,n,token.data,"status=DONE")==0&&wf_lines_with2(buf,n,token.data,"status=FAILED")==0{
   if wf_id_replay_binding(buf,n,rid.data,fid.data)!=1{p("WFLOW resume ERROR identity binding incomplete; no dispatch\n");rc=WF_EVIDENCE_ERROR}
   if rc==0{if wf_flow_known(cx,fid.data)!=1{p("WFLOW resume ERROR unknown flow=");p(fid.data);p(" -- fail loud, nothing fabricated\n");rc=0-1}}
   if rc==0{let ran:i64=wf_run_ver(buf,n,token.data);if cx[4]>0&&ran>0&&ran!=cx[4]{if wf_emit_drift(cx[2] as *u8,rid.data,ran,cx[4])!=0{rc=WF_EVIDENCE_ERROR}else{p("WFLOW-RESUME-WARN version drift run=");p(rid.data);p(" ran=");pn(ran);p(" now=");pn(cx[4]);p("\n")}}}
   if rc==0{let first:i64=wf_max_ok_step(buf,n,token.data);p("WFLOW-RESUME run=");p(rid.data);p(" from-step=");pn(first+1);p(" flow=");p(fid.data);p("\n");let run:i64=wf_run_from(cx,rid.data,fid.data,first);if run==WF_EVIDENCE_ERROR{rc=run}else{resumed=resumed+1}}
  };wf_text_free(token);wf_text_free(rid);wf_text_free(fid)
 };wf_text_free(status)
};if rc!=0{break};ls=le+1
};sys_munmap_direct(buf,sz[0]+1);sys_munmap_direct(sz as *u8,__size_of(i64));if rc!=0{return rc};return resumed
1225}
1226
1227func wf_lines_load(path: *u8, arr: *i64, cap: i64) -> i64 {
  let szp: *i64 = sys_mmap(8) as *i64
  let buf: *u8 = wf_read_snapshot(path, szp)
  if (buf as i64) == 0 { return 0 - 1 }
  let n: i64 = szp[0]
  if n == 0 { p("WFLOW files ERROR missing or empty definition file -- fail loud\n" as *u8); return 0 - 1 }
  var cnt: i64 = 0
  var ls: i64 = 0
  while ls < n {
      var le: i64 = ls
      while le < n { if buf[le] == (10 as u8) { break } le = le + 1 }
      var keep: i64 = 0
      if le > ls { keep = 1 }
      if keep == 1 { if buf[ls] == (35 as u8) { keep = 0 } }
      if keep == 1 { if buf[ls] == (64 as u8) { keep = 0 } }
      if keep == 1 {
          if cnt >= cap { p("WFLOW files ERROR too many definition rows -- fail loud\n" as *u8); return 0 - 1 }
          let s: *u8 = sys_mmap(le - ls + 2)
          var k: i64 = 0
          while ls + k < le { s[k] = buf[ls+k]; k = k + 1 }
          s[k] = 0 as u8
          arr[cnt] = s as i64
          cnt = cnt + 1
      }
      ls = le + 1
  }
  return cnt
1254}
1255func wf_cx_files(flowsp: *u8, stepsp: *u8, led: *u8, cat: *u8, cx: *i64, flows: *i64) -> i64 {
  let steps: *i64 = sys_mmap(8 * 128) as *i64
  let nf: i64 = wf_lines_load(flowsp, flows, 64)
  if nf < 1 { return 0 - 1 }
  let nst: i64 = wf_lines_load(stepsp, steps, 128)
  if nst < 1 { return 0 - 1 }
  if wf_load_flows(flows, nf) < 0 { return 0 - 1 }
  if wf_load(steps, nst) < 0 { return 0 - 1 }
  cx[0] = steps as i64
  cx[1] = nst
  cx[2] = led as i64
  cx[3] = 0
  var usecat: i64 = 1
  if cat[0] == (45 as u8) { if cat[1] == (0 as u8) { usecat = 0 } }
  if usecat == 1 { cx[3] = cat as i64 }
  cx[4] = wf_defs_version(flowsp)
  return nf
1272}
1273func wf_fire_files(flowsp: *u8, stepsp: *u8, led: *u8, cat: *u8, ev: *u8) -> i64 {
  let cx: *i64 = sys_mmap(40) as *i64
  let flows: *i64 = sys_mmap(8 * 64) as *i64
  let nf: i64 = wf_cx_files(flowsp, stepsp, led, cat, cx, flows)
  if nf < 0 { return 0 - 1 }
  return wf_fire(cx, flows, nf, ev)
1279}
1280func wf_resume_files(flowsp: *u8, stepsp: *u8, led: *u8, cat: *u8) -> i64 {
  let cx: *i64 = sys_mmap(40) as *i64
  let flows: *i64 = sys_mmap(8 * 64) as *i64
  let nf: i64 = wf_cx_files(flowsp, stepsp, led, cat, cx, flows)
  if nf < 0 { return 0 - 1 }
  return wf_resume(cx)
1286}
1287
1288// ---- R6 run observability: board + 0-JS HTML derived ENTIRELY by ledger replay (nothing self-reported) ----
1289func wf_state_name(st: i64) -> *u8 {
  if st == 3 { return "DONE" as *u8 }
  if st == 2 { return "FAILED" as *u8 }
  if st == 1 { return "PARKED" as *u8 }
  return "RUNNING" as *u8
1294}
1295// derived state: DONE > FAILED > PARKED(waiting approval) > RUNNING(in-flight)
1296func wf_run_state(buf: *u8, n: i64, tok: *u8) -> i64 {
  if wf_lines_with2(buf, n, tok, "status=DONE" as *u8) > 0 { return 3 }
  if wf_lines_with2(buf, n, tok, "status=FAILED" as *u8) > 0 { return 2 }
  if wf_lines_with2(buf, n, tok, "status=WAIT" as *u8) > 0 { return 1 }
  return 0
1301}
1302// collect distinct run ids from START lines; returns count, -1 loud over cap
1303func wf_obs_runs(buf:*u8,n:i64,rids:*i64,cap:i64)->i64{
var count:i64=0;var ls:i64=0
while ls<n{var le:i64=ls;while le<n{if buf[le]==(10 as u8){break};le=le+1};if le==n{return 0-1}
if wf_text_match(buf,ls,le,"WFRUN rid=",10)==1{let st:*WfOwnedText=wf_id_field(buf,ls,le," status=");if st==(0 as *WfOwnedText){return 0-1}
 if seq(st.data,"START")==1{let r:*WfOwnedText=wf_id_field(buf,ls,le," rid=");if r==(0 as *WfOwnedText){wf_text_free(st);return 0-1};var dup:i64=0;var i:i64=0;while i<count{if seq(rids[i] as *u8,r.data)==1{dup=1};i=i+1}
  if dup==0{if count>=cap{wf_text_free(r);wf_text_free(st);return 0-1};rids[count]=r.data as i64;count=count+1;sys_munmap_direct(r as *u8,__size_of(WfOwnedText))}else{wf_text_free(r)}
 };wf_text_free(st)
};ls=le+1
};return count
1312}
1313
1314func wf_obs_tok(tok: *u8, rid: *u8) -> i64 {
  var q: i64 = 0
  q = wf_cat(tok, q, " rid=" as *u8)
  q = wf_cat(tok, q, rid)
  tok[q] = 32 as u8
  tok[q+1] = 0 as u8
  return q
1321}
1322func wf_obs_release(buf:*u8,n:i64,rids:*i64,slots:i64)->i64{
if (rids as i64)>0{var i:i64=0;while i<slots{let rid:*u8=rids[i] as *u8;if (rid as i64)>0{sys_munmap_direct(rid,slen(rid)+1)};i=i+1};sys_munmap_direct(rids as *u8,slots*__size_of(i64))}
if (buf as i64)>0{sys_munmap_direct(buf,n+1)};return 0
1325}
1326func wf_obs_board(led: *u8) -> i64 {
  let szp: *i64 = sys_mmap_try(__size_of(i64)) as *i64
  if (szp as i64)<=0{return WF_EVIDENCE_ERROR}
  let buf: *u8 = wf_read_snapshot(led, szp)
  if (buf as i64) == 0 { sys_munmap_direct(szp as *u8,8); return 0 - 1 }
  let n: i64 = szp[0]
  sys_munmap_direct(szp as *u8,8)
  let slots:i64=n/slen("WFRUN rid=")+1
  let rids:*i64=sys_mmap_try(slots*__size_of(i64)) as *i64
  if (rids as i64)<=0{wf_obs_release(buf,n,0 as *i64,0);return 0-1}
  let cnt:i64=wf_obs_runs(buf,n,rids,slots)
  if cnt < 0 { wf_obs_release(buf,n,rids,slots);return 0 - 1 }
  let tok:*u8=sys_mmap_try(n+1)
  if (tok as i64)<=0{wf_obs_release(buf,n,rids,slots);return 0-1}
  var d: i64 = 0
  var f: i64 = 0
  var w: i64 = 0
  var ru: i64 = 0
  var i: i64 = 0
  while i < cnt {
      let rid: *u8 = rids[i] as *u8
      wf_obs_tok(tok, rid)
      let st: i64 = wf_run_state(buf, n, tok)
      let oks: i64 = wf_lines_with2(buf, n, tok, "status=OK" as *u8)
      let ats: i64 = wf_lines_with2(buf, n, tok, "status=ATT" as *u8)
      p("WFOBS run=" as *u8); p(rid)
      p(" state=" as *u8); p(wf_state_name(st))
      p(" oksteps=" as *u8); pn(oks)
      p(" attempts=" as *u8); pn(ats)
      p("\n" as *u8)
      if st == 3 { d = d + 1 }
      if st == 2 { f = f + 1 }
      if st == 1 { w = w + 1 }
      if st == 0 { ru = ru + 1 }
      i = i + 1
  }
  p("WFOBS-TOTALS runs=" as *u8); pn(cnt); p(" done=" as *u8); pn(d); p(" failed=" as *u8); pn(f); p(" parked=" as *u8); pn(w); p(" running=" as *u8); pn(ru); p("\n" as *u8)
  sys_munmap_direct(tok,n+1);wf_obs_release(buf,n,rids,slots)
  return cnt
1365}
1366// 0-JS sovereign run board page; unquoted HTML5 attrs + rgb() colors (no quote/hash/bang literals)
1367func wf_obs_html(led: *u8, out: *u8) -> i64 {
  let szp: *i64 = sys_mmap_try(__size_of(i64)) as *i64
  if (szp as i64)<=0{return WF_EVIDENCE_ERROR}
  let buf: *u8 = wf_read_snapshot(led, szp)
  if (buf as i64) == 0 { sys_munmap_direct(szp as *u8,8); return 0 - 1 }
  let n: i64 = szp[0]
  sys_munmap_direct(szp as *u8,8)
  let slots:i64=n/slen("WFRUN rid=")+1
  let rids:*i64=sys_mmap_try(slots*__size_of(i64)) as *i64
  if (rids as i64)<=0{wf_obs_release(buf,n,0 as *i64,0);return 0-1}
  let cnt:i64=wf_obs_runs(buf,n,rids,slots)
  if cnt < 0 { wf_obs_release(buf,n,rids,slots);return 0 - 1 }
  var needed:i64=slen("<")+slen("DOCTYPE html><html><head><meta charset=utf-8><title>Nishi Workflow Runs</title><style>body{font-family:monospace;background:rgb(16,17,22);color:rgb(222,224,230);margin:2em}table{border-collapse:collapse}td,th{border:1px solid rgb(60,62,72);padding:6px 12px}h1{color:rgb(140,190,255)}.DONE{color:rgb(90,210,130)}.FAILED{color:rgb(245,95,95)}.PARKED{color:rgb(235,195,95)}.RUNNING{color:rgb(120,175,245)}</style></head><body><h1>Nishi Workflow Runs</h1><p>Derived by REPLAY of the event-sourced run ledger, nothing self-reported.</p><table><tr><th>run</th><th>state</th><th>ok steps</th><th>attempts</th></tr>")+slen("<tr><td>")+slen("</td><td class=")+slen(">")+slen("</td><td>")+slen("</td><td>")+slen("</td></tr>")+slen("</table><p>runs ")+slen(" &middot; done ")+slen(" &middot; failed ")+slen(" &middot; parked ")+slen(" &middot; running ")+slen("</p><p>Generated by nx_wflow_engine wf_obs_html (sovereign, 0-JS, ledger replay).</p></body></html>")+2+slen("9223372036854775807")*5
  var measure:i64=0
  while measure<cnt{let escaped:*WfOwnedText=wf_id_html(rids[measure] as *u8);if escaped==(0 as *WfOwnedText){wf_obs_release(buf,n,rids,slots);return 0-1};let extra:i64=escaped.len+slen("<tr><td></td><td class=></td><td></td><td></td></tr>")+slen("RUNNING")*2+slen("9223372036854775807")*2;if extra>WF_TEXT_I64_MAX-needed{wf_text_free(escaped);wf_obs_release(buf,n,rids,slots);return 0-1};needed=needed+extra;wf_text_free(escaped);measure=measure+1}
  let hb:*u8=sys_mmap_try(needed)
  if (hb as i64)<=0{wf_obs_release(buf,n,rids,slots);return 0-1}
  var o: i64 = 0
  o = wf_cat(hb, o, "<" as *u8)
  hb[o] = 33 as u8
  o = o + 1
  o = wf_cat(hb, o, "DOCTYPE html><html><head><meta charset=utf-8><title>Nishi Workflow Runs</title><style>body{font-family:monospace;background:rgb(16,17,22);color:rgb(222,224,230);margin:2em}table{border-collapse:collapse}td,th{border:1px solid rgb(60,62,72);padding:6px 12px}h1{color:rgb(140,190,255)}.DONE{color:rgb(90,210,130)}.FAILED{color:rgb(245,95,95)}.PARKED{color:rgb(235,195,95)}.RUNNING{color:rgb(120,175,245)}</style></head><body><h1>Nishi Workflow Runs</h1><p>Derived by REPLAY of the event-sourced run ledger, nothing self-reported.</p><table><tr><th>run</th><th>state</th><th>ok steps</th><th>attempts</th></tr>" as *u8)
  let tok:*u8=sys_mmap_try(n+1)
  if (tok as i64)<=0{sys_munmap_direct(hb,needed);wf_obs_release(buf,n,rids,slots);return 0-1}
  var d: i64 = 0
  var f: i64 = 0
  var w: i64 = 0
  var ru: i64 = 0
  var i: i64 = 0
  while i < cnt {
      let rid: *u8 = rids[i] as *u8
      wf_obs_tok(tok, rid)
      let st: i64 = wf_run_state(buf, n, tok)
      let oks: i64 = wf_lines_with2(buf, n, tok, "status=OK" as *u8)
      let ats: i64 = wf_lines_with2(buf, n, tok, "status=ATT" as *u8)
      o = wf_cat(hb, o, "<tr><td>" as *u8)
      let escaped:*WfOwnedText=wf_id_html(rid)
      if escaped==(0 as *WfOwnedText){sys_munmap_direct(tok,n+1);sys_munmap_direct(hb,needed);wf_obs_release(buf,n,rids,slots);return 0-1}
      o=wf_cat(hb,o,escaped.data);wf_text_free(escaped)
      o = wf_cat(hb, o, "</td><td class=" as *u8)
      o = wf_cat(hb, o, wf_state_name(st))
      o = wf_cat(hb, o, ">" as *u8)
      o = wf_cat(hb, o, wf_state_name(st))
      o = wf_cat(hb, o, "</td><td>" as *u8)
      o = wf_catn(hb, o, oks)
      o = wf_cat(hb, o, "</td><td>" as *u8)
      o = wf_catn(hb, o, ats)
      o = wf_cat(hb, o, "</td></tr>" as *u8)
      if st == 3 { d = d + 1 }
      if st == 2 { f = f + 1 }
      if st == 1 { w = w + 1 }
      if st == 0 { ru = ru + 1 }
      i = i + 1
  }
  o = wf_cat(hb, o, "</table><p>runs " as *u8)
  o = wf_catn(hb, o, cnt)
  o = wf_cat(hb, o, " &middot; done " as *u8)
  o = wf_catn(hb, o, d)
  o = wf_cat(hb, o, " &middot; failed " as *u8)
  o = wf_catn(hb, o, f)
  o = wf_cat(hb, o, " &middot; parked " as *u8)
  o = wf_catn(hb, o, w)
  o = wf_cat(hb, o, " &middot; running " as *u8)
  o = wf_catn(hb, o, ru)
  o = wf_cat(hb, o, "</p><p>Generated by nx_wflow_engine wf_obs_html (sovereign, 0-JS, ledger replay).</p></body></html>" as *u8)
  let write_rc:i64=wf_writeall(out,hb,o)
  sys_munmap_direct(tok,n+1);sys_munmap_direct(hb,needed);wf_obs_release(buf,n,rids,slots)
  if write_rc!=0{return 0-1};return cnt
1435}
1436
1437func wf_selftest() -> i64 {
  p("=== NX-WFLOW-ENGINE SELFTEST (R1 keystone: multi-step durable runs, retry, ledger-dedup, crash-resume) ===\n" as *u8)
  var ok: i64 = 1
1440
  // fresh unique ledger under /tmp (getpid is broken on this backend -- use epoch seconds)
  let led: *u8 = sys_mmap(128)
  var lo: i64 = 0
  lo = wf_cat(led, lo, "/tmp/wflow_st_" as *u8)
  lo = wf_catn(led, lo, wf_now())
  lo = wf_cat(led, lo, ".log" as *u8)
  led[lo] = 0 as u8
1448
  let flows: *i64 = sys_mmap(8 * 8) as *i64
  flows[0] = "f1|deal-stage|to=Negotiation" as *u8 as i64
  flows[1] = "f2|giving-received|-" as *u8 as i64
  flows[2] = "f3|ops-check|-" as *u8 as i64
  let steps: *i64 = sys_mmap(8 * 16) as *i64
  steps[0] = "f1|1|create-task|-|Prepare contract and pricing|1" as *u8 as i64
  steps[1] = "f1|2|send|card|Congrats on reaching Negotiation|2" as *u8 as i64
  steps[2] = "f1|3|notify|-|Deal advanced|1" as *u8 as i64
  steps[3] = "f2|1|probe-fail|-|3|4" as *u8 as i64
  steps[4] = "f2|2|update-field|-|status=thanked|1" as *u8 as i64
  steps[5] = "f3|1|probe-fail|-|9|2" as *u8 as i64
  steps[6] = "f3|2|notify|-|never reached|1" as *u8 as i64
1461
  let cx: *i64 = sys_mmap(32) as *i64
  cx[0] = steps as i64
  cx[1] = 7
  cx[2] = led as i64
  cx[3] = 0
1467
  // T1 whole-set validation, loud refusals
  let l1: i64 = wf_load(steps, 7)
  let l2: i64 = wf_load_flows(flows, 3)
  p("  T1 load steps=" as *u8); pn(l1); p(" flows=" as *u8); pn(l2); p("\n" as *u8)
  if l1 != 7 { ok = 0 }
  if l2 != 3 { ok = 0 }
  let bad1: *i64 = sys_mmap(16) as *i64
  bad1[0] = "f9|1|explode|-|boom|1" as *u8 as i64
  if wf_load(bad1, 1) != (0 - 1) { ok = 0 }
  let bad2: *i64 = sys_mmap(16) as *i64
  bad2[0] = "f9|1|send|fax|x|1" as *u8 as i64
  if wf_load(bad2, 1) != (0 - 1) { ok = 0 }
  let bad3: *i64 = sys_mmap(16) as *i64
  bad3[0] = "f9|1|notify|-|x|0" as *u8 as i64
  if wf_load(bad3, 1) != (0 - 1) { ok = 0 }
1483
  let ridE1:*WfOwnedText=wf_id_hex("E1","f1");let tokE1:*WfOwnedText=wf_id_tok(ridE1.data)
  let ridE2:*WfOwnedText=wf_id_hex("E2","f2");let tokE2:*WfOwnedText=wf_id_tok(ridE2.data)
  let ridE3:*WfOwnedText=wf_id_hex("E3","f3");let tokE3:*WfOwnedText=wf_id_tok(ridE3.data)
  // T2 fire -> 3-step run completes in order, DONE recorded
  let r1: i64 = wf_fire(cx, flows, 3, "deal-stage~deal=Riverside Contract~to=Negotiation~id=E1" as *u8)
  let szp: *i64 = sys_mmap(8) as *i64
  var buf: *u8 = wf_read_snapshot(led, szp)
  let okE1: i64 = wf_lines_with2(buf, szp[0], tokE1.data, "status=OK" as *u8)
  let dnE1: i64 = wf_lines_with2(buf, szp[0], tokE1.data, "status=DONE" as *u8)
  p("  T2 fire started=" as *u8); pn(r1); p(" okSteps=" as *u8); pn(okE1); p(" done=" as *u8); pn(dnE1); p("\n" as *u8)
  if r1 != 1 { ok = 0 }
  if okE1 != 3 { ok = 0 }
  if dnE1 != 1 { ok = 0 }
1497
  // T3 durable dedup: same event id never starts twice (survives across processes via the ledger)
  let r2: i64 = wf_fire(cx, flows, 3, "deal-stage~deal=Riverside Contract~to=Negotiation~id=E1" as *u8)
  buf = wf_read_snapshot(led, szp)
  let stE1: i64 = wf_lines_with2(buf, szp[0], tokE1.data, "status=START" as *u8)
  p("  T3 refire rc=" as *u8); pn(r2); p(" starts=" as *u8); pn(stE1); p("\n" as *u8)
  if r2 != (0 - 100) { ok = 0 }
  if stE1 != 1 { ok = 0 }
1505
  // T4 per-step retry: probe-fail 3 with max 4 succeeds on attempt 3
  let r3: i64 = wf_fire(cx, flows, 3, "giving-received~person=Rose~amount=500~id=E2" as *u8)
  buf = wf_read_snapshot(led, szp)
  let ok3: i64 = wf_lines_with2(buf, szp[0], tokE2.data, "status=OK att=3" as *u8)
  let atts: i64 = wf_lines_with2(buf, szp[0], tokE2.data, "status=ATT" as *u8)
  let dnE2: i64 = wf_lines_with2(buf, szp[0], tokE2.data, "status=DONE" as *u8)
  p("  T4 retry started=" as *u8); pn(r3); p(" okAtt3=" as *u8); pn(ok3); p(" attempts=" as *u8); pn(atts); p(" done=" as *u8); pn(dnE2); p("\n" as *u8)
  if r3 != 1 { ok = 0 }
  if ok3 != 1 { ok = 0 }
  if atts != 4 { ok = 0 }
  if dnE2 != 1 { ok = 0 }
1517
  // T5 exhausted retries -> FAILED at step, later steps never run
  let r4: i64 = wf_fire(cx, flows, 3, "ops-check~probe=edge~id=E3" as *u8)
  buf = wf_read_snapshot(led, szp)
  let fsE3: i64 = wf_lines_with2(buf, szp[0], tokE3.data, "status=FAILSTEP" as *u8)
  let flE3: i64 = wf_lines_with2(buf, szp[0], tokE3.data, "status=FAILED" as *u8)
  let s2E3: i64 = wf_lines_with2(buf, szp[0], tokE3.data, " step=2 " as *u8)
  let dnE3: i64 = wf_lines_with2(buf, szp[0], tokE3.data, "status=DONE" as *u8)
  p("  T5 exhaust rc=" as *u8); pn(r4); p(" failstep=" as *u8); pn(fsE3); p(" failed=" as *u8); pn(flE3); p(" step2lines=" as *u8); pn(s2E3); p(" done=" as *u8); pn(dnE3); p("\n" as *u8)
  if fsE3 != 1 { ok = 0 }
  if flE3 != 1 { ok = 0 }
  if s2E3 != 0 { ok = 0 }
  if dnE3 != 0 { ok = 0 }
1530
  // T6 THE CROWN -- durable crash-resume: synthetic run crashed after step 1; resume completes 2..3
  // WITHOUT re-executing step 1 (event-sourced replay, additive only)
  wf_emit(led, "E9.f1" as *u8, "f1" as *u8, 0, "START" as *u8, 0)
  wf_emit_var(led,"E9.f1","id","E9")
  wf_emit(led, "E9.f1" as *u8, "f1" as *u8, 1, "ATT" as *u8, 1)
  wf_emit(led, "E9.f1" as *u8, "f1" as *u8, 1, "OK" as *u8, 1)
  let rs: i64 = wf_resume(cx)
  buf = wf_read_snapshot(led, szp)
  let okE9: i64 = wf_lines_with2(buf, szp[0], " rid=E9.f1 " as *u8, "status=OK" as *u8)
  let ok1E9: i64 = wf_lines_with2(buf, szp[0], " rid=E9.f1 " as *u8, " step=1 status=OK" as *u8)
  let dnE9: i64 = wf_lines_with2(buf, szp[0], " rid=E9.f1 " as *u8, "status=DONE" as *u8)
  p("  T6 resume resumed=" as *u8); pn(rs); p(" okSteps=" as *u8); pn(okE9); p(" step1okOnce=" as *u8); pn(ok1E9); p(" done=" as *u8); pn(dnE9); p("\n" as *u8)
  if rs != 1 { ok = 0 }
  if okE9 != 3 { ok = 0 }
  if ok1E9 != 1 { ok = 0 }
  if dnE9 != 1 { ok = 0 }
1547
  // T7 liar-kill negatives: unknown-flow in-flight run refuses loudly (no fabricated DONE);
  // nonsense status never appears
  wf_emit(led, "EX.zz" as *u8, "zz" as *u8, 0, "START" as *u8, 0)
  wf_emit_var(led,"EX.zz","id","EX")
  let rz: i64 = wf_resume(cx)
  buf = wf_read_snapshot(led, szp)
  let dnZZ: i64 = wf_lines_with2(buf, szp[0], " rid=EX.zz " as *u8, "status=DONE" as *u8)
  let ban: i64 = wf_lines_with2(buf, szp[0], "WFRUN" as *u8, "status=BANANA" as *u8)
  p("  T7 negctl resumeRc=" as *u8); pn(rz); p(" zzDone=" as *u8); pn(dnZZ); p(" nonsense=" as *u8); pn(ban); p("\n" as *u8)
  if rz != (0 - 1) { ok = 0 }
  if dnZZ != 0 { ok = 0 }
  if ban != 0 { ok = 0 }
1560
  p("NX-WFLOW-ENGINE-SELFTEST ledger=" as *u8); p(led)
  p(" runs: done=3 failed=1 dedup=1 resumed=1 " as *u8)
  if ok == 1 { p("verdict=GREEN\n" as *u8); return 0 }
  p("verdict=RED\n" as *u8)
  return 1
1566}
nx_wflow_engine.nx source

`nx_wflow_engine.nx` source
