nx_wflow_execution_engine_candidate_t346.nx source
↩ module page · 1610 lines · 71428 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 {
35 let n: i64 = slen(hay)
36 let m: i64 = slen(needle)
37 if m == 0 { return 0 }
38 var i: i64 = 0
39 while i + m <= n {
40 var k: i64 = 0
41 var hit: i64 = 1
42 while k < m { if hay[i+k] != needle[k] { hit = 0; k = m } else { k = k + 1 } }
43 if hit == 1 { return 1 }
44 i = i + 1
45 }
46 return 0
47}
48func wf_evid(ev: *u8, out: *u8, cap: i64) -> i64 {
49 let n: i64 = slen(ev)
50 var i: i64 = 0
51 while i + 3 < n {
52 if ev[i] == (126 as u8) { if ev[i+1] == (105 as u8) { if ev[i+2] == (100 as u8) { if ev[i+3] == (61 as u8) {
53 var q: i64 = i + 4
54 var t: i64 = 0
55 while q < n { if ev[q] == (126 as u8) { break } if t < cap - 1 { out[t] = ev[q]; t = t + 1 } q = q + 1 }
56 out[t] = 0 as u8
57 return 1
58 } } } }
59 i = i + 1
60 }
61 out[0] = 0 as u8
62 return 0
63}
64
65// ---- small locals ----
66func 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 }
67func wf_catn(dst: *u8, off: i64, v: i64) -> i64 {
68 var x: i64 = off
69 var m: i64 = v
70 if m < 0 { dst[x] = 45 as u8; x = x + 1; m = 0 - m }
71 if m == 0 { dst[x] = 48 as u8; return x + 1 }
72 let t: *u8 = sys_mmap(24)
73 var k: i64 = 0
74 while m > 0 { t[k] = (48 + (m - (m/10)*10)) as u8; m = m / 10; k = k + 1 }
75 var j: i64 = 0
76 while j < k { dst[x] = t[k-1-j]; x = x + 1; j = j + 1 }
77 return x
78}
79func wf_atoi(s: *u8) -> i64 {
80 var v: i64 = 0
81 var i: i64 = 0
82 var any: i64 = 0
83 while s[i] != (0 as u8) {
84 let c: i64 = s[i] as i64
85 if c >= 48 { if c <= 57 { v = v*10 + (c - 48); any = 1 } }
86 if c < 48 { return 0 - 1 }
87 if c > 57 { return 0 - 1 }
88 i = i + 1
89 }
90 if any == 0 { return 0 - 1 }
91 return v
92}
93// raw __syscall(201) returns a broken constant on this backend (the getpid -25 class of quirk) --
94// use the proven clock_gettime wrapper so concurrent selftests never share a ledger path
95func wf_now() -> i64 { return sys_now_realtime_sec() }
96
97// ---- durable append-only ledger (flock + O_APPEND; one line per transition) ----
98// Required transition evidence: -4 must stop execution, never enter action retries.
99// fsync covers file contents; this does not claim directory-entry crash durability.
100const WF_EVIDENCE_ERROR: i64 = 0 - 4
101func wf_append(path: *u8, line: *u8) -> i64 {
102 let fd: i64 = __syscall(257, 0 - 100, path as i64, WF_MAGIC_1089, 420, 0, 0)
103 if fd < 0 { p("WFLOW evidence ERROR open failed\n"); return WF_EVIDENCE_ERROR }
104 var locked: i64 = sys_flock(fd, SYS_LOCK_EX)
105 while locked == (0 - 4) { locked = sys_flock(fd, SYS_LOCK_EX) }
106 var rc: i64 = 0
107 if locked != 0 { rc = WF_EVIDENCE_ERROR }
108 let n: i64 = slen(line)
109 var off: i64 = 0
110 while off < n {
111 if rc != 0 { break }
112 let w: i64 = sys_write(fd, (line as i64 + off) as *u8, n - off)
113 if w == (0 - 4) { continue }
114 if w <= 0 { rc = WF_EVIDENCE_ERROR } else { off = off + w }
115 }
116 if rc == 0 {
117 var sync: i64 = sys_fsync(fd)
118 while sync == (0 - 4) { sync = sys_fsync(fd) }
119 if sync != 0 { rc = WF_EVIDENCE_ERROR }
120 }
121 if locked == 0 { if sys_flock(fd, SYS_LOCK_UN) != 0 { rc = WF_EVIDENCE_ERROR } }
122 if sys_close(fd) != 0 { rc = WF_EVIDENCE_ERROR }
123 if rc != 0 { p("WFLOW evidence ERROR transition not confirmed; execution stops, reconcile ledger\n") }
124 return rc
125}
126const WF_LEDCAP: i64 = 1048576
127import "nx_wflow_text_owned.nx"
128func wf_readall(path: *u8, szp: *i64) -> *u8 {
129 let buf: *u8 = sys_mmap(WF_LEDCAP)
130 szp[0] = 0
131 let fd: i64 = __syscall(257, 0 - 100, path as i64, 0, 0, 0, 0)
132 if fd < 0 { if fd == (0 - 2) { return buf } return 0 as *u8 }
133 var off: i64 = 0
134 var go: i64 = 1
135 while go == 1 {
136 let r: i64 = __syscall(0, fd, (buf as i64) + off, WF_LEDCAP - off, 0, 0, 0)
137 if r == (0 - 4) { continue }
138 if r < 0 { __syscall(3, fd, 0, 0, 0, 0, 0); return 0 as *u8 }
139 if r == 0 { go = 0 }
140 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 } }
141 }
142 __syscall(3, fd, 0, 0, 0, 0, 0)
143 szp[0] = off
144 return buf
145}
146func wf_writeall(path: *u8, buf: *u8, n: i64) -> i64 {
147 let fd: i64 = __syscall(257, 0 - 100, path as i64, 577, 420, 0, 0)
148 if fd < 0 { p("WFLOW write ERROR cannot open out -- fail loud\n" as *u8); return 0 - 1 }
149 let w: i64 = sys_write(fd, buf, n)
150 __syscall(3, fd, 0, 0, 0, 0, 0)
151 if w != n { p("WFLOW write ERROR short write -- fail loud\n" as *u8); return 0 - 1 }
152 return 0
153}
154
155// ---- ledger replay queries ----
156func wf_has_rng(buf: *u8, ls: i64, le: i64, needle: *u8) -> i64 {
157 let m: i64 = slen(needle)
158 if m == 0 { return 0 }
159 var i: i64 = ls
160 while i + m <= le {
161 var k: i64 = 0
162 var hit: i64 = 1
163 while k < m { if buf[i+k] != needle[k] { hit = 0; k = m } else { k = k + 1 } }
164 if hit == 1 { return 1 }
165 i = i + 1
166 }
167 return 0
168}
169// count ledger LINES containing BOTH substrings
170func wf_lines_with2(buf: *u8, n: i64, a: *u8, b: *u8) -> i64 {
171 var c: i64 = 0
172 var ls: i64 = 0
173 while ls < n {
174 var le: i64 = ls
175 while le < n { if buf[le] == (10 as u8) { break } le = le + 1 }
176 if wf_has_rng(buf, ls, le, a) == 1 { if wf_has_rng(buf, ls, le, b) == 1 { c = c + 1 } }
177 ls = le + 1
178 }
179 return c
180}
181// parse the number after key within [ls,le); -1 if absent
182func wf_num_after(buf: *u8, ls: i64, le: i64, key: *u8) -> i64 {
183 let m: i64 = slen(key)
184 var i: i64 = ls
185 while i + m <= le {
186 var k: i64 = 0
187 var hit: i64 = 1
188 while k < m { if buf[i+k] != key[k] { hit = 0; k = m } else { k = k + 1 } }
189 if hit == 1 {
190 var q: i64 = i + m
191 var v: i64 = 0
192 var any: i64 = 0
193 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 } }
194 if any == 1 { return v }
195 return 0 - 1
196 }
197 i = i + 1
198 }
199 return 0 - 1
200}
201// copy the space-terminated value after key within [ls,le) into out; 1 if found
202func wf_val_after(buf: *u8, ls: i64, le: i64, key: *u8, out: *u8, cap: i64) -> i64 {
203 let m: i64 = slen(key)
204 var i: i64 = ls
205 while i + m <= le {
206 var k: i64 = 0
207 var hit: i64 = 1
208 while k < m { if buf[i+k] != key[k] { hit = 0; k = m } else { k = k + 1 } }
209 if hit == 1 {
210 var q: i64 = i + m
211 var t: i64 = 0
212 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 }
213 out[t] = 0 as u8
214 return 1
215 }
216 i = i + 1
217 }
218 out[0] = 0 as u8
219 return 0
220}
221// highest step idx with status=OK for this run token
222func wf_max_ok_step(buf: *u8, n: i64, ridtok: *u8) -> i64 {
223 var mx: i64 = 0
224 var ls: i64 = 0
225 while ls < n {
226 var le: i64 = ls
227 while le < n { if buf[le] == (10 as u8) { break } le = le + 1 }
228 if wf_has_rng(buf, ls, le, ridtok) == 1 { if wf_has_rng(buf, ls, le, "status=OK" as *u8) == 1 {
229 let v: i64 = wf_num_after(buf, ls, le, " step=" as *u8)
230 if v > mx { mx = v }
231 } }
232 ls = le + 1
233 }
234 return mx
235}
236
237// ---- definition validation (whole-set, loud, nothing half-loaded) ----
238func wf_action_ok(a: *u8) -> i64 {
239 if seq(a, "create-task" as *u8) == 1 { return 1 }
240 if seq(a, "send" as *u8) == 1 { return 1 }
241 if seq(a, "notify" as *u8) == 1 { return 1 }
242 if seq(a, "update-field" as *u8) == 1 { return 1 }
243 if seq(a, "probe-fail" as *u8) == 1 { return 1 }
244 if seq(a, "exec-organ" as *u8) == 1 { return 1 }
245 if seq(a, "approve" as *u8) == 1 { return 1 }
246 if seq(a, "set-var" as *u8) == 1 { return 1 }
247 if seq(a, "for-each" as *u8) == 1 { return 1 }
248 return 0
249}
250func wf_load(steps: *i64, n: i64) -> i64 {
251 let a: *u8 = sys_mmap(64)
252 let ch: *u8 = sys_mmap(64)
253 let ix: *u8 = sys_mmap(32)
254 let ma: *u8 = sys_mmap(32)
255 var i: i64 = 0
256 while i < n {
257 let r: *u8 = steps[i] as *u8
258 pipe_field(r, 1, ix, 32)
259 if wf_atoi(ix) < 1 { p("WFLOW load ERROR bad step idx -- fail loud\n" as *u8); return 0 - 1 }
260 pipe_field(r, 2, a, 64)
261 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 }
262 if seq(a, "send" as *u8) == 1 {
263 pipe_field(r, 3, ch, 64)
264 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 }
265 }
266 if seq(a, "exec-organ" as *u8) == 1 {
267 pipe_field(r, 3, ch, 64)
268 var chok: i64 = 1
269 if ch[0] == (0 as u8) { chok = 0 }
270 if ch[0] == (45 as u8) { if ch[1] == (0 as u8) { chok = 0 } }
271 if ch[0] == (62 as u8) { chok = 0 }
272 var gsep: i64 = 0
273 var gi2: i64 = 0
274 while ch[gi2] != (0 as u8) { if ch[gi2] == (62 as u8) { gsep = gi2 } gi2 = gi2 + 1 }
275 if gsep > 0 { if ch[gsep+1] == (0 as u8) { chok = 0 } }
276 if chok == 0 { p("WFLOW load ERROR exec-organ needs connector or connector>var -- fail loud\n" as *u8); return 0 - 1 }
277 }
278 if seq(a, "approve" as *u8) == 1 {
279 pipe_field(r, 3, ch, 64)
280 var apok: i64 = 1
281 if ch[0] == (0 as u8) { apok = 0 }
282 if ch[0] == (45 as u8) { if ch[1] == (0 as u8) { apok = 0 } }
283 if apok == 0 { p("WFLOW load ERROR approve needs an approver name -- fail loud\n" as *u8); return 0 - 1 }
284 }
285 if seq(a, "set-var" as *u8) == 1 {
286 let ag: *u8 = sys_mmap(512)
287 pipe_field(r, 4, ag, 512)
288 var haseq: i64 = 0
289 var gi: i64 = 0
290 while ag[gi] != (0 as u8) { if ag[gi] == (61 as u8) { haseq = 1 } gi = gi + 1 }
291 if haseq == 0 { p("WFLOW load ERROR set-var needs key=value -- fail loud\n" as *u8); return 0 - 1 }
292 }
293 if seq(a, "for-each" as *u8) == 1 {
294 pipe_field(r, 3, ch, 64)
295 var feok: i64 = 1
296 if ch[0] == (0 as u8) { feok = 0 }
297 if ch[0] == (45 as u8) { if ch[1] == (0 as u8) { feok = 0 } }
298 if ch[0] == (62 as u8) { feok = 0 }
299 let ag2: *u8 = sys_mmap(512)
300 pipe_field(r, 4, ag2, 512)
301 if ag2[0] == (0 as u8) { feok = 0 }
302 if feok == 0 { p("WFLOW load ERROR for-each needs connector and a list arg -- fail loud\n" as *u8); return 0 - 1 }
303 }
304 pipe_field(r, 5, ma, 32)
305 if wf_atoi(ma) < 1 { p("WFLOW load ERROR max-attempts under 1 -- fail loud\n" as *u8); return 0 - 1 }
306 // R9: optional 7th field = per-step condition, must be k=v or dash
307 let cnd: *u8 = sys_mmap(256)
308 pipe_field(r, 6, cnd, 256)
309 var cok: i64 = 1
310 if cnd[0] != (0 as u8) {
311 if cnd[0] == (45 as u8) { if cnd[1] == (0 as u8) { cok = 1 } else { cok = 0 } } else { cok = 0 }
312 if cok == 0 {
313 var hq: i64 = 0
314 var ci: i64 = 0
315 while cnd[ci] != (0 as u8) { if cnd[ci] == (61 as u8) { hq = 1 } ci = ci + 1 }
316 if hq == 1 { cok = 1 }
317 }
318 }
319 if cok == 0 { p("WFLOW load ERROR step condition must be k=v or dash -- fail loud\n" as *u8); return 0 - 1 }
320 // R10: optional 8th field = on-fail target step idx (forward-only, > own idx)
321 let onf: *u8 = sys_mmap(32)
322 pipe_field(r, 7, onf, 32)
323 var ofok: i64 = 1
324 if onf[0] != (0 as u8) {
325 var isdash: i64 = 0
326 if onf[0] == (45 as u8) { if onf[1] == (0 as u8) { isdash = 1 } }
327 if isdash == 0 {
328 let tj: i64 = wf_atoi(onf)
329 if tj < 1 { ofok = 0 }
330 pipe_field(r, 1, ix, 32)
331 if tj <= wf_atoi(ix) { ofok = 0 }
332 }
333 }
334 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 }
335 i = i + 1
336 }
337 return n
338}
339func wf_load_flows(flows: *i64, n: i64) -> i64 {
340 let k: *u8 = sys_mmap(128)
341 var i: i64 = 0
342 while i < n {
343 let r: *u8 = flows[i] as *u8
344 pipe_field(r, 1, k, 128)
345 if k[0] == (0 as u8) { p("WFLOW load ERROR flow without event-kind -- fail loud\n" as *u8); return 0 - 1 }
346 i = i + 1
347 }
348 return n
349}
350// does flow header match event? (kind equal + cond k=v present, or cond '-')
351func wf_match(fhdr: *u8, ev: *u8) -> i64 {
352 let fk: *u8 = sys_mmap(128)
353 let ek: *u8 = sys_mmap(128)
354 pipe_field(fhdr, 1, fk, 128)
355 var i: i64 = 0
356 var t: i64 = 0
357 var go: i64 = 1
358 while go == 1 {
359 let c: i64 = ev[i] as i64
360 if c == 0 { go = 0 }
361 if c == 126 { go = 0 }
362 if go == 1 { if t < 127 { ek[t] = ev[i]; t = t + 1 } i = i + 1 }
363 }
364 ek[t] = 0 as u8
365 if seq(fk, ek) == 0 { return 0 }
366 let cond: *u8 = sys_mmap(256)
367 pipe_field(fhdr, 2, cond, 256)
368 if cond[0] == (45 as u8) { return 1 }
369 if cond[0] == (0 as u8) { return 1 }
370 let nb: *u8 = sys_mmap(300)
371 nb[0] = 126 as u8
372 var q: i64 = 0
373 while cond[q] != (0 as u8) { nb[q+1] = cond[q]; q = q + 1 }
374 nb[q+1] = 0 as u8
375 return wf_has(ev, nb)
376}
377
378// ---- one durable transition line ----
379func wf_emit(led: *u8, rid: *u8, fid: *u8, step: i64, st: *u8, att: i64) -> i64 {
380 let ln: *u8 = sys_mmap(512)
381 var o: i64 = 0
382 o = wf_cat(ln, o, "WFRUN rid=" as *u8)
383 o = wf_cat(ln, o, rid)
384 o = wf_cat(ln, o, " flow=" as *u8)
385 o = wf_cat(ln, o, fid)
386 o = wf_cat(ln, o, " step=" as *u8)
387 o = wf_catn(ln, o, step)
388 o = wf_cat(ln, o, " status=" as *u8)
389 o = wf_cat(ln, o, st)
390 o = wf_cat(ln, o, " att=" as *u8)
391 o = wf_catn(ln, o, att)
392 ln[o] = 10 as u8
393 ln[o+1] = 0 as u8
394 return wf_append(led, ln)
395}
396
397// ---- R3 connector layer: GREEN fail-closed catalog, any blessed organ as a workflow step ----
398// catalog rows: name<TAB>elfpath<TAB>GREEN. No catalog / unknown name / flag not exactly GREEN = refuse.
399func wf_conn_resolve(cx: *i64, name: *u8, out: *u8) -> i64 {
400 let conf: *u8 = cx[3] as *u8
401 if (conf as i64) == 0 { p("WFLOW exec-organ ERROR no connector catalog loaded -- fail closed\n" as *u8); return 0 }
402 let szp: *i64 = sys_mmap(8) as *i64
403 let buf: *u8 = wf_readall(conf, szp)
404 if (buf as i64) == 0 { return 0 }
405 let n: i64 = szp[0]
406 if n == 0 { p("WFLOW exec-organ ERROR connector catalog missing or empty -- fail closed\n" as *u8); return 0 }
407 let nm: i64 = slen(name)
408 var ls: i64 = 0
409 while ls < n {
410 var le: i64 = ls
411 while le < n { if buf[le] == (10 as u8) { break } le = le + 1 }
412 var t1: i64 = ls
413 while t1 < le { if buf[t1] == (9 as u8) { break } t1 = t1 + 1 }
414 var hit: i64 = 0
415 if t1 - ls == nm {
416 hit = 1
417 var k: i64 = 0
418 while k < nm { if buf[ls+k] != name[k] { hit = 0; k = nm } else { k = k + 1 } }
419 }
420 if hit == 1 {
421 var t2: i64 = t1 + 1
422 while t2 < le { if buf[t2] == (9 as u8) { break } t2 = t2 + 1 }
423 let fl0: i64 = t2 + 1
424 var okflag: i64 = 0
425 if le - fl0 == 5 {
426 okflag = 1
427 let g: *u8 = "GREEN" as *u8
428 var k2: i64 = 0
429 while k2 < 5 { if buf[fl0+k2] != g[k2] { okflag = 0; k2 = 5 } else { k2 = k2 + 1 } }
430 }
431 if okflag == 0 { p("WFLOW exec-organ ERROR connector=" as *u8); p(name); p(" not GREEN -- fail closed\n" as *u8); return 0 }
432 var q: i64 = t1 + 1
433 var o: i64 = 0
434 while q < t2 { if o < 500 { out[o] = buf[q]; o = o + 1 } q = q + 1 }
435 out[o] = 0 as u8
436 if o == 0 { return 0 }
437 return 1
438 }
439 ls = le + 1
440 }
441 p("WFLOW exec-organ ERROR connector unknown=" as *u8); p(name); p(" -- fail closed\n" as *u8)
442 return 0
443}
444func wf_rd32(b: *u8, off: i64) -> i64 {
445 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)
446}
447// run the resolved organ as the step: argv = arg split on ~ ('-' or empty = no args); exit 0 = step OK.
448// R8b: channel `connector>varname` CAPTURES child stdout (bounded 480B, newlines to spaces, loud over-cap)
449// into the run variable plane -- step OUTPUT becomes later steps' INPUT.
450import "nx_wflow_execution_evidence_candidate_t346.nx"
451func wf_exec_organ(cx:*i64,name0:*u8,arg:*u8,rid:*u8)->i64{return wf_exec_organ_observed(cx,name0,arg,rid,0,0)}
452func wf_exec_organ_observed(cx: *i64, name0: *u8, arg: *u8, rid: *u8, step:i64, attempt:i64) -> i64 {
453 let name: *u8 = sys_mmap(128)
454 let capv: *u8 = sys_mmap(128)
455 var ni: i64 = 0
456 var ci: i64 = 0
457 var seen: i64 = 0
458 var i0: i64 = 0
459 while name0[i0] != (0 as u8) {
460 if name0[i0] == (62 as u8) { seen = 1 } else {
461 if seen == 0 { if ni < 126 { name[ni] = name0[i0]; ni = ni + 1 } } else { if ci < 126 { capv[ci] = name0[i0]; ci = ci + 1 } }
462 }
463 i0 = i0 + 1
464 }
465 name[ni] = 0 as u8
466 capv[ci] = 0 as u8
467 var docap: i64 = 0
468 if ci > 0 { docap = 1 }
469 let elf: *u8 = sys_mmap(512)
470 if wf_conn_resolve(cx, name, elf) == 0 { if wfx_record(cx,rid,name,"-",step,attempt,WFX_REFUSED,0,0,0,"connector-unresolved")!=0{return WF_EVIDENCE_ERROR};return 0 }
471 let av: *i64 = sys_mmap(8 * 16) as *i64
472 av[0] = elf as i64
473 var na: i64 = 1
474 var skip: i64 = 0
475 if arg[0] == (0 as u8) { skip = 1 }
476 if arg[0] == (45 as u8) { if arg[1] == (0 as u8) { skip = 1 } }
477 if skip == 0 {
478 let ab: *u8 = sys_mmap(WF_MAGIC_1024)
479 var i: i64 = 0
480 var over: i64 = 0
481 while arg[i] != (0 as u8) { if i < 1000 { ab[i] = arg[i] } else { over = 1 } i = i + 1 }
482 if over == 1 { p("WFLOW exec-organ ERROR argv too long -- fail closed\n" as *u8); if wfx_record(cx,rid,name,elf,step,attempt,WFX_REFUSED,0,0,0,"argv-too-long")!=0{return WF_EVIDENCE_ERROR};return 0 }
483 ab[i] = 0 as u8
484 let alen: i64 = i
485 var st0: i64 = 0
486 var j: i64 = 0
487 while j <= alen {
488 var cut: i64 = 0
489 if j == alen { cut = 1 }
490 if cut == 0 { if ab[j] == (126 as u8) { cut = 1 } }
491 if cut == 1 {
492 ab[j] = 0 as u8
493 if na >= 14 { p("WFLOW exec-organ ERROR too many args -- fail closed\n" as *u8); if wfx_record(cx,rid,name,elf,step,attempt,WFX_REFUSED,0,0,0,"too-many-args")!=0{return WF_EVIDENCE_ERROR};return 0 }
494 av[na] = (ab as i64) + st0
495 na = na + 1
496 st0 = j + 1
497 }
498 j = j + 1
499 }
500 }
501 av[na] = 0
502 let fb: *u8 = sys_mmap(16)
503 if docap == 1 {
504 if __syscall(293, fb as i64, 0, 0, 0, 0, 0) != 0 { p("WFLOW capture ERROR pipe failed -- fail closed\n" as *u8); if wfx_record(cx,rid,name,elf,step,attempt,WFX_REFUSED,0,0,0,"pipe-failed")!=0{return WF_EVIDENCE_ERROR};return 0 }
505 }
506 if wfx_record(cx,rid,name,elf,step,attempt,WFX_BEGIN,-1,0,0,"dispatch-intent")!=0{if docap==1{sys_close(wf_rd32(fb,0));sys_close(wf_rd32(fb,4))};return WF_EVIDENCE_ERROR}
507 let pid: i64 = sys_fork()
508 if pid == 0 {
509 if docap == 1 {
510 let pw: i64 = wf_rd32(fb, 4)
511 sys_dup3(pw, 1, 0)
512 __syscall(3, wf_rd32(fb, 0), 0, 0, 0, 0, 0)
513 __syscall(3, pw, 0, 0, 0, 0, 0)
514 }
515 let envp: *i64 = sys_mmap(16) as *i64
516 envp[0] = "PATH=/usr/bin:/bin" as *u8 as i64
517 envp[1] = 0
518 sys_execve(elf, av, envp)
519 sys_exit(127)
520 }
521 if pid < 0 { p("WFLOW exec-organ ERROR fork failed -- fail closed\n" as *u8); if wfx_record(cx,rid,name,elf,step,attempt,WFX_REFUSED,0,0,0,"fork-failed")!=0{return WF_EVIDENCE_ERROR};return 0 }
522 let cbuf: *u8 = sys_mmap(WF_MAGIC_1024)
523 var total: i64 = 0
524 if docap == 1 {
525 let pr: i64 = wf_rd32(fb, 0)
526 __syscall(3, wf_rd32(fb, 4), 0, 0, 0, 0, 0)
527 var go: i64 = 1
528 while go == 1 {
529 let rr: i64 = __syscall(0, pr, (cbuf as i64) + total, 1023 - total, 0, 0, 0)
530 if rr <= 0 { go = 0 }
531 if go == 1 { total = total + rr; if total >= 1023 { go = 0 } }
532 }
533 __syscall(3, pr, 0, 0, 0, 0, 0)
534 }
535 let st: *i64 = sys_mmap(16) as *i64
536 var waited: i64 = sys_wait4(pid, st, 0)
537 while waited == (0 - 4) { waited = sys_wait4(pid, st, 0) }
538 if waited != pid { if wfx_record(cx,rid,name,elf,step,attempt,WFX_WAIT_UNKNOWN,1,0,0,"child-wait-failed")!=0{return WF_EVIDENCE_ERROR};return 0 }
539 let ec: i64 = wait_status_rc(st[0])
540 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)
541 if wfx_record(cx,rid,name,elf,step,attempt,WFX_COMPLETE,1,1,ec,"child-exit")!=0{return WF_EVIDENCE_ERROR}
542 if ec != 0 { return 0 }
543 if docap == 1 {
544 if total > 480 { p("WFLOW capture ERROR output over 480 bytes -- fail loud\n" as *u8); return 0 }
545 var t: i64 = 0
546 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 }
547 while total > 0 { if cbuf[total-1] == (32 as u8) { total = total - 1 } else { break } }
548 cbuf[total] = 0 as u8
549 let ledc: *u8 = cx[2] as *u8
550 if wf_emit_var(ledc, rid, capv, cbuf) != 0 { return WF_EVIDENCE_ERROR }
551 p("WFLOW-CAPTURE run=" as *u8); p(rid); p(" var=" as *u8); p(capv); p(" v=" as *u8); p(cbuf); p("\n" as *u8)
552 }
553 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 {
558 let number:*WfOwnedText=wf_text_decimal(n)
559 if number==(0 as *WfOwnedText){return WF_EVIDENCE_ERROR}
560 let rn:i64=slen(rid);let vn:i64=slen(val)
561 var bytes:i64=slen("WFITER rid=")+slen(" n=")+slen(" item=")+2
562 if number.len>WF_TEXT_I64_MAX-bytes{wf_text_free(number);return WF_EVIDENCE_ERROR};bytes=bytes+number.len
563 if rn>WF_TEXT_I64_MAX-bytes{wf_text_free(number);return WF_EVIDENCE_ERROR};bytes=bytes+rn
564 if vn>WF_TEXT_I64_MAX-bytes{wf_text_free(number);return WF_EVIDENCE_ERROR};bytes=bytes+vn
565 let line:*u8=sys_mmap_try(bytes)
566 if (line as i64)<=0{wf_text_free(number);return WF_EVIDENCE_ERROR}
567 var at:i64=wf_cat(line,0,"WFITER rid=")
568 at=wf_cat(line,at,rid);at=wf_cat(line,at," n=");at=wf_cat(line,at,number.data)
569 at=wf_cat(line,at," item=");at=wf_cat(line,at,val);line[at]=10 as u8;line[at+1]=0
570 var rc:i64=wf_append(led,line)
571 if sys_munmap_direct(line,bytes)!=0{rc=WF_EVIDENCE_ERROR}
572 if wf_text_free(number)!=0{rc=WF_EVIDENCE_ERROR}
573 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 {
580 let led:*u8=cx[2] as *u8;let total:i64=slen(argraw)
581 var sep:i64=0;while sep<total{if argraw[sep]==(126 as u8){break};sep=sep+1}
582 let list:*WfOwnedText=wf_text_slice(argraw,0,sep)
583 if list==(0 as *WfOwnedText){return WF_EVIDENCE_ERROR}
584 var tpl:*WfOwnedText=0 as *WfOwnedText
585 if sep<total{tpl=wf_text_slice(argraw,sep+1,total-sep-1)}else{tpl=wf_text_slice("{item}",0,slen("{item}"))}
586 if tpl==(0 as *WfOwnedText){wf_text_free(list);return WF_EVIDENCE_ERROR}
587 let expanded:*WfOwnedText=wf_subst_owned(cx,rid,list.data)
588 wf_text_free(list)
589 if expanded==(0 as *WfOwnedText){wf_text_free(tpl);return 0}
590 var rc:i64=1;var start:i64=0;var cursor:i64=0;var item:i64=0
591 while cursor<=expanded.len{
592 var cut:i64=0
593 if cursor==expanded.len{cut=1}else{if expanded.data[cursor]==(44 as u8){cut=1}}
594 if cut==1{
595 expanded.data[cursor]=0
596 if cursor>start{
597 let value:*u8=((expanded.data as i64)+start) as *u8;item=item+1
598 if wf_emit_var(led,rid,"item",value)!=0{rc=WF_EVIDENCE_ERROR;break}
599 if wf_emit_iter(led,rid,item,value)!=0{rc=WF_EVIDENCE_ERROR;break}
600 let prepared:*WfOwnedText=wf_subst_owned(cx,rid,tpl.data)
601 if prepared==(0 as *WfOwnedText){rc=0;break}
602 let result:i64=wf_exec_organ(cx,chspec,prepared.data,rid)
603 if wf_text_free(prepared)!=0{rc=WF_EVIDENCE_ERROR;break}
604 if result!=1{rc=result;break}
605 }
606 start=cursor+1
607 }
608 cursor=cursor+1
609 }
610 if wf_text_free(expanded)!=0{rc=WF_EVIDENCE_ERROR}
611 if wf_text_free(tpl)!=0{rc=WF_EVIDENCE_ERROR}
612 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 {
617 p("WFLOW-ACTION action=" as *u8); p(action)
618 if seq(action, "send" as *u8) == 1 { p(" channel=" as *u8); p(ch); p(" route=" as *u8); p(sr_route(ch)) }
619 p(" arg=" as *u8); p(arg)
620 p(" att=" as *u8); pn(att)
621 p("\n" as *u8)
622 if seq(action, "probe-fail" as *u8) == 1 {
623 let k: i64 = wf_atoi(arg)
624 if att < k { return 0 }
625 return 1
626 }
627 if seq(action, "exec-organ" as *u8) == 1 { return wf_exec_organ(cx, ch, arg, rid) }
628 if seq(action, "for-each" as *u8) == 1 { return wf_for_each(cx, ch, arg, rid) }
629 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 {
637 let led: *u8 = cx[2] as *u8
638 let szp: *i64 = sys_mmap(8) as *i64
639 let buf: *u8 = wf_readall(led, szp)
640 if (buf as i64) == 0 { return 0 }
641 let tok: *u8 = sys_mmap(256)
642 var q: i64 = 0
643 q = wf_cat(tok, q, " rid=" as *u8)
644 q = wf_cat(tok, q, rid)
645 tok[q] = 32 as u8
646 tok[q+1] = 0 as u8
647 let nb: *u8 = sys_mmap(128)
648 var o: i64 = 0
649 o = wf_cat(nb, o, " step=" as *u8)
650 o = wf_catn(nb, o, idx)
651 o = wf_cat(nb, o, " decision=DENIED" as *u8)
652 nb[o] = 0 as u8
653 if wf_lines_with2(buf, szp[0], tok, nb) > 0 { return 0 - 1 }
654 var o2: i64 = 0
655 o2 = wf_cat(nb, o2, " step=" as *u8)
656 o2 = wf_catn(nb, o2, idx)
657 o2 = wf_cat(nb, o2, " decision=APPROVED" as *u8)
658 nb[o2] = 0 as u8
659 if wf_lines_with2(buf, szp[0], tok, nb) > 0 { return 1 }
660 return 0
661}
662// the operator surface: append a decision record (only APPROVED or DENIED accepted, loud else)
663func wf_decide(cx: *i64, rid: *u8, idx: i64, decision: *u8, who: *u8) -> i64 {
664 var okd: i64 = 0
665 if seq(decision, "APPROVED" as *u8) == 1 { okd = 1 }
666 if seq(decision, "DENIED" as *u8) == 1 { okd = 1 }
667 if okd == 0 { p("WFLOW decide ERROR decision must be APPROVED or DENIED -- fail loud\n" as *u8); return 0 - 1 }
668 let led: *u8 = cx[2] as *u8
669 let ln: *u8 = sys_mmap(512)
670 var o: i64 = 0
671 o = wf_cat(ln, o, "WFDEC rid=" as *u8)
672 o = wf_cat(ln, o, rid)
673 o = wf_cat(ln, o, " step=" as *u8)
674 o = wf_catn(ln, o, idx)
675 o = wf_cat(ln, o, " decision=" as *u8)
676 o = wf_cat(ln, o, decision)
677 o = wf_cat(ln, o, " by=" as *u8)
678 o = wf_cat(ln, o, who)
679 ln[o] = 10 as u8
680 ln[o+1] = 0 as u8
681 return wf_append(led, ln)
682}
683
684// ---- R8 data passing: run-scoped variables on the SAME event-sourced ledger ----
685// WFVAR rid=<rid> k=<key> v=<value-to-end-of-line>. Trigger-event fields auto-bind at fire; a set-var
686// step writes derived vars; {key} placeholders in any step arg substitute at execution; LAST write wins;
687// unknown placeholder = LOUD step failure (the send_merge law). Values must not contain ~ or newline.
688func wf_emit_var(led: *u8, rid: *u8, k: *u8, v: *u8) -> i64 {
689 // Allocate the complete serialized record, including newline and terminator.
690 var bytes: i64 = slen("WFVAR rid=" as *u8)+slen(" k=" as *u8)+slen(" v=" as *u8)+2
691 let rn: i64 = slen(rid);let kn: i64 = slen(k);let vn: i64 = slen(v)
692 if rn>9223372036854775807-bytes{return 0-1};bytes=bytes+rn
693 if kn>9223372036854775807-bytes{return 0-1};bytes=bytes+kn
694 if vn>9223372036854775807-bytes{return 0-1};bytes=bytes+vn
695 let ln: *u8 = sys_mmap_try(bytes)
696 if (ln as i64)<=0{p("WFLOW variable ERROR allocation refused; record not appended\n" as *u8);return 0-1}
697 var o: i64 = 0
698 o = wf_cat(ln, o, "WFVAR rid=" as *u8)
699 o = wf_cat(ln, o, rid)
700 o = wf_cat(ln, o, " k=" as *u8)
701 o = wf_cat(ln, o, k)
702 o = wf_cat(ln, o, " v=" as *u8)
703 o = wf_cat(ln, o, v)
704 ln[o] = 10 as u8
705 ln[o+1] = 0 as u8
706 let rc:i64=wf_append(led,ln)
707 let released:i64=sys_munmap_direct(ln,bytes)
708 if released!=0{return 0-1}
709 return rc
710}
711// copy value after key to END OF LINE (values may contain spaces)
712func wf_val_line_end(buf: *u8, ls: i64, le: i64, key: *u8, out: *u8, cap: i64) -> i64 {
713 let m: i64 = slen(key)
714 var i: i64 = ls
715 while i + m <= le {
716 var k: i64 = 0
717 var hit: i64 = 1
718 while k < m { if buf[i+k] != key[k] { hit = 0; k = m } else { k = k + 1 } }
719 if hit == 1 {
720 var q: i64 = i + m
721 var t: i64 = 0
722 while q < le { if t < cap - 1 { out[t] = buf[q]; t = t + 1 } q = q + 1 }
723 out[t] = 0 as u8
724 return 1
725 }
726 i = i + 1
727 }
728 out[0] = 0 as u8
729 return 0
730}
731func wf_obs_tok_fwd(tok: *u8, rid: *u8) -> i64 {
732 var q: i64 = 0
733 q = wf_cat(tok, q, " rid=" as *u8)
734 q = wf_cat(tok, q, rid)
735 tok[q] = 32 as u8
736 tok[q+1] = 0 as u8
737 return q
738}
739// Latest WFVAR value: 1 found, 0 missing, -1 unresolved or insufficient capacity; no partial output.
740func wf_var_get(cx: *i64, rid: *u8, key: *u8, out: *u8, cap: i64) -> i64 {
741 if cap<=0{return 0-1};out[0]=0
742 let owned:*WfOwnedText=wf_var_owned(cx,rid,key)
743 if owned==(0 as *WfOwnedText){return 0-1}
744 var rc:i64=owned.status
745 if owned.status==1{
746 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}
747 }
748 if wf_text_free(owned)!=0{out[0]=0;return 0-1}
749 return rc
750}
751// Substitute {key} from one owned ledger snapshot; -1 on unresolved input, allocation or insufficient output.
752func wf_subst(cx: *i64, rid: *u8, src: *u8, out: *u8, cap: i64) -> i64 {
753 if cap<=0{return 0-1};out[0]=0
754 let owned:*WfOwnedText=wf_subst_owned(cx,rid,src)
755 if owned==(0 as *WfOwnedText){return 0-1}
756 var rc:i64=owned.len
757 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}
758 if wf_text_free(owned)!=0{out[0]=0;return 0-1}
759 return rc
760}
761// bind every k=v field of the trigger event as an initial run variable (segment 0 = kind, skipped)
762func wf_bind_event(cx: *i64, rid: *u8, ev: *u8) -> i64 {
763 let n:i64=slen(ev)
764 if n==9223372036854775807{return WF_EVIDENCE_ERROR}
765 let bytes:i64=n+1
766 let kb:*u8=sys_mmap_try(bytes)
767 if (kb as i64)<=0{return WF_EVIDENCE_ERROR}
768 let vb:*u8=sys_mmap_try(bytes)
769 if (vb as i64)<=0{sys_munmap_direct(kb,bytes);return WF_EVIDENCE_ERROR}
770 var i:i64=0
771 var bound:i64=0
772 var result:i64=0
773 while i<n {
774 if ev[i]!=(126 as u8){i=i+1}
775 else {
776 var j:i64=i+1
777 var kn:i64=0
778 var haseq:i64=0
779 while j<n {
780 if ev[j]==(126 as u8){break}
781 if ev[j]==(61 as u8){haseq=1;j=j+1;break}
782 kb[kn]=ev[j];kn=kn+1;j=j+1
783 }
784 kb[kn]=0 as u8
785 if haseq==1 {
786 var vn:i64=0
787 while j<n {
788 if ev[j]==(126 as u8){break}
789 vb[vn]=ev[j];vn=vn+1;j=j+1
790 }
791 vb[vn]=0 as u8
792 if kn>0 {
793 if wf_emit_var(cx[2] as *u8,rid,kb,vb)!=0{result=WF_EVIDENCE_ERROR;break}
794 bound=bound+1
795 }
796 }
797 i=j
798 }
799 }
800 let kr:i64=sys_munmap_direct(kb,bytes)
801 let vr:i64=sys_munmap_direct(vb,bytes)
802 if result!=0||kr!=0||vr!=0{return WF_EVIDENCE_ERROR}
803 return bound
804}
805
806// ---- R12 versioning: @version N directive in the flows file; runs stamp WFVER; resume flags drift ----
807func wf_defs_version(path: *u8) -> i64 {
808 let szp: *i64 = sys_mmap(8) as *i64
809 let buf: *u8 = wf_readall(path, szp)
810 if (buf as i64) == 0 { return 0 }
811 let n: i64 = szp[0]
812 var ls: i64 = 0
813 while ls < n {
814 var le: i64 = ls
815 while le < n { if buf[le] == (10 as u8) { break } le = le + 1 }
816 if wf_has_rng(buf, ls, le, "@version " as *u8) == 1 {
817 let v: i64 = wf_num_after(buf, ls, le, "@version " as *u8)
818 if v > 0 { return v }
819 }
820 ls = le + 1
821 }
822 return 0
823}
824func wf_emit_ver(led: *u8, rid: *u8, ver: i64) -> i64 {
825 let ln: *u8 = sys_mmap(512)
826 var o: i64 = 0
827 o = wf_cat(ln, o, "WFVER rid=" as *u8)
828 o = wf_cat(ln, o, rid)
829 o = wf_cat(ln, o, " ver=" as *u8)
830 o = wf_catn(ln, o, ver)
831 ln[o] = 10 as u8
832 ln[o+1] = 0 as u8
833 return wf_append(led, ln)
834}
835func wf_emit_drift(led: *u8, rid: *u8, ranv: i64, nowv: i64) -> i64 {
836 let ln: *u8 = sys_mmap(512)
837 var o: i64 = 0
838 o = wf_cat(ln, o, "WFVERDRIFT rid=" as *u8)
839 o = wf_cat(ln, o, rid)
840 o = wf_cat(ln, o, " ran=" as *u8)
841 o = wf_catn(ln, o, ranv)
842 o = wf_cat(ln, o, " now=" as *u8)
843 o = wf_catn(ln, o, nowv)
844 ln[o] = 10 as u8
845 ln[o+1] = 0 as u8
846 return wf_append(led, ln)
847}
848// the version a run was STARTED under (last WFVER line for the rid; 0 = unversioned)
849func wf_run_ver(buf: *u8, n: i64, ridtok: *u8) -> i64 {
850 var v: i64 = 0
851 var ls: i64 = 0
852 while ls < n {
853 var le: i64 = ls
854 while le < n { if buf[le] == (10 as u8) { break } le = le + 1 }
855 if wf_has_rng(buf, ls, le, "WFVER " as *u8) == 1 { if wf_has_rng(buf, ls, le, ridtok) == 1 {
856 let x: i64 = wf_num_after(buf, ls, le, " ver=" as *u8)
857 if x > 0 { v = x }
858 } }
859 ls = le + 1
860 }
861 return v
862}
863
864// ---- R13 templates: gallery = <dir>/index (name|description rows) + <name>.flows/.steps pairs ----
865func wf_tpl_exists(path: *u8) -> i64 {
866 let fd: i64 = __syscall(257, 0 - 100, path as i64, 0, 0, 0, 0)
867 if fd < 0 { return 0 }
868 __syscall(3, fd, 0, 0, 0, 0, 0)
869 return 1
870}
871func wf_tpl_path(dir: *u8, name: *u8, ext: *u8, out: *u8) -> i64 {
872 var o: i64 = 0
873 o = wf_cat(out, o, dir)
874 out[o] = 47 as u8
875 o = o + 1
876 o = wf_cat(out, o, name)
877 o = wf_cat(out, o, ext)
878 out[o] = 0 as u8
879 return o
880}
881func wf_tpl_index_path(dir: *u8, out: *u8) -> i64 {
882 var o: i64 = 0
883 o = wf_cat(out, o, dir)
884 out[o] = 47 as u8
885 o = o + 1
886 o = wf_cat(out, o, "index" as *u8)
887 out[o] = 0 as u8
888 return o
889}
890// list the gallery: every index row printed with file-existence + version; returns USABLE count, -1 loud
891func wf_tpl_list(dir: *u8) -> i64 {
892 let ip: *u8 = sys_mmap(512)
893 wf_tpl_index_path(dir, ip)
894 let szp: *i64 = sys_mmap(8) as *i64
895 let buf: *u8 = wf_readall(ip, szp)
896 if (buf as i64) == 0 { return 0 - 1 }
897 let n: i64 = szp[0]
898 if n == 0 { p("WFLOW templates ERROR missing gallery index -- fail loud\n" as *u8); return 0 - 1 }
899 let nm: *u8 = sys_mmap(128)
900 let ds: *u8 = sys_mmap(512)
901 let fp2: *u8 = sys_mmap(512)
902 let sp2: *u8 = sys_mmap(512)
903 var cnt: i64 = 0
904 var ls: i64 = 0
905 while ls < n {
906 var le: i64 = ls
907 while le < n { if buf[le] == (10 as u8) { break } le = le + 1 }
908 var keep: i64 = 0
909 if le > ls { keep = 1 }
910 if keep == 1 { if buf[ls] == (35 as u8) { keep = 0 } }
911 if keep == 1 {
912 let row: *u8 = sys_mmap(le - ls + 2)
913 var k: i64 = 0
914 while ls + k < le { row[k] = buf[ls+k]; k = k + 1 }
915 row[k] = 0 as u8
916 pipe_field(row, 0, nm, 128)
917 pipe_field(row, 1, ds, 512)
918 wf_tpl_path(dir, nm, ".flows" as *u8, fp2)
919 wf_tpl_path(dir, nm, ".steps" as *u8, sp2)
920 let fe: i64 = wf_tpl_exists(fp2)
921 let se: i64 = wf_tpl_exists(sp2)
922 let v: i64 = wf_defs_version(fp2)
923 p("WFTPL name=" as *u8); p(nm)
924 p(" ver=" as *u8); pn(v)
925 p(" flows=" as *u8); pn(fe)
926 p(" steps=" as *u8); pn(se)
927 p(" desc=" as *u8); p(ds)
928 p("\n" as *u8)
929 if fe == 1 { if se == 1 { cnt = cnt + 1 } }
930 }
931 ls = le + 1
932 }
933 p("WFTPL-TOTAL usable=" as *u8); pn(cnt); p("\n" as *u8)
934 return cnt
935}
936func wf_tpl_indexed(dir: *u8, name: *u8) -> i64 {
937 let ip: *u8 = sys_mmap(512)
938 wf_tpl_index_path(dir, ip)
939 let szp: *i64 = sys_mmap(8) as *i64
940 let buf: *u8 = wf_readall(ip, szp)
941 if (buf as i64) == 0 { return 0 - 1 }
942 let n: i64 = szp[0]
943 if n == 0 { return 0 - 1 }
944 let nm: *u8 = sys_mmap(128)
945 var ls: i64 = 0
946 while ls < n {
947 var le: i64 = ls
948 while le < n { if buf[le] == (10 as u8) { break } le = le + 1 }
949 if le > ls { if buf[ls] != (35 as u8) {
950 let row: *u8 = sys_mmap(le - ls + 2)
951 var k: i64 = 0
952 while ls + k < le { row[k] = buf[ls+k]; k = k + 1 }
953 row[k] = 0 as u8
954 pipe_field(row, 0, nm, 128)
955 if seq(nm, name) == 1 { return 1 }
956 } }
957 ls = le + 1
958 }
959 return 0
960}
961// instantiate: fail-closed to INDEXED templates; byte-copy both files; whole-set validate the COPY
962// (a broken template never lands silently). Returns the template version (0 = unversioned), -1 loud.
963func wf_tpl_instantiate(dir: *u8, name: *u8, df: *u8, ds: *u8) -> i64 {
964 let ix: i64 = wf_tpl_indexed(dir, name)
965 if ix != 1 { p("WFLOW instantiate ERROR template not in gallery index -- fail closed\n" as *u8); return 0 - 1 }
966 let fp2: *u8 = sys_mmap(512)
967 let sp2: *u8 = sys_mmap(512)
968 wf_tpl_path(dir, name, ".flows" as *u8, fp2)
969 wf_tpl_path(dir, name, ".steps" as *u8, sp2)
970 let szp: *i64 = sys_mmap(8) as *i64
971 let fb: *u8 = wf_readall(fp2, szp)
972 if (fb as i64) == 0 { return 0 - 1 }
973 if szp[0] == 0 { p("WFLOW instantiate ERROR template flows file missing -- fail loud\n" as *u8); return 0 - 1 }
974 if wf_writeall(df, fb, szp[0]) != 0 { return 0 - 1 }
975 let sb: *u8 = wf_readall(sp2, szp)
976 if (sb as i64) == 0 { return 0 - 1 }
977 if szp[0] == 0 { p("WFLOW instantiate ERROR template steps file missing -- fail loud\n" as *u8); return 0 - 1 }
978 if wf_writeall(ds, sb, szp[0]) != 0 { return 0 - 1 }
979 let fl: *i64 = sys_mmap(8 * 64) as *i64
980 let st2: *i64 = sys_mmap(8 * 128) as *i64
981 let nf: i64 = wf_lines_load(df, fl, 64)
982 if nf < 1 { return 0 - 1 }
983 let nst: i64 = wf_lines_load(ds, st2, 128)
984 if nst < 1 { return 0 - 1 }
985 if wf_load_flows(fl, nf) < 0 { return 0 - 1 }
986 if wf_load(st2, nst) < 0 { return 0 - 1 }
987 let v: i64 = wf_defs_version(df)
988 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)
989 return v
990}
991
992// R9 branching: per-step condition k=v against the run variable plane. Missing variable or
993// mismatch = 0 (the step SKIPs, recorded in the ledger); match = 1. Equality only, v1.
994func wf_cond_ok(cx: *i64, rid: *u8, cond: *u8) -> i64 {
995 let n:i64=slen(cond);var split:i64=0
996 while split<n{if cond[split]==(61 as u8){break};split=split+1}
997 if split==0{return 0};if split==n{return 0}
998 let key:*WfOwnedText=wf_text_slice(cond,0,split)
999 if key==(0 as *WfOwnedText){return 0-1}
1000 let value:*WfOwnedText=wf_var_owned(cx,rid,key.data)
1001 wf_text_free(key)
1002 if value==(0 as *WfOwnedText){return 0-1}
1003 var result:i64=0
1004 if value.status==1{result=seq(value.data,((cond as i64)+split+1) as *u8)}
1005 if value.status<0{result=0-1}
1006 if wf_text_free(value)!=0{return 0-1};return result
1007}
1008
1009// ---- the durable executor: run flow fid for run rid, starting AFTER step s0 ----
1010// cx bundle: cx[0]=steps ptr, cx[1]=nsteps, cx[2]=ledger path
1011func wf_run_from(cx: *i64, rid: *u8, fid: *u8, s0: i64) -> i64 {
1012 let steps:*i64=cx[0] as *i64;var i:i64=0;var longest:i64=0
1013 while i<cx[1]{let n:i64=slen(steps[i] as *u8);if n>longest{longest=n};i=i+1}
1014 if longest==WF_TEXT_I64_MAX{return WF_EVIDENCE_ERROR}
1015 let scratch:*WfRunScratch=wf_run_scratch_new(longest+1)
1016 if scratch==(0 as *WfRunScratch){return WF_EVIDENCE_ERROR}
1017 let rc:i64=wf_run_from_owned(cx,rid,fid,s0,scratch)
1018 if wf_run_scratch_free(scratch)!=0{return WF_EVIDENCE_ERROR}
1019 return rc
1020}
1021
1022func wf_run_from_owned(cx: *i64, rid: *u8, fid: *u8, s0: i64, scratch:*WfRunScratch) -> i64 {
1023 let steps: *i64 = cx[0] as *i64
1024 let ns: i64 = cx[1]
1025 let led: *u8 = cx[2] as *u8
1026 let f2: *u8 = scratch.fields.fid
1027 let ix: *u8 = scratch.fields.index
1028 let a: *u8 = scratch.fields.action
1029 let ch: *u8 = scratch.fields.channel
1030 let arg: *u8 = scratch.fields.arg
1031 let ma: *u8 = scratch.fields.retry
1032 var i: i64 = 0
1033 var done: i64 = 0
1034 var failed: i64 = 0
1035 var parked: i64 = 0
1036 var jumpto: i64 = 0
1037 while i < ns {
1038 if wf_text_free(scratch.prepared)!=0{return WF_EVIDENCE_ERROR};scratch.prepared=0 as *WfOwnedText
1039 let r: *u8 = steps[i] as *u8
1040 pipe_field(r, 0, f2, scratch.cap)
1041 var use: i64 = 0
1042 if seq(f2, fid) == 1 { use = 1 }
1043 var idx: i64 = 0
1044 if use == 1 { pipe_field(r, 1, ix, scratch.cap); idx = wf_atoi(ix); if idx <= s0 { use = 0 } }
1045 // R10: an active on-fail jump skips the remaining normal-path steps (recorded) until the target
1046 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 } } }
1047 if use == 1 {
1048 pipe_field(r, 2, a, scratch.cap)
1049 pipe_field(r, 3, ch, scratch.cap)
1050 pipe_field(r, 4, arg, scratch.cap)
1051 pipe_field(r, 5, ma, scratch.cap)
1052 // R9: optional per-step condition (7th field, k=v) vs the variable plane -> SKIP on mismatch
1053 let cond: *u8 = scratch.fields.condition
1054 pipe_field(r, 6, cond, scratch.cap)
1055 var argx: *u8 = arg
1056 var mode: i64 = 0
1057 var docond: i64 = 0
1058 if cond[0] != (0 as u8) { docond = 1 }
1059 if cond[0] == (45 as u8) { if cond[1] == (0 as u8) { docond = 0 } }
1060 if docond == 1 { let condition:i64=wf_cond_ok(cx,rid,cond);if condition==0{mode=4};if condition<0{mode=3} }
1061 // R8: resolve {key} placeholders from the run variable plane; unknown = LOUD step failure.
1062 // for-each bypasses step-level subst ({item} binds inside the loop) -- the loop substitutes
1063 // the list once and the per-item template each iteration.
1064 if mode == 0 {
1065 var rawarg: i64 = 0
1066 if seq(a, "for-each" as *u8) == 1 { rawarg = 1 }
1067 if rawarg==0 {
1068 scratch.prepared=wf_subst_owned(cx,rid,arg)
1069 if scratch.prepared==(0 as *WfOwnedText){mode=3}else{argx=scratch.prepared.data}
1070 }
1071 }
1072 if mode == 4 {
1073 p("WFLOW-SKIP run=" as *u8); p(rid); p(" step=" as *u8); pn(idx); p(" cond=" as *u8); p(cond); p("\n" as *u8)
1074 if wf_emit(led, rid, fid, idx, "SKIP" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
1075 }
1076 if mode == 0 { if seq(a, "set-var" as *u8) == 1 { mode = 2 } }
1077 if mode == 0 { if seq(a, "approve" as *u8) == 1 { mode = 1 } }
1078 if mode == 3 {
1079 if wf_emit(led, rid, fid, idx, "FAILSTEP" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
1080 if wf_emit(led, rid, fid, idx, "FAILED" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
1081 failed = 1
1082 i = ns
1083 }
1084 if mode == 2 {
1085 var e:i64=0
1086 while argx[e]!=(0 as u8){if argx[e]==(61 as u8){break};e=e+1}
1087 if argx[e]==(61 as u8){if e>0{
1088 argx[e]=0
1089 let kb:*u8=argx
1090 let vb:*u8=((argx as i64)+e+1) as *u8
1091 if wf_emit_var(led, rid, kb, vb) != 0 { return WF_EVIDENCE_ERROR }
1092 p("WFLOW-SETVAR run=" as *u8); p(rid); p(" k=" as *u8); p(kb); p(" v=" as *u8); p(vb); p("\n" as *u8)
1093 if wf_emit(led, rid, fid, idx, "OK" as *u8, 1) != 0 { return WF_EVIDENCE_ERROR }
1094 done = done + 1
1095 } else { mode = 3 } } else { mode = 3 }
1096 if mode == 3 {
1097 p("WFLOW set-var ERROR needs key=value -- fail loud\n" as *u8)
1098 if wf_emit(led, rid, fid, idx, "FAILSTEP" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
1099 if wf_emit(led, rid, fid, idx, "FAILED" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
1100 failed = 1
1101 i = ns
1102 }
1103 }
1104 if mode == 1 {
1105 let dec: i64 = wf_decision(cx, rid, idx)
1106 if dec == 1 {
1107 p("WFLOW-APPROVAL granted run=" as *u8); p(rid); p(" step=" as *u8); pn(idx); p("\n" as *u8)
1108 if wf_emit(led, rid, fid, idx, "OK" as *u8, 1) != 0 { return WF_EVIDENCE_ERROR }
1109 done = done + 1
1110 }
1111 if dec == (0 - 1) {
1112 p("WFLOW-APPROVAL DENIED run=" as *u8); p(rid); p(" step=" as *u8); pn(idx); p(" -- run fails (deny wins)\n" as *u8)
1113 if wf_emit(led, rid, fid, idx, "FAILSTEP" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
1114 if wf_emit(led, rid, fid, idx, "FAILED" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
1115 failed = 1
1116 i = ns
1117 }
1118 if dec == 0 {
1119 p("WFLOW-APPROVAL pending run=" as *u8); p(rid); p(" step=" as *u8); pn(idx)
1120 p(" approver=" as *u8); p(ch); p(" what=" as *u8); p(argx); p(" -- run PARKED\n" as *u8)
1121 if wf_emit(led, rid, fid, idx, "WAIT" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
1122 parked = 1
1123 i = ns
1124 }
1125 }
1126 if mode == 0 {
1127 let maxa: i64 = wf_atoi(ma)
1128 var att: i64 = 1
1129 var okd: i64 = 0
1130 while att <= maxa {
1131 if wf_emit(led, rid, fid, idx, "ATT" as *u8, att) != 0 { return WF_EVIDENCE_ERROR }
1132 let rc: i64 = wf_exec(cx, a, ch, argx, att, rid)
1133 if rc == WF_EVIDENCE_ERROR { return rc }
1134 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 }
1135 }
1136 if okd == 0 {
1137 if wf_emit(led, rid, fid, idx, "FAILSTEP" as *u8, maxa) != 0 { return WF_EVIDENCE_ERROR }
1138 // R10: on-fail routing (8th field) -- exhausted ACTION failures only; the failure stays
1139 // recorded (FAILSTEP + ONFAIL), wf_error vars land on the plane, execution jumps forward
1140 let onf: *u8 = scratch.fields.onfail
1141 pipe_field(r, 7, onf, scratch.cap)
1142 var tj: i64 = 0
1143 var haveof: i64 = 0
1144 if onf[0] != (0 as u8) { haveof = 1 }
1145 if onf[0] == (45 as u8) { if onf[1] == (0 as u8) { haveof = 0 } }
1146 if haveof == 1 { tj = wf_atoi(onf) }
1147 if tj > 0 {
1148 if wf_emit(led, rid, fid, idx, "ONFAIL" as *u8, tj) != 0 { return WF_EVIDENCE_ERROR }
1149 if wf_emit_var(led, rid, "wf_error" as *u8, "yes" as *u8) != 0 { return WF_EVIDENCE_ERROR }
1150 let sv: *u8 = sys_mmap(32)
1151 var so: i64 = 0
1152 so = wf_cat(sv, so, "step" as *u8)
1153 so = wf_catn(sv, so, idx)
1154 sv[so] = 0 as u8
1155 if wf_emit_var(led, rid, "wf_error_step" as *u8, sv) != 0 { return WF_EVIDENCE_ERROR }
1156 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)
1157 jumpto = tj
1158 } else {
1159 if wf_emit(led, rid, fid, idx, "FAILED" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
1160 failed = 1
1161 i = ns
1162 }
1163 } else { done = done + 1 }
1164 }
1165 }
1166 i = i + 1
1167 }
1168 if failed == 1 { return 0 - 2 }
1169 if parked == 1 { return 0 - 3 }
1170 if wf_emit(led, rid, fid, 0, "DONE" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
1171 return done
1172}
1173
1174// ---- fire an event: durable ledger-deduped, one run per matched flow ----
1175func wf_fire(cx: *i64, flows: *i64, nf: i64, ev: *u8) -> i64 {
1176 let evid: *u8 = sys_mmap(128)
1177 if wf_evid(ev, evid, 128) == 0 { p("WFLOW fire ERROR event has no id -- fail loud\n" as *u8); return 0 - 1 }
1178 let led: *u8 = cx[2] as *u8
1179 let szp: *i64 = sys_mmap(8) as *i64
1180 let buf: *u8 = wf_readall(led, szp)
1181 if (buf as i64) == 0 { return 0 - 1 }
1182 let fid: *u8 = sys_mmap(64)
1183 let rid: *u8 = sys_mmap(200)
1184 let tok: *u8 = sys_mmap(256)
1185 var started: i64 = 0
1186 var matched: i64 = 0
1187 var f: i64 = 0
1188 while f < nf {
1189 let fh: *u8 = flows[f] as *u8
1190 if wf_match(fh, ev) == 1 {
1191 matched = matched + 1
1192 pipe_field(fh, 0, fid, 64)
1193 var o: i64 = 0
1194 o = wf_cat(rid, o, evid)
1195 rid[o] = 46 as u8
1196 o = o + 1
1197 o = wf_cat(rid, o, fid)
1198 rid[o] = 0 as u8
1199 var q: i64 = 0
1200 q = wf_cat(tok, q, " rid=" as *u8)
1201 q = wf_cat(tok, q, rid)
1202 tok[q] = 32 as u8
1203 tok[q+1] = 0 as u8
1204 if wf_lines_with2(buf, szp[0], tok, "status=START" as *u8) > 0 {
1205 p("WFLOW dedup: run " as *u8); p(rid); p(" already started (durable) -- skipped\n" as *u8)
1206 } else {
1207 if wf_emit(led, rid, fid, 0, "START" as *u8, 0) != 0 { return WF_EVIDENCE_ERROR }
1208 if cx[4] > 0 { if wf_emit_ver(led, rid, cx[4]) != 0 { return WF_EVIDENCE_ERROR } }
1209 if wf_bind_event(cx, rid, ev) < 0 { return WF_EVIDENCE_ERROR }
1210 let run_rc: i64 = wf_run_from(cx, rid, fid, 0)
1211 if run_rc == WF_EVIDENCE_ERROR { return run_rc }
1212 started = started + 1
1213 }
1214 }
1215 f = f + 1
1216 }
1217 if matched > 0 { if started == 0 { return 0 - 100 } }
1218 return started
1219}
1220
1221// any step row for this flow?
1222func wf_flow_known(cx: *i64, fid: *u8) -> i64 {
1223 let steps: *i64 = cx[0] as *i64
1224 let ns: i64 = cx[1]
1225 let f2: *u8 = sys_mmap(64)
1226 var i: i64 = 0
1227 while i < ns {
1228 pipe_field(steps[i] as *u8, 0, f2, 64)
1229 if seq(f2, fid) == 1 { return 1 }
1230 i = i + 1
1231 }
1232 return 0
1233}
1234
1235// ---- crash recovery: complete every in-flight run from its first non-OK step ----
1236func wf_resume(cx: *i64) -> i64 {
1237 let led: *u8 = cx[2] as *u8
1238 let szp: *i64 = sys_mmap(8) as *i64
1239 let buf: *u8 = wf_readall(led, szp)
1240 if (buf as i64) == 0 { return 0 - 1 }
1241 let n: i64 = szp[0]
1242 let rid: *u8 = sys_mmap(200)
1243 let fid: *u8 = sys_mmap(64)
1244 let tok: *u8 = sys_mmap(256)
1245 var resumed: i64 = 0
1246 var ls: i64 = 0
1247 while ls < n {
1248 var le: i64 = ls
1249 while le < n { if buf[le] == (10 as u8) { break } le = le + 1 }
1250 if wf_has_rng(buf, ls, le, "status=START" as *u8) == 1 {
1251 wf_val_after(buf, ls, le, " rid=" as *u8, rid, 200)
1252 wf_val_after(buf, ls, le, " flow=" as *u8, fid, 64)
1253 var q: i64 = 0
1254 q = wf_cat(tok, q, " rid=" as *u8)
1255 q = wf_cat(tok, q, rid)
1256 tok[q] = 32 as u8
1257 tok[q+1] = 0 as u8
1258 let dn: i64 = wf_lines_with2(buf, n, tok, "status=DONE" as *u8)
1259 let fl: i64 = wf_lines_with2(buf, n, tok, "status=FAILED" as *u8)
1260 if dn == 0 { if fl == 0 {
1261 if wf_flow_known(cx, fid) == 0 { p("WFLOW resume ERROR unknown flow=" as *u8); p(fid); p(" -- fail loud, nothing fabricated\n" as *u8); return 0 - 1 }
1262 // R12: flag version drift (run started under a different definitions version) -- resume
1263 // proceeds (additive semantics) but the drift is RECORDED and printed
1264 let curv: i64 = cx[4]
1265 if curv > 0 {
1266 let ranv: i64 = wf_run_ver(buf, n, tok)
1267 if ranv > 0 { if ranv != curv {
1268 if wf_emit_drift(led, rid, ranv, curv) != 0 { return WF_EVIDENCE_ERROR }
1269 p("WFLOW-RESUME-WARN version drift run=" as *u8); p(rid); p(" ran=" as *u8); pn(ranv); p(" now=" as *u8); pn(curv); p("\n" as *u8)
1270 } }
1271 }
1272 let s0: i64 = wf_max_ok_step(buf, n, tok)
1273 p("WFLOW-RESUME run=" as *u8); p(rid); p(" from-step=" as *u8); pn(s0 + 1); p("\n" as *u8)
1274 let run_rc: i64 = wf_run_from(cx, rid, fid, s0)
1275 if run_rc == WF_EVIDENCE_ERROR { return run_rc }
1276 resumed = resumed + 1
1277 } }
1278 }
1279 ls = le + 1
1280 }
1281 return resumed
1282}
1283
1284// ---- definitions as DATA FILES: load rows, validate whole, fire/resume (the production entry path) ----
1285// skips comment (leading 35) and empty lines; -1 loud on missing/empty file or row-cap
1286func wf_lines_load(path: *u8, arr: *i64, cap: i64) -> i64 {
1287 let szp: *i64 = sys_mmap(8) as *i64
1288 let buf: *u8 = wf_readall(path, szp)
1289 if (buf as i64) == 0 { return 0 - 1 }
1290 let n: i64 = szp[0]
1291 if n == 0 { p("WFLOW files ERROR missing or empty definition file -- fail loud\n" as *u8); return 0 - 1 }
1292 var cnt: i64 = 0
1293 var ls: i64 = 0
1294 while ls < n {
1295 var le: i64 = ls
1296 while le < n { if buf[le] == (10 as u8) { break } le = le + 1 }
1297 var keep: i64 = 0
1298 if le > ls { keep = 1 }
1299 if keep == 1 { if buf[ls] == (35 as u8) { keep = 0 } }
1300 if keep == 1 { if buf[ls] == (64 as u8) { keep = 0 } }
1301 if keep == 1 {
1302 if cnt >= cap { p("WFLOW files ERROR too many definition rows -- fail loud\n" as *u8); return 0 - 1 }
1303 let s: *u8 = sys_mmap(le - ls + 2)
1304 var k: i64 = 0
1305 while ls + k < le { s[k] = buf[ls+k]; k = k + 1 }
1306 s[k] = 0 as u8
1307 arr[cnt] = s as i64
1308 cnt = cnt + 1
1309 }
1310 ls = le + 1
1311 }
1312 return cnt
1313}
1314func wf_cx_files(flowsp: *u8, stepsp: *u8, led: *u8, cat: *u8, cx: *i64, flows: *i64) -> i64 {
1315 let steps: *i64 = sys_mmap(8 * 128) as *i64
1316 let nf: i64 = wf_lines_load(flowsp, flows, 64)
1317 if nf < 1 { return 0 - 1 }
1318 let nst: i64 = wf_lines_load(stepsp, steps, 128)
1319 if nst < 1 { return 0 - 1 }
1320 if wf_load_flows(flows, nf) < 0 { return 0 - 1 }
1321 if wf_load(steps, nst) < 0 { return 0 - 1 }
1322 cx[0] = steps as i64
1323 cx[1] = nst
1324 cx[2] = led as i64
1325 cx[3] = 0
1326 var usecat: i64 = 1
1327 if cat[0] == (45 as u8) { if cat[1] == (0 as u8) { usecat = 0 } }
1328 if usecat == 1 { cx[3] = cat as i64 }
1329 cx[4] = wf_defs_version(flowsp)
1330 return nf
1331}
1332func wf_fire_files(flowsp: *u8, stepsp: *u8, led: *u8, cat: *u8, ev: *u8) -> i64 {
1333 let cx: *i64 = sys_mmap(40) as *i64
1334 let flows: *i64 = sys_mmap(8 * 64) as *i64
1335 let nf: i64 = wf_cx_files(flowsp, stepsp, led, cat, cx, flows)
1336 if nf < 0 { return 0 - 1 }
1337 return wf_fire(cx, flows, nf, ev)
1338}
1339func wf_resume_files(flowsp: *u8, stepsp: *u8, led: *u8, cat: *u8) -> i64 {
1340 let cx: *i64 = sys_mmap(40) as *i64
1341 let flows: *i64 = sys_mmap(8 * 64) as *i64
1342 let nf: i64 = wf_cx_files(flowsp, stepsp, led, cat, cx, flows)
1343 if nf < 0 { return 0 - 1 }
1344 return wf_resume(cx)
1345}
1346
1347// ---- R6 run observability: board + 0-JS HTML derived ENTIRELY by ledger replay (nothing self-reported) ----
1348func wf_state_name(st: i64) -> *u8 {
1349 if st == 3 { return "DONE" as *u8 }
1350 if st == 2 { return "FAILED" as *u8 }
1351 if st == 1 { return "PARKED" as *u8 }
1352 return "RUNNING" as *u8
1353}
1354// derived state: DONE > FAILED > PARKED(waiting approval) > RUNNING(in-flight)
1355func wf_run_state(buf: *u8, n: i64, tok: *u8) -> i64 {
1356 if wf_lines_with2(buf, n, tok, "status=DONE" as *u8) > 0 { return 3 }
1357 if wf_lines_with2(buf, n, tok, "status=FAILED" as *u8) > 0 { return 2 }
1358 if wf_lines_with2(buf, n, tok, "status=WAIT" as *u8) > 0 { return 1 }
1359 return 0
1360}
1361// collect distinct run ids from START lines; returns count, -1 loud over cap
1362func wf_obs_runs(buf: *u8, n: i64, rids: *i64, cap: i64) -> i64 {
1363 var cnt: i64 = 0
1364 var ls: i64 = 0
1365 while ls < n {
1366 var le: i64 = ls
1367 while le < n { if buf[le] == (10 as u8) { break } le = le + 1 }
1368 if wf_has_rng(buf, ls, le, "status=START" as *u8) == 1 {
1369 let r: *u8 = sys_mmap(200)
1370 wf_val_after(buf, ls, le, " rid=" as *u8, r, 200)
1371 var dup: i64 = 0
1372 var i: i64 = 0
1373 while i < cnt { if seq(rids[i] as *u8, r) == 1 { dup = 1; i = cnt } else { i = i + 1 } }
1374 if dup == 0 {
1375 if cnt >= cap { p("WFLOW obs ERROR run cap exceeded -- fail loud\n" as *u8); return 0 - 1 }
1376 rids[cnt] = r as i64
1377 cnt = cnt + 1
1378 }
1379 }
1380 ls = le + 1
1381 }
1382 return cnt
1383}
1384func wf_obs_tok(tok: *u8, rid: *u8) -> i64 {
1385 var q: i64 = 0
1386 q = wf_cat(tok, q, " rid=" as *u8)
1387 q = wf_cat(tok, q, rid)
1388 tok[q] = 32 as u8
1389 tok[q+1] = 0 as u8
1390 return q
1391}
1392func wf_obs_board(led: *u8) -> i64 {
1393 let szp: *i64 = sys_mmap(8) as *i64
1394 let buf: *u8 = wf_readall(led, szp)
1395 if (buf as i64) == 0 { return 0 - 1 }
1396 let n: i64 = szp[0]
1397 let rids: *i64 = sys_mmap(8 * 128) as *i64
1398 let cnt: i64 = wf_obs_runs(buf, n, rids, 128)
1399 if cnt < 0 { return 0 - 1 }
1400 let tok: *u8 = sys_mmap(256)
1401 var d: i64 = 0
1402 var f: i64 = 0
1403 var w: i64 = 0
1404 var ru: i64 = 0
1405 var i: i64 = 0
1406 while i < cnt {
1407 let rid: *u8 = rids[i] as *u8
1408 wf_obs_tok(tok, rid)
1409 let st: i64 = wf_run_state(buf, n, tok)
1410 let oks: i64 = wf_lines_with2(buf, n, tok, "status=OK" as *u8)
1411 let ats: i64 = wf_lines_with2(buf, n, tok, "status=ATT" as *u8)
1412 p("WFOBS run=" as *u8); p(rid)
1413 p(" state=" as *u8); p(wf_state_name(st))
1414 p(" oksteps=" as *u8); pn(oks)
1415 p(" attempts=" as *u8); pn(ats)
1416 p("\n" as *u8)
1417 if st == 3 { d = d + 1 }
1418 if st == 2 { f = f + 1 }
1419 if st == 1 { w = w + 1 }
1420 if st == 0 { ru = ru + 1 }
1421 i = i + 1
1422 }
1423 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)
1424 return cnt
1425}
1426// 0-JS sovereign run board page; unquoted HTML5 attrs + rgb() colors (no quote/hash/bang literals)
1427func wf_obs_html(led: *u8, out: *u8) -> i64 {
1428 let szp: *i64 = sys_mmap(8) as *i64
1429 let buf: *u8 = wf_readall(led, szp)
1430 if (buf as i64) == 0 { return 0 - 1 }
1431 let n: i64 = szp[0]
1432 let rids: *i64 = sys_mmap(8 * 128) as *i64
1433 let cnt: i64 = wf_obs_runs(buf, n, rids, 128)
1434 if cnt < 0 { return 0 - 1 }
1435 let hb: *u8 = sys_mmap(WF_MAGIC_65536)
1436 var o: i64 = 0
1437 o = wf_cat(hb, o, "<" as *u8)
1438 hb[o] = 33 as u8
1439 o = o + 1
1440 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)
1441 let tok: *u8 = sys_mmap(256)
1442 var d: i64 = 0
1443 var f: i64 = 0
1444 var w: i64 = 0
1445 var ru: i64 = 0
1446 var i: i64 = 0
1447 while i < cnt {
1448 if o > WF_MAGIC_60000 { p("WFLOW obs ERROR page over cap -- fail loud\n" as *u8); return 0 - 1 }
1449 let rid: *u8 = rids[i] as *u8
1450 wf_obs_tok(tok, rid)
1451 let st: i64 = wf_run_state(buf, n, tok)
1452 let oks: i64 = wf_lines_with2(buf, n, tok, "status=OK" as *u8)
1453 let ats: i64 = wf_lines_with2(buf, n, tok, "status=ATT" as *u8)
1454 o = wf_cat(hb, o, "<tr><td>" as *u8)
1455 o = wf_cat(hb, o, rid)
1456 o = wf_cat(hb, o, "</td><td class=" as *u8)
1457 o = wf_cat(hb, o, wf_state_name(st))
1458 o = wf_cat(hb, o, ">" as *u8)
1459 o = wf_cat(hb, o, wf_state_name(st))
1460 o = wf_cat(hb, o, "</td><td>" as *u8)
1461 o = wf_catn(hb, o, oks)
1462 o = wf_cat(hb, o, "</td><td>" as *u8)
1463 o = wf_catn(hb, o, ats)
1464 o = wf_cat(hb, o, "</td></tr>" as *u8)
1465 if st == 3 { d = d + 1 }
1466 if st == 2 { f = f + 1 }
1467 if st == 1 { w = w + 1 }
1468 if st == 0 { ru = ru + 1 }
1469 i = i + 1
1470 }
1471 o = wf_cat(hb, o, "</table><p>runs " as *u8)
1472 o = wf_catn(hb, o, cnt)
1473 o = wf_cat(hb, o, " · done " as *u8)
1474 o = wf_catn(hb, o, d)
1475 o = wf_cat(hb, o, " · failed " as *u8)
1476 o = wf_catn(hb, o, f)
1477 o = wf_cat(hb, o, " · parked " as *u8)
1478 o = wf_catn(hb, o, w)
1479 o = wf_cat(hb, o, " · running " as *u8)
1480 o = wf_catn(hb, o, ru)
1481 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)
1482 if wf_writeall(out, hb, o) != 0 { return 0 - 1 }
1483 return cnt
1484}
1485
1486func wf_selftest() -> i64 {
1487 p("=== NX-WFLOW-ENGINE SELFTEST (R1 keystone: multi-step durable runs, retry, ledger-dedup, crash-resume) ===\n" as *u8)
1488 var ok: i64 = 1
1489
1490 // fresh unique ledger under /tmp (getpid is broken on this backend -- use epoch seconds)
1491 let led: *u8 = sys_mmap(128)
1492 var lo: i64 = 0
1493 lo = wf_cat(led, lo, "/tmp/wflow_st_" as *u8)
1494 lo = wf_catn(led, lo, wf_now())
1495 lo = wf_cat(led, lo, ".log" as *u8)
1496 led[lo] = 0 as u8
1497
1498 let flows: *i64 = sys_mmap(8 * 8) as *i64
1499 flows[0] = "f1|deal-stage|to=Negotiation" as *u8 as i64
1500 flows[1] = "f2|giving-received|-" as *u8 as i64
1501 flows[2] = "f3|ops-check|-" as *u8 as i64
1502 let steps: *i64 = sys_mmap(8 * 16) as *i64
1503 steps[0] = "f1|1|create-task|-|Prepare contract and pricing|1" as *u8 as i64
1504 steps[1] = "f1|2|send|card|Congrats on reaching Negotiation|2" as *u8 as i64
1505 steps[2] = "f1|3|notify|-|Deal advanced|1" as *u8 as i64
1506 steps[3] = "f2|1|probe-fail|-|3|4" as *u8 as i64
1507 steps[4] = "f2|2|update-field|-|status=thanked|1" as *u8 as i64
1508 steps[5] = "f3|1|probe-fail|-|9|2" as *u8 as i64
1509 steps[6] = "f3|2|notify|-|never reached|1" as *u8 as i64
1510
1511 let cx: *i64 = sys_mmap(32) as *i64
1512 cx[0] = steps as i64
1513 cx[1] = 7
1514 cx[2] = led as i64
1515 cx[3] = 0
1516
1517 // T1 whole-set validation, loud refusals
1518 let l1: i64 = wf_load(steps, 7)
1519 let l2: i64 = wf_load_flows(flows, 3)
1520 p(" T1 load steps=" as *u8); pn(l1); p(" flows=" as *u8); pn(l2); p("\n" as *u8)
1521 if l1 != 7 { ok = 0 }
1522 if l2 != 3 { ok = 0 }
1523 let bad1: *i64 = sys_mmap(16) as *i64
1524 bad1[0] = "f9|1|explode|-|boom|1" as *u8 as i64
1525 if wf_load(bad1, 1) != (0 - 1) { ok = 0 }
1526 let bad2: *i64 = sys_mmap(16) as *i64
1527 bad2[0] = "f9|1|send|fax|x|1" as *u8 as i64
1528 if wf_load(bad2, 1) != (0 - 1) { ok = 0 }
1529 let bad3: *i64 = sys_mmap(16) as *i64
1530 bad3[0] = "f9|1|notify|-|x|0" as *u8 as i64
1531 if wf_load(bad3, 1) != (0 - 1) { ok = 0 }
1532
1533 // T2 fire -> 3-step run completes in order, DONE recorded
1534 let r1: i64 = wf_fire(cx, flows, 3, "deal-stage~deal=Riverside Contract~to=Negotiation~id=E1" as *u8)
1535 let szp: *i64 = sys_mmap(8) as *i64
1536 var buf: *u8 = wf_readall(led, szp)
1537 let okE1: i64 = wf_lines_with2(buf, szp[0], " rid=E1.f1 " as *u8, "status=OK" as *u8)
1538 let dnE1: i64 = wf_lines_with2(buf, szp[0], " rid=E1.f1 " as *u8, "status=DONE" as *u8)
1539 p(" T2 fire started=" as *u8); pn(r1); p(" okSteps=" as *u8); pn(okE1); p(" done=" as *u8); pn(dnE1); p("\n" as *u8)
1540 if r1 != 1 { ok = 0 }
1541 if okE1 != 3 { ok = 0 }
1542 if dnE1 != 1 { ok = 0 }
1543
1544 // T3 durable dedup: same event id never starts twice (survives across processes via the ledger)
1545 let r2: i64 = wf_fire(cx, flows, 3, "deal-stage~deal=Riverside Contract~to=Negotiation~id=E1" as *u8)
1546 buf = wf_readall(led, szp)
1547 let stE1: i64 = wf_lines_with2(buf, szp[0], " rid=E1.f1 " as *u8, "status=START" as *u8)
1548 p(" T3 refire rc=" as *u8); pn(r2); p(" starts=" as *u8); pn(stE1); p("\n" as *u8)
1549 if r2 != (0 - 100) { ok = 0 }
1550 if stE1 != 1 { ok = 0 }
1551
1552 // T4 per-step retry: probe-fail 3 with max 4 succeeds on attempt 3
1553 let r3: i64 = wf_fire(cx, flows, 3, "giving-received~person=Rose~amount=500~id=E2" as *u8)
1554 buf = wf_readall(led, szp)
1555 let ok3: i64 = wf_lines_with2(buf, szp[0], " rid=E2.f2 " as *u8, "status=OK att=3" as *u8)
1556 let atts: i64 = wf_lines_with2(buf, szp[0], " rid=E2.f2 " as *u8, "status=ATT" as *u8)
1557 let dnE2: i64 = wf_lines_with2(buf, szp[0], " rid=E2.f2 " as *u8, "status=DONE" as *u8)
1558 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)
1559 if r3 != 1 { ok = 0 }
1560 if ok3 != 1 { ok = 0 }
1561 if atts != 4 { ok = 0 }
1562 if dnE2 != 1 { ok = 0 }
1563
1564 // T5 exhausted retries -> FAILED at step, later steps never run
1565 let r4: i64 = wf_fire(cx, flows, 3, "ops-check~probe=edge~id=E3" as *u8)
1566 buf = wf_readall(led, szp)
1567 let fsE3: i64 = wf_lines_with2(buf, szp[0], " rid=E3.f3 " as *u8, "status=FAILSTEP" as *u8)
1568 let flE3: i64 = wf_lines_with2(buf, szp[0], " rid=E3.f3 " as *u8, "status=FAILED" as *u8)
1569 let s2E3: i64 = wf_lines_with2(buf, szp[0], " rid=E3.f3 " as *u8, " step=2 " as *u8)
1570 let dnE3: i64 = wf_lines_with2(buf, szp[0], " rid=E3.f3 " as *u8, "status=DONE" as *u8)
1571 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)
1572 if fsE3 != 1 { ok = 0 }
1573 if flE3 != 1 { ok = 0 }
1574 if s2E3 != 0 { ok = 0 }
1575 if dnE3 != 0 { ok = 0 }
1576
1577 // T6 THE CROWN -- durable crash-resume: synthetic run crashed after step 1; resume completes 2..3
1578 // WITHOUT re-executing step 1 (event-sourced replay, additive only)
1579 wf_emit(led, "E9.f1" as *u8, "f1" as *u8, 0, "START" as *u8, 0)
1580 wf_emit(led, "E9.f1" as *u8, "f1" as *u8, 1, "ATT" as *u8, 1)
1581 wf_emit(led, "E9.f1" as *u8, "f1" as *u8, 1, "OK" as *u8, 1)
1582 let rs: i64 = wf_resume(cx)
1583 buf = wf_readall(led, szp)
1584 let okE9: i64 = wf_lines_with2(buf, szp[0], " rid=E9.f1 " as *u8, "status=OK" as *u8)
1585 let ok1E9: i64 = wf_lines_with2(buf, szp[0], " rid=E9.f1 " as *u8, " step=1 status=OK" as *u8)
1586 let dnE9: i64 = wf_lines_with2(buf, szp[0], " rid=E9.f1 " as *u8, "status=DONE" as *u8)
1587 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)
1588 if rs != 1 { ok = 0 }
1589 if okE9 != 3 { ok = 0 }
1590 if ok1E9 != 1 { ok = 0 }
1591 if dnE9 != 1 { ok = 0 }
1592
1593 // T7 liar-kill negatives: unknown-flow in-flight run refuses loudly (no fabricated DONE);
1594 // nonsense status never appears
1595 wf_emit(led, "EX.zz" as *u8, "zz" as *u8, 0, "START" as *u8, 0)
1596 let rz: i64 = wf_resume(cx)
1597 buf = wf_readall(led, szp)
1598 let dnZZ: i64 = wf_lines_with2(buf, szp[0], " rid=EX.zz " as *u8, "status=DONE" as *u8)
1599 let ban: i64 = wf_lines_with2(buf, szp[0], "WFRUN" as *u8, "status=BANANA" as *u8)
1600 p(" T7 negctl resumeRc=" as *u8); pn(rz); p(" zzDone=" as *u8); pn(dnZZ); p(" nonsense=" as *u8); pn(ban); p("\n" as *u8)
1601 if rz != (0 - 1) { ok = 0 }
1602 if dnZZ != 0 { ok = 0 }
1603 if ban != 0 { ok = 0 }
1604
1605 p("NX-WFLOW-ENGINE-SELFTEST ledger=" as *u8); p(led)
1606 p(" runs: done=3 failed=1 dedup=1 resumed=1 " as *u8)
1607 if ok == 1 { p("verdict=GREEN\n" as *u8); return 0 }
1608 p("verdict=RED\n" as *u8)
1609 return 1
1610}