nx_pm.nx source

1// nx_pm.nx -- THE NISHI PM: the coordinator that routes work to the ONE Accountable role per the RACI, so 2// workstreams / publishing / fixing / optimizing never compete for the same work. The RACI registry IS the 3// PM's routing table (single source of truth). For a work request (activity + payload), the PM: 4// 1. looks up the Accountable role in knowledge/registry/nishi_raci.tsv (exactly one, validated by nx_raci), 5// 2. ROUTES it to that role's intake -- change-apply roles (doctor=fix, engineer=optimize) -> nx_role_queue 6// submit (their isolated serialized queue); other roles -> recorded as that role's to perform, 7// 3. LOGS the routing to the pmcoord channel in the sovereign seg-store knowledge/store/roles-* (the one coordination record; NO TSV). 8// This is "the PM coordinates all the publishing and workstreams" -- RACI-driven, no collisions, auditable. 9// 10// usage: nx_pm route <activity> <target> <oldfile> <newfile> <requester> (target/files used only for apply-roles) 11// nx_pm accountable <activity> 12// license_tier: ORIGINAL 13import "nx_syscalls.nx" 14import "nx_role_queue.nx" // rq_submit / rq_field / rq_streq / rq_cat / rq_n / rq_w + (transitively) da_read 15import "nx_role_store.nx" // rs_append -- PM routing audit log now lives in the SOVEREIGN seg-store (NO TSV) 16import "nx_raci_sov.nx" // raci_accountable_for -- RACI ownership now read from the SOVEREIGN seg-store (NO TSV) 17const PM_MAGIC_2048: i64 = 2048 18 19const PM_COORD: *u8 = "knowledge/roles/pm_coord.tsv" // RETIRED file path (kept for provenance); routing now -> rs_append("pmcoord") 20 21func pm_has(s: *u8, ch: i64) -> i64 { var i: i64=0; while s[i]!=(0 as u8){ if (s[i] as i64)==ch {return 1} i=i+1 } return 0 } 22 23// find the Accountable role for `activity` -> out_role, FROM THE SOVEREIGN STORE (knowledge/store/raci; 24// reader-flip 2026-07-16, was a nishi_raci.tsv scan). returns 1 if found. 25func pm_accountable(activity: *u8, out_role: *u8) -> i64 { 26 return raci_accountable_for(activity, out_role) 27} 28 29// LOG a PM routing to the pmcoord CHANNEL in the sovereign seg-store (knowledge/store/roles-*). Same record 30// (ts<TAB>activity<TAB>role<TAB>requester<TAB>disp), seg-store record instead of a pm_coord.tsv append line. 31func pm_record(activity: *u8, role: *u8, requester: *u8, disp: *u8) -> i64 { 32 let r: *u8 = sys_mmap(PM_MAGIC_2048); var o: i64 = 0 33 o = rq_n(r, o, sys_now_realtime_sec()); r[o]=9 as u8; o=o+1 34 o = rq_cat(r, o, activity); r[o]=9 as u8; o=o+1 35 o = rq_cat(r, o, role); r[o]=9 as u8; o=o+1 36 o = rq_cat(r, o, requester); r[o]=9 as u8; o=o+1 37 o = rq_cat(r, o, disp) 38 rs_append("pmcoord" as *u8, r, o) 39 return 0 40} 41 42// route a work request to its Accountable role. returns 1 routed, 0 no-accountable. 43func pm_route(activity: *u8, target: *u8, oldf: *u8, newf: *u8, requester: *u8) -> i64 { 44 let role: *u8 = sys_mmap(64) 45 if pm_accountable(activity, role) == 0 { 46 rq_w("PM: NO accountable role for activity '" as *u8); rq_w(activity); rq_w("' (RACI gap)\n" as *u8) 47 pm_record(activity, "?" as *u8, requester, "UNROUTED-no-accountable" as *u8) 48 return 0 49 } 50 var disp: *u8 = "owner-performs" as *u8 51 // change-apply roles get the work queued into their serialized intake; others are recorded as owner 52 if rq_streq(role, "doctor" as *u8) == 1 { rq_submit(role, target, oldf, newf, requester); disp = "queued@role_queue" as *u8 } 53 if rq_streq(role, "engineer" as *u8) == 1 { rq_submit(role, target, oldf, newf, requester); disp = "queued@role_queue" as *u8 } 54 pm_record(activity, role, requester, disp) 55 rq_w("PM: '" as *u8); rq_w(activity); rq_w("' -> accountable=" as *u8); rq_w(role); rq_w(" -> " as *u8); rq_w(disp); rq_w("\n" as *u8) 56 return 1 57} 58 59// fork + execve a tool with up to 5 string args; wait; return child exit code. 60func pm_run(path: *u8, a1: *u8, a2: *u8, a3: *u8, a4: *u8, a5: *u8) -> i64 { 61 let pid: i64 = sys_fork() 62 if pid == 0 { 63 let av: *i64 = sys_mmap(8*7) as *i64 64 av[0]=path as i64; av[1]=a1 as i64; av[2]=a2 as i64; av[3]=a3 as i64; av[4]=a4 as i64; av[5]=a5 as i64; av[6]=0 65 let ev: *i64 = sys_mmap(16) as *i64; ev[0]="PATH=/usr/bin:/bin" as *u8 as i64; ev[1]=0 66 sys_execve(path, av, ev); sys_exit(127) 67 } 68 let st: *i64 = sys_mmap(16) as *i64; sys_wait4(pid, st, 0) 69 return (st[0] >> 8) & 0xff 70} 71 72// route a PUBLISH to the Publisher's intake (nx_pub_submit), per the RACI (publish = A:publisher). returns 1. 73func pm_publish(src: *u8, dest: *u8, site: *u8, requester: *u8, policy: *u8) -> i64 { 74 let role: *u8 = sys_mmap(64) 75 if pm_accountable("publish" as *u8, role) == 0 { rq_w("PM: no accountable role for publish (RACI gap)\n" as *u8); return 0 } 76 let rc: i64 = pm_run("_offc/nx_pub_submit.elf" as *u8, src, dest, site, requester, policy) 77 pm_record("publish" as *u8, role, requester, "submitted@pub_submit" as *u8) 78 rq_w("PM: 'publish' -> accountable=" as *u8); rq_w(role); rq_w(" -> nx_pub_submit exit=" as *u8); let b: *u8=sys_mmap(16); let e: i64=rq_n(b,0,rc); sys_write(1,b,e); rq_w("\n" as *u8) 79 return 1 80} 81 82func main(argc: i64, argv: *i64) -> i64 { 83 if argc < 3 { rq_w("usage: nx_pm route <activity> <target> <oldfile> <newfile> <requester> | accountable <activity>\n" as *u8); return 2 } 84 let cmd: *u8 = argv[1] as *u8 85 if rq_streq(cmd, "accountable" as *u8) == 1 { 86 let role: *u8 = sys_mmap(64) 87 if pm_accountable(argv[2] as *u8, role) == 1 { rq_w("accountable(" as *u8); rq_w(argv[2] as *u8); rq_w(")=" as *u8); rq_w(role); rq_w("\n" as *u8); return 0 } 88 rq_w("no accountable role for " as *u8); rq_w(argv[2] as *u8); rq_w("\n" as *u8); return 1 89 } 90 if rq_streq(cmd, "route" as *u8) == 1 { 91 if argc < 7 { rq_w("usage: nx_pm route <activity> <target> <oldfile> <newfile> <requester>\n" as *u8); return 2 } 92 let rc: i64 = pm_route(argv[2] as *u8, argv[3] as *u8, argv[4] as *u8, argv[5] as *u8, argv[6] as *u8) 93 if rc == 1 { return 0 } 94 return 1 95 } 96 if rq_streq(cmd, "publish" as *u8) == 1 { 97 if argc < 7 { rq_w("usage: nx_pm publish <src> <dest> <site> <requester> <policy>\n" as *u8); return 2 } 98 let rcp: i64 = pm_publish(argv[2] as *u8, argv[3] as *u8, argv[4] as *u8, argv[5] as *u8, argv[6] as *u8) 99 if rcp == 1 { return 0 } 100 return 1 101 } 102 rq_w("unknown subcommand\n" as *u8); return 2 103}