code wiki / _hdl_build / nx_publisher.nx
`nx_publisher.nx` source

↩ module page · 1404 lines · 79632 B
1// nx_publisher.nx -- THE NISHI PUBLISHER, R1: the single INTAKE. pub_submit() enqueues a publish REQUEST to a
2// durable, concurrency-safe queue so every workstream DECOUPLES from the live site (workstreams REQUEST; the
3// publisher SHIPS). The queue append reuses nx_framed_append's fa_appendz = O_APPEND + flock(LOCK_EX) +
4// write-until-complete, so N concurrent submitters can NEVER tear or lose a record -- the exact collision class
5// that caused the sites.elf outage. Each request carries a sha256 of the SOURCE CONTENT (honest identity ->
6// later enables idempotent dedup + post-deploy verification). One framed TSV line per request:
7//   PENDING<TAB>site<TAB>requester<TAB>sha256<TAB>src<TAB>dest<TAB>policy
8// This is R1 of [[project-nishi-publisher-role-2026-06-20]] (R0 census GREEN). license_tier: ORIGINAL
9import "nx_syscalls.nx"
10import "nx_connect.nx"   // bounded connect: a raw sys_connect hangs ~127s on a black-holed host
11import "nx_framed_append.nx"
12import "nx_sha256.nx"
13import "nx_arbiter.nx"
14const PUB_MAGIC_50000: i64 = 50000
15const PUB_MAGIC_4294967296: i64 = 4294967296
16const PUB_MAGIC_100000: i64 = 100000
17const PUB_MAGIC_1000000: i64 = 1000000
18const PUB_MAGIC_262144: i64 = 262144
19const PUB_MAGIC_4096: i64 = 4096
20
21const PUB_REC_CAP: i64 = 4096
22
23func pub_queue_default() -> *u8 { return "knowledge/publish/queue.tsv" as *u8 }
24
25// idempotent mkdir of the canonical queue dir (ignore EEXIST). raw x86_64 mkdir = 83 (mirrors nx_arbiter fl_mkdir).
26func pub_init() -> i64 { let d: *u8 = "knowledge/publish" as *u8; __syscall(83, d as i64, 493, 0, 0, 0, 0); return 0 }
27
28func pub_hexd(v: i64) -> i64 { if v < 10 { return 48 + v } return 87 + v }   // 0-9 then a-f
29
30// write nul-terminated `s` to fd (no framing) -- used ONLY by the unlocked negative control below.
31func pub_uw(fd: i64, s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(fd, s, n); return 0 }
32func pub_unap() -> i64 { let ts: *i64 = sys_mmap(16) as *i64; ts[0] = 0; ts[1] = PUB_MAGIC_50000; __syscall(35, ts as i64, 0, 0, 0, 0, 0); return 0 }
33
34// sha256 of the file at `path` -> 64 lowercase hex chars into outhex (NUL-terminated). 1 if read+hashed, 0 if missing.
35func pub_sha_file(path: *u8, outhex: *u8) -> i64 {
36    let lenp: *i64 = sys_mmap(8) as *i64
37    let data: *u8 = sys_read_file(path, lenp)
38    if (data as i64) == 0 { return 0 }
39    let dig: *u8 = sys_mmap(32)
40    sha256_digest(data, lenp[0], dig)
41    sys_munmap(data, PUB_MAGIC_4294967296 + 16)
42    var i: i64 = 0
43    while i < 32 {
44        let b: i64 = dig[i] & 0xff
45        outhex[i*2]     = pub_hexd((b >> 4) & 0xf) as u8
46        outhex[i*2 + 1] = pub_hexd(b & 0xf) as u8
47        i = i + 1
48    }
49    outhex[64] = 0 as u8
50    return 1
51}
52
53// assemble the request record into `rec` (NUL-terminated); returns its byte length. shared by both write paths.
54func pub_build_rec(rec: *u8, site: *u8, requester: *u8, hex: *u8, src: *u8, dest: *u8, policy: *u8) -> i64 {
55    var o: i64 = 0
56    o = fa_cat(rec, o, "PENDING" as *u8); rec[o] = 9 as u8; o = o + 1
57    o = fa_cat(rec, o, site);             rec[o] = 9 as u8; o = o + 1
58    o = fa_cat(rec, o, requester);        rec[o] = 9 as u8; o = o + 1
59    o = fa_cat(rec, o, hex);              rec[o] = 9 as u8; o = o + 1
60    o = fa_cat(rec, o, src);              rec[o] = 9 as u8; o = o + 1
61    o = fa_cat(rec, o, dest);             rec[o] = 9 as u8; o = o + 1
62    o = fa_cat(rec, o, policy)
63    rec[o] = 0 as u8
64    return o
65}
66
67// THE INTAKE. Enqueue a request to `qpath` as ONE atomic framed line. Returns 1 on durable append, else <=0
68// (src missing = -9; fa_appendz error otherwise). Concurrency-safe by construction (fa_appendz holds LOCK_EX).
69func pub_submit_to(qpath: *u8, src: *u8, dest: *u8, site: *u8, requester: *u8, policy: *u8) -> i64 {
70    let hex: *u8 = sys_mmap(72)
71    if pub_sha_file(src, hex) == 0 { return 0 - 9 }
72    let rec: *u8 = sys_mmap(PUB_REC_CAP)
73    pub_build_rec(rec, site, requester, hex, src, dest, policy)
74    let wr: i64 = fa_appendz(qpath, rec, PUB_REC_CAP)
75    if wr > 0 { return 1 }
76    return wr
77}
78
79// convenience: enqueue to the canonical shared queue (ensures the dir exists first).
80func pub_submit(src: *u8, dest: *u8, site: *u8, requester: *u8, policy: *u8) -> i64 {
81    pub_init()
82    return pub_submit_to(pub_queue_default(), src, dest, site, requester, policy)
83}
84
85// NEGATIVE CONTROL ONLY (the gate uses this to PROVE the framed path is load-bearing): assemble the SAME record
86// but emit it as SEVERAL separate writes with NO flock -> concurrent callers interleave -> TORN lines. This is
87// exactly the pre-fix outage bug; the real publisher NEVER calls it. Returns 0.
88func pub_append_unlocked(qpath: *u8, src: *u8, dest: *u8, site: *u8, requester: *u8, policy: *u8) -> i64 {
89    let hex: *u8 = sys_mmap(72)
90    pub_sha_file(src, hex)
91    let fd: i64 = sys_openat_append(qpath, 0x1a4)
92    if fd < 0 { return 0 - 1 }
93    pub_uw(fd, "PENDING\t" as *u8); pub_uw(fd, site); pub_uw(fd, "\t" as *u8)
94    pub_unap(); pub_uw(fd, requester); pub_uw(fd, "\t" as *u8); pub_uw(fd, hex); pub_uw(fd, "\t" as *u8)
95    pub_unap(); pub_uw(fd, src); pub_uw(fd, "\t" as *u8); pub_uw(fd, dest); pub_uw(fd, "\t" as *u8); pub_uw(fd, policy)
96    pub_uw(fd, "\n" as *u8)
97    sys_close(fd)
98    return 0
99}
100
101// ============================================================================================================
102// R2 -- THE PUBLISHER CORE LOOP: dequeue PENDING -> serialize (fl_acquire per critical section) -> idempotent
103// sha-skip (already-in-ledger -> no-op, rule #10) -> record to the append-only ledger. The fl_acquire makes the
104// check-then-record ATOMIC across concurrent publisher instances, so the same request is NEVER double-published.
105// ============================================================================================================
106func pub_ledger_default() -> *u8 { return "knowledge/publish/ledger.tsv" as *u8 }
107func pub_nap_us(us: i64) -> i64 { let ts: *i64 = sys_mmap(16) as *i64; ts[0]=0; ts[1]=us*1000; __syscall(35, ts as i64, 0, 0, 0, 0, 0); return 0 }
108
109// 1 iff NUL-terminated a == b.
110func pub_streq(a: *u8, b: *u8) -> i64 {
111    var i: i64 = 0
112    while a[i] != (0 as u8) { if a[i] != b[i] { return 0 } i = i + 1 }
113    if b[i] != (0 as u8) { return 0 }
114    return 1
115}
116// copy the idx-th TAB-delimited field of line[0..linelen) into out (NUL-terminated). returns 1.
117func pub_field(line: *u8, linelen: i64, idx: i64, out: *u8) -> i64 {
118    var f: i64 = 0
119    var i: i64 = 0
120    while f < idx { if i < linelen { if line[i] == (9 as u8) { f = f + 1 } i = i + 1 } else { f = idx; i = linelen } }
121    var o: i64 = 0
122    var stop: i64 = 0
123    while i < linelen {
124        if stop == 0 { if line[i] == (9 as u8) { stop = 1 } }
125        if stop == 0 { out[o] = line[i]; o = o + 1 }
126        i = i + 1
127    }
128    out[o] = 0 as u8
129    return 1
130}
131// 1 iff the ledger already has a line whose field0==sha AND field1==dest (the idempotency key).
132func pub_led_has(ledpath: *u8, sha: *u8, dest: *u8) -> i64 {
133    let lenp: *i64 = sys_mmap(8) as *i64
134    let data: *u8 = sys_read_file(ledpath, lenp)
135    if (data as i64) == 0 { return 0 }
136    let n: i64 = lenp[0]
137    let f0: *u8 = sys_mmap(256)
138    let f1: *u8 = sys_mmap(256)
139    var i: i64 = 0
140    var ls: i64 = 0
141    var found: i64 = 0
142    while i < n {
143        if data[i] == (10 as u8) {
144            let line: *u8 = ((data as i64) + ls) as *u8
145            pub_field(line, i - ls, 0, f0)
146            pub_field(line, i - ls, 1, f1)
147            if pub_streq(f0, sha) == 1 { if pub_streq(f1, dest) == 1 { found = 1 } }
148            ls = i + 1
149        }
150        i = i + 1
151    }
152    sys_munmap(data, PUB_MAGIC_4294967296 + 16)
153    return found
154}
155// append a PUBLISHED ledger record (idempotency key first): sha<TAB>dest<TAB>site<TAB>PUBLISHED.
156func pub_record_ledger(ledpath: *u8, sha: *u8, dest: *u8, site: *u8) -> i64 {
157    let rec: *u8 = sys_mmap(PUB_REC_CAP)
158    var o: i64 = 0
159    o = fa_cat(rec, o, sha);  rec[o]=9 as u8; o=o+1
160    o = fa_cat(rec, o, dest); rec[o]=9 as u8; o=o+1
161    o = fa_cat(rec, o, site); rec[o]=9 as u8; o=o+1
162    o = fa_cat(rec, o, "PUBLISHED" as *u8); rec[o]=9 as u8; o=o+1   // outcome (field3)
163    o = fa_catn(rec, o, sys_now_realtime_sec()); rec[o]=9 as u8; o=o+1   // pub_ts (field4) -- enables DORA
164    o = fa_catn(rec, o, 0)                                               // lead_sec (field5); live submit-ts plumb = v2
165    rec[o] = 0 as u8
166    return fa_appendz(ledpath, rec, PUB_REC_CAP)
167}
168// Process the queue ONCE. For each PENDING request: (under lockres if locked==1) skip if already in the ledger,
169// else record it. racewin_us = a deliberate window between check and record so a gate can force the race; the
170// REAL publisher passes 0. Returns the number newly published this pass. lockres = the serialize resource
171// (real publisher: keyed by site; gate: per-wsid). Idempotent: re-running over the same queue publishes 0.
172func pub_run(qpath: *u8, ledpath: *u8, lockres: *u8, locked: i64, racewin_us: i64) -> i64 {
173    let lenp: *i64 = sys_mmap(8) as *i64
174    let data: *u8 = sys_read_file(qpath, lenp)
175    if (data as i64) == 0 { return 0 }
176    let n: i64 = lenp[0]
177    let st: *u8 = sys_mmap(64)
178    let sha: *u8 = sys_mmap(256)
179    let dest: *u8 = sys_mmap(256)
180    let site: *u8 = sys_mmap(256)
181    var i: i64 = 0
182    var ls: i64 = 0
183    var published: i64 = 0
184    while i < n {
185        if data[i] == (10 as u8) {
186            let line: *u8 = ((data as i64) + ls) as *u8
187            let ll: i64 = i - ls
188            pub_field(line, ll, 0, st)
189            if pub_streq(st, "PENDING" as *u8) == 1 {
190                pub_field(line, ll, 1, site)
191                pub_field(line, ll, 3, sha)
192                pub_field(line, ll, 5, dest)
193                var lfd: i64 = 0 - 1
194                if locked == 1 { lfd = fl_acquire(lockres, PUB_MAGIC_100000, 1) }
195                if pub_led_has(ledpath, sha, dest) == 0 {
196                    if racewin_us > 0 { pub_nap_us(racewin_us) }
197                    pub_record_ledger(ledpath, sha, dest, site)
198                    published = published + 1
199                }
200                if locked == 1 { fl_release(lfd) }
201            }
202            ls = i + 1
203        }
204        i = i + 1
205    }
206    return published
207}
208
209// ============================================================================================================
210// R3 -- STAGE + VERIFY-BEFORE-PROMOTE (fail-closed). The publisher NEVER copies an artifact straight to live: it
211// stages it, RE-HASHES the staged bytes, and promotes ONLY if the hash still equals the approved sha. If the
212// artifact was corrupted/swapped between submit and publish (TOCTOU), the hashes differ -> REJECT, live untouched.
213// (The HTTP-level "does it actually serve" check = nx_url_truth_gate, wired at R8 against the real site.)
214// ============================================================================================================
215func pub_exists(path: *u8) -> i64 { let fd: i64 = sys_openat_rd(path); if fd < 0 { return 0 } sys_close(fd); return 1 }
216
217// copy whole file src -> dst (O_TRUNC). returns bytes copied, or <0 (-1 read fail, -2 open-dst fail).
218func pub_copy(src: *u8, dst: *u8) -> i64 {
219    let lenp: *i64 = sys_mmap(8) as *i64
220    let data: *u8 = sys_read_file(src, lenp)
221    if (data as i64) == 0 { return 0 - 1 }
222    let fd: i64 = sys_openat_wr(dst, 0x1a4)
223    if fd < 0 { return 0 - 2 }
224    sys_write(fd, data, lenp[0])
225    sys_close(fd)
226    return lenp[0]
227}
228// 1 iff sha256(file at path) equals expect_hex.
229func pub_sha_match(path: *u8, expect_hex: *u8) -> i64 {
230    let h: *u8 = sys_mmap(72)
231    if pub_sha_file(path, h) == 0 { return 0 }
232    return pub_streq(h, expect_hex)
233}
234// ============================================================================================================
235// R4 -- ATOMIC promote (never half-published) + ROLLBACK (keep the previous live version, restore on demand).
236// ============================================================================================================
237func pub_pcat(base: *u8, suffix: *u8, out: *u8) -> i64 {
238    var o: i64 = 0; var i: i64 = 0
239    while base[i] != (0 as u8) { out[o]=base[i]; o=o+1; i=i+1 }
240    i = 0; while suffix[i] != (0 as u8) { out[o]=suffix[i]; o=o+1; i=i+1 }
241    out[o] = 0 as u8; return o
242}
243func pub_rename(old: *u8, neu: *u8) -> i64 { return __syscall(82, old as i64, neu as i64, 0, 0, 0, 0) }   // x86_64 rename
244
245// ATOMIC promote: copy stage -> live.tmp, then rename(live.tmp, live). rename is atomic on one filesystem, so a
246// concurrent reader sees the OLD or the NEW live -- never a half-written file. returns 1, or <0 on error.
247func pub_promote_atomic(stage: *u8, live: *u8) -> i64 {
248    let tmp: *u8 = sys_mmap(640); pub_pcat(live, ".tmp" as *u8, tmp)
249    if pub_copy(stage, tmp) < 0 { return 0 - 1 }
250    if pub_rename(tmp, live) != 0 { return 0 - 2 }
251    return 1
252}
253// keep the current live as live.prev BEFORE overwriting it (the rollback target). no-op if live absent.
254func pub_backup_prev(live: *u8) -> i64 {
255    if pub_exists(live) == 0 { return 0 }
256    let prev: *u8 = sys_mmap(640); pub_pcat(live, ".prev" as *u8, prev)
257    pub_copy(live, prev)
258    return 1
259}
260// restore live from live.prev (atomically, via a temp + rename). returns 1 if rolled back, 0 if no prev exists.
261func pub_rollback(live: *u8) -> i64 {
262    let prev: *u8 = sys_mmap(640); pub_pcat(live, ".prev" as *u8, prev)
263    if pub_exists(prev) == 0 { return 0 }
264    let tmp: *u8 = sys_mmap(640); pub_pcat(live, ".rbk" as *u8, tmp)
265    if pub_copy(prev, tmp) < 0 { return 0 - 1 }
266    if pub_rename(tmp, live) != 0 { return 0 - 2 }
267    return 1
268}
269
270// STAGE -> VERIFY -> (keep prev) -> ATOMIC PROMOTE. Returns 1 if PROMOTED, 0 if REJECTED (fail-closed, live NOT
271// written), <0 on a copy error. live is updated ONLY after the staged bytes verify against expect_sha, and only
272// via an atomic rename; the prior live is preserved as live.prev so pub_rollback can restore it.
273// R8-asset rung (operator 2026-06-21 "if it cant then build hardware rung up each rung"): create dest PARENT
274// subdirs so the publisher can ship NESTED artifacts (e.g. reader/<slug>/cover.png), not just flat files.
275// mkdirat each path prefix in place (NUL-terminate at each '/', mkdir, restore); EEXIST is harmless. Additive:
276// flat dests have no '/' after the root so this is a no-op for them.
277func pub_mkdirp(path: *u8) -> i64 {
278    var i: i64 = 1
279    while path[i] != (0 as u8) {
280        if path[i] == (0x2f as u8) { path[i] = 0 as u8; __syscall(258, 0-100, path, 0x1ed, 0, 0, 0); path[i] = 0x2f as u8 }
281        i = i + 1
282    }
283    return 0
284}
285func pub_deploy(expect_sha: *u8, src: *u8, stagepath: *u8, livepath: *u8) -> i64 {
286    pub_mkdirp(stagepath)                                         // ensure dest subdirs exist (nested artifacts)
287    pub_mkdirp(livepath)
288    if pub_copy(src, stagepath) < 0 { return 0 - 1 }              // STAGE
289    if pub_sha_match(stagepath, expect_sha) == 0 { return 0 }     // VERIFY -> FAIL-CLOSED: do not promote
290    pub_backup_prev(livepath)                                     // KEEP-PREVIOUS (rollback target)
291    if pub_promote_atomic(stagepath, livepath) < 0 { return 0 - 2 }   // ATOMIC PROMOTE
292    return 1
293}
294
295// ============================================================================================================
296// R7 -- THE NO-BYPASS LAW (SEAM audit). Workstreams must publish via pub_submit ONLY; the raw transport organs
297// (nx_aw_send / nx_aw_push) are publisher-INTERNAL. This audit flags any source that references the raw transport
298// and is NOT allowlisted (the publisher itself) -> "no more direct publishing from workstreams" is made true by
299// a MECHANICAL gate, not a promise (mirrors the writer's mode-SEAM audit).
300// ============================================================================================================
301// substring search ('match' is a NishiLang reserved word -> use 'hit').
302func pub_substr(hay: *u8, haylen: i64, needle: *u8) -> i64 {
303    var nl: i64 = 0; while needle[nl] != (0 as u8) { nl = nl + 1 }
304    if nl == 0 { return 0 }
305    var i: i64 = 0
306    while i + nl <= haylen {
307        var hit: i64 = 1; var j: i64 = 0
308        while j < nl { if hay[i+j] != needle[j] { hit = 0; j = nl } else { j = j + 1 } }
309        if hit == 1 { return 1 }
310        i = i + 1
311    }
312    return 0
313}
314func pub_file_has(path: *u8, needle: *u8) -> i64 {
315    let lenp: *i64 = sys_mmap(8) as *i64
316    let data: *u8 = sys_read_file(path, lenp)
317    if (data as i64) == 0 { return 0 }
318    return pub_substr(data, lenp[0], needle)
319}
320// 1 iff the file is a direct-publish BYPASS: references a raw-transport organ AND is not allowlisted.
321func pub_audit_file(path: *u8, allowlisted: i64) -> i64 {
322    if allowlisted == 1 { return 0 }
323    if pub_file_has(path, "nx_aw_send" as *u8) == 1 { return 1 }
324    if pub_file_has(path, "nx_aw_push" as *u8) == 1 { return 1 }
325    return 0
326}
327// STRICT no-bypass audit: workstreams may touch publishing ONLY via pub_submit. This flags a non-allowlisted source
328// that references ANY live-touching primitive -- raw transport OR the publisher's own deploy/promote/rollback/
329// progressive-delivery/ledger internals -- so the law covers the WHOLE dangerous surface, not just transport.
330// Reads the file ONCE, scans all forbidden symbols. (allowlisted = the sanctioned publisher family, set by caller.)
331func pub_audit_file_strict(path: *u8, allowlisted: i64) -> i64 {
332    if allowlisted == 1 { return 0 }
333    let lenp: *i64 = sys_mmap(8) as *i64
334    let data: *u8 = sys_read_file(path, lenp)
335    if (data as i64)==0 { return 0 }
336    let n: i64 = lenp[0]
337    if pub_substr(data, n, "nx_aw_send" as *u8)==1 { return 1 }          // raw transport
338    if pub_substr(data, n, "nx_aw_push" as *u8)==1 { return 1 }
339    if pub_substr(data, n, "pub_deploy" as *u8)==1 { return 1 }          // direct stage->promote
340    if pub_substr(data, n, "pub_promote_atomic" as *u8)==1 { return 1 }  // direct atomic live-write
341    if pub_substr(data, n, "pub_rollback" as *u8)==1 { return 1 }        // direct rollback
342    if pub_substr(data, n, "pub_blue_green_flip" as *u8)==1 { return 1 } // direct blue-green flip
343    if pub_substr(data, n, "pub_canary_rollout" as *u8)==1 { return 1 }  // direct canary
344    if pub_substr(data, n, "pub_record_ledger" as *u8)==1 { return 1 }   // direct ledger write
345    if pub_substr(data, n, "pub_record_outcome" as *u8)==1 { return 1 }
346    return 0
347}
348// scan a NUL-terminated forbidden symbol over a code buffer -> 1 if found. (the strict forbidden set, one place.)
349func pub_forbidden_in(code: *u8, n: i64) -> i64 {
350    if pub_substr(code, n, "nx_aw_send" as *u8)==1 { return 1 }
351    if pub_substr(code, n, "nx_aw_push" as *u8)==1 { return 1 }
352    if pub_substr(code, n, "pub_deploy" as *u8)==1 { return 1 }
353    if pub_substr(code, n, "pub_promote_atomic" as *u8)==1 { return 1 }
354    if pub_substr(code, n, "pub_rollback" as *u8)==1 { return 1 }
355    if pub_substr(code, n, "pub_blue_green_flip" as *u8)==1 { return 1 }
356    if pub_substr(code, n, "pub_canary_rollout" as *u8)==1 { return 1 }
357    if pub_substr(code, n, "pub_record_ledger" as *u8)==1 { return 1 }
358    if pub_substr(code, n, "pub_record_outcome" as *u8)==1 { return 1 }
359    return 0
360}
361// COMMENT-AWARE strict audit: strips // line-comments to end-of-line BEFORE matching, so a comment-only mention of a
362// primitive (e.g. nx_runpath's WIRING-PLAN comment) is NOT a false bypass. Reduces the coarse audit's false positives.
363// (Limitation: a "//" inside a same-line string literal would also be stripped -> rare false-negative; noted.)
364func pub_audit_src_strict(path: *u8, allowlisted: i64) -> i64 {
365    if allowlisted == 1 { return 0 }
366    let lenp: *i64 = sys_mmap(8) as *i64
367    let data: *u8 = sys_read_file(path, lenp)
368    if (data as i64)==0 { return 0 }
369    let n: i64 = lenp[0]
370    let code: *u8 = sys_mmap(n + 16)
371    var i: i64=0; var o: i64=0; var incomment: i64=0
372    while i < n {
373        let c: i64 = data[i] & 0xff
374        if c == 10 { incomment=0; code[o]=10 as u8; o=o+1 }
375        else {
376            if incomment == 0 {
377                if c == 47 { if i+1 < n { if (data[i+1] & 0xff) == 47 { incomment = 1 } } }
378                if incomment == 0 { code[o]=c as u8; o=o+1 }
379            }
380        }
381        i = i + 1
382    }
383    return pub_forbidden_in(code, o)
384}
385
386// ============================================================================================================
387// R8 -- the PRODUCTION loop: compose the whole pipeline. dequeue PENDING → serialize → idempotent-skip → DEPLOY
388// (stage→verify→atomic-promote, R3/R4) → record ledger. This is what a workstream's pub_submit'd request flows
389// through; the only remaining step to go truly live is the sovereign transport push of liveroot → the host.
390// ============================================================================================================
391func pub_join(root: *u8, name: *u8, out: *u8) -> i64 {
392    var o: i64 = pub_pcat(root, "/" as *u8, out)
393    var i: i64 = 0
394    while name[i] != (0 as u8) { out[o]=name[i]; o=o+1; i=i+1 }
395    out[o] = 0 as u8
396    return o
397}
398// publish ONE request end-to-end. returns 1 published, 0 idempotent-skip, <0 deploy reject/error.
399func pub_publish_one(sha: *u8, src: *u8, dest: *u8, site: *u8, stageroot: *u8, liveroot: *u8, ledpath: *u8) -> i64 {
400    if pub_led_has(ledpath, sha, dest) == 1 { return 0 }              // idempotent (rule #10)
401    let stagepath: *u8 = sys_mmap(700); pub_join(stageroot, dest, stagepath)
402    let livepath: *u8 = sys_mmap(700); pub_join(liveroot, dest, livepath)
403    let r: i64 = pub_deploy(sha, src, stagepath, livepath)            // STAGE→VERIFY→ATOMIC-PROMOTE
404    if r == 1 { pub_record_ledger(ledpath, sha, dest, site); return 1 }
405    return r
406}
407// process the queue, publishing each PENDING request under the serialize lock. returns #published this pass.
408func pub_run_full(qpath: *u8, ledpath: *u8, stageroot: *u8, liveroot: *u8, lockres: *u8) -> i64 {
409    let lenp: *i64 = sys_mmap(8) as *i64
410    let data: *u8 = sys_read_file(qpath, lenp)
411    if (data as i64) == 0 { return 0 }
412    let n: i64 = lenp[0]
413    let st: *u8 = sys_mmap(64)
414    let sha: *u8 = sys_mmap(256)
415    let src: *u8 = sys_mmap(512)
416    let dest: *u8 = sys_mmap(256)
417    let site: *u8 = sys_mmap(256)
418    var i: i64 = 0
419    var ls: i64 = 0
420    var published: i64 = 0
421    while i < n {
422        if data[i] == (10 as u8) {
423            let line: *u8 = ((data as i64) + ls) as *u8
424            let ll: i64 = i - ls
425            pub_field(line, ll, 0, st)
426            if pub_streq(st, "PENDING" as *u8) == 1 {
427                pub_field(line, ll, 1, site)
428                pub_field(line, ll, 3, sha)
429                pub_field(line, ll, 4, src)
430                pub_field(line, ll, 5, dest)
431                let lfd: i64 = fl_acquire(lockres, PUB_MAGIC_100000, 1)
432                let pr: i64 = pub_publish_one(sha, src, dest, site, stageroot, liveroot, ledpath)
433                fl_release(lfd)
434                if pr == 1 { published = published + 1 }
435            }
436            ls = i + 1
437        }
438        i = i + 1
439    }
440    return published
441}
442
443// ============================================================================================================
444// R6 -- RELEASE POLICY / approval gate. Internal publishes proceed; OUTWARD-facing publishes require an explicit
445// operator approval of the EXACT artifact (an approval token keyed by sha, created out-of-band) -> the publisher
446// will not push outward, unapproved content BY CONSTRUCTION. This is the operator-gating made mechanical.
447// ============================================================================================================
448func pub_policy_allows(policy: *u8, sha: *u8, approval_dir: *u8) -> i64 {
449    if pub_streq(policy, "internal" as *u8) == 1 { return 1 }   // internal lane: no approval needed
450    let apath: *u8 = sys_mmap(700); pub_join(approval_dir, sha, apath)
451    if pub_exists(apath) == 1 { return 1 }                       // outward: approved iff the operator-ok token exists
452    return 0                                                     // outward + unapproved -> HELD (fail-closed)
453}
454// the operator approves a specific artifact (by content sha) out-of-band: create approval_dir/<sha>.
455func pub_approve(sha: *u8, approval_dir: *u8) -> i64 {
456    let apath: *u8 = sys_mmap(700); pub_join(approval_dir, sha, apath)
457    let fd: i64 = sys_openat_wr(apath, 0x1a4); if fd < 0 { return 0 }
458    sys_close(fd); return 1
459}
460
461// GOVERNED production loop = pub_run_full + the R6 policy gate enforced IN the loop. An OUTWARD request is HELD
462// (left PENDING, not published) until the operator approves its exact artifact; internal requests proceed. This
463// is the full pipeline a real outward publish runs through -- operator-gating by construction, not by promise.
464func pub_run_governed(qpath: *u8, ledpath: *u8, stageroot: *u8, liveroot: *u8, lockres: *u8, approval_dir: *u8) -> i64 {
465    let lenp: *i64 = sys_mmap(8) as *i64
466    let data: *u8 = sys_read_file(qpath, lenp)
467    if (data as i64) == 0 { return 0 }
468    let n: i64 = lenp[0]
469    let st: *u8 = sys_mmap(64)
470    let sha: *u8 = sys_mmap(256)
471    let src: *u8 = sys_mmap(512)
472    let dest: *u8 = sys_mmap(256)
473    let site: *u8 = sys_mmap(256)
474    let policy: *u8 = sys_mmap(64)
475    var i: i64 = 0
476    var ls: i64 = 0
477    var published: i64 = 0
478    while i < n {
479        if data[i] == (10 as u8) {
480            let line: *u8 = ((data as i64) + ls) as *u8
481            let ll: i64 = i - ls
482            pub_field(line, ll, 0, st)
483            if pub_streq(st, "PENDING" as *u8) == 1 {
484                pub_field(line, ll, 1, site)
485                pub_field(line, ll, 3, sha)
486                pub_field(line, ll, 4, src)
487                pub_field(line, ll, 5, dest)
488                pub_field(line, ll, 6, policy)
489                if pub_policy_allows(policy, sha, approval_dir) == 1 {       // R6: outward needs operator-ok
490                    let lfd: i64 = fl_acquire(lockres, PUB_MAGIC_100000, 1)
491                    let pr: i64 = pub_publish_one(sha, src, dest, site, stageroot, liveroot, ledpath)
492                    fl_release(lfd)
493                    if pr == 1 { published = published + 1 }
494                }
495            }
496            ls = i + 1
497        }
498        i = i + 1
499    }
500    return published
501}
502
503// ============================================================================================================
504// R10 -- UN-STUCKABLE DRAIN (operator: "make sure the publisher cant get stuck again as a whole even if
505// somethings get stuck"). pub_run_governed publishes each item IN-PROCESS, so a single item that CRASHES
506// pub_publish_one -- a malformed queue field overflowing a fixed buffer, a pathological dest, a corrupt source --
507// takes the WHOLE drain down, and every later item is stuck forever (head-of-line blocking = the stuck root cause).
508// pub_run_resilient FORKS a child per item: the child does ALL the risky work (extract the user-data fields,
509// policy-check, stage/verify/promote) and exits; the PARENT only reaps it and CONTINUES, no matter how the child
510// died. A crashed (signalled) or failed (exit 2) item is QUARANTINED to a dead-letter file -- with its RAW queue
511// line, so the parent never has to parse a hostile field -- for the autonomous Dr+Engineer self-heal
512// (nx_pub_selfheal) to root-cause. It can NEVER block the rest. pub_deploy's atomic promote already guarantees a
513// crashed child cannot leave a half-written live file. ADDITIVE: pub_run_governed is unchanged; callers opt in.
514// ============================================================================================================
515func pub_dlcat(rec: *u8, o: i64, s: *u8) -> i64 { var w: i64=o; var i: i64=0; while s[i]!=(0 as u8){rec[w]=s[i];w=w+1;i=i+1} return w }
516func pub_dlnum(rec: *u8, o: i64, v: i64) -> i64 { var w: i64=o; var m: i64=v; if m==0{rec[w]=48 as u8;return w+1} let t:*u8=sys_mmap(24); var k:i64=0; while m>0{t[k]=(48+(m%10)) as u8;m=m/10;k=k+1} var j:i64=0; while j<k{rec[w]=t[k-1-j];w=w+1;j=j+1} return w }
517
518// quarantine one stuck item by its RAW queue line (no field parsing in the crash-proof parent) + a reason+detail.
519func pub_dl_raw(dlpath: *u8, reason: *u8, detail: i64, line: *u8, ll: i64) -> i64 {
520    let rec: *u8 = sys_mmap(PUB_REC_CAP)
521    var o: i64 = 0
522    o = pub_dlcat(rec, o, "STUCK" as *u8); rec[o]=9 as u8; o=o+1
523    o = pub_dlcat(rec, o, reason);         rec[o]=9 as u8; o=o+1
524    o = pub_dlnum(rec, o, detail);         rec[o]=9 as u8; o=o+1
525    var k: i64 = 0
526    while k < ll { if o < PUB_REC_CAP - 2 { rec[o]=line[k]; o=o+1 } k=k+1 }
527    rec[o]=10 as u8; o=o+1; rec[o]=0 as u8
528    fa_appendz(dlpath, rec, PUB_REC_CAP)
529    return 1
530}
531
532// UN-STUCKABLE governed drain. outcounts[0..4] = published, held(outward-unapproved), failed, crashed, skipped.
533// Returns #published. NO single item can halt the loop -- each is published in its own process.
534func pub_run_resilient(qpath: *u8, ledpath: *u8, stageroot: *u8, liveroot: *u8, lockres: *u8, approval_dir: *u8, dlpath: *u8, outcounts: *i64) -> i64 {
535    outcounts[0]=0; outcounts[1]=0; outcounts[2]=0; outcounts[3]=0; outcounts[4]=0
536    let dlt: i64 = sys_openat_wr(dlpath, 0x1a4); if dlt >= 0 { sys_close(dlt) }   // fresh dead-letter for THIS pass (no accumulation -> the self-heal never re-acts on stale records)
537    let lenp: *i64 = sys_mmap(8) as *i64
538    let data: *u8 = sys_read_file(qpath, lenp)
539    if (data as i64) == 0 { return 0 }
540    let n: i64 = lenp[0]
541    let st: *u8 = sys_mmap(64)
542    var i: i64 = 0; var ls: i64 = 0
543    while i < n {
544        if data[i] == (10 as u8) {
545            let line: *u8 = ((data as i64) + ls) as *u8
546            let ll: i64 = i - ls
547            if ll > 0 {
548                pub_field(line, ll, 0, st)                                 // status only (machine-written, short)
549                if pub_streq(st, "PENDING" as *u8) == 1 {
550                    let lfd: i64 = fl_acquire(lockres, PUB_MAGIC_100000, 1)
551                    let pid: i64 = sys_fork()
552                    if pid == 0 {
553                        // CHILD: every hostile/heavy operation is isolated here.
554                        let sha: *u8 = sys_mmap(256); let src: *u8 = sys_mmap(512); let dest: *u8 = sys_mmap(256)
555                        let site: *u8 = sys_mmap(256); let policy: *u8 = sys_mmap(64)
556                        pub_field(line, ll, 1, site); pub_field(line, ll, 3, sha); pub_field(line, ll, 4, src)
557                        pub_field(line, ll, 5, dest); pub_field(line, ll, 6, policy)
558                        if pub_policy_allows(policy, sha, approval_dir) == 0 { sys_exit(3) }   // outward unapproved -> HELD
559                        if pub_led_has(ledpath, sha, dest) == 1 { sys_exit(1) }                // already published (this exact sha) -> SKIP
560                        let cur: *u8 = sys_mmap(72)                                            // STALE-DUP guard: if the source's CURRENT
561                        if pub_sha_file(src, cur) == 1 { if pub_led_has(ledpath, cur, dest) == 1 { sys_exit(1) } }   // content is already live at dest, the queued (stale) sha is just log cruft -> SKIP cleanly, do not fail/quarantine
562                        let pr: i64 = pub_publish_one(sha, src, dest, site, stageroot, liveroot, ledpath)
563                        if pr == 1 { sys_exit(0) }                                             // PUBLISHED
564                        sys_exit(2)                                                            // FAILED (verify/copy/promote)
565                    }
566                    let stx: *i64 = sys_mmap(16) as *i64; stx[0] = 0
567                    sys_wait4(pid, stx, 0)
568                    fl_release(lfd)
569                    let raw: i64 = stx[0]
570                    let sig: i64 = raw & 0x7f
571                    if sig != 0 {
572                        outcounts[3] = outcounts[3] + 1
573                        pub_dl_raw(dlpath, "CRASH" as *u8, sig, line, ll)
574                    } else {
575                        let code: i64 = (raw >> 8) & 0xff
576                        if code == 0 { outcounts[0] = outcounts[0] + 1 }
577                        else { if code == 1 { outcounts[4] = outcounts[4] + 1 }
578                        else { if code == 3 { outcounts[1] = outcounts[1] + 1 }
579                        else { outcounts[2] = outcounts[2] + 1; pub_dl_raw(dlpath, "FAIL" as *u8, code, line, ll) } } }
580                    }
581                }
582            }
583            ls = i + 1
584        }
585        i = i + 1
586    }
587    return outcounts[0]
588}
589
590// ============================================================================================================
591// R11 -- AUTONOMOUS Dr+ENGINEER SELF-HEAL over the dead-letter (operator: "flag the dr and engineer to partner on
592// fixing the root causes of the stuck autonomously and automatically"). The ENGINEER re-diagnoses each quarantined
593// item AGAINST THE LIVE FILESYSTEM (grounded, not guessed -- rule #4): a CRASH = a publisher-code bug on that input;
594// a FAIL with a missing source = the artifact moved/was deleted; a FAIL whose source no longer hashes to the
595// submitted sha = the artifact changed after submit (TOCTOU); otherwise a transient deploy error. The DOCTOR then
596// assigns the remedy: a code-bug is FLAGGED to the engineer (and stays quarantined so it can't crash again); a
597// missing/changed source is FLAGGED to the owning workstream to re-stage/re-submit; a transient error is left to
598// RETRY on the next idempotent drain pass. Output is a structured report the two roles act on -- automatic, no human.
599// ============================================================================================================
600func pub_sh_cause(reason: *u8, sha: *u8, src: *u8, ledpath: *u8, dest: *u8) -> i64 {
601    if pub_streq(reason, "CRASH" as *u8) == 1 { return 1 }                    // PUBLISHER-BUG (code crashed on this input)
602    if pub_exists(src) == 0 { return 2 }                                      // SRC-MISSING
603    let h: *u8 = sys_mmap(72)
604    if pub_sha_file(src, h) == 1 { if pub_streq(h, sha) == 0 {                // current content differs from the QUEUED sha
605        if pub_led_has(ledpath, h, dest) == 1 { return 5 }                    // ALREADY-LIVE: current content already published to dest -> stale dup (prune, do NOT republish)
606        return 3                                                             // ARTIFACT-CHANGED: current content not yet live -> republish
607    } }
608    return 4                                                                  // DEPLOY-ERROR (sha matches but failed -> transient retry)
609}
610func pub_sh_cause_name(c: i64) -> *u8 { if c==1 { return "PUBLISHER-BUG" as *u8 } if c==2 { return "SRC-MISSING" as *u8 } if c==3 { return "ARTIFACT-CHANGED" as *u8 } if c==5 { return "ALREADY-LIVE" as *u8 } return "DEPLOY-ERROR" as *u8 }
611func pub_sh_action(c: i64) -> *u8 { if c==1 { return "FLAG-ENGINEER" as *u8 } if c==2 { return "FLAG-OWNER-RESTAGE" as *u8 } if c==3 { return "FLAG-OWNER-RESUBMIT" as *u8 } if c==5 { return "PRUNE-STALE-DUP" as *u8 } return "RETRY-NEXT-PASS" as *u8 }
612
613// process every STUCK record in the dead-letter -> diagnose + remedy, append to reportpath. returns #processed.
614// dead-letter layout (pub_dl_raw): STUCK<TAB>reason<TAB>detail<TAB>PENDING<TAB>site<TAB>req<TAB>sha<TAB>src<TAB>dest<TAB>policy
615// DOCTOR auto-remedy: an INTERNAL artifact that merely CHANGED since submit is re-queued with its CURRENT sha (the
616// owning ws's clear intent to publish the current bytes) -- BOUNDED to once per (sha,dest) via resub_led so it can
617// NEVER loop; OUTWARD changed/everything-else is FLAGGED (outward still needs operator approval, by construction).
618// requeue_path==0 disables auto-resubmit (pure diagnosis, e.g. a dry-run).
619func pub_selfheal(dlpath: *u8, reportpath: *u8, requeue_path: *u8, resub_led: *u8, ledpath: *u8) -> i64 {
620    let lenp: *i64 = sys_mmap(8) as *i64
621    let data: *u8 = sys_read_file(dlpath, lenp)
622    if (data as i64) == 0 { return 0 }
623    let n: i64 = lenp[0]
624    let st: *u8 = sys_mmap(64); let reason: *u8 = sys_mmap(64); let sha: *u8 = sys_mmap(256)
625    let src: *u8 = sys_mmap(512); let dest: *u8 = sys_mmap(256); let policy: *u8 = sys_mmap(64)
626    let site: *u8 = sys_mmap(256); let req: *u8 = sys_mmap(256)
627    var i: i64 = 0; var ls: i64 = 0; var processed: i64 = 0
628    while i < n {
629        if data[i] == (10 as u8) {
630            let line: *u8 = ((data as i64) + ls) as *u8
631            let ll: i64 = i - ls
632            if ll > 0 {
633                pub_field(line, ll, 0, st)
634                if pub_streq(st, "STUCK" as *u8) == 1 {
635                    pub_field(line, ll, 1, reason)
636                    pub_field(line, ll, 6, sha); pub_field(line, ll, 7, src); pub_field(line, ll, 8, dest)
637                    let cause: i64 = pub_sh_cause(reason, sha, src, ledpath, dest)
638                    var action: *u8 = pub_sh_action(cause)
639                    if cause == 3 {                                            // ARTIFACT-CHANGED (genuinely not-yet-live)
640                        pub_field(line, ll, 9, policy)
641                        if (requeue_path as i64) != 0 { if pub_streq(policy, "internal" as *u8) == 1 {
642                            if pub_led_has(resub_led, sha, dest) == 1 { action = "ALREADY-RESUBMITTED" as *u8 }
643                            else {
644                                pub_field(line, ll, 4, site); pub_field(line, ll, 5, req)
645                                pub_submit_to(requeue_path, src, dest, site, req, "internal" as *u8)   // re-queue with the CURRENT sha
646                                pub_record_ledger(resub_led, sha, dest, site)                          // bound: never resubmit this stale (sha,dest) again
647                                action = "AUTO-RESUBMITTED" as *u8
648                            }
649                        } }
650                    }
651                    let rec: *u8 = sys_mmap(PUB_REC_CAP); var o: i64 = 0
652                    o = pub_dlcat(rec, o, pub_sh_cause_name(cause)); rec[o]=9 as u8; o=o+1
653                    o = pub_dlcat(rec, o, action);                   rec[o]=9 as u8; o=o+1
654                    o = pub_dlcat(rec, o, dest);                     rec[o]=9 as u8; o=o+1
655                    o = pub_dlcat(rec, o, sha)
656                    rec[o]=10 as u8; o=o+1; rec[o]=0 as u8
657                    fa_appendz(reportpath, rec, PUB_REC_CAP)
658                    processed = processed + 1
659                }
660            }
661            ls = i + 1
662        }
663        i = i + 1
664    }
665    return processed
666}
667
668// ============================================================================================================
669// R9 -- POST-DEPLOY HTTP SMOKE TEST + HEALTH-GATED AUTO-ROLLBACK. pub_deploy proves the staged BYTES hash to the
670// approved sha; it CANNOT prove the live URL actually SERVES them. That blind spot is exactly the sites.elf class
671// of failure: the deploy "succeeded" on disk while the service did not serve (down / wedged / wrong path). pub_smoke
672// does a REAL loopback HTTP GET of the served path and verifies the SERVED body hashes to the approved sha;
673// pub_health_gate auto-rolls-back to the last-good live if the smoke fails -> the publisher never leaves a broken
674// URL live AND never falsely reports a publish that the world can't actually fetch. Sovereign client (our own
675// socket syscalls; mirrors nx_pub_serve_gate's cg_get with a read-to-close loop so the body sha is exact).
676// ============================================================================================================
677func pub_nap_ms(ms: i64) -> i64 { let ts: *i64 = sys_mmap(16) as *i64; ts[0]=0; ts[1]=ms*PUB_MAGIC_1000000; __syscall(35, ts as i64, 0, 0, 0, 0, 0); return 0 }
678
679// sha256(data[0..len)) -> 64 lowercase hex chars into outhex (NUL-terminated). byte-range twin of pub_sha_file.
680func pub_hash_bytes(data: *u8, len: i64, outhex: *u8) -> i64 {
681    let dig: *u8 = sys_mmap(32)
682    sha256_digest(data, len, dig)
683    var i: i64 = 0
684    while i < 32 {
685        let b: i64 = dig[i] & 0xff
686        outhex[i*2]     = pub_hexd((b >> 4) & 0xf) as u8
687        outhex[i*2 + 1] = pub_hexd(b & 0xf) as u8
688        i = i + 1
689    }
690    outhex[64] = 0 as u8
691    return 1
692}
693// offset of the HTTP body (just past the CRLFCRLF header terminator) in resp[0..n), or -1 if not found.
694func pub_body_off(resp: *u8, n: i64) -> i64 {
695    var i: i64 = 0
696    while i + 3 < n {
697        if resp[i]==(13 as u8) { if resp[i+1]==(10 as u8) { if resp[i+2]==(13 as u8) { if resp[i+3]==(10 as u8) { return i + 4 } } } }
698        i = i + 1
699    }
700    return 0 - 1
701}
702// sovereign loopback HTTP/1.0 client: connect 127.0.0.1:port (retry until the server is up), GET /<path>, read the
703// FULL response (loop until the server closes / RCVTIMEO). returns bytes read, or -1 if the server never answered.
704func pub_http_get(port: i64, path: *u8, resp: *u8, cap: i64) -> i64 {
705    var tries: i64 = 0
706    while tries < 80 {
707        let fd: i64 = sys_socket(2, 1, 0)
708        if fd >= 0 {
709            let tv: *u8 = sys_mmap(16); tv[0]=2 as u8; var tz: i64=1; while tz<16 { tv[tz]=0 as u8; tz=tz+1 }   // 2s
710            sys_setsockopt(fd, 1, 20, tv, 16)                  // SO_RCVTIMEO -> sys_read can't block forever
711            let addr: *u8 = sys_mmap(16)
712            addr[0]=2 as u8; addr[1]=0 as u8; addr[2]=((port>>8)&0xff) as u8; addr[3]=(port&0xff) as u8
713            addr[4]=127 as u8; addr[5]=0 as u8; addr[6]=0 as u8; addr[7]=1 as u8
714            var z: i64=8; while z<16 { addr[z]=0 as u8; z=z+1 }
715            if nx_connect_bounded(fd, addr, 16, NX_CONN_DEFAULT_MS) == 0 {
716                let r: *u8 = sys_mmap(512); var o: i64=0
717                o = fa_cat(r, o, "GET /" as *u8); o = fa_cat(r, o, path); o = fa_cat(r, o, " HTTP/1.0\r\nConnection: close\r\n\r\n" as *u8)
718                sys_write(fd, r, o)
719                var got: i64 = 0
720                var done: i64 = 0
721                while done == 0 {
722                    let rd: i64 = sys_read(fd, ((resp as i64)+got) as *u8, cap-got)
723                    if rd <= 0 { done = 1 } else { got = got + rd; if got >= cap { done = 1 } }
724                }
725                sys_close(fd)
726                return got
727            }
728            sys_close(fd)
729        }
730        pub_nap_ms(20)
731        tries = tries + 1
732    }
733    return 0 - 1
734}
735// THE POST-DEPLOY SMOKE TEST: GET the served path; pass iff status is 200 AND the served body hashes to expect_sha.
736// returns 1 = the live URL truly serves the approved artifact, 0 = it does not (no answer / non-200 / wrong bytes).
737func pub_smoke(port: i64, urlpath: *u8, expect_sha: *u8) -> i64 {
738    let resp: *u8 = sys_mmap(PUB_MAGIC_262144)
739    let n: i64 = pub_http_get(port, urlpath, resp, PUB_MAGIC_262144)
740    if n <= 0 { return 0 }                                       // server down / wedged -> serve failure
741    if pub_substr(resp, n, "200 OK" as *u8) == 0 { return 0 }    // non-200 (e.g. 404 wrong path) -> serve failure
742    let bo: i64 = pub_body_off(resp, n)
743    if bo < 0 { return 0 }
744    let h: *u8 = sys_mmap(72)
745    pub_hash_bytes(((resp as i64)+bo) as *u8, n - bo, h)
746    return pub_streq(h, expect_sha)                              // served body must hash to the approved artifact
747}
748// HEALTH GATE: after a promote, confirm the live URL serves the approved bytes; if not, AUTO-ROLLBACK to last-good.
749// returns 1 = healthy (smoke passed), 0 = unhealthy (smoke failed -> rolled back; the URL is not left broken).
750func pub_health_gate(expect_sha: *u8, livepath: *u8, port: i64, urlpath: *u8) -> i64 {
751    if pub_smoke(port, urlpath, expect_sha) == 1 { return 1 }
752    pub_rollback(livepath)
753    return 0
754}
755// THE SAFE DEPLOY: stage -> verify(sha) -> backup-prev -> atomic-promote (pub_deploy), THEN post-deploy HTTP smoke
756// at the live URL; if the served result doesn't match, auto-rollback. returns 1 = PUBLISHED + served-verified,
757// 0 = REJECTED (pre-promote verify failed, OR the live URL didn't serve it -> rolled back), <0 = copy error.
758func pub_deploy_smoke(expect_sha: *u8, src: *u8, stagepath: *u8, livepath: *u8, port: i64, urlpath: *u8) -> i64 {
759    let d: i64 = pub_deploy(expect_sha, src, stagepath, livepath)
760    if d != 1 { return d }                                       // pre-promote verify failed/error: live untouched
761    return pub_health_gate(expect_sha, livepath, port, urlpath)  // 1 = healthy, 0 = unhealthy (rolled back)
762}
763
764// ============================================================================================================
765// R10 -- OBSERVABILITY: STATUS (what is queued / published / ready / held) + DRY-RUN (plan a run, NO side effects).
766// Both are READ-ONLY over the queue + ledger. dry-run is the operator's "what will this do?" before a real publish:
767// it computes the EXACT set pub_run_governed would publish without writing live or appending the ledger.
768// ============================================================================================================
769// number of records (lines) in the ledger = things PUBLISHED so far.
770func pub_count_ledger(ledpath: *u8) -> i64 {
771    let lenp: *i64 = sys_mmap(8) as *i64
772    let data: *u8 = sys_read_file(ledpath, lenp)
773    if (data as i64)==0 { return 0 }
774    let n: i64 = lenp[0]
775    var i: i64=0; var c: i64=0
776    while i<n { if data[i]==(10 as u8) { c=c+1 } i=i+1 }
777    return c
778}
779// THE PLAN PREDICATE (shared by dry-run; mirrors the wet pub_run_governed decision): a PENDING request WOULD be
780// published iff policy allows (internal, or outward+approved) AND it is not already in the ledger. Fills out[]:
781// out[0]=total PENDING, out[1]=ready(would publish), out[2]=held(policy denied), out[3]=idempotent-skip(in ledger).
782func pub_plan(qpath: *u8, ledpath: *u8, approval_dir: *u8, out: *i64) -> i64 {
783    out[0]=0; out[1]=0; out[2]=0; out[3]=0
784    let lenp: *i64 = sys_mmap(8) as *i64
785    let data: *u8 = sys_read_file(qpath, lenp)
786    if (data as i64)==0 { return 0 }
787    let n: i64 = lenp[0]
788    let st: *u8 = sys_mmap(64); let sha: *u8 = sys_mmap(256); let dest: *u8 = sys_mmap(256); let policy: *u8 = sys_mmap(64)
789    var i: i64=0; var ls: i64=0
790    while i<n {
791        if data[i]==(10 as u8) {
792            let line: *u8 = ((data as i64)+ls) as *u8
793            let ll: i64 = i-ls
794            pub_field(line, ll, 0, st)
795            if pub_streq(st, "PENDING" as *u8)==1 {
796                out[0]=out[0]+1
797                pub_field(line, ll, 3, sha)
798                pub_field(line, ll, 5, dest)
799                pub_field(line, ll, 6, policy)
800                if pub_policy_allows(policy, sha, approval_dir)==1 {
801                    if pub_led_has(ledpath, sha, dest)==1 { out[3]=out[3]+1 } else { out[1]=out[1]+1 }
802                } else { out[2]=out[2]+1 }
803            }
804            ls=i+1
805        }
806        i=i+1
807    }
808    return out[1]
809}
810// DRY-RUN: how many requests a real pub_run_governed WOULD publish now, with ZERO side effects (no live write, no
811// ledger append). Equals the wet run's published count (artifacts were sha-verified at submit). out[] = full plan.
812func pub_dryrun(qpath: *u8, ledpath: *u8, approval_dir: *u8, out: *i64) -> i64 {
813    return pub_plan(qpath, ledpath, approval_dir, out)
814}
815// STATUS snapshot. out[0]=PENDING requests, out[1]=PUBLISHED(ledger records), out[2]=ready, out[3]=held,
816// out[4]=idempotent-already-published. (Pure read; the live dashboard is this data, formatted by the caller.)
817func pub_status(qpath: *u8, ledpath: *u8, approval_dir: *u8, out: *i64) -> i64 {
818    let p: *i64 = sys_mmap(64) as *i64
819    pub_plan(qpath, ledpath, approval_dir, p)
820    out[0]=p[0]; out[2]=p[1]; out[3]=p[2]; out[4]=p[3]
821    out[1]=pub_count_ledger(ledpath)
822    return 0
823}
824
825// ============================================================================================================
826// R11 -- AUDIT QUERY (provenance/history lookups over the append-only ledger) + CHANGELOG (release notes). All
827// READ-ONLY: the ledger IS the source of truth for what was published, when-ordered, with what content sha.
828// ============================================================================================================
829// is `dest` currently published? returns 1 + the LATEST sha into out_sha (ledger is append-only, last match wins),
830// else 0 (out_sha untouched). This is the "what is live at this path, and from which artifact?" provenance query.
831func pub_audit_query(ledpath: *u8, dest: *u8, out_sha: *u8) -> i64 {
832    let lenp: *i64 = sys_mmap(8) as *i64
833    let data: *u8 = sys_read_file(ledpath, lenp)
834    if (data as i64)==0 { return 0 }
835    let n: i64 = lenp[0]
836    let f0: *u8 = sys_mmap(256); let f1: *u8 = sys_mmap(256)
837    var i: i64=0; var ls: i64=0; var found: i64=0
838    while i<n {
839        if data[i]==(10 as u8) {
840            let line: *u8 = ((data as i64)+ls) as *u8
841            pub_field(line, i-ls, 0, f0)
842            pub_field(line, i-ls, 1, f1)
843            if pub_streq(f1, dest)==1 {
844                var k: i64=0; while f0[k]!=(0 as u8){ out_sha[k]=f0[k]; k=k+1 } out_sha[k]=0 as u8
845                found=1
846            }
847            ls=i+1
848        }
849        i=i+1
850    }
851    return found
852}
853// how many releases recorded for `site` (field2). audit aggregate.
854func pub_query_count_site(ledpath: *u8, site: *u8) -> i64 {
855    let lenp: *i64 = sys_mmap(8) as *i64
856    let data: *u8 = sys_read_file(ledpath, lenp)
857    if (data as i64)==0 { return 0 }
858    let n: i64 = lenp[0]
859    let f2: *u8 = sys_mmap(256)
860    var i: i64=0; var ls: i64=0; var c: i64=0
861    while i<n {
862        if data[i]==(10 as u8) {
863            let line: *u8 = ((data as i64)+ls) as *u8
864            pub_field(line, i-ls, 2, f2)
865            if pub_streq(f2, site)==1 { c=c+1 }
866            ls=i+1
867        }
868        i=i+1
869    }
870    return c
871}
872// parse a leading decimal int from a NUL-terminated string (ignores non-digits).
873func pub_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 }
874
875// record a ledger event with an EXPLICIT outcome + timestamp + lead (used by tests/recovery to log non-PUBLISHED
876// outcomes; the live path uses pub_record_ledger = PUBLISHED + now()). record = sha dest site outcome pub_ts lead.
877func pub_record_outcome(ledpath: *u8, sha: *u8, dest: *u8, site: *u8, outcome: *u8, ts: i64, lead: i64) -> i64 {
878    let rec: *u8 = sys_mmap(PUB_REC_CAP)
879    var o: i64 = 0
880    o = fa_cat(rec, o, sha);  rec[o]=9 as u8; o=o+1
881    o = fa_cat(rec, o, dest); rec[o]=9 as u8; o=o+1
882    o = fa_cat(rec, o, site); rec[o]=9 as u8; o=o+1
883    o = fa_cat(rec, o, outcome); rec[o]=9 as u8; o=o+1
884    o = fa_catn(rec, o, ts); rec[o]=9 as u8; o=o+1
885    o = fa_catn(rec, o, lead)
886    rec[o] = 0 as u8
887    return fa_appendz(ledpath, rec, PUB_REC_CAP)
888}
889
890// ============================================================================================================
891// R12 -- DORA METRICS (the release-engineering yardstick) computed from the ledger event-log + deploy NOTIFY.
892// The ledger records sha dest site outcome pub_ts lead_sec, so the publisher can compute the four DORA metrics:
893//   out[0]=deploys  out[1]=successes  out[2]=failures  out[3]=change_fail_permille  out[4]=span_sec
894//   out[5]=deploys_per_ksec  out[6]=avg_MTTR_sec(resolved failures)  out[7]=resolved_failures  out[8]=avg_lead_sec
895// MTTR pairs each failure with the NEXT later success for the SAME dest (the recovery). change-fail-rate and
896// frequency are live now; lead-time is computed but reads 0 on live records until submit-ts is plumbed (v2).
897// ============================================================================================================
898func pub_dora(ledpath: *u8, out: *i64) -> i64 {
899    var z: i64=0; while z<9 { out[z]=0; z=z+1 }
900    let lenp: *i64 = sys_mmap(8) as *i64
901    let data: *u8 = sys_read_file(ledpath, lenp)
902    if (data as i64)==0 { return 0 }
903    let n: i64 = lenp[0]
904    let oc: *u8 = sys_mmap(64); let de: *u8 = sys_mmap(256); let tsf: *u8 = sys_mmap(64); let ldf: *u8 = sys_mmap(64)
905    let oc2: *u8 = sys_mmap(64); let de2: *u8 = sys_mmap(256); let ts2f: *u8 = sys_mmap(64)
906    var mints: i64=0; var maxts: i64=0; var havets: i64=0
907    var deploys: i64=0; var succ: i64=0; var fail: i64=0
908    var lead_sum: i64=0; var mttr_sum: i64=0; var mttr_cnt: i64=0
909    var i: i64=0; var ls: i64=0
910    while i<n {
911        if data[i]==(10 as u8) {
912            let line: *u8 = ((data as i64)+ls) as *u8
913            let ll: i64 = i-ls
914            pub_field(line, ll, 3, oc)
915            pub_field(line, ll, 4, tsf); let ts: i64 = pub_atoi(tsf)
916            deploys = deploys + 1
917            if havets==0 { mints=ts; maxts=ts; havets=1 } else { if ts<mints {mints=ts} if ts>maxts {maxts=ts} }
918            if pub_streq(oc, "PUBLISHED" as *u8)==1 {
919                succ=succ+1
920                pub_field(line, ll, 5, ldf); lead_sum = lead_sum + pub_atoi(ldf)
921            } else {
922                fail=fail+1
923                pub_field(line, ll, 1, de)
924                var best: i64=0; var found: i64=0
925                var j: i64=0; var js: i64=0
926                while j<n {
927                    if data[j]==(10 as u8) {
928                        let l2: *u8 = ((data as i64)+js) as *u8
929                        let ll2: i64 = j-js
930                        pub_field(l2, ll2, 3, oc2)
931                        if pub_streq(oc2, "PUBLISHED" as *u8)==1 {
932                            pub_field(l2, ll2, 1, de2)
933                            if pub_streq(de2, de)==1 {
934                                pub_field(l2, ll2, 4, ts2f); let ts2: i64 = pub_atoi(ts2f)
935                                if ts2 > ts { if found==0 { best=ts2; found=1 } else { if ts2<best { best=ts2 } } }
936                            }
937                        }
938                        js=j+1
939                    }
940                    j=j+1
941                }
942                if found==1 { mttr_sum = mttr_sum + (best - ts); mttr_cnt = mttr_cnt + 1 }
943            }
944            ls=i+1
945        }
946        i=i+1
947    }
948    out[0]=deploys; out[1]=succ; out[2]=fail
949    if deploys>0 { out[3] = fail*1000/deploys }
950    out[4] = maxts - mints
951    if out[4] > 0 { out[5] = deploys*1000/out[4] }
952    out[6] = 0; if mttr_cnt>0 { out[6] = mttr_sum/mttr_cnt }
953    out[7] = mttr_cnt
954    out[8] = 0; if succ>0 { out[8] = lead_sum/succ }
955    return deploys
956}
957
958// DEPLOY NOTIFICATION: append a framed event to a notify channel (a file daemons/operators watch -- the sovereign
959// equivalent of an alert webhook). One atomic framed line: kind<TAB>dest<TAB>sha<TAB>ts. returns the append result.
960func pub_notify(notifypath: *u8, kind: *u8, dest: *u8, sha: *u8) -> i64 {
961    let rec: *u8 = sys_mmap(PUB_REC_CAP)
962    var o: i64 = 0
963    o = fa_cat(rec, o, kind); rec[o]=9 as u8; o=o+1
964    o = fa_cat(rec, o, dest); rec[o]=9 as u8; o=o+1
965    o = fa_cat(rec, o, sha);  rec[o]=9 as u8; o=o+1
966    o = fa_catn(rec, o, sys_now_realtime_sec())
967    rec[o] = 0 as u8
968    return fa_appendz(notifypath, rec, PUB_REC_CAP)
969}
970
971// CHANGELOG / release notes: project the ledger into human-readable "dest sha site" lines into out (NUL-terminated).
972// returns the byte length. Bounded by cap (stops appending if a line would overflow). Pure read + format.
973func pub_changelog(ledpath: *u8, out: *u8, cap: i64) -> i64 {
974    let lenp: *i64 = sys_mmap(8) as *i64
975    let data: *u8 = sys_read_file(ledpath, lenp)
976    if (data as i64)==0 { out[0]=0 as u8; return 0 }
977    let n: i64 = lenp[0]
978    let f0: *u8 = sys_mmap(256); let f1: *u8 = sys_mmap(256); let f2: *u8 = sys_mmap(256)
979    var i: i64=0; var ls: i64=0; var o: i64=0
980    while i<n {
981        if data[i]==(10 as u8) {
982            let line: *u8 = ((data as i64)+ls) as *u8
983            pub_field(line, i-ls, 1, f1)   // dest
984            pub_field(line, i-ls, 0, f0)   // sha
985            pub_field(line, i-ls, 2, f2)   // site
986            if o + 320 < cap {
987                o = fa_cat(out, o, f1); out[o]=32 as u8; o=o+1
988                o = fa_cat(out, o, f0); out[o]=32 as u8; o=o+1
989                o = fa_cat(out, o, f2); out[o]=10 as u8; o=o+1
990            }
991            ls=i+1
992        }
993        i=i+1
994    }
995    out[o]=0 as u8
996    return o
997}
998
999// ============================================================================================================
1000// R13 -- GOVERNANCE II: N-of-M multi-approver QUORUM (beyond R6's single operator-ok) + release SCHEDULE
1001// (deploy window + a FREEZE override). Both file-based, sovereign, fail-closed.
1002// ============================================================================================================
1003// path of the per-artifact approvers file: <approval_dir>/<sha>.approvers (one distinct approver per line).
1004func pub_approvers_path(approval_dir: *u8, sha: *u8, out: *u8) -> i64 {
1005    let tmp: *u8 = sys_mmap(700); pub_join(approval_dir, sha, tmp)
1006    pub_pcat(tmp, ".approvers" as *u8, out)
1007    return 0
1008}
1009// 1 iff a whole line == needle exists in the file.
1010func pub_has_line(path: *u8, needle: *u8) -> i64 {
1011    let lenp: *i64 = sys_mmap(8) as *i64
1012    let data: *u8 = sys_read_file(path, lenp)
1013    if (data as i64)==0 { return 0 }
1014    let n: i64 = lenp[0]
1015    let f: *u8 = sys_mmap(256)
1016    var i: i64=0; var ls: i64=0
1017    while i<n {
1018        if data[i]==(10 as u8) {
1019            let line: *u8 = ((data as i64)+ls) as *u8
1020            pub_field(line, i-ls, 0, f)
1021            if pub_streq(f, needle)==1 { return 1 }
1022            ls=i+1
1023        }
1024        i=i+1
1025    }
1026    return 0
1027}
1028// record an approval by a named approver for an artifact (DEDUP: the same approver counts once). returns 1 if newly
1029// recorded, 0 if this approver already approved.
1030func pub_approve_quorum(approval_dir: *u8, sha: *u8, approver: *u8) -> i64 {
1031    let p: *u8 = sys_mmap(700); pub_approvers_path(approval_dir, sha, p)
1032    if pub_has_line(p, approver)==1 { return 0 }
1033    let rec: *u8 = sys_mmap(256); var o: i64=0; o=fa_cat(rec, o, approver); rec[o]=0 as u8
1034    fa_appendz(p, rec, 256)
1035    return 1
1036}
1037// 1 iff at least `need` DISTINCT approvers have approved the artifact (N-of-M). fail-closed (no file -> 0).
1038func pub_quorum_met(approval_dir: *u8, sha: *u8, need: i64) -> i64 {
1039    let p: *u8 = sys_mmap(700); pub_approvers_path(approval_dir, sha, p)
1040    let lenp: *i64 = sys_mmap(8) as *i64
1041    let data: *u8 = sys_read_file(p, lenp)
1042    if (data as i64)==0 { return 0 }
1043    let n: i64 = lenp[0]
1044    var i: i64=0; var c: i64=0
1045    while i<n { if data[i]==(10 as u8) { c=c+1 } i=i+1 }
1046    if c >= need { return 1 }
1047    return 0
1048}
1049// 1 iff a freeze flag file is present (deploys frozen).
1050func pub_frozen(freeze_path: *u8) -> i64 { return pub_exists(freeze_path) }
1051// release SCHEDULE: allow a deploy iff NOT frozen AND now is within [win_start, win_end]. fail-closed; freeze wins.
1052func pub_schedule_allows(now: i64, win_start: i64, win_end: i64, freeze_path: *u8) -> i64 {
1053    if pub_exists(freeze_path)==1 { return 0 }
1054    if now < win_start { return 0 }
1055    if now > win_end { return 0 }
1056    return 1
1057}
1058
1059// ============================================================================================================
1060// R14 -- PROGRESSIVE DELIVERY: BLUE-GREEN (two materialized slots + zero-copy instant atomic flip) and CANARY
1061// (a separate cohort slot, served-smoke-gated before full promotion -> a bad version reaches 0% of main).
1062// ============================================================================================================
1063// path of a color slot: <slotroot>/<dest>.<color>  (e.g. page.html.blue).
1064func pub_bg_cpath(slotroot: *u8, dest: *u8, color: *u8, out: *u8) -> i64 {
1065    var o: i64 = pub_join(slotroot, dest, out)
1066    out[o]=46 as u8; o=o+1
1067    var i: i64=0; while color[i]!=(0 as u8){ out[o]=color[i]; o=o+1; i=i+1 }
1068    out[o]=0 as u8
1069    return o
1070}
1071// materialize a color slot from src + verify its bytes against expect_sha. returns 1 ok, 0 sha-mismatch, <0 copy err.
1072func pub_blue_green_stage(slotroot: *u8, dest: *u8, color: *u8, src: *u8, expect_sha: *u8) -> i64 {
1073    let cp: *u8 = sys_mmap(700); pub_bg_cpath(slotroot, dest, color, cp)
1074    if pub_copy(src, cp) < 0 { return 0 - 1 }
1075    if pub_sha_match(cp, expect_sha)==0 { return 0 }
1076    return 1
1077}
1078// THE FLIP: atomically repoint the active path <slotroot>/<dest> to the given color via a relative symlink + rename
1079// (rename is atomic -> a reader sees old-or-new, never broken). ZERO-COPY: the color files are never rewritten; the
1080// flip (and flip-back, for instant rollback) is an O(1) pointer swap. returns 1, or <0 on rename error.
1081func pub_blue_green_flip(slotroot: *u8, dest: *u8, color: *u8) -> i64 {
1082    let tgt: *u8 = sys_mmap(700); var o: i64=0; var i: i64=0   // relative target "<dest>.<color>"
1083    while dest[i]!=(0 as u8){ tgt[o]=dest[i]; o=o+1; i=i+1 } tgt[o]=46 as u8; o=o+1
1084    i=0; while color[i]!=(0 as u8){ tgt[o]=color[i]; o=o+1; i=i+1 } tgt[o]=0 as u8
1085    let active: *u8 = sys_mmap(700); pub_join(slotroot, dest, active)
1086    let tmp: *u8 = sys_mmap(700); pub_pcat(active, ".flip" as *u8, tmp)
1087    __syscall(87, tmp as i64, 0, 0, 0, 0, 0)                    // unlink any stale tmp link
1088    __syscall(88, tgt as i64, tmp as i64, 0, 0, 0, 0)          // symlink(target, tmp)  [x86_64 symlink=88]
1089    if pub_rename(tmp, active) != 0 { return 0 - 2 }           // atomic replace -> instant flip
1090    return 1
1091}
1092// CANARY rollout: stage the new version to a SEPARATE canary slot, run `cohort` served-smoke checks against the
1093// canary URL, and promote to main ONLY if every check passes; otherwise ABORT -> main is never touched (a bad
1094// version reaches 0% of main). The smoke is the SERVED-result gate (catches disk-OK-but-URL-bad, not just bytes).
1095// returns 1 promoted, 0 aborted (canary unhealthy; main untouched), <0 stage error. urlbase = the served dest path.
1096func pub_canary_rollout(expect_sha: *u8, src: *u8, slotroot: *u8, dest: *u8, urlbase: *u8, port: i64, cohort: i64) -> i64 {
1097    let cp: *u8 = sys_mmap(700); pub_bg_cpath(slotroot, dest, "canary" as *u8, cp)
1098    if pub_copy(src, cp) < 0 { return 0 - 1 }                  // stage to the canary slot
1099    let curl: *u8 = sys_mmap(700); var o: i64=0; var i: i64=0  // canary URL = "<urlbase>.canary"
1100    while urlbase[i]!=(0 as u8){ curl[o]=urlbase[i]; o=o+1; i=i+1 } curl[o]=46 as u8; o=o+1
1101    let cc: *u8 = "canary" as *u8; i=0; while cc[i]!=(0 as u8){ curl[o]=cc[i]; o=o+1; i=i+1 } curl[o]=0 as u8
1102    var k: i64=0
1103    while k < cohort {
1104        if pub_smoke(port, curl, expect_sha)==0 { return 0 }   // canary unhealthy -> ABORT, main untouched
1105        k=k+1
1106    }
1107    let mp: *u8 = sys_mmap(700); pub_join(slotroot, dest, mp)
1108    if pub_promote_atomic(cp, mp) < 0 { return 0 - 2 }         // healthy -> full rollout (atomic)
1109    return 1
1110}
1111
1112// ============================================================================================================
1113// R15 -- MULTI-ENV PROMOTE-CHAIN (dev->stage->prod), MULTI-TARGET FAN-OUT, and FEATURE FLAGS (decouple deploy
1114// from release). roots/targets are passed as a *i64 array of string pointers (roots[i] = an env/target root path).
1115// ============================================================================================================
1116// promote an artifact through an ORDERED chain of env roots; each env stages+verifies+promotes before the next.
1117// returns the number of envs reached (== nroots if it reached prod; < nroots if it stopped at a failed verify ->
1118// the later envs (incl prod) are NEVER touched = fail-closed ordered promotion).
1119func pub_promote_chain(expect_sha: *u8, src: *u8, dest: *u8, roots: *i64, nroots: i64, stageroot: *u8) -> i64 {
1120    var i: i64 = 0
1121    while i < nroots {
1122        let envroot: *u8 = roots[i] as *u8
1123        let stagep: *u8 = sys_mmap(700); pub_join(stageroot, dest, stagep)
1124        let livep: *u8 = sys_mmap(700); pub_join(envroot, dest, livep)
1125        if pub_deploy(expect_sha, src, stagep, livep) != 1 { return i }   // stopped at env i; envs >= i untouched
1126        i = i + 1
1127    }
1128    return nroots
1129}
1130// fan out one artifact to N targets ALL-OR-NOTHING: stage+verify ONCE, and only if it verifies do we atomically
1131// promote to each target -> no target ever receives unverified bytes (fail-closed). returns #targets promoted.
1132func pub_fanout(expect_sha: *u8, src: *u8, dest: *u8, targets: *i64, ntargets: i64, stageroot: *u8) -> i64 {
1133    let stagep: *u8 = sys_mmap(700); pub_join(stageroot, dest, stagep)
1134    if pub_copy(src, stagep) < 0 { return 0 - 1 }
1135    if pub_sha_match(stagep, expect_sha) == 0 { return 0 }     // fail-closed: NO target gets bad bytes
1136    var i: i64 = 0; var done: i64 = 0
1137    while i < ntargets {
1138        let troot: *u8 = targets[i] as *u8
1139        let livep: *u8 = sys_mmap(700); pub_join(troot, dest, livep)
1140        if pub_promote_atomic(stagep, livep) >= 0 { done = done + 1 }
1141        i = i + 1
1142    }
1143    return done
1144}
1145// FEATURE FLAGS: a flag store decouples DEPLOY (artifact shipped) from RELEASE (flag on). Append-only (last wins).
1146func pub_flag_set(store: *u8, name: *u8, val: i64) -> i64 {
1147    let rec: *u8 = sys_mmap(256); var o: i64=0
1148    o = fa_cat(rec, o, name); rec[o]=9 as u8; o=o+1
1149    o = fa_catn(rec, o, val); rec[o]=0 as u8
1150    return fa_appendz(store, rec, 256)
1151}
1152// read a flag: the LAST recorded value for `name`, or 0 (OFF) if never set (fail-safe default = off).
1153func pub_flag_get(store: *u8, name: *u8) -> i64 {
1154    let lenp: *i64 = sys_mmap(8) as *i64
1155    let data: *u8 = sys_read_file(store, lenp)
1156    if (data as i64)==0 { return 0 }
1157    let n: i64 = lenp[0]
1158    let f0: *u8 = sys_mmap(128); let f1: *u8 = sys_mmap(64)
1159    var i: i64=0; var ls: i64=0; var val: i64=0
1160    while i<n {
1161        if data[i]==(10 as u8) {
1162            let line: *u8 = ((data as i64)+ls) as *u8
1163            pub_field(line, i-ls, 0, f0)
1164            if pub_streq(f0, name)==1 { pub_field(line, i-ls, 1, f1); val = pub_atoi(f1) }
1165            ls=i+1
1166        }
1167        i=i+1
1168    }
1169    return val
1170}
1171
1172// ============================================================================================================
1173// R16 -- ROLLING update (incremental, halt-on-failure), artifact RETENTION/GC (never-brick), bounded RETRY.
1174// ============================================================================================================
1175// update instances ONE AT A TIME; verify each; STOP at the first failure (the remaining instances are never touched
1176// -> a bad rollout is halted mid-roll, blast-limited). returns the number of instances successfully updated.
1177func pub_rolling(expect_sha: *u8, src: *u8, dest: *u8, targets: *i64, ntargets: i64, stageroot: *u8) -> i64 {
1178    var i: i64 = 0
1179    while i < ntargets {
1180        let troot: *u8 = targets[i] as *u8
1181        let stagep: *u8 = sys_mmap(700); pub_join(stageroot, dest, stagep)
1182        let livep: *u8 = sys_mmap(700); pub_join(troot, dest, livep)
1183        if pub_deploy(expect_sha, src, stagep, livep) != 1 { return i }   // halt: instances >= i untouched
1184        i = i + 1
1185    }
1186    return ntargets
1187}
1188// RETENTION/GC: remove the publisher's TRANSIENT leftovers for a live path (.tmp/.flip/.rbk). It NEVER removes the
1189// live file or its .prev (the rollback target) -> never-brick #26 by construction. returns 1.
1190func pub_retain(livepath: *u8) -> i64 {
1191    let tmp: *u8 = sys_mmap(700); pub_pcat(livepath, ".tmp" as *u8, tmp); __syscall(87, tmp as i64, 0, 0, 0, 0, 0)
1192    let flip: *u8 = sys_mmap(700); pub_pcat(livepath, ".flip" as *u8, flip); __syscall(87, flip as i64, 0, 0, 0, 0, 0)
1193    let rbk: *u8 = sys_mmap(700); pub_pcat(livepath, ".rbk" as *u8, rbk); __syscall(87, rbk as i64, 0, 0, 0, 0, 0)
1194    return 1
1195}
1196// bounded RETRY with growing backoff: attempt up to maxtries; the op "succeeds" on attempt #succeed_at. returns the
1197// winning attempt (1..maxtries), or 0 if exhausted (GIVES UP -- bounded, never an infinite loop). graceful-degradation.
1198func pub_retry(succeed_at: i64, maxtries: i64, base_backoff_ms: i64) -> i64 {
1199    var k: i64 = 1
1200    while k <= maxtries {
1201        if k >= succeed_at { return k }
1202        pub_nap_ms(base_backoff_ms * k)         // linear backoff before the next attempt
1203        k = k + 1
1204    }
1205    return 0
1206}
1207
1208// ============================================================================================================
1209// R17 -- request PRIORITY lanes (field7 of the queue record) + per-environment concurrency groups (LOCKGROUP).
1210// ============================================================================================================
1211// submit with an explicit priority (field7). Higher = more urgent. (pub_submit_to records carry no field7 = pri 0.)
1212func pub_submit_pri(qpath: *u8, src: *u8, dest: *u8, site: *u8, requester: *u8, policy: *u8, pri: i64) -> i64 {
1213    let hex: *u8 = sys_mmap(72)
1214    if pub_sha_file(src, hex) == 0 { return 0 - 9 }
1215    let rec: *u8 = sys_mmap(PUB_REC_CAP)
1216    var o: i64 = pub_build_rec(rec, site, requester, hex, src, dest, policy)   // 7 fields, rec[o]=0
1217    rec[o]=9 as u8; o=o+1
1218    o = fa_catn(rec, o, pri); rec[o]=0 as u8                                    // field7 = priority
1219    return fa_appendz(qpath, rec, PUB_REC_CAP)
1220}
1221// pick the highest-priority PENDING request: returns its priority (>=0), writes its dest into out_dest; -1 if none.
1222func pub_priority_pick(qpath: *u8, out_dest: *u8) -> i64 {
1223    out_dest[0] = 0 as u8
1224    let lenp: *i64 = sys_mmap(8) as *i64
1225    let data: *u8 = sys_read_file(qpath, lenp)
1226    if (data as i64)==0 { return 0 - 1 }
1227    let n: i64 = lenp[0]
1228    let st: *u8 = sys_mmap(64); let dest: *u8 = sys_mmap(256); let prif: *u8 = sys_mmap(64)
1229    var i: i64=0; var ls: i64=0; var best: i64 = 0 - 1
1230    while i<n {
1231        if data[i]==(10 as u8) {
1232            let line: *u8 = ((data as i64)+ls) as *u8
1233            let ll: i64 = i-ls
1234            pub_field(line, ll, 0, st)
1235            if pub_streq(st, "PENDING" as *u8)==1 {
1236                pub_field(line, ll, 7, prif)
1237                let pri: i64 = pub_atoi(prif)
1238                if pri > best {
1239                    best = pri
1240                    pub_field(line, ll, 5, dest)
1241                    var k: i64=0; while dest[k]!=(0 as u8){ out_dest[k]=dest[k]; k=k+1 } out_dest[k]=0 as u8
1242                }
1243            }
1244            ls=i+1
1245        }
1246        i=i+1
1247    }
1248    return best
1249}
1250// LOCKGROUP: derive a per-site lock resource so DIFFERENT sites publish in parallel while the SAME site serializes.
1251func pub_lockgroup_res(site: *u8, out: *u8) -> i64 {
1252    var o: i64 = 0; let p: *u8 = "publish:" as *u8
1253    while p[o]!=(0 as u8){ out[o]=p[o]; o=o+1 }
1254    var i: i64=0; while site[i]!=(0 as u8){ out[o]=site[i]; o=o+1; i=i+1 }
1255    out[o]=0 as u8
1256    return o
1257}
1258// acquire the per-site publish lock (returns the fl_acquire fd); release with pub_lockgroup_release.
1259func pub_lockgroup_acquire(site: *u8) -> i64 {
1260    let res: *u8 = sys_mmap(320); pub_lockgroup_res(site, res)
1261    return fl_acquire(res, PUB_MAGIC_100000, 1)
1262}
1263func pub_lockgroup_release(fd: i64) -> i64 { return fl_release(fd) }
1264
1265// ============================================================================================================
1266// R18 -- PUBLIC-DOMAIN WIRING (pub_dns): the publisher EMITS a stable domain->endpoint mapping for hosting to apply
1267// (DNS A / stable reverse-proxy), so going public is NOT a fragile daemon-swap. The publisher owns the EMISSION +
1268// a stable internal endpoint; hosting APPLIES the DNS record (coordinated). config = one parseable directive line.
1269// ============================================================================================================
1270func pub_dns_config(domain: *u8, host: *u8, port: i64, out: *u8) -> i64 {
1271    var o: i64 = 0
1272    o = fa_cat(out, o, "publish-endpoint domain=" as *u8); o = fa_cat(out, o, domain)
1273    o = fa_cat(out, o, " host=" as *u8); o = fa_cat(out, o, host)
1274    o = fa_cat(out, o, " port=" as *u8); o = fa_catn(out, o, port)
1275    o = fa_cat(out, o, " proto=https" as *u8)
1276    out[o] = 0 as u8
1277    return o
1278}
1279// persist the mapping (idempotent overwrite) for hosting to apply. returns 1.
1280func pub_dns_emit(domain: *u8, host: *u8, port: i64, path: *u8) -> i64 {
1281    let buf: *u8 = sys_mmap(512); let n: i64 = pub_dns_config(domain, host, port, buf)
1282    let fd: i64 = sys_openat_wr(path, 0x1a4); if fd<0 { return 0 }
1283    sys_write(fd, buf, n); sys_close(fd)
1284    return 1
1285}
1286
1287// ============================================================================================================
1288// R19 -- CONTINUOUS DRAIN (the daemon loop): repeatedly run the GOVERNED pipeline so the publisher keeps the live
1289// queue drained as workstreams submit. Each cycle is idempotent + serialized (pub_run_governed), so a daemon is
1290// just this loop with poll-sleep. pub_run_drain is the BOUNDED form (stops when a cycle publishes 0 = drained, or
1291// after max_cycles) -- gateable; a real daemon passes a large/poll cap. returns total published across cycles.
1292// ============================================================================================================
1293func pub_run_drain(qpath: *u8, ledpath: *u8, stageroot: *u8, liveroot: *u8, lockres: *u8, approval_dir: *u8, max_cycles: i64, sleep_ms: i64) -> i64 {
1294    var total: i64 = 0
1295    var c: i64 = 0
1296    while c < max_cycles {
1297        let n: i64 = pub_run_governed(qpath, ledpath, stageroot, liveroot, lockres, approval_dir)
1298        total = total + n
1299        if n == 0 { return total }          // queue drained this cycle (a daemon would poll-sleep and re-check)
1300        c = c + 1
1301        if sleep_ms > 0 { pub_nap_ms(sleep_ms) }
1302    }
1303    return total
1304}
1305
1306// ============================================================================================================
1307// R20 -- CALL-AWARE no-bypass audit. The comment-aware audit still over-flags STRING-LITERAL mentions (e.g. an
1308// HTML page that prints "<code>nx_aw_push</code>" in its body). A real bypass is a CALL: the symbol immediately
1309// followed by '('. This strips // comments AND matches only "symbol(" -> the TRUE direct-caller count.
1310// ============================================================================================================
1311func pub_forbidden_call(code: *u8, n: i64) -> i64 {
1312    if pub_substr(code, n, "nx_aw_send(" as *u8)==1 { return 1 }
1313    if pub_substr(code, n, "nx_aw_push(" as *u8)==1 { return 1 }
1314    if pub_substr(code, n, "pub_deploy(" as *u8)==1 { return 1 }
1315    if pub_substr(code, n, "pub_promote_atomic(" as *u8)==1 { return 1 }
1316    if pub_substr(code, n, "pub_rollback(" as *u8)==1 { return 1 }
1317    if pub_substr(code, n, "pub_blue_green_flip(" as *u8)==1 { return 1 }
1318    if pub_substr(code, n, "pub_canary_rollout(" as *u8)==1 { return 1 }
1319    if pub_substr(code, n, "pub_record_ledger(" as *u8)==1 { return 1 }
1320    if pub_substr(code, n, "pub_record_outcome(" as *u8)==1 { return 1 }
1321    return 0
1322}
1323func pub_audit_call_strict(path: *u8, allowlisted: i64) -> i64 {
1324    if allowlisted == 1 { return 0 }
1325    let lenp: *i64 = sys_mmap(8) as *i64
1326    let data: *u8 = sys_read_file(path, lenp)
1327    if (data as i64)==0 { return 0 }
1328    let n: i64 = lenp[0]
1329    let code: *u8 = sys_mmap(n + 16)
1330    var i: i64=0; var o: i64=0; var incomment: i64=0
1331    while i < n {
1332        let c: i64 = data[i] & 0xff
1333        if c == 10 { incomment=0; code[o]=10 as u8; o=o+1 }
1334        else {
1335            if incomment == 0 {
1336                if c == 47 { if i+1 < n { if (data[i+1] & 0xff) == 47 { incomment = 1 } } }
1337                if incomment == 0 { code[o]=c as u8; o=o+1 }
1338            }
1339        }
1340        i = i + 1
1341    }
1342    return pub_forbidden_call(code, o)
1343}
1344
1345// ============================================================================================================
1346// R21 -- BEST-PRACTICE CONTINUOUS DAEMON (researched + cited: knowledge/fetched/dmn_*.raw). Operator: "run as we add
1347// things, not start/stop". The sovereign researcher (nx_dmn_research_fetch, 13/13 sources) determined the practice:
1348//  · SUPERVISED auto-restart, not manual start/stop (systemd/runit "supervision") -> the supervisor runs+restarts it;
1349//  · CRASH-ONLY (dmn_b_crashonly): safe to kill+restart anytime; recover BY restart -> our drain is idempotent;
1350//  · EVENT-DRIVEN via inotify (dmn_c_inotify): act as work ARRIVES, not fixed polling (poll-timeout as the fallback);
1351//  · WATCHDOG/HEARTBEAT (dmn_d_*): emit a heartbeat so a HANG (the wedge) is detected, not just a crash;
1352//  · GRACEFUL stop (dmn_e_*): on a stop signal, finish the current drain + exit cleanly.
1353// pub_daemon_serve is the loop; the pieces are individually gated. The SUPERVISOR is the hosting workstream's
1354// (hostop_supervise emits the command); this daemon is supervision-COMPATIBLE by being crash-only + heartbeat + graceful.
1355// ============================================================================================================
1356// LIVENESS: write the current epoch to the heartbeat file (a watchdog reads it to detect a hang). returns the epoch.
1357func pub_daemon_heartbeat(hbpath: *u8) -> i64 {
1358    let now: i64 = sys_now_realtime_sec()
1359    let buf: *u8 = sys_mmap(24); let o: i64 = fa_catn(buf, 0, now); buf[o]=0 as u8
1360    let fd: i64 = sys_openat_wr(hbpath, 0x1a4); if fd < 0 { return 0 - 1 }
1361    sys_write(fd, buf, o); sys_close(fd)
1362    return now
1363}
1364// WATCHDOG view: seconds since the last heartbeat (now - hb); -1 if no heartbeat. A supervisor restarts if age > limit.
1365func pub_daemon_hb_age(hbpath: *u8, now: i64) -> i64 {
1366    let lenp: *i64 = sys_mmap(8) as *i64
1367    let d: *u8 = sys_read_file(hbpath, lenp)
1368    if (d as i64) == 0 { return 0 - 1 }
1369    return now - pub_atoi(d)
1370}
1371// EVENT-DRIVEN: inotify-watch a directory (queue dir) for writes/creates/moves. returns the inotify fd, or <0.
1372func pub_daemon_watch_init(dirpath: *u8) -> i64 {
1373    let infd: i64 = __syscall(294, 0, 0, 0, 0, 0, 0)              // inotify_init1(0)
1374    if infd < 0 { return infd }
1375    __syscall(254, infd, dirpath as i64, 0x18a, 0, 0, 0)         // add_watch: IN_MODIFY|IN_CLOSE_WRITE|IN_MOVED_TO|IN_CREATE
1376    return infd
1377}
1378// wait for a watched change OR timeout_ms (poll fallback). returns 1 if a change arrived (events drained), 0 on timeout.
1379func pub_daemon_wait(infd: i64, timeout_ms: i64) -> i64 {
1380    let pfd: *u8 = sys_mmap(16)
1381    pfd[0]=(infd & 0xff) as u8; pfd[1]=((infd>>8)&0xff) as u8; pfd[2]=((infd>>16)&0xff) as u8; pfd[3]=((infd>>24)&0xff) as u8
1382    pfd[4]=1 as u8; pfd[5]=0 as u8; pfd[6]=0 as u8; pfd[7]=0 as u8   // events=POLLIN
1383    let r: i64 = __syscall(7, pfd as i64, 1, timeout_ms, 0, 0, 0)    // poll(fds,1,timeout_ms)
1384    if r <= 0 { return 0 }                                            // timeout/err -> no change
1385    let buf: *u8 = sys_mmap(PUB_MAGIC_4096); sys_read(infd, buf, PUB_MAGIC_4096)         // drain events
1386    return 1
1387}
1388// GRACEFUL: a stop is requested iff the stop-flag file exists (a SIGTERM handler or supervisor creates it).
1389func pub_daemon_should_stop(stopflag: *u8) -> i64 { return pub_exists(stopflag) }
1390// THE DAEMON LOOP: crash-only idempotent drain, event-driven (inotify), heartbeat liveness, graceful stop. max_cycles
1391// large = effectively continuous (the supervisor restarts on death); bounded = gateable. returns total published.
1392func pub_daemon_serve(qpath: *u8, ledpath: *u8, stageroot: *u8, liveroot: *u8, lockres: *u8, approval: *u8, watchdir: *u8, hbpath: *u8, stopflag: *u8, max_cycles: i64, poll_ms: i64) -> i64 {
1393    let infd: i64 = pub_daemon_watch_init(watchdir)
1394    var cycles: i64 = 0; var total: i64 = 0
1395    while cycles < max_cycles {
1396        if pub_daemon_should_stop(stopflag) == 1 { return total }    // graceful stop
1397        total = total + pub_run_governed(qpath, ledpath, stageroot, liveroot, lockres, approval)   // crash-only idempotent drain
1398        pub_daemon_heartbeat(hbpath)                                 // liveness
1399        if pub_daemon_should_stop(stopflag) == 1 { return total }
1400        if infd >= 0 { pub_daemon_wait(infd, poll_ms) }              // event-driven wait (act as work arrives)
1401        cycles = cycles + 1
1402    }
1403    return total
1404}
nx_publisher.nx source

`nx_publisher.nx` source
