nx_workflow_lib.nx source
↩ module page · 199 lines · 9253 B
1// nx_workflow_lib.nx -- CONSOLIDATED workflow tool (MCP name: nx_workflow, tool #5 of the 15), LIBRARY
2// half. READ-ONLY first increment over the WMS registry SSOT (knowledge/store/ws-): the operator's
3// "workstreams callable from anywhere" ask -- a crash needs only this board + a resume trigger.
4// board <prefix> -> every workstream: state / empire / id / last_touched (+ per-state totals)
5// show <prefix> <id> -> one workstream, every field labeled
6// Composes the WMS store lib (ws_get_p / ws_field / WS record schema) -- no reinvention. spawn/dispatch/
7// resume = WRITE increments behind their own cap, after this read surface is proven.
8// license_tier: ORIGINAL
9import "nx_syscalls.nx"
10import "nx_itoa_lib.nx" // shared MSB-first emitter (zero-alloc)
11import "_hdl_build/nx_workstream_store.nx" // ws_get_p / ws_field / the ws: record schema (SSOT lib)
12
13const WF_REC_CAP: i64 = 8192 // one ws:<id> record working buffer
14const WF_IDS_CAP: i64 = 262144 // ws:ids index working buffer
15const WF_FIELD_CAP: i64 = 1024 // one extracted field
16const WF_KEY_CAP: i64 = 512
17const WF_ASCII_0: i64 = 48 // '0'
18const WF_F_EMPIRE: i64 = 1 // record field slots (schema: id empire state last_touched memory code deps)
19const WF_F_STATE: i64 = 2
20const WF_F_TOUCHED: i64 = 3
21const WF_F_MEMORY: i64 = 4
22const WF_F_CODE: i64 = 5
23const WF_SUM_MAX: i64 = 40 // summary caps the ACTIVE listing (loudly) -- agent-context-sized
24const WF_F_DEPS: i64 = 6
25
26func wf_puts(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 }
27// MIGRATED to the shared emitter (debt 1785563586). The old body mmapped a scratch buffer
28// per call and never freed it. At PAGE granularity that is 4096B leaked PER CALL -- the
29// defect that took 28.5GB of a 36GB host in nx_ts_lumadiff (2MB input, ~3.66M calls).
30// nxi_* is MSB-first, allocates NOTHING, and emits identical bytes including the sign.
31func wf_putn(v: i64) -> i64 { nxi_out(v); return 0 }
32func wf_seq(a: *u8, b: *u8) -> i64 { var i: i64 = 0; while a[i] != (0 as u8) { if a[i] != b[i] { return 0 } i = i + 1 } if b[i] != (0 as u8) { return 0 } return 1 }
33// build "ws:<id>" into out
34func wf_key(out: *u8, id: *u8, idlen: i64) -> i64 {
35 var o: i64 = 0
36 let pre: *u8 = "ws:" as *u8
37 var i: i64 = 0
38 while pre[i] != (0 as u8) { out[o] = pre[i]; o = o + 1; i = i + 1 }
39 i = 0
40 while i < idlen { out[o] = id[i]; o = o + 1; i = i + 1 }
41 out[o] = 0 as u8
42 return o
43}
44// print one labeled field of rec
45func wf_show_field(rec: *u8, rlen: i64, f: i64, label: *u8) -> i64 {
46 let fb: *u8 = sys_mmap(WF_FIELD_CAP)
47 ws_field(rec, rlen, f, fb)
48 wf_puts(label); wf_puts(fb); wf_puts("\n" as *u8)
49 return 0
50}
51
52// board: walk ws:ids, one line per workstream "<state>\t<empire>\t<id>\t<touched>"; totals + verdict.
53// Returns the workstream count; -1 if the registry index is absent (not a WMS store).
54func wf_board(prefix: *u8) -> i64 {
55 let po: *i64 = sys_mmap(16) as *i64
56 let lo: *i64 = sys_mmap(16) as *i64
57 if ws_get_p(prefix, "ws:ids" as *u8, po, lo) != 1 {
58 wf_puts("NX-WORKFLOW: no ws:ids index under this prefix (not a WMS registry?)\n" as *u8)
59 return 0 - 1
60 }
61 // copy the ids index (the record buffer is remapped by later ss_get calls)
62 let ids: *u8 = sys_mmap(WF_IDS_CAP)
63 var iln: i64 = lo[0]
64 if iln > WF_IDS_CAP - 1 { iln = WF_IDS_CAP - 1 }
65 let src: *u8 = po[0] as *u8
66 var c: i64 = 0
67 while c < iln { ids[c] = src[c]; c = c + 1 }
68 wf_puts("=== NX-WORKFLOW BOARD (WMS registry SSOT) ===\n" as *u8)
69 wf_puts("state\tempire\tid\tlast_touched\n" as *u8)
70 let kb: *u8 = sys_mmap(WF_KEY_CAP)
71 let fb: *u8 = sys_mmap(WF_FIELD_CAP)
72 var cnt: i64 = 0
73 var active: i64 = 0
74 var blocked: i64 = 0
75 var ls: i64 = 0
76 var i: i64 = 0
77 while i <= iln {
78 var eot: i64 = 0
79 if i == iln { eot = 1 } else { if ids[i] == (9 as u8) { eot = 1 } }
80 if eot == 1 {
81 if i > ls {
82 wf_key(kb, (ids as i64 + ls) as *u8, i - ls)
83 if ws_get_p(prefix, kb, po, lo) == 1 {
84 let rec: *u8 = po[0] as *u8
85 let rl: i64 = lo[0]
86 ws_field(rec, rl, WF_F_STATE, fb)
87 wf_puts(fb); wf_puts("\t" as *u8)
88 if wf_seq(fb, "ACTIVE" as *u8) == 1 { active = active + 1 }
89 if wf_seq(fb, "BLOCKED" as *u8) == 1 { blocked = blocked + 1 }
90 ws_field(rec, rl, WF_F_EMPIRE, fb)
91 wf_puts(fb); wf_puts("\t" as *u8)
92 sys_write(1, (ids as i64 + ls) as *u8, i - ls)
93 wf_puts("\t" as *u8)
94 ws_field(rec, rl, WF_F_TOUCHED, fb)
95 wf_puts(fb); wf_puts("\n" as *u8)
96 cnt = cnt + 1
97 }
98 }
99 ls = i + 1
100 }
101 i = i + 1
102 }
103 wf_puts("workstreams=" as *u8); wf_putn(cnt)
104 wf_puts(" active=" as *u8); wf_putn(active)
105 wf_puts(" blocked=" as *u8); wf_putn(blocked)
106 wf_puts("\n" as *u8)
107 return cnt
108}
109// summary: the AGENT-SIZED board (07-17 -- the full board is ~65KB and blows agent contexts over MCP).
110// Same ws:ids walk, but emits ONLY the per-state totals + the ACTIVE rows capped at WF_SUM_MAX
111// (cap is LOUD: "active-shown=N of M"). Returns total count, -1 absent-index.
112func wf_summary(prefix: *u8) -> i64 {
113 let po: *i64 = sys_mmap(16) as *i64
114 let lo: *i64 = sys_mmap(16) as *i64
115 // snapshot-once (ss_open) + ss_hget per key: the audit-scale cure for the O(ids x segs) walk
116 // (per-key ss_get re-scanned every segment per id -> ~50s at 1971 ids x 100 segs -> edge timeout)
117 let h: *i64 = ss_open(prefix)
118 if ss_hget(h, "ws:ids" as *u8, po, lo) != 1 {
119 wf_puts("NX-WORKFLOW: no ws:ids index under this prefix (not a WMS registry?)\n" as *u8)
120 return 0 - 1
121 }
122 let ids: *u8 = sys_mmap(WF_IDS_CAP)
123 var iln: i64 = lo[0]
124 if iln > WF_IDS_CAP - 1 { iln = WF_IDS_CAP - 1 }
125 let src: *u8 = po[0] as *u8
126 var c: i64 = 0
127 while c < iln { ids[c] = src[c]; c = c + 1 }
128 wf_puts("=== NX-WORKFLOW SUMMARY (agent-sized; full rows = `board`) ===\n" as *u8)
129 let kb: *u8 = sys_mmap(WF_KEY_CAP)
130 let fb: *u8 = sys_mmap(WF_FIELD_CAP)
131 let eb: *u8 = sys_mmap(WF_FIELD_CAP) // empire in its OWN buffer (fb still holds state for the tallies)
132 var cnt: i64 = 0
133 var active: i64 = 0
134 var blocked: i64 = 0
135 var done: i64 = 0
136 var shown: i64 = 0
137 var ls: i64 = 0
138 var i: i64 = 0
139 while i <= iln {
140 var eot: i64 = 0
141 if i == iln { eot = 1 } else { if ids[i] == (9 as u8) { eot = 1 } }
142 if eot == 1 {
143 if i > ls {
144 wf_key(kb, (ids as i64 + ls) as *u8, i - ls)
145 if ss_hget(h, kb, po, lo) == 1 {
146 let rec: *u8 = po[0] as *u8
147 let rl: i64 = lo[0]
148 ws_field(rec, rl, WF_F_STATE, fb)
149 if wf_seq(fb, "ACTIVE" as *u8) == 1 {
150 active = active + 1
151 if shown < WF_SUM_MAX {
152 wf_puts("ACTIVE\t" as *u8)
153 ws_field(rec, rl, WF_F_EMPIRE, eb)
154 wf_puts(eb); wf_puts("\t" as *u8)
155 sys_write(1, (ids as i64 + ls) as *u8, i - ls)
156 wf_puts("\n" as *u8)
157 shown = shown + 1
158 }
159 }
160 if wf_seq(fb, "BLOCKED" as *u8) == 1 { blocked = blocked + 1 }
161 if wf_seq(fb, "DONE" as *u8) == 1 { done = done + 1 }
162 cnt = cnt + 1
163 }
164 }
165 ls = i + 1
166 }
167 i = i + 1
168 }
169 wf_puts("workstreams=" as *u8); wf_putn(cnt)
170 wf_puts(" active=" as *u8); wf_putn(active)
171 wf_puts(" blocked=" as *u8); wf_putn(blocked)
172 wf_puts(" done=" as *u8); wf_putn(done)
173 wf_puts(" active-shown=" as *u8); wf_putn(shown)
174 wf_puts(" of " as *u8); wf_putn(active)
175 wf_puts("\n" as *u8)
176 return cnt
177}
178// show: every field of one workstream, labeled. 1 found | 0 tombstoned | -1 absent (the store's honest codes).
179func wf_show(prefix: *u8, id: *u8) -> i64 {
180 let kb: *u8 = sys_mmap(WF_KEY_CAP)
181 var n: i64 = 0
182 while id[n] != (0 as u8) { n = n + 1 }
183 wf_key(kb, id, n)
184 let po: *i64 = sys_mmap(16) as *i64
185 let lo: *i64 = sys_mmap(16) as *i64
186 let r: i64 = ws_get_p(prefix, kb, po, lo)
187 if r < 0 { wf_puts("NX-WORKFLOW ABSENT: no such workstream id\n" as *u8); return r }
188 if r == 0 { wf_puts("NX-WORKFLOW TOMBSTONED: workstream was retired\n" as *u8); return r }
189 let rec: *u8 = po[0] as *u8
190 let rl: i64 = lo[0]
191 wf_puts("id: " as *u8); wf_puts(id); wf_puts("\n" as *u8)
192 wf_show_field(rec, rl, WF_F_EMPIRE, "empire: " as *u8)
193 wf_show_field(rec, rl, WF_F_STATE, "state: " as *u8)
194 wf_show_field(rec, rl, WF_F_TOUCHED, "touched: " as *u8)
195 wf_show_field(rec, rl, WF_F_MEMORY, "memory: " as *u8)
196 wf_show_field(rec, rl, WF_F_CODE, "code: " as *u8)
197 wf_show_field(rec, rl, WF_F_DEPS, "deps: " as *u8)
198 return 1
199}