nx_heartbeat_monitor_gate.nx source
↩ module page · 168 lines · 9575 B
1// nx_heartbeat_monitor_gate.nx -- the REFEREE for WMS-M1 (liveness / heartbeat monitor).
2//
3// Proves the nx_heartbeat_monitor detector (a) FLAGS exactly the crashed stream as
4// STALLED, (b) NEGATIVE CONTROL: the two freshly-beating streams are NOT flagged --
5// a detector that returns STALLED for everything FAILS this lane; one that returns
6// ALIVE for everything FAILS the core lane -- so the same scan call cross-checks both,
7// (c) TAMPER: the stalled stream RESUMES beating (a fresh max-epoch beat) -> STALLED
8// CLEARS -- proves the monitor tracks the LATEST beat, not a sticky first reading,
9// (d) malformed line surfaced (count > 0, ws_ledger no-silent-skip discipline).
10// No-false-green: GREEN requires ALL lanes.
11//
12// TIME IS INJECTED (NOW / THRESHOLD are gate constants; every beat is placed at a
13// controlled epoch via hb_beat_at) -- the gate does NOT depend on real elapsed
14// wall-clock, so it is deterministic and re-runnable.
15//
16// WRITE DISCIPLINE (the torn-line bug): the gate's OWN evidence record is assembled
17// into ONE buffer and emitted with a SINGLE locked fa_appendz -- to a UNIQUE scratch
18// path (epoch+pid) so concurrent gate runs never collide (eat our own dogfood). The
19// row-by-row lines go to fd 1 (the terminal -- uncontended, same exemption the
20// ws_ledger gate uses for stdout).
21//
22// Exit 0 GREEN, 1 RED. Sovereign: only nx_syscalls + nx_heartbeat_monitor.
23// license_tier: ORIGINAL
24import "nx_syscalls.nx"
25import "nx_heartbeat_monitor.nx"
26
27const G_NOW: i64 = 1000 // gate-injected "current" epoch
28const G_THRESH: i64 = 60 // staleness threshold (gate parameter, not baked in detector)
29const G_EVCAP: i64 = 512 // bounded evidence record size
30
31// stdout-only writers (terminal is uncontended; the FILE record uses fa_appendz).
32func gp(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 }
33func gn(v: i64) -> i64 { let bb: *u8=sys_mmap(28); var m: i64=v; if m<0{m=0-m; sys_write(1,"-\x00" as *u8,1)}; let t: *u8=sys_mmap(28); var k: i64=0; if m==0{t[0]=48 as u8;k=1}; while m>0{t[k]=(48+(m%10)) as u8;m=m/10;k=k+1}; var i: i64=0; while i<k{bb[i]=t[k-1-i];i=i+1}; sys_write(1,bb,k); return 0 }
34
35// build the UNIQUE scratch evidence path "knowledge/status/heartbeat_monitor_gate.<epoch>.<pid>.tmp"
36// is NOT used -- per the directive the canonical evidence log is the stable path; the
37// per-run uniqueness requirement is satisfied by appending (additive) to the stable log
38// via the LOCKED fa_appendz (safe to run concurrently with itself). We ALSO emit a
39// per-run unique tmp twin so concurrent runs never even contend on the same fd offset.
40func build_unique_tmp(buf: *u8, epoch: i64, pid: i64) -> i64 {
41 var o: i64 = 0
42 o = fa_cat(buf, o, "knowledge/status/heartbeat_monitor_gate.\x00" as *u8)
43 o = fa_catn(buf, o, epoch)
44 o = fa_cat(buf, o, ".\x00" as *u8)
45 o = fa_catn(buf, o, pid)
46 o = fa_cat(buf, o, ".tmp\x00" as *u8)
47 buf[o] = 0 as u8
48 return o
49}
50
51func main() -> i64 {
52 let pid: i64 = __syscall(39, 0, 0, 0, 0, 0, 0)
53 let now0: i64 = sys_now_realtime_sec()
54
55 gp("=== WMS-M1 HEARTBEAT-MONITOR GATE (liveness detector) ===\n" as *u8)
56 gp(" now_inj=\x00" as *u8); gn(G_NOW)
57 gp(" threshold=\x00" as *u8); gn(G_THRESH)
58 gp(" pid=\x00" as *u8); gn(pid)
59 gp("\n\x00" as *u8)
60
61 // fresh heartbeat channel each run (truncate via O_TRUNC write-open)
62 let chan: *u8 = "/tmp/hbm_gate.log\x00" as *u8
63 let f0: i64 = sys_openat_wr(chan, 0x1a4)
64 if f0 > 0 { sys_close(f0) }
65
66 // ===== three streams beat at CONTROLLED epochs (the real locked-write path) =====
67 // ws=1 @ 990 (age 10, fresh) ws=2 @ 985 (age 15, fresh) ws=3 @ 900 (age 100, STALLED)
68 hb_beat_at(chan, 1, 990, 0, pid)
69 hb_beat_at(chan, 2, 985, 0, pid)
70 hb_beat_at(chan, 3, 900, 0, pid)
71
72 let ids: *i64 = sys_mmap(8 * 4) as *i64
73 ids[0] = 1; ids[1] = 2; ids[2] = 3
74 let vd: *i64 = sys_mmap(8 * 4) as *i64
75 let outs: *i64 = sys_mmap(32) as *i64
76
77 // ===== PRE scan: exactly ws=3 STALLED; ws=1 & ws=2 ALIVE (neg-control) =====
78 hbm_scan(chan, G_NOW, G_THRESH, ids, 3, vd, outs)
79 let v1_pre: i64 = vd[0]
80 let v2_pre: i64 = vd[1]
81 let v3_pre: i64 = vd[2]
82 let pre_stalled: i64 = outs[0]
83 let pre_alive: i64 = outs[1]
84
85 // ===== TAMPER / RESUME: ws=3 beats fresh @ 995 (age 5) -> must clear STALLED =====
86 hb_beat_at(chan, 3, 995, 1, pid)
87 hbm_scan(chan, G_NOW, G_THRESH, ids, 3, vd, outs)
88 let v3_post: i64 = vd[2]
89 let post_stalled: i64 = outs[0]
90
91 // ===== MALFORMED: append one raw broken line (missing " END") -> must be flagged =====
92 let mfd: i64 = sys_openat_append(chan, 0x1a4)
93 if mfd > 0 {
94 sys_write(mfd, "HBX ws=9 epoch=900 seq=0 actor=0 BROKEN\n\x00" as *u8, 40)
95 sys_close(mfd)
96 }
97 hbm_scan(chan, G_NOW, G_THRESH, ids, 3, vd, outs)
98 let flagged: i64 = outs[2]
99
100 // ===== rows (stdout) =====
101 gp("HBM row=core_stall ws3_pre=\x00" as *u8); gn(v3_pre); gp(" want=STALLED(1)\x00" as *u8)
102 if v3_pre == HB_STALLED { gp(" verdict=PASS\n\x00" as *u8) } else { gp(" verdict=FAIL\n\x00" as *u8) }
103
104 gp("HBM row=neg_control ws1_pre=\x00" as *u8); gn(v1_pre); gp(" ws2_pre=\x00" as *u8); gn(v2_pre); gp(" want=ALIVE(0)\x00" as *u8)
105 if v1_pre == HB_ALIVE { if v2_pre == HB_ALIVE { gp(" verdict=PASS(fresh-not-flagged)\n\x00" as *u8) } else { gp(" verdict=FAIL\n\x00" as *u8) } } else { gp(" verdict=FAIL\n\x00" as *u8) }
106
107 gp("HBM row=count stalled=\x00" as *u8); gn(pre_stalled); gp(" alive=\x00" as *u8); gn(pre_alive); gp(" want=1/2\x00" as *u8)
108 if pre_stalled == 1 { if pre_alive == 2 { gp(" verdict=PASS\n\x00" as *u8) } else { gp(" verdict=FAIL\n\x00" as *u8) } } else { gp(" verdict=FAIL\n\x00" as *u8) }
109
110 gp("HBM row=tamper ws3_post=\x00" as *u8); gn(v3_post); gp(" post_stalled=\x00" as *u8); gn(post_stalled); gp(" want=ALIVE(0)/0\x00" as *u8)
111 if v3_post == HB_ALIVE { if post_stalled == 0 { gp(" verdict=PASS(resume-clears)\n\x00" as *u8) } else { gp(" verdict=FAIL\n\x00" as *u8) } } else { gp(" verdict=FAIL\n\x00" as *u8) }
112
113 gp("HBM row=malformed_flag flagged=\x00" as *u8); gn(flagged); gp(" want>=1\x00" as *u8)
114 if flagged >= 1 { gp(" verdict=PASS(corruption-surfaced)\n\x00" as *u8) } else { gp(" verdict=FAIL(corruption-swallowed)\n\x00" as *u8) }
115
116 // ===== verdict (no false green -- every lane required) =====
117 var green: i64 = 1
118 if v3_pre != HB_STALLED { green = 0 } // core: the crashed stream IS caught
119 if v1_pre != HB_ALIVE { green = 0 } // NEG-CONTROL: fresh stream NOT flagged
120 if v2_pre != HB_ALIVE { green = 0 } // NEG-CONTROL: fresh stream NOT flagged
121 if pre_stalled != 1 { green = 0 } // exactly one stalled
122 if pre_alive != 2 { green = 0 }
123 if v3_post != HB_ALIVE { green = 0 } // TAMPER: resume clears STALLED
124 if post_stalled != 0 { green = 0 }
125 if flagged < 1 { green = 0 } // corruption surfaced (no silent skip)
126
127 // ===== evidence record: ONE buffer -> ONE locked fa_appendz (write discipline) =====
128 // stable canonical log (additive, locked-safe) + a per-run UNIQUE tmp twin so two
129 // concurrent gate runs never even share an fd offset.
130 let rec: *u8 = sys_mmap(G_EVCAP + 32)
131 var o: i64 = 0
132 o = fa_cat(rec, o, "HEARTBEAT-MONITOR verdict=\x00" as *u8)
133 if green == 1 { o = fa_cat(rec, o, "GREEN\x00" as *u8) } else { o = fa_cat(rec, o, "RED\x00" as *u8) }
134 o = fa_cat(rec, o, " now=\x00" as *u8); o = fa_catn(rec, o, G_NOW)
135 o = fa_cat(rec, o, " threshold=\x00" as *u8); o = fa_catn(rec, o, G_THRESH)
136 o = fa_cat(rec, o, " stalled_pre=\x00" as *u8); o = fa_catn(rec, o, pre_stalled)
137 o = fa_cat(rec, o, " alive_pre=\x00" as *u8); o = fa_catn(rec, o, pre_alive)
138 o = fa_cat(rec, o, " ws3_pre=\x00" as *u8); o = fa_catn(rec, o, v3_pre)
139 o = fa_cat(rec, o, " ws3_post=\x00" as *u8); o = fa_catn(rec, o, v3_post)
140 o = fa_cat(rec, o, " stalled_post=\x00" as *u8); o = fa_catn(rec, o, post_stalled)
141 o = fa_cat(rec, o, " flagged=\x00" as *u8); o = fa_catn(rec, o, flagged)
142 o = fa_cat(rec, o, " epoch=\x00" as *u8); o = fa_catn(rec, o, now0)
143 o = fa_cat(rec, o, " pid=\x00" as *u8); o = fa_catn(rec, o, pid)
144 if green == 0 {
145 o = fa_cat(rec, o, " reason=\x00" as *u8)
146 if v3_pre != HB_STALLED { o = fa_cat(rec, o, "core-not-caught \x00" as *u8) }
147 if v1_pre != HB_ALIVE { o = fa_cat(rec, o, "neg-ws1-flagged \x00" as *u8) }
148 if v2_pre != HB_ALIVE { o = fa_cat(rec, o, "neg-ws2-flagged \x00" as *u8) }
149 if pre_stalled != 1 { o = fa_cat(rec, o, "count-stalled-wrong \x00" as *u8) }
150 if pre_alive != 2 { o = fa_cat(rec, o, "count-alive-wrong \x00" as *u8) }
151 if v3_post != HB_ALIVE { o = fa_cat(rec, o, "resume-did-not-clear \x00" as *u8) }
152 if post_stalled != 0 { o = fa_cat(rec, o, "post-stalled-nonzero \x00" as *u8) }
153 if flagged < 1 { o = fa_cat(rec, o, "malformed-swallowed \x00" as *u8) }
154 }
155 rec[o] = 0 as u8
156
157 // canonical stable log -- locked atomic append (one write).
158 fa_appendz("knowledge/status/heartbeat_monitor_gate.log\x00" as *u8, rec, G_EVCAP)
159 // per-run unique tmp twin -- never shares an fd offset with a concurrent run.
160 let upath: *u8 = sys_mmap(256)
161 build_unique_tmp(upath, now0, pid)
162 fa_appendz(upath, rec, G_EVCAP)
163
164 gp("HEARTBEAT-MONITOR verdict=\x00" as *u8)
165 if green == 1 { gp("GREEN\n\x00" as *u8) } else { gp("RED\n\x00" as *u8) }
166 if green == 1 { return 0 }
167 return 1
168}