nx_dual_companion_compose_test.nx source
↩ module page · 200 lines · 8624 B
1// nx_dual_companion_compose_test.nx -- two-actor concurrent inference proof.
2//
3// The parallel-companion north star concretized: NARRATOR + NPC actors
4// running interleaved cooperative steps, each generating 3 tokens
5// autoregressively, output ordered through one multiplexer.
6//
7// What this proves end-to-end:
8// 1. Two independent LLM-actor contexts can share the same scheduler +
9// message bus + multiplexer without state corruption.
10// 2. Cooperative pick_next interleaves them by priority + fairness.
11// 3. nx_lr_actor_reset_for_next_token lets each actor roll into the
12// next token using its own previous output (autoregressive loop).
13// 4. Multiplexer correctly orders concurrent tokens by stream class
14// priority (NARRATOR LLM_TEXT > NPC LLM_TEXT same class, FIFO
15// tie-break + arrival-time tracking).
16// 5. The other-agent's embedding + matmul + sample primitives are
17// invoked 12 times (6 tokens x 2 actors x 3 primitive calls each)
18// without leak / overflow / state cross-contamination.
19//
20// This is the smallest substrate-honest demonstration of the
21// [[feedback-parallel-companion-multimodal-dnd-real-time]] cardinal.
22
23import "nx_syscalls.nx"
24import "nx_tier.nx"
25import "nx_tensor.nx"
26import "nx_actor.nx"
27import "nx_message.nx"
28import "nx_stream_multiplexer.nx"
29import "nx_actor_role_llm.nx"
30
31// Fill helper -- deterministic pattern so each row of the embedding
32// table is distinct and matmul output is reproducible.
33func _fill_q10(t: *NxTensor, base: i64, stride: i64) -> i64 {
34 let p: *i64 = t.storage as *i64
35 var i: nx_int = 0
36 while i < t.numel {
37 p[i] = base + (i as i64) * stride
38 i = i + 1
39 }
40 return 0
41}
42
43func main() -> i64 {
44 let now: nx_size = 1000000
45 let TOKENS_PER_ACTOR: nx_int = 3
46
47 // 1: build shared synthetic 8-vocab x 4-hidden weights
48 let shape_buf: *u8 = sys_mmap(16)
49 let shape: *i64 = shape_buf as *i64
50 shape[0] = 8
51 shape[1] = 4
52 let err_buf: *u8 = sys_mmap(8)
53 let err_p: *i64 = err_buf as *i64
54 let embed_table: *NxTensor = nx_t_alloc(NX_DT_I64, shape, 2, err_p)
55 if err_p[0] != NX_T_OK { return 1 }
56 _fill_q10(embed_table, 50, 3)
57
58 let shape2_buf: *u8 = sys_mmap(16)
59 let shape2: *i64 = shape2_buf as *i64
60 shape2[0] = 4
61 shape2[1] = 8
62 let out_proj: *NxTensor = nx_t_alloc(NX_DT_I64, shape2, 2, err_p)
63 if err_p[0] != NX_T_OK { return 2 }
64 _fill_q10(out_proj, 30, 2)
65
66 // 2: scheduler + message bus + multiplexer
67 let sched: *NxActorScheduler = nx_ac_sched_new(8, 1000000, 16, now)
68 let bus: *NxMessageBus = nx_ms_bus_new(8, 16, 16)
69 let mux: *NxStreamMultiplexer = nx_sm_new(32, 50000)
70
71 // 3: spawn NARRATOR (priority 80) + NPC (priority 60) + LISTENER (priority 50)
72 let NARR: nx_int = 8001
73 let NPC: nx_int = 8002
74 let LIS: nx_int = 8099
75 nx_ac_spawn(sched, NARR, 1, 80, 0, now)
76 nx_ac_spawn(sched, NPC, 1, 60, 0, now)
77 nx_ac_spawn(sched, LIS, 4, 50, 0, now)
78 nx_ms_register(bus, NARR)
79 nx_ms_register(bus, NPC)
80 nx_ms_register(bus, LIS)
81 nx_ms_subscribe(bus, LIS, NX_MS_KIND_LLM_TOKEN)
82
83 // 4: two LLM actor contexts with different initial prompts + seeds
84 let narr_ctx: *NxLlmActorCtx = nx_lr_actor_new(embed_table, out_proj, 1, 8, 4, 1024, 3, 17)
85 let npc_ctx: *NxLlmActorCtx = nx_lr_actor_new(embed_table, out_proj, 5, 8, 4, 1024, 3, 23)
86 if (narr_ctx as i64) == 0 { return 3 }
87 if (npc_ctx as i64) == 0 { return 4 }
88
89 // 5: token-history arrays for verification
90 let narr_hist_buf: *u8 = sys_mmap(TOKENS_PER_ACTOR * 8)
91 let narr_hist: *i64 = narr_hist_buf as *i64
92 let npc_hist_buf: *u8 = sys_mmap(TOKENS_PER_ACTOR * 8)
93 let npc_hist: *i64 = npc_hist_buf as *i64
94
95 // 6: drive both actors interleaved for 3 tokens each.
96 // Strategy: for each token round (1..3):
97 // * run NARRATOR through 5 phases -> emit token
98 // * push narrator's token to multiplexer
99 // * reset narrator with new prompt = generated token
100 // * run NPC through 5 phases -> emit token
101 // * push npc's token to multiplexer
102 // * reset npc with new prompt
103 //
104 // This is cooperative-interleaved (not preemptive) which matches
105 // NishiLang/qemu's actual concurrency model.
106 var round: nx_int = 0
107 var tick: nx_size = now + 100
108 while round < TOKENS_PER_ACTOR {
109 // NARRATOR token round
110 var p: nx_int = 0
111 while p < 5 { // INIT->EMBED->PROJ->SAMPLE->EMIT->DONE = 5 steps
112 nx_lr_actor_step(narr_ctx, sched, bus, NARR, tick)
113 tick = tick + 50
114 p = p + 1
115 }
116 let narr_tok: nx_int = nx_lr_actor_token(narr_ctx)
117 if narr_tok < 0 { return 10 + round }
118 if narr_tok >= 8 { return 20 + round }
119 narr_hist[round] = narr_tok as i64
120 nx_sm_push(mux, NARR, NX_SM_CLASS_LLM_TEXT, 100 + round, 8, narr_tok as nx_size, tick)
121
122 // NPC token round
123 var q: nx_int = 0
124 while q < 5 {
125 nx_lr_actor_step(npc_ctx, sched, bus, NPC, tick)
126 tick = tick + 50
127 q = q + 1
128 }
129 let npc_tok: nx_int = nx_lr_actor_token(npc_ctx)
130 if npc_tok < 0 { return 30 + round }
131 if npc_tok >= 8 { return 40 + round }
132 npc_hist[round] = npc_tok as i64
133 nx_sm_push(mux, NPC, NX_SM_CLASS_LLM_TEXT, 200 + round, 8, npc_tok as nx_size, tick)
134
135 // Reset both for next token (unless this was last round)
136 if round < (TOKENS_PER_ACTOR - 1) {
137 if nx_lr_actor_reset_for_next_token(narr_ctx, sched, NARR, narr_tok) != NX_LR_AC_V_STEPPED { return 50 }
138 if nx_lr_actor_reset_for_next_token(npc_ctx, sched, NPC, npc_tok) != NX_LR_AC_V_STEPPED { return 51 }
139 }
140 round = round + 1
141 }
142
143 // 7: both actors should be COMPLETED (final phase, no further reset)
144 let na: *NxActor = nx_ac_find(sched, NARR)
145 let nb: *NxActor = nx_ac_find(sched, NPC)
146 if na.state != NX_AC_STATE_COMPLETED { return 60 }
147 if nb.state != NX_AC_STATE_COMPLETED { return 61 }
148 if nx_lr_actor_is_done(narr_ctx) != 1 { return 62 }
149 if nx_lr_actor_is_done(npc_ctx) != 1 { return 63 }
150
151 // 8: 6 tokens in multiplexer total (3 narrator + 3 npc)
152 if nx_sm_pending_count(mux) != 6 { return 64 }
153
154 // 9: LISTENER mailbox should have 6 LLM_TOKEN messages (3 fanouts each)
155 if nx_ms_pending(bus, LIS) != 6 { return 65 }
156 // Drain + verify each msg carries a valid token
157 var drained: nx_int = 0
158 while drained < 6 {
159 let m: *NxMessage = nx_ms_receive(bus, LIS)
160 if (m as i64) == 0 { return 70 + drained }
161 if m.kind != NX_MS_KIND_LLM_TOKEN { return 80 + drained }
162 let t: nx_int = m.payload_handle as nx_int
163 if t < 0 { return 90 + drained }
164 if t >= 8 { return 100 + drained }
165 drained = drained + 1
166 }
167
168 // 10: refused reset paths
169 let stray_ctx: *NxLlmActorCtx = nx_lr_actor_new(embed_table, out_proj, 0, 8, 4, 1024, 3, 99)
170 // Not in DONE -> INVALID
171 if nx_lr_actor_reset_for_next_token(stray_ctx, sched, 9999, 1) != NX_LR_AC_V_INVALID { return 110 }
172 // Force it to DONE then try invalid token id
173 stray_ctx.current_phase = NX_LR_PHASE_DONE
174 if nx_lr_actor_reset_for_next_token(stray_ctx, sched, 9999, -1) != NX_LR_AC_V_INVALID { return 111 }
175 if nx_lr_actor_reset_for_next_token(stray_ctx, sched, 9999, 8) != NX_LR_AC_V_INVALID { return 112 }
176 if nx_lr_actor_reset_for_next_token(stray_ctx, sched, 9999, 7) != NX_LR_AC_V_STEPPED { return 113 }
177
178 // 11: null guard
179 let null_ctx: *NxLlmActorCtx = (0 as i64) as *NxLlmActorCtx
180 if nx_lr_actor_reset_for_next_token(null_ctx, sched, 1, 1) != NX_LR_AC_V_NULL { return 114 }
181
182 // 12: drain multiplexer + verify NARRATOR tokens come out first (LLM_TEXT priority is
183 // same for both -- but starvation-boost favors older chunks). We pushed narrator
184 // first each round so older arrival wins tie-break: token positions in delivery
185 // should be NARR_T0, NPC_T0, NARR_T1, NPC_T1, NARR_T2, NPC_T2 in arrival order.
186 let oc_buf: *u8 = sys_mmap(24)
187 let oc: *i64 = oc_buf as *i64
188 let os: *i64 = (oc_buf + 8) as *i64
189 let oid: *i64 = (oc_buf + 16) as *i64
190 var consumed: nx_int = 0
191 while consumed < 6 {
192 let v: nx_int = nx_sm_next(mux, oc, os, oid, tick + (consumed as nx_size) * 100)
193 if v != NX_SM_V_DELIVERED { return 120 + consumed }
194 if oc[0] != (NX_SM_CLASS_LLM_TEXT as i64) { return 130 + consumed }
195 consumed = consumed + 1
196 }
197 if nx_sm_pending_count(mux) != 0 { return 140 }
198
199 return 0
200}