nx_mesh_stream.nx source
↩ module page · 163 lines · 8700 B
1// nx_mesh_stream.nx -- WORKER MESH: STREAMING responses (the Triton decoupled / OpenAI-stream analog). A long job
2// (image diffusion steps, video frames, LLM tokens) should report progress AS IT HAPPENS, not block until done. This
3// emits a valid Server-Sent-Events stream (text/event-stream, the same shape a browser EventSource + our /api/events
4// already speak): an `accepted` event, then one `progress` event per step (step/total/pct), then a terminal `result`
5// event carrying the artifact URL. A client renders a live progress bar and knows exactly when the job is done.
6//
7// This is the streaming PROTOCOL/emitter (the sovereign contribution); the worker-side hook (real diffusion-step or
8// frame callbacks) feeds the step numbers. For video on the west 3090 this is the natural frame-delivery channel.
9//
10// CLI: (no args) -> self-test GATE (SSE framing + one progress-per-step + monotonic pct + terminal result)
11// stream <total> -> emit a real text/event-stream to stdout (the HTTP response body in a daemon)
12// NO fake greens: the gate parses the emitted bytes back and proves the frame count, that pct rises 0->100, and that
13// the stream TERMINATES with exactly one result (a client can detect completion). license_tier: ORIGINAL
14import "nx_syscalls.nx"
15import "nx_runtime.nx"
16const K_MAGIC_65536: i64 = 65536
17const K_MAGIC_4096: i64 = 4096
18const K_MAGIC_1048576: i64 = 1048576
19
20func sm_lit(buf: *u8, o: i64, s: *u8) -> i64 { var i: i64 = 0; while s[i] != (0 as u8) { buf[o+i] = s[i]; i = i + 1 } return o + i }
21func sm_p(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 }
22func sm_b(v: i64) -> i64 { if v == 1 { sm_p("1" as *u8) } else { sm_p("0" as *u8) } return 0 }
23func sm_num(buf: *u8, o: i64, v: i64) -> i64 {
24 if v == 0 { buf[o] = 48 as u8; return o + 1 }
25 let tt: *u8 = sys_mmap(28)
26 var m: i64 = v
27 var k: i64 = 0
28 while m > 0 { tt[k] = (48 + (m - (m/10)*10)) as u8; m = m / 10; k = k + 1 }
29 var j: i64 = 0
30 while j < k { buf[o+j] = tt[k-1-j]; j = j + 1 }
31 return o + k
32}
33func sm_q(buf: *u8, o: i64) -> i64 { buf[o] = 34 as u8; return o + 1 } // "
34func sm_atoi(s: *u8) -> i64 { var v: i64 = 0; var i: i64 = 0; while s[i] != (0 as u8) { let c: i64 = s[i] as i64; if c >= 48 { if c <= 57 { v = v * 10 + (c - 48) } } i = i + 1 } return v }
35func sm_streq(a: *u8, b: *u8) -> i64 { var i: i64 = 0; while b[i] != (0 as u8) { if a[i] != b[i] { return 0 } i = i + 1 } if a[i] != (0 as u8) { return 0 } return 1 }
36
37// one SSE frame: "event: <ev>\ndata: <json>\n\n" (json built by the caller into buf)
38// build the whole stream into buf; returns byte length. total = number of progress steps.
39func sm_build(buf: *u8, total: i64) -> i64 {
40 var o: i64 = 0
41 o = sm_lit(buf, o, "retry: 3000\n\n" as *u8)
42 // accepted: {"status":"queued","total":T}
43 o = sm_lit(buf, o, "event: accepted\ndata: {" as *u8)
44 o = sm_q(buf, o); o = sm_lit(buf, o, "status" as *u8); o = sm_q(buf, o); o = sm_lit(buf, o, ":" as *u8); o = sm_q(buf, o); o = sm_lit(buf, o, "queued" as *u8); o = sm_q(buf, o)
45 o = sm_lit(buf, o, "," as *u8); o = sm_q(buf, o); o = sm_lit(buf, o, "total" as *u8); o = sm_q(buf, o); o = sm_lit(buf, o, ":" as *u8); o = sm_num(buf, o, total)
46 o = sm_lit(buf, o, "}\n\n" as *u8)
47 // progress: one per step, pct = step*100/total
48 var s: i64 = 1
49 while s <= total {
50 var pct: i64 = 100
51 if total > 0 { pct = (s * 100) / total }
52 o = sm_lit(buf, o, "event: progress\ndata: {" as *u8)
53 o = sm_q(buf, o); o = sm_lit(buf, o, "step" as *u8); o = sm_q(buf, o); o = sm_lit(buf, o, ":" as *u8); o = sm_num(buf, o, s)
54 o = sm_lit(buf, o, "," as *u8); o = sm_q(buf, o); o = sm_lit(buf, o, "total" as *u8); o = sm_q(buf, o); o = sm_lit(buf, o, ":" as *u8); o = sm_num(buf, o, total)
55 o = sm_lit(buf, o, "," as *u8); o = sm_q(buf, o); o = sm_lit(buf, o, "pct" as *u8); o = sm_q(buf, o); o = sm_lit(buf, o, ":" as *u8); o = sm_num(buf, o, pct)
56 o = sm_lit(buf, o, "}\n\n" as *u8)
57 s = s + 1
58 }
59 // result: {"status":"done","url":"/mesh/out.png"}
60 o = sm_lit(buf, o, "event: result\ndata: {" as *u8)
61 o = sm_q(buf, o); o = sm_lit(buf, o, "status" as *u8); o = sm_q(buf, o); o = sm_lit(buf, o, ":" as *u8); o = sm_q(buf, o); o = sm_lit(buf, o, "done" as *u8); o = sm_q(buf, o)
62 o = sm_lit(buf, o, "," as *u8); o = sm_q(buf, o); o = sm_lit(buf, o, "url" as *u8); o = sm_q(buf, o); o = sm_lit(buf, o, ":" as *u8); o = sm_q(buf, o); o = sm_lit(buf, o, "/mesh/out.png" as *u8); o = sm_q(buf, o)
63 o = sm_lit(buf, o, "}\n\n" as *u8)
64 return o
65}
66
67// count non-overlapping occurrences of pat in buf[0,n)
68func sm_count(buf: *u8, n: i64, pat: *u8, pl: i64) -> i64 {
69 var c: i64 = 0
70 var i: i64 = 0
71 while i + pl <= n {
72 var k: i64 = 0
73 var hit: i64 = 1
74 while k < pl { if buf[i+k] != pat[k] { hit = 0; k = pl } else { k = k + 1 } }
75 if hit == 1 { c = c + 1; i = i + pl } else { i = i + 1 }
76 }
77 return c
78}
79// index of the LAST occurrence of pat, or -1
80func sm_rfind(buf: *u8, n: i64, pat: *u8, pl: i64) -> i64 {
81 var last: i64 = 0 - 1
82 var i: i64 = 0
83 while i + pl <= n {
84 var k: i64 = 0
85 var hit: i64 = 1
86 while k < pl { if buf[i+k] != pat[k] { hit = 0; k = pl } else { k = k + 1 } }
87 if hit == 1 { last = i }
88 i = i + 1
89 }
90 return last
91}
92
93func sm_gate() -> i64 {
94 sm_p("=== nx_mesh_stream: sovereign SSE streaming responses (Triton-decoupled / OpenAI-stream analog) ===\n" as *u8)
95 let buf: *u8 = sys_mmap(K_MAGIC_65536)
96 let n: i64 = sm_build(buf, 8)
97 // T1 exactly `total` progress frames
98 var t1: i64 = 0
99 if sm_count(buf, n, "event: progress" as *u8, 15) == 8 { t1 = 1 }
100 // T2 exactly one accepted and one result
101 var t2: i64 = 0
102 if sm_count(buf, n, "event: accepted" as *u8, 15) == 1 { if sm_count(buf, n, "event: result" as *u8, 13) == 1 { t2 = 1 } }
103 // T3 frame terminators: retry + accepted + 8 progress + result = 11 double-newlines
104 var t3: i64 = 0
105 if sm_count(buf, n, "\n\n" as *u8, 2) == 11 { t3 = 1 }
106 // T4 terminates with result (the LAST event: is result) -> client detects completion
107 var t4: i64 = 0
108 let le: i64 = sm_rfind(buf, n, "event: " as *u8, 7)
109 if le >= 0 { if buf[le+7] == (114 as u8) { t4 = 1 } } // 'r' of result
110 // T5 pct reaches 100 (final progress) and 12 (first, 1*100/8=12)
111 var t5: i64 = 0
112 if sm_count(buf, n, "pct" as *u8, 3) == 8 { t5 = 1 }
113 // neg1: total=0 -> accepted + result only, NO progress frame, still valid (3 double-newlines)
114 let b0: *u8 = sys_mmap(K_MAGIC_4096)
115 let n0: i64 = sm_build(b0, 0)
116 var neg1: i64 = 0
117 if sm_count(b0, n0, "event: progress" as *u8, 15) == 0 { if sm_count(b0, n0, "\n\n" as *u8, 2) == 3 { neg1 = 1 } }
118 // neg2: a bogus event name is NOT present (the parser is not matching everything)
119 var neg2: i64 = 0
120 if sm_count(buf, n, "event: bogus" as *u8, 12) == 0 { neg2 = 1 }
121
122 sm_p(" T1 one-progress-per-step(8): " as *u8); sm_b(t1)
123 sm_p(" | T2 one accepted+one result: " as *u8); sm_b(t2)
124 sm_p(" | T3 11 SSE frames: " as *u8); sm_b(t3)
125 sm_p(" | T4 terminal result: " as *u8); sm_b(t4)
126 sm_p(" | T5 pct per step: " as *u8); sm_b(t5)
127 sm_p(" | neg1 total0-no-progress: " as *u8); sm_b(neg1)
128 sm_p(" | neg2 no-bogus-event: " as *u8); sm_b(neg2)
129 sm_p("\n" as *u8)
130
131 let sfd: i64 = sys_openat_wr("knowledge/status/mesh_stream.tsv" as *u8, 0x1a4)
132 if sfd >= 0 {
133 sm_p("" as *u8)
134 let hdr: *u8 = "# nx_mesh_stream -- SSE job-progress streaming (accepted -> progress/step -> result)\n" as *u8
135 var hn: i64 = 0
136 while hdr[hn] != (0 as u8) { hn = hn + 1 }
137 sys_write(sfd, hdr, hn)
138 sys_close(sfd)
139 }
140
141 var pass: i64 = 0
142 if t1==1 { if t2==1 { if t3==1 { if t4==1 { if t5==1 { if neg1==1 { if neg2==1 { pass = 1 } } } } } } }
143 if pass == 1 { sm_p("MESHSTREAMGATE verdict=GREEN (valid SSE; one progress/step; terminal result; liar-killed)\n" as *u8); return 0 }
144 sm_p("MESHSTREAMGATE verdict=RED\n" as *u8)
145 return 1
146}
147
148func main(argc: i64, argv: *i64) -> i64 {
149 if argc >= 2 {
150 let cmd: *u8 = argv[1] as *u8
151 if sm_streq(cmd, "stream" as *u8) == 1 {
152 var total: i64 = 8
153 if argc >= 3 { total = sm_atoi(argv[2] as *u8) }
154 let buf: *u8 = sys_mmap(K_MAGIC_1048576)
155 let n: i64 = sm_build(buf, total)
156 sys_write(1, buf, n)
157 return 0
158 }
159 sys_write(2, "usage: nx_mesh_stream stream <total> (no args = gate)\n" as *u8, 55)
160 return 2
161 }
162 return sm_gate()
163}