nx_watch.nx source
↩ module page · 208 lines · 8104 B
1// nx_watch.nx -- SOVEREIGN JOB / LIVENESS MONITOR (operator 2026-07-16: "our monitors are garbage
2// -- WSL-polling Claude-Code Monitor flakes; build a STATE-OF-THE-ART Nishi monitor as MCP + RACI +
3// API + workstream capability"). ROOT CAUSE of the flake: watching = polling WSL for "is it done"
4// through fragile shell `until` conditions (screen-noise, exit-code weirdness, compound tests). THE
5// SOVEREIGN FIX: a job writes ITS OWN terminal state into the sovereign store; nx_watch reports it
6// DETERMINISTICALLY from the store -- zero polling, zero WSL dependence, works cross-session.
7// mark <ns> <id> <state> job records its state (RUNNING at start, GREEN/RED/DONE at end) + ts
8// beat <ns> <id> heartbeat for a long job (liveness; staleness = crash signal)
9// status <ns> <id> print state + age_ms; EXIT CODE = verdict (0 GREEN/DONE, 1 RED,
10// 2 RUNNING, 3 STALLED[running+stale], 4 ABSENT) -- scriptable, no grep
11// wait <ns> <id> <max_ms> block until terminal (GREEN/RED/DONE) or timeout; sovereign sleep poll
12// Store = seg_store via nx_sov_ledger (knowledge/store/watch_<ns>, key=id -> "STATE|us"). The
13// staleness verdict is PURE (stored ts vs now) = gate-testable without a real clock race.
14// RACI: supervisor|monitor (owns this). MCP: status/list = read verbs (broad-expose tier).
15// license_tier: ORIGINAL
16import "nx_sov_ledger.nx"
17import "nx_lib_std.nx"
18const NW_MAGIC_2000: i64 = 2000
19
20const NW_STALE_MS: i64 = 120000 // a RUNNING/BEAT job silent > 2min = STALLED (data-driven default)
21
22const NW_GREEN: i64 = 0
23const NW_RED: i64 = 1
24const NW_RUNNING: i64 = 2
25const NW_STALLED: i64 = 3
26const NW_ABSENT: i64 = 4
27
28func nw_prefix(ns: *u8, out: *u8) -> i64 {
29 let p: *u8 = "knowledge/store/watch_" as *u8
30 var o: i64 = 0
31 var i: i64 = 0
32 while p[i] != (0 as u8) { out[o] = p[i]; o = o + 1; i = i + 1 }
33 i = 0
34 while ns[i] != (0 as u8) { out[o] = ns[i]; o = o + 1; i = i + 1 }
35 out[o] = 0 as u8
36 return o
37}
38
39// state word -> class: 0=terminal-green 1=terminal-red 2=live(running/beat/custom). -1 malformed.
40// ANY unknown non-empty word = a CUSTOM PROGRESS STATE = live (jobs announce phases like ADMIT-WAIT,
41// T1-RUN; a custom word is a liveness signal, age-gated to STALLED like RUNNING). The old behavior
42// returned -1 for unknown words and nw_status printed ABSENT -- conflating "present with a custom
43// state" and "no such key" = a fail-loud violation that cost a false "watch is broken" debug arc
44// (2026-07-16; witness nx_watch_read_probe: ledger read fine, nw_class was the miss).
45func nw_class(word: *u8, wl: i64) -> i64 {
46 // first char discriminates: G(REEN)/D(ONE)=green, R(ED|UNNING) needs 2nd char, B(EAT)=live
47 if wl <= 0 { return 0 - 1 }
48 let c0: i64 = word[0] as i64
49 if c0 == 71 { return 0 } // G
50 if c0 == 68 { return 0 } // D(ONE)
51 if c0 == 66 { return 2 } // B(EAT)
52 if c0 == 82 {
53 if wl >= 2 {
54 let c1: i64 = word[1] as i64
55 if c1 == 69 { return 1 } // RE(D)
56 if c1 == 85 { return 2 } // RU(NNING)
57 }
58 return 2
59 }
60 return 2 // custom progress word = live
61}
62
63// PURE verdict: state class + age vs stale threshold -> the exit-code verdict. gate-testable.
64func nw_verdict(cls: i64, age_ms: i64, stale_ms: i64) -> i64 {
65 if cls == 0 { return NW_GREEN }
66 if cls == 1 { return NW_RED }
67 if cls == 2 {
68 if age_ms > stale_ms { return NW_STALLED }
69 return NW_RUNNING
70 }
71 return NW_ABSENT
72}
73
74// write "STATE|us" for a job.
75func nw_mark(ns: *u8, id: *u8, state: *u8) -> i64 {
76 let prefix: *u8 = sys_mmap(256) as *u8
77 nw_prefix(ns, prefix)
78 let val: *u8 = sys_mmap(128) as *u8
79 var o: i64 = 0
80 var i: i64 = 0
81 while state[i] != (0 as u8) { val[o] = state[i]; o = o + 1; i = i + 1 }
82 val[o] = 124 as u8; o = o + 1
83 let now: i64 = sys_now_us()
84 o = o + std_itoa(now, val + o)
85 val[o] = 0 as u8
86 sov_put_str(prefix, id, val)
87 return 0
88}
89
90// read a job -> class + age_ms. returns class (-1 absent); ageout[0]=age_ms.
91func nw_read(ns: *u8, id: *u8, wordout: *u8, ageout: *i64) -> i64 {
92 let prefix: *u8 = sys_mmap(256) as *u8
93 nw_prefix(ns, prefix)
94 let v: *u8 = sys_mmap(128) as *u8
95 let n: i64 = sov_get_copy(prefix, id, v, 127)
96 if n <= 0 { return 0 - 1 }
97 // split "STATE|us"
98 var bar: i64 = 0 - 1
99 var i: i64 = 0
100 while i < n { if v[i] == (124 as u8) { bar = i; i = n } else { i = i + 1 } }
101 if bar < 0 { return 0 - 1 }
102 var wl: i64 = 0
103 while wl < bar { wordout[wl] = v[wl]; wl = wl + 1 }
104 wordout[wl] = 0 as u8
105 var us: i64 = 0
106 var p: i64 = bar + 1
107 while p < n { let c: i64 = v[p] as i64; if c >= 48 && c <= 57 { us = us * 10 + (c - 48) } p = p + 1 }
108 let now: i64 = sys_now_us()
109 var age: i64 = (now - us) / 1000
110 if age < 0 { age = 0 }
111 ageout[0] = age
112 return nw_class(wordout, wl)
113}
114
115func nw_status(ns: *u8, id: *u8) -> i64 {
116 let word: *u8 = sys_mmap(64) as *u8
117 let age: *i64 = sys_mmap(8) as *i64
118 let cls: i64 = nw_read(ns, id, word, age)
119 if cls < 0 {
120 std_puts("WATCH " as *u8)
121 std_puts(id)
122 std_putln(" ABSENT" as *u8)
123 return NW_ABSENT
124 }
125 let v: i64 = nw_verdict(cls, age[0], NW_STALE_MS)
126 std_puts("WATCH " as *u8)
127 std_puts(id)
128 std_puts(" state=" as *u8)
129 std_puts(word)
130 std_puts(" age_ms=" as *u8)
131 std_pdec(age[0])
132 std_puts(" -> " as *u8)
133 if v == NW_GREEN { std_putln("GREEN" as *u8) }
134 if v == NW_RED { std_putln("RED" as *u8) }
135 if v == NW_RUNNING { std_putln("RUNNING" as *u8) }
136 if v == NW_STALLED { std_putln("STALLED (silent past threshold = crash signal)" as *u8) }
137 return v
138}
139
140// sovereign wait: poll the STORE (not WSL) until terminal or timeout.
141func nw_wait(ns: *u8, id: *u8, max_ms: i64, poll_ms: i64) -> i64 {
142 let word: *u8 = sys_mmap(64) as *u8
143 let age: *i64 = sys_mmap(8) as *i64
144 var waited: i64 = 0
145 var go: i64 = 1
146 while go == 1 {
147 let cls: i64 = nw_read(ns, id, word, age)
148 if cls == 0 { return NW_GREEN }
149 if cls == 1 { return NW_RED }
150 if waited >= max_ms { go = 0 }
151 if go == 1 { sys_sleep_ms(poll_ms); waited = waited + poll_ms }
152 }
153 return nw_status(ns, id)
154}
155
156func main(argc: i64, argv: *i64) -> i64 {
157 if argc < 3 {
158 std_putln("usage: nx_watch mark <ns> <id> <state> | beat <ns> <id> | status <ns> <id> | wait <ns> <id> <max_ms>" as *u8)
159 sys_exit(4)
160 return 4
161 }
162 let a1: i64 = argv[1]
163 let verb: *u8 = a1 as *u8
164 let a2: i64 = argv[2]
165 let ns: *u8 = a2 as *u8
166 if std_streq(verb, "mark" as *u8) == 1 {
167 if argc < 5 { std_putln("mark needs <ns> <id> <state>" as *u8); sys_exit(4); return 4 }
168 let a3: i64 = argv[3]
169 let id: *u8 = a3 as *u8
170 let a4: i64 = argv[4]
171 let st: *u8 = a4 as *u8
172 nw_mark(ns, id, st)
173 std_putln("WATCH marked" as *u8)
174 sys_exit(0)
175 return 0
176 }
177 if std_streq(verb, "beat" as *u8) == 1 {
178 if argc < 4 { std_putln("beat needs <ns> <id>" as *u8); sys_exit(4); return 4 }
179 let b3: i64 = argv[3]
180 let id2: *u8 = b3 as *u8
181 nw_mark(ns, id2, "BEAT" as *u8)
182 std_putln("WATCH beat" as *u8)
183 sys_exit(0)
184 return 0
185 }
186 if std_streq(verb, "status" as *u8) == 1 {
187 if argc < 4 { std_putln("status needs <ns> <id>" as *u8); sys_exit(4); return 4 }
188 let c3: i64 = argv[3]
189 let id3: *u8 = c3 as *u8
190 let v: i64 = nw_status(ns, id3)
191 sys_exit(v)
192 return v
193 }
194 if std_streq(verb, "wait" as *u8) == 1 {
195 if argc < 5 { std_putln("wait needs <ns> <id> <max_ms>" as *u8); sys_exit(4); return 4 }
196 let d3: i64 = argv[3]
197 let id4: *u8 = d3 as *u8
198 let d4: i64 = argv[4]
199 let mp: *u8 = d4 as *u8
200 let maxms: i64 = std_atoi(mp)
201 let v2: i64 = nw_wait(ns, id4, maxms, NW_MAGIC_2000)
202 sys_exit(v2)
203 return v2
204 }
205 std_putln("unknown verb" as *u8)
206 sys_exit(4)
207 return 4
208}