nx_message.nx source
↩ module page · 351 lines · 12793 B
1// nx_message.nx -- typed inter-actor message passing substrate.
2//
3// Per CARDINAL [[feedback-parallel-companion-multimodal-dnd-real-time]]:
4// concurrent actors need to communicate. The LLM-actor emits tokens
5// that the audio-TTS-actor + the on-screen-text-renderer-actor both
6// consume. The game-action-actor emits dice rolls that the LLM-actor
7// receives as context. Without typed message passing, every actor
8// shares mutable state -- classic data-race anti-pattern.
9//
10// V1 ships sealed message-kind enum + actor-addressed mailbox + ring
11// buffer + fan-out (one sender, multiple subscribers) + delivery
12// verdict + content-addressed payload handles.
13
14import "nx_syscalls.nx"
15import "nx_tier.nx"
16
17// ===== Sealed enum: NxMessageKind =================================
18
19const NX_MS_KIND_LLM_TOKEN: nx_int = 0
20const NX_MS_KIND_LLM_DONE: nx_int = 1
21const NX_MS_KIND_TOOL_CALL: nx_int = 2
22const NX_MS_KIND_TOOL_RESULT: nx_int = 3
23const NX_MS_KIND_IMAGE_TILE: nx_int = 4
24const NX_MS_KIND_AUDIO_FRAME: nx_int = 5
25const NX_MS_KIND_GAME_ACTION: nx_int = 6
26const NX_MS_KIND_GAME_STATE: nx_int = 7
27const NX_MS_KIND_OPERATOR_INPUT: nx_int = 8
28const NX_MS_KIND_SYSTEM_EVENT: nx_int = 9
29const NX_MS_KIND_ERROR_REPORT: nx_int = 10
30const NX_MS_KIND_N: nx_int = 11
31
32// ===== Sealed enum: NxDeliveryMode ================================
33
34const NX_MS_MODE_UNICAST: nx_int = 0 // exactly one recipient
35const NX_MS_MODE_FANOUT: nx_int = 1 // all subscribers
36const NX_MS_MODE_BROADCAST: nx_int = 2 // every actor in scheduler
37const NX_MS_MODE_N: nx_int = 3
38
39// ===== Sealed enum: NxMessageVerdict ==============================
40
41const NX_MS_V_DELIVERED: nx_int = 0
42const NX_MS_V_QUEUED: nx_int = 1 // recipient not yet ready
43const NX_MS_V_DROPPED_FULL: nx_int = 2 // mailbox full
44const NX_MS_V_DROPPED_NO_ROUTE: nx_int = 3 // no recipient subscribed
45const NX_MS_V_INVALID: nx_int = 4
46const NX_MS_V_NULL: nx_int = 5
47const NX_MS_V_N: nx_int = 6
48
49// ===== Struct: NxMessage ==========================================
50
51struct NxMessage {
52 message_id: nx_int,
53 sender_actor_id: nx_int,
54 recipient_actor_id: nx_int, // 0 if fanout/broadcast
55 kind: nx_int,
56 payload_handle: nx_size,
57 payload_bytes: nx_size,
58 sent_at_us: nx_size,
59 delivered_at_us: nx_size,
60}
61
62const NX_MS_M_BYTES: nx_int = 64
63
64struct NxMailbox {
65 actor_id: nx_int,
66 messages: *u8, // ring of NxMessage
67 head: nx_int,
68 tail: nx_int,
69 capacity: nx_int,
70 n_pending: nx_int,
71 n_delivered_total: nx_int,
72 n_dropped_total: nx_int,
73}
74
75const NX_MS_BX_BYTES: nx_int = 64
76
77// V1 subscription record: an actor subscribed to a specific message kind.
78// Topic = (kind). Future versions can add per-sender filters.
79struct NxSubscription {
80 actor_id: nx_int,
81 kind: nx_int,
82}
83
84const NX_MS_SU_BYTES: nx_int = 16
85
86struct NxMessageBus {
87 mailboxes: *u8,
88 n_mailboxes: nx_int,
89 mb_capacity: nx_int,
90 subscriptions: *u8,
91 n_subscriptions: nx_int,
92 sub_capacity: nx_int,
93 next_message_id: nx_int,
94 per_mailbox_capacity: nx_int,
95}
96
97const NX_MS_BU_BYTES: nx_int = 64
98
99// ===== Validators =================================================
100
101func nx_ms_kind_is_valid(k: nx_int) -> nx_int {
102 if k < 0 { return 0 }
103 if k >= NX_MS_KIND_N { return 0 }
104 return 1
105}
106
107func nx_ms_mode_is_valid(m: nx_int) -> nx_int {
108 if m < 0 { return 0 }
109 if m >= NX_MS_MODE_N { return 0 }
110 return 1
111}
112
113func nx_ms_v_is_valid(v: nx_int) -> nx_int {
114 if v < 0 { return 0 }
115 if v >= NX_MS_V_N { return 0 }
116 return 1
117}
118
119// ===== Constructor ================================================
120
121func nx_ms_bus_new(mb_capacity: nx_int,
122 sub_capacity: nx_int,
123 per_mailbox_capacity: nx_int) -> *NxMessageBus {
124 if mb_capacity <= 0 { return 0 as *NxMessageBus }
125 if sub_capacity <= 0 { return 0 as *NxMessageBus }
126 if per_mailbox_capacity <= 0 { return 0 as *NxMessageBus }
127 let raw: *u8 = sys_mmap(NX_MS_BU_BYTES)
128 let b: *NxMessageBus = raw as *NxMessageBus
129 b.mailboxes = sys_mmap(mb_capacity * NX_MS_BX_BYTES)
130 b.n_mailboxes = 0
131 b.mb_capacity = mb_capacity
132 b.subscriptions = sys_mmap(sub_capacity * NX_MS_SU_BYTES)
133 b.n_subscriptions = 0
134 b.sub_capacity = sub_capacity
135 b.next_message_id = 1
136 b.per_mailbox_capacity = per_mailbox_capacity
137 return b
138}
139
140func _ms_mb_at(b: *NxMessageBus, idx: nx_int) -> *NxMailbox {
141 if idx < 0 { return 0 as *NxMailbox }
142 if idx >= b.n_mailboxes { return 0 as *NxMailbox }
143 let off: nx_int = idx * NX_MS_BX_BYTES
144 return (b.mailboxes + off) as *NxMailbox
145}
146
147func _ms_sub_at(b: *NxMessageBus, idx: nx_int) -> *NxSubscription {
148 if idx < 0 { return 0 as *NxSubscription }
149 if idx >= b.n_subscriptions { return 0 as *NxSubscription }
150 let off: nx_int = idx * NX_MS_SU_BYTES
151 return (b.subscriptions + off) as *NxSubscription
152}
153
154// ===== Mailbox registration =======================================
155
156func nx_ms_register(b: *NxMessageBus, actor_id: nx_int) -> nx_int {
157 if (b as i64) == 0 { return NX_MS_V_NULL }
158 if actor_id == 0 { return NX_MS_V_INVALID }
159 if b.n_mailboxes >= b.mb_capacity { return NX_MS_V_INVALID }
160 let off: nx_int = b.n_mailboxes * NX_MS_BX_BYTES
161 let mb: *NxMailbox = (b.mailboxes + off) as *NxMailbox
162 mb.actor_id = actor_id
163 mb.messages = sys_mmap(b.per_mailbox_capacity * NX_MS_M_BYTES)
164 mb.head = 0
165 mb.tail = 0
166 mb.capacity = b.per_mailbox_capacity
167 mb.n_pending = 0
168 mb.n_delivered_total = 0
169 mb.n_dropped_total = 0
170 b.n_mailboxes = b.n_mailboxes + 1
171 return NX_MS_V_DELIVERED
172}
173
174func nx_ms_find_mailbox(b: *NxMessageBus, actor_id: nx_int) -> *NxMailbox {
175 if (b as i64) == 0 { return 0 as *NxMailbox }
176 var i: nx_int = 0
177 while i < b.n_mailboxes {
178 let mb: *NxMailbox = _ms_mb_at(b, i)
179 if mb.actor_id == actor_id { return mb }
180 i = i + 1
181 }
182 return 0 as *NxMailbox
183}
184
185// ===== Subscribe ==================================================
186
187func nx_ms_subscribe(b: *NxMessageBus,
188 actor_id: nx_int,
189 kind: nx_int) -> nx_int {
190 if (b as i64) == 0 { return NX_MS_V_NULL }
191 if nx_ms_kind_is_valid(kind) == 0 { return NX_MS_V_INVALID }
192 if actor_id == 0 { return NX_MS_V_INVALID }
193 if b.n_subscriptions >= b.sub_capacity { return NX_MS_V_INVALID }
194 let off: nx_int = b.n_subscriptions * NX_MS_SU_BYTES
195 let s: *NxSubscription = (b.subscriptions + off) as *NxSubscription
196 s.actor_id = actor_id
197 s.kind = kind
198 b.n_subscriptions = b.n_subscriptions + 1
199 return NX_MS_V_DELIVERED
200}
201
202// ===== Deposit into mailbox ring ==================================
203
204func _ms_mb_push(mb: *NxMailbox,
205 message_id: nx_int,
206 sender: nx_int,
207 recipient: nx_int,
208 kind: nx_int,
209 payload_handle: nx_size,
210 payload_bytes: nx_size,
211 now_us: nx_size) -> nx_int {
212 if mb.n_pending >= mb.capacity {
213 mb.n_dropped_total = mb.n_dropped_total + 1
214 return NX_MS_V_DROPPED_FULL
215 }
216 let off: nx_int = mb.tail * NX_MS_M_BYTES
217 let m: *NxMessage = (mb.messages + off) as *NxMessage
218 m.message_id = message_id
219 m.sender_actor_id = sender
220 m.recipient_actor_id = recipient
221 m.kind = kind
222 m.payload_handle = payload_handle
223 m.payload_bytes = payload_bytes
224 m.sent_at_us = now_us
225 m.delivered_at_us = now_us
226 mb.tail = (mb.tail + 1) % mb.capacity
227 mb.n_pending = mb.n_pending + 1
228 mb.n_delivered_total = mb.n_delivered_total + 1
229 return NX_MS_V_DELIVERED
230}
231
232// ===== Send (unicast / fanout / broadcast) ========================
233
234func nx_ms_send_unicast(b: *NxMessageBus,
235 sender: nx_int,
236 recipient: nx_int,
237 kind: nx_int,
238 payload_handle: nx_size,
239 payload_bytes: nx_size,
240 now_us: nx_size) -> nx_int {
241 if (b as i64) == 0 { return NX_MS_V_NULL }
242 if nx_ms_kind_is_valid(kind) == 0 { return NX_MS_V_INVALID }
243 let mb: *NxMailbox = nx_ms_find_mailbox(b, recipient)
244 if (mb as i64) == 0 { return NX_MS_V_DROPPED_NO_ROUTE }
245 let msg_id: nx_int = b.next_message_id
246 b.next_message_id = b.next_message_id + 1
247 return _ms_mb_push(mb, msg_id, sender, recipient, kind, payload_handle, payload_bytes, now_us)
248}
249
250func nx_ms_send_fanout(b: *NxMessageBus,
251 sender: nx_int,
252 kind: nx_int,
253 payload_handle: nx_size,
254 payload_bytes: nx_size,
255 now_us: nx_size) -> nx_int {
256 if (b as i64) == 0 { return NX_MS_V_NULL }
257 if nx_ms_kind_is_valid(kind) == 0 { return NX_MS_V_INVALID }
258 var any_delivered: nx_int = 0
259 var i: nx_int = 0
260 while i < b.n_subscriptions {
261 let s: *NxSubscription = _ms_sub_at(b, i)
262 if s.kind == kind {
263 let mb: *NxMailbox = nx_ms_find_mailbox(b, s.actor_id)
264 if (mb as i64) != 0 {
265 let msg_id: nx_int = b.next_message_id
266 b.next_message_id = b.next_message_id + 1
267 _ms_mb_push(mb, msg_id, sender, s.actor_id, kind, payload_handle, payload_bytes, now_us)
268 any_delivered = 1
269 }
270 }
271 i = i + 1
272 }
273 if any_delivered == 0 { return NX_MS_V_DROPPED_NO_ROUTE }
274 return NX_MS_V_DELIVERED
275}
276
277func nx_ms_send_broadcast(b: *NxMessageBus,
278 sender: nx_int,
279 kind: nx_int,
280 payload_handle: nx_size,
281 payload_bytes: nx_size,
282 now_us: nx_size) -> nx_int {
283 if (b as i64) == 0 { return NX_MS_V_NULL }
284 if nx_ms_kind_is_valid(kind) == 0 { return NX_MS_V_INVALID }
285 if b.n_mailboxes == 0 { return NX_MS_V_DROPPED_NO_ROUTE }
286 var i: nx_int = 0
287 while i < b.n_mailboxes {
288 let mb: *NxMailbox = _ms_mb_at(b, i)
289 if mb.actor_id != sender {
290 let msg_id: nx_int = b.next_message_id
291 b.next_message_id = b.next_message_id + 1
292 _ms_mb_push(mb, msg_id, sender, mb.actor_id, kind, payload_handle, payload_bytes, now_us)
293 }
294 i = i + 1
295 }
296 return NX_MS_V_DELIVERED
297}
298
299// ===== Receive ====================================================
300//
301// Pops the oldest message from recipient's mailbox. Returns a pointer
302// to the popped record in the mailbox ring. Caller may copy out
303// before the slot gets overwritten by the next push.
304
305func nx_ms_receive(b: *NxMessageBus, recipient: nx_int) -> *NxMessage {
306 if (b as i64) == 0 { return 0 as *NxMessage }
307 let mb: *NxMailbox = nx_ms_find_mailbox(b, recipient)
308 if (mb as i64) == 0 { return 0 as *NxMessage }
309 if mb.n_pending == 0 { return 0 as *NxMessage }
310 let off: nx_int = mb.head * NX_MS_M_BYTES
311 let m: *NxMessage = (mb.messages + off) as *NxMessage
312 mb.head = (mb.head + 1) % mb.capacity
313 mb.n_pending = mb.n_pending - 1
314 return m
315}
316
317// ===== Accessors ==================================================
318
319func nx_ms_pending(b: *NxMessageBus, recipient: nx_int) -> nx_int {
320 if (b as i64) == 0 { return 0 }
321 let mb: *NxMailbox = nx_ms_find_mailbox(b, recipient)
322 if (mb as i64) == 0 { return 0 }
323 return mb.n_pending
324}
325
326func nx_ms_delivered_total(b: *NxMessageBus, recipient: nx_int) -> nx_int {
327 if (b as i64) == 0 { return 0 }
328 let mb: *NxMailbox = nx_ms_find_mailbox(b, recipient)
329 if (mb as i64) == 0 { return 0 }
330 return mb.n_delivered_total
331}
332
333func nx_ms_dropped_total(b: *NxMessageBus, recipient: nx_int) -> nx_int {
334 if (b as i64) == 0 { return 0 }
335 let mb: *NxMailbox = nx_ms_find_mailbox(b, recipient)
336 if (mb as i64) == 0 { return 0 }
337 return mb.n_dropped_total
338}
339
340func nx_ms_subscription_count(b: *NxMessageBus, kind: nx_int) -> nx_int {
341 if (b as i64) == 0 { return 0 }
342 if nx_ms_kind_is_valid(kind) == 0 { return 0 }
343 var count: nx_int = 0
344 var i: nx_int = 0
345 while i < b.n_subscriptions {
346 let s: *NxSubscription = _ms_sub_at(b, i)
347 if s.kind == kind { count = count + 1 }
348 i = i + 1
349 }
350 return count
351}