nx_task.nx source
↩ module page · 51 lines · 3095 B
1// nx_task.nx -- MCP Tasks-primitive parity, sovereign: dispatch a long-running task (get a handle), POLL its status,
2// and retrieve the RESULT. State machine on seg_store via the canonical nx_registry (additive -> the full task history
3// is kept). Actual async execution composes nx_daemon (a worker daemon does the work then task_set ...done...). Record
4// (tab-separated): id <TAB> status <TAB> payload <TAB> result. status = pending | running | done | failed.
5// license_tier: ORIGINAL
6import "nx_registry.nx"
7import "nx_tabrec.nx"
8const TASK_MAGIC_4096: i64 = 4096
9
10const TASK_KP: *u8 = "task:" as *u8
11const TASK_IDX: *u8 = "__taskidx__" as *u8
12const TASK_PREFIX: *u8 = "knowledge/taskq-" as *u8
13
14// dispatch: store the task pending with its payload, empty result. Returns the handle (the id) via reg_put rc.
15func task_dispatch_pfx(prefix: *u8, id: *u8, payload: *u8) -> i64 {
16 let rec: *u8 = sys_mmap(TASK_MAGIC_4096); var o: i64=0
17 o=tr_cat(rec,o,id); o=tr_tab(rec,o); o=tr_cat(rec,o,"pending" as *u8); o=tr_tab(rec,o); o=tr_cat(rec,o,payload); o=tr_tab(rec,o)
18 return reg_put(prefix, TASK_KP, TASK_IDX, id, rec, o)
19}
20
21// set status (+ result), preserving id + payload from the current record. Writes a NEW version (history kept).
22func task_set_pfx(prefix: *u8, id: *u8, status: *u8, result: *u8) -> i64 {
23 let po: *i64=sys_mmap(16) as *i64; let lo: *i64=sys_mmap(16) as *i64
24 if reg_get(prefix, TASK_KP, id, po, lo) != 1 { return 0 - 1 }
25 let rec: *u8 = po[0] as *u8; let rlen: i64 = lo[0]
26 let f2: *i64 = sys_mmap(16) as *i64
27 let nrec: *u8 = sys_mmap(TASK_MAGIC_4096); var o: i64=0
28 o=tr_cat(nrec,o,id); o=tr_tab(nrec,o); o=tr_cat(nrec,o,status); o=tr_tab(nrec,o)
29 if tr_field(rec, 0, rlen, 2, f2)==1 { var i: i64=0; while i<f2[1] { nrec[o]=rec[f2[0]+i]; o=o+1; i=i+1 } }
30 o=tr_tab(nrec,o); o=tr_cat(nrec,o,result)
31 return reg_put(prefix, TASK_KP, TASK_IDX, id, nrec, o)
32}
33
34// poll: copy the current status field into sbuf (NUL-terminated); returns its length (0 if absent).
35func task_status_pfx(prefix: *u8, id: *u8, sbuf: *u8) -> i64 {
36 let po: *i64=sys_mmap(16) as *i64; let lo: *i64=sys_mmap(16) as *i64
37 if reg_get(prefix, TASK_KP, id, po, lo) != 1 { sbuf[0]=0 as u8; return 0 }
38 let rec: *u8 = po[0] as *u8; let rlen: i64 = lo[0]
39 let f2: *i64 = sys_mmap(16) as *i64
40 if tr_field(rec, 0, rlen, 1, f2)==0 { sbuf[0]=0 as u8; return 0 }
41 var i: i64=0; while i<f2[1] { sbuf[i]=rec[f2[0]+i]; i=i+1 } sbuf[f2[1]]=0 as u8
42 return f2[1]
43}
44
45func task_get_pfx(prefix: *u8, id: *u8, ptrout: *i64, lenout: *i64) -> i64 { return reg_get(prefix, TASK_KP, id, ptrout, lenout) }
46func task_list_pfx(prefix: *u8, idxbuf: *u8, cap: i64) -> i64 { return reg_index(prefix, TASK_IDX, idxbuf, cap) }
47
48// production-prefix wrappers
49func task_dispatch(id: *u8, payload: *u8) -> i64 { return task_dispatch_pfx(TASK_PREFIX, id, payload) }
50func task_set(id: *u8, status: *u8, result: *u8) -> i64 { return task_set_pfx(TASK_PREFIX, id, status, result) }
51func task_status(id: *u8, sbuf: *u8) -> i64 { return task_status_pfx(TASK_PREFIX, id, sbuf) }