Mercurial

comparison mrjunejune/inference_bridge.c @ 260:1f9877b637e9

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Add Copilot-powered cyberpunk JRPG chat Integrate the production JRPG chat with Seobeo streaming, Deita persistence, and a Bazel-managed Copilot SDK and LiteLLM inference stack. Co-authored-by: Copilot <[email protected]>
author MrJuneJune <mrjunejune@users.noreply.github.com>
date Wed, 05 Aug 2026 09:19:41 -0700
parents
children b401627fc49e
comparison
equal deleted inserted replaced
259:667156fcd3e3 260:1f9877b637e9
1 #include "mrjunejune/inference_bridge.h"
2
3 #include <errno.h>
4 #include <pthread.h>
5 #include <signal.h>
6 #include <stdatomic.h>
7 #include <stdio.h>
8 #include <stdlib.h>
9 #include <string.h>
10 #include <sys/wait.h>
11 #include <time.h>
12 #include <unistd.h>
13
14 #define INFERENCE_EVENT_MAX (1024 * 1024 + 4096)
15 #define INFERENCE_READY_TIMEOUT_MS 15000
16
17 struct Inference_Bridge {
18 char *sidecar_path;
19 char *copilot_cli_path;
20 FILE *p_commands;
21 FILE *p_events;
22 pid_t child_pid;
23 pthread_t reader_thread;
24 pthread_mutex_t write_mutex;
25 pthread_mutex_t ready_mutex;
26 pthread_cond_t ready_condition;
27 boolean write_mutex_initialized;
28 boolean ready_mutex_initialized;
29 boolean ready_condition_initialized;
30 boolean reader_started;
31 _Atomic boolean running;
32 _Atomic boolean ready;
33 Inference_Event_Handler handler;
34 void *p_user_data;
35 };
36
37 static int64 Inference_Monotonic_Milliseconds(void)
38 {
39 struct timespec now;
40 clock_gettime(CLOCK_MONOTONIC, &now);
41 return (int64)now.tv_sec * 1000 + now.tv_nsec / 1000000;
42 }
43
44 static const char *Inference_JSON_String(
45 Dowa_JSON_Entry *object,
46 const char *key)
47 {
48 char *value = Dowa_JSON_Get_String(object, key);
49 return value ? value : "";
50 }
51
52 static boolean Inference_JSON_Boolean(
53 Dowa_JSON_Entry *object,
54 const char *key)
55 {
56 Dowa_JSON_Value *p_value = Dowa_JSON_Get(object, key);
57 return p_value && p_value->type == DOWA_JSON_BOOL
58 ? p_value->bool_val
59 : FALSE;
60 }
61
62 static int64 Inference_JSON_Integer(
63 Dowa_JSON_Entry *object,
64 const char *key)
65 {
66 Dowa_JSON_Value *p_value = Dowa_JSON_Get(object, key);
67 return p_value && p_value->type == DOWA_JSON_NUMBER
68 ? (int64)p_value->num_val
69 : 0;
70 }
71
72 static void Inference_Bridge_Handle_Line(
73 Inference_Bridge *p_bridge,
74 const char *line,
75 size_t length)
76 {
77 Dowa_Arena *p_arena = Dowa_Arena_Create(length * 3 + 4096);
78 if (!p_arena)
79 return;
80 Dowa_JSON_Value parsed = Dowa_JSON_Parse(line, (int32)length, p_arena);
81 if (parsed.type != DOWA_JSON_OBJECT)
82 {
83 Dowa_Arena_Free(p_arena);
84 return;
85 }
86 Dowa_JSON_Entry *object = parsed.object_val;
87 const char *type = Inference_JSON_String(object, "type");
88 if (strcmp(type, "ready") == 0)
89 {
90 atomic_store(&p_bridge->ready, TRUE);
91 pthread_mutex_lock(&p_bridge->ready_mutex);
92 pthread_cond_broadcast(&p_bridge->ready_condition);
93 pthread_mutex_unlock(&p_bridge->ready_mutex);
94 }
95
96 Inference_Event event = {
97 .type = type,
98 .request_id = Inference_JSON_String(object, "request_id"),
99 .conversation_id = Inference_JSON_String(object, "conversation_id"),
100 .delta = Inference_JSON_String(object, "delta"),
101 .content = Inference_JSON_String(object, "content"),
102 .failed = Inference_JSON_Boolean(object, "failed"),
103 .aborted = Inference_JSON_Boolean(object, "aborted"),
104 };
105
106 Dowa_JSON_Value *p_error = Dowa_JSON_Get(object, "error");
107 if (p_error && p_error->type == DOWA_JSON_OBJECT)
108 {
109 Dowa_JSON_Entry *error = p_error->object_val;
110 event.error_code = Inference_JSON_String(error, "code");
111 event.error_message = Inference_JSON_String(error, "message");
112 }
113 else
114 {
115 event.error_code = "";
116 event.error_message = "";
117 }
118
119 Dowa_JSON_Value *p_usage = Dowa_JSON_Get(object, "usage");
120 if (p_usage && p_usage->type == DOWA_JSON_OBJECT)
121 {
122 Dowa_JSON_Entry *usage = p_usage->object_val;
123 event.input_tokens = Inference_JSON_Integer(usage, "input_tokens");
124 event.output_tokens = Inference_JSON_Integer(usage, "output_tokens");
125 }
126
127 if (p_bridge->handler)
128 p_bridge->handler(&event, p_bridge->p_user_data);
129 Dowa_Arena_Free(p_arena);
130 }
131
132 static void *Inference_Bridge_Read_Events(void *p_context)
133 {
134 Inference_Bridge *p_bridge = p_context;
135 char *line = NULL;
136 size_t capacity = 0;
137 while (atomic_load(&p_bridge->running))
138 {
139 ssize_t amount = getline(&line, &capacity, p_bridge->p_events);
140 if (amount < 0)
141 break;
142 if ((size_t)amount > INFERENCE_EVENT_MAX)
143 continue;
144 size_t length = (size_t)amount;
145 while (length > 0 &&
146 (line[length - 1] == '\n' || line[length - 1] == '\r'))
147 length--;
148 if (length > 0)
149 Inference_Bridge_Handle_Line(p_bridge, line, length);
150 }
151 free(line);
152 atomic_store(&p_bridge->ready, FALSE);
153 atomic_store(&p_bridge->running, FALSE);
154 if (p_bridge->handler)
155 {
156 Inference_Event closed = {
157 .type = "bridge.closed",
158 .error_code = "sidecar_closed",
159 .error_message = "Inference sidecar connection closed",
160 .failed = TRUE,
161 };
162 p_bridge->handler(&closed, p_bridge->p_user_data);
163 }
164 pthread_mutex_lock(&p_bridge->ready_mutex);
165 pthread_cond_broadcast(&p_bridge->ready_condition);
166 pthread_mutex_unlock(&p_bridge->ready_mutex);
167 return NULL;
168 }
169
170 static boolean Inference_Bridge_Write(
171 Inference_Bridge *p_bridge,
172 const char *payload)
173 {
174 if (!p_bridge || !payload || !atomic_load(&p_bridge->running))
175 return FALSE;
176 pthread_mutex_lock(&p_bridge->write_mutex);
177 boolean success =
178 fputs(payload, p_bridge->p_commands) >= 0 &&
179 fputc('\n', p_bridge->p_commands) != EOF &&
180 fflush(p_bridge->p_commands) == 0;
181 pthread_mutex_unlock(&p_bridge->write_mutex);
182 return success;
183 }
184
185 static boolean Inference_Bridge_Command(
186 Inference_Bridge *p_bridge,
187 const char *command,
188 const char *request_id,
189 const char *conversation_id,
190 const char *prompt)
191 {
192 if (!command || !request_id)
193 return FALSE;
194 size_t input_length =
195 strlen(command) + strlen(request_id) +
196 strlen(conversation_id ? conversation_id : "") +
197 strlen(prompt ? prompt : "");
198 if (input_length > (((size_t)-1) - 4096) / 12)
199 return FALSE;
200 Dowa_Arena *p_arena = Dowa_Arena_Create(input_length * 12 + 4096);
201 if (!p_arena)
202 return FALSE;
203 char *escaped_command = Dowa_JSON_Escape_String(command, 0, p_arena);
204 char *escaped_request = Dowa_JSON_Escape_String(request_id, 0, p_arena);
205 char *escaped_conversation = Dowa_JSON_Escape_String(
206 conversation_id ? conversation_id : "", 0, p_arena);
207 char *escaped_prompt = prompt
208 ? Dowa_JSON_Escape_String(prompt, 0, p_arena)
209 : NULL;
210 if (!escaped_command || !escaped_request || !escaped_conversation ||
211 (prompt && !escaped_prompt))
212 {
213 Dowa_Arena_Free(p_arena);
214 return FALSE;
215 }
216 size_t capacity =
217 strlen(escaped_command) + strlen(escaped_request) +
218 strlen(escaped_conversation) +
219 (escaped_prompt ? strlen(escaped_prompt) : 0) + 160;
220 char *payload = Dowa_Arena_Allocate(p_arena, capacity);
221 if (!payload)
222 {
223 Dowa_Arena_Free(p_arena);
224 return FALSE;
225 }
226 if (escaped_prompt)
227 {
228 snprintf(
229 payload,
230 capacity,
231 "{\"command\":\"%s\",\"request_id\":\"%s\","
232 "\"conversation_id\":\"%s\",\"prompt\":\"%s\"}",
233 escaped_command,
234 escaped_request,
235 escaped_conversation,
236 escaped_prompt);
237 }
238 else
239 {
240 snprintf(
241 payload,
242 capacity,
243 "{\"command\":\"%s\",\"request_id\":\"%s\","
244 "\"conversation_id\":\"%s\"}",
245 escaped_command,
246 escaped_request,
247 escaped_conversation);
248 }
249 boolean success = Inference_Bridge_Write(p_bridge, payload);
250 Dowa_Arena_Free(p_arena);
251 return success;
252 }
253
254 Inference_Bridge *Inference_Bridge_Create(
255 const char *sidecar_path,
256 const char *copilot_cli_path,
257 Inference_Event_Handler handler,
258 void *p_user_data)
259 {
260 if (!sidecar_path || !copilot_cli_path)
261 return NULL;
262 Inference_Bridge *p_bridge = calloc(1, sizeof(*p_bridge));
263 if (!p_bridge)
264 return NULL;
265 p_bridge->sidecar_path = strdup(sidecar_path);
266 p_bridge->copilot_cli_path = strdup(copilot_cli_path);
267 p_bridge->handler = handler;
268 p_bridge->p_user_data = p_user_data;
269 p_bridge->child_pid = -1;
270 if (!p_bridge->sidecar_path || !p_bridge->copilot_cli_path)
271 {
272 Inference_Bridge_Destroy(p_bridge);
273 return NULL;
274 }
275 if (pthread_mutex_init(&p_bridge->write_mutex, NULL) != 0)
276 {
277 Inference_Bridge_Destroy(p_bridge);
278 return NULL;
279 }
280 p_bridge->write_mutex_initialized = TRUE;
281 if (pthread_mutex_init(&p_bridge->ready_mutex, NULL) != 0)
282 {
283 Inference_Bridge_Destroy(p_bridge);
284 return NULL;
285 }
286 p_bridge->ready_mutex_initialized = TRUE;
287 if (pthread_cond_init(&p_bridge->ready_condition, NULL) != 0)
288 {
289 Inference_Bridge_Destroy(p_bridge);
290 return NULL;
291 }
292 p_bridge->ready_condition_initialized = TRUE;
293 return p_bridge;
294 }
295
296 static void Inference_Bridge_Stop_Process(Inference_Bridge *p_bridge)
297 {
298 if (!p_bridge)
299 return;
300 if (atomic_load(&p_bridge->running) && p_bridge->p_commands)
301 Inference_Bridge_Command(
302 p_bridge, "shutdown", "server-shutdown", "", NULL);
303 if (p_bridge->p_commands)
304 {
305 fclose(p_bridge->p_commands);
306 p_bridge->p_commands = NULL;
307 }
308
309 if (p_bridge->child_pid > 0)
310 {
311 int status = 0;
312 pid_t result = 0;
313 for (int attempt = 0; attempt < 20; attempt++)
314 {
315 result = waitpid(p_bridge->child_pid, &status, WNOHANG);
316 if (result != 0)
317 break;
318 usleep(50000);
319 }
320 if (result == 0)
321 {
322 kill(p_bridge->child_pid, SIGTERM);
323 waitpid(p_bridge->child_pid, &status, 0);
324 }
325 p_bridge->child_pid = -1;
326 }
327 atomic_store(&p_bridge->running, FALSE);
328 atomic_store(&p_bridge->ready, FALSE);
329 if (p_bridge->reader_started)
330 {
331 pthread_join(p_bridge->reader_thread, NULL);
332 p_bridge->reader_started = FALSE;
333 }
334 if (p_bridge->p_events)
335 {
336 fclose(p_bridge->p_events);
337 p_bridge->p_events = NULL;
338 }
339 }
340
341 boolean Inference_Bridge_Start(Inference_Bridge *p_bridge)
342 {
343 if (!p_bridge || atomic_load(&p_bridge->running))
344 return FALSE;
345 int commands[2];
346 int events[2];
347 if (pipe(commands) != 0)
348 return FALSE;
349 if (pipe(events) != 0)
350 {
351 close(commands[0]);
352 close(commands[1]);
353 return FALSE;
354 }
355
356 pid_t child = fork();
357 if (child < 0)
358 {
359 close(commands[0]);
360 close(commands[1]);
361 close(events[0]);
362 close(events[1]);
363 return FALSE;
364 }
365 if (child == 0)
366 {
367 dup2(commands[0], STDIN_FILENO);
368 dup2(events[1], STDOUT_FILENO);
369 close(commands[0]);
370 close(commands[1]);
371 close(events[0]);
372 close(events[1]);
373 execl(
374 p_bridge->sidecar_path,
375 p_bridge->sidecar_path,
376 p_bridge->copilot_cli_path,
377 NULL);
378 _exit(127);
379 }
380
381 close(commands[0]);
382 close(events[1]);
383 p_bridge->p_commands = fdopen(commands[1], "w");
384 p_bridge->p_events = fdopen(events[0], "r");
385 if (!p_bridge->p_commands || !p_bridge->p_events)
386 {
387 if (p_bridge->p_commands)
388 fclose(p_bridge->p_commands);
389 else
390 close(commands[1]);
391 if (p_bridge->p_events)
392 fclose(p_bridge->p_events);
393 else
394 close(events[0]);
395 kill(child, SIGTERM);
396 waitpid(child, NULL, 0);
397 return FALSE;
398 }
399 setvbuf(p_bridge->p_commands, NULL, _IOLBF, 0);
400 p_bridge->child_pid = child;
401 atomic_store(&p_bridge->running, TRUE);
402 atomic_store(&p_bridge->ready, FALSE);
403 if (pthread_create(
404 &p_bridge->reader_thread,
405 NULL,
406 Inference_Bridge_Read_Events,
407 p_bridge) != 0)
408 {
409 atomic_store(&p_bridge->running, FALSE);
410 fclose(p_bridge->p_commands);
411 fclose(p_bridge->p_events);
412 kill(child, SIGTERM);
413 waitpid(child, NULL, 0);
414 p_bridge->p_commands = NULL;
415 p_bridge->p_events = NULL;
416 p_bridge->child_pid = -1;
417 return FALSE;
418 }
419 p_bridge->reader_started = TRUE;
420
421 int64 deadline =
422 Inference_Monotonic_Milliseconds() + INFERENCE_READY_TIMEOUT_MS;
423 pthread_mutex_lock(&p_bridge->ready_mutex);
424 while (atomic_load(&p_bridge->running) &&
425 !atomic_load(&p_bridge->ready))
426 {
427 int64 remaining = deadline - Inference_Monotonic_Milliseconds();
428 if (remaining <= 0)
429 break;
430 struct timespec timeout;
431 clock_gettime(CLOCK_REALTIME, &timeout);
432 timeout.tv_sec += remaining / 1000;
433 timeout.tv_nsec += (remaining % 1000) * 1000000;
434 if (timeout.tv_nsec >= 1000000000)
435 {
436 timeout.tv_sec++;
437 timeout.tv_nsec -= 1000000000;
438 }
439 pthread_cond_timedwait(
440 &p_bridge->ready_condition, &p_bridge->ready_mutex, &timeout);
441 }
442 boolean ready = atomic_load(&p_bridge->ready);
443 pthread_mutex_unlock(&p_bridge->ready_mutex);
444 if (!ready)
445 {
446 Inference_Bridge_Stop_Process(p_bridge);
447 return FALSE;
448 }
449 return TRUE;
450 }
451
452 boolean Inference_Bridge_Is_Ready(const Inference_Bridge *p_bridge)
453 {
454 return p_bridge &&
455 atomic_load(&p_bridge->running) &&
456 atomic_load(&p_bridge->ready);
457 }
458
459 boolean Inference_Bridge_Start_Turn(
460 Inference_Bridge *p_bridge,
461 const char *request_id,
462 const char *conversation_id,
463 const char *prompt)
464 {
465 return Inference_Bridge_Command(
466 p_bridge, "turn.start", request_id, conversation_id, prompt);
467 }
468
469 boolean Inference_Bridge_Abort_Turn(
470 Inference_Bridge *p_bridge,
471 const char *request_id,
472 const char *conversation_id)
473 {
474 return Inference_Bridge_Command(
475 p_bridge, "turn.abort", request_id, conversation_id, NULL);
476 }
477
478 boolean Inference_Bridge_Delete_Conversation(
479 Inference_Bridge *p_bridge,
480 const char *request_id,
481 const char *conversation_id)
482 {
483 return Inference_Bridge_Command(
484 p_bridge, "conversation.delete", request_id, conversation_id, NULL);
485 }
486
487 void Inference_Bridge_Destroy(Inference_Bridge *p_bridge)
488 {
489 if (!p_bridge)
490 return;
491 Inference_Bridge_Stop_Process(p_bridge);
492 if (p_bridge->ready_condition_initialized)
493 pthread_cond_destroy(&p_bridge->ready_condition);
494 if (p_bridge->ready_mutex_initialized)
495 pthread_mutex_destroy(&p_bridge->ready_mutex);
496 if (p_bridge->write_mutex_initialized)
497 pthread_mutex_destroy(&p_bridge->write_mutex);
498 free(p_bridge->sidecar_path);
499 free(p_bridge->copilot_cli_path);
500 free(p_bridge);
501 }