689 lines
24 KiB
C
689 lines
24 KiB
C
#include "hygate.h"
|
|
#include <stdio.h>
|
|
#include <stdlib.h>
|
|
#include <string.h>
|
|
#include <unistd.h>
|
|
#include <time.h>
|
|
#include <errno.h>
|
|
#include <json-c/json.h>
|
|
#include <curl/curl.h>
|
|
|
|
/* ------------------------------------------------------------------ */
|
|
/* Streaming context */
|
|
/* ------------------------------------------------------------------ */
|
|
struct tcst {
|
|
char *key;
|
|
char *prev_args;
|
|
bool emitted;
|
|
};
|
|
|
|
typedef struct {
|
|
int client_fd;
|
|
bool streaming;
|
|
char completion_id[64];
|
|
long created;
|
|
const char *model;
|
|
strbuf line; /* SSE line buffer (used for both modes) */
|
|
strbuf emitted_content; /* streaming: content emitted so far */
|
|
strbuf emitted_reasoning;/* streaming: reasoning emitted so far */
|
|
struct tcst *tc_state;
|
|
int tc_state_n;
|
|
int tc_state_cap;
|
|
bool any_toolcall;
|
|
int include_usage;
|
|
char *final_content; /* non-streaming accumulation */
|
|
char *final_reasoning;
|
|
tool_call_t *final_tc;
|
|
int final_n_tc;
|
|
int status;
|
|
bool headers_sent;
|
|
} sse_ctx_t;
|
|
|
|
/* ------------------------------------------------------------------ */
|
|
/* Low-level response writers */
|
|
/* ------------------------------------------------------------------ */
|
|
static void send_sse_headers(int fd) {
|
|
const char *h =
|
|
"HTTP/1.1 200 OK\r\n"
|
|
"Content-Type: text/event-stream\r\n"
|
|
"Cache-Control: no-cache\r\n"
|
|
"Connection: keep-alive\r\n"
|
|
"Access-Control-Allow-Origin: *\r\n"
|
|
"Access-Control-Allow-Methods: GET, POST, OPTIONS\r\n"
|
|
"Access-Control-Allow-Headers: Content-Type, Authorization, api-key, X-User-Agent\r\n"
|
|
"\r\n";
|
|
write_all(fd, h, strlen(h));
|
|
}
|
|
|
|
static void send_options(int fd) {
|
|
const char *h =
|
|
"HTTP/1.1 200 OK\r\n"
|
|
"Access-Control-Allow-Origin: *\r\n"
|
|
"Access-Control-Allow-Methods: GET, POST, OPTIONS\r\n"
|
|
"Access-Control-Allow-Headers: Content-Type, Authorization, api-key, X-User-Agent\r\n"
|
|
"Connection: close\r\n"
|
|
"Content-Length: 0\r\n\r\n";
|
|
write_all(fd, h, strlen(h));
|
|
}
|
|
|
|
static void send_simple(int fd, int code, const char *codestr, const char *body) {
|
|
char header[512];
|
|
int hl = snprintf(header, sizeof header,
|
|
"HTTP/1.1 %d %s\r\n"
|
|
"Content-Type: text/plain; charset=utf-8\r\n"
|
|
"Access-Control-Allow-Origin: *\r\n"
|
|
"Access-Control-Allow-Methods: GET, POST, OPTIONS\r\n"
|
|
"Access-Control-Allow-Headers: Content-Type, Authorization, api-key, X-User-Agent\r\n"
|
|
"Connection: close\r\n"
|
|
"Content-Length: %zu\r\n\r\n", code, codestr, strlen(body));
|
|
write_all(fd, header, hl);
|
|
write_all(fd, body, strlen(body));
|
|
}
|
|
|
|
static void send_json(int fd, const char *json) {
|
|
char header[512];
|
|
int hl = snprintf(header, sizeof header,
|
|
"HTTP/1.1 200 OK\r\n"
|
|
"Content-Type: application/json\r\n"
|
|
"Access-Control-Allow-Origin: *\r\n"
|
|
"Access-Control-Allow-Methods: GET, POST, OPTIONS\r\n"
|
|
"Access-Control-Allow-Headers: Content-Type, Authorization, api-key, X-User-Agent\r\n"
|
|
"Connection: close\r\n"
|
|
"Content-Length: %zu\r\n\r\n", strlen(json));
|
|
write_all(fd, header, hl);
|
|
write_all(fd, json, strlen(json));
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ */
|
|
/* SSE chunk emitter */
|
|
/* ------------------------------------------------------------------ */
|
|
static void emit_chunk(sse_ctx_t *ctx, const char *role, const char *content,
|
|
const char *reasoning, const char *toolcalls_json,
|
|
const char *finish_reason) {
|
|
json_object *root = json_object_new_object();
|
|
json_object_object_add(root, "id", json_object_new_string(ctx->completion_id));
|
|
json_object_object_add(root, "object", json_object_new_string("chat.completion.chunk"));
|
|
json_object_object_add(root, "created", json_object_new_int64(ctx->created));
|
|
json_object_object_add(root, "model", json_object_new_string(ctx->model));
|
|
json_object *choices = json_object_new_array();
|
|
json_object *choice = json_object_new_object();
|
|
json_object_object_add(choice, "index", json_object_new_int(0));
|
|
json_object *delta = json_object_new_object();
|
|
if (role) json_object_object_add(delta, "role", json_object_new_string(role));
|
|
if (content) json_object_object_add(delta, "content", json_object_new_string(content));
|
|
if (reasoning) json_object_object_add(delta, "reasoning_content", json_object_new_string(reasoning));
|
|
if (toolcalls_json) {
|
|
json_object *tc = json_tokener_parse(toolcalls_json);
|
|
if (tc) json_object_object_add(delta, "tool_calls", tc);
|
|
}
|
|
json_object_object_add(choice, "delta", delta);
|
|
if (finish_reason)
|
|
json_object_object_add(choice, "finish_reason", json_object_new_string(finish_reason));
|
|
json_object_array_add(choices, choice);
|
|
json_object_object_add(root, "choices", choices);
|
|
|
|
const char *s = json_object_to_json_string(root);
|
|
if (s) {
|
|
size_t need = strlen(s) + 32;
|
|
char *buf = (char *)malloc(need);
|
|
int bl = snprintf(buf, need, "data: %s\n\n", s);
|
|
write_all(ctx->client_fd, buf, (size_t)bl);
|
|
free(buf);
|
|
}
|
|
json_object_put(root);
|
|
}
|
|
|
|
static void emit_usage_chunk(sse_ctx_t *ctx, int prompt_tokens, int completion_tokens) {
|
|
json_object *root = json_object_new_object();
|
|
json_object_object_add(root, "id", json_object_new_string(ctx->completion_id));
|
|
json_object_object_add(root, "object", json_object_new_string("chat.completion.chunk"));
|
|
json_object_object_add(root, "created", json_object_new_int64(ctx->created));
|
|
json_object_object_add(root, "model", json_object_new_string(ctx->model));
|
|
json_object *choices = json_object_new_array();
|
|
json_object_object_add(root, "choices", choices);
|
|
json_object *usage = json_object_new_object();
|
|
json_object_object_add(usage, "prompt_tokens", json_object_new_int(prompt_tokens));
|
|
json_object_object_add(usage, "completion_tokens", json_object_new_int(completion_tokens));
|
|
json_object_object_add(usage, "total_tokens",
|
|
json_object_new_int(prompt_tokens + completion_tokens));
|
|
json_object_object_add(root, "usage", usage);
|
|
const char *j = json_object_to_json_string(root);
|
|
if (j) {
|
|
size_t need = strlen(j) + 32;
|
|
char *buf = (char *)malloc(need);
|
|
int bl = snprintf(buf, need, "data: %s\n\n", j);
|
|
write_all(ctx->client_fd, buf, (size_t)bl);
|
|
free(buf);
|
|
}
|
|
json_object_put(root);
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ */
|
|
/* SSE line processing */
|
|
/* ------------------------------------------------------------------ */
|
|
static void handle_stream_delta(sse_ctx_t *ctx, char *c, char *r,
|
|
tool_call_t *tc, int ntc);
|
|
|
|
static void process_one_line(sse_ctx_t *ctx, char *line) {
|
|
while (*line == ' ' || *line == '\t' || *line == '\r' || *line == '\n') line++;
|
|
if (*line == 0) return;
|
|
char *e = line + strlen(line) - 1;
|
|
while (e > line && (*e == ' ' || *e == '\t' || *e == '\r' || *e == '\n')) *e-- = 0;
|
|
if (strncmp(line, "data: ", 6) != 0) return;
|
|
const char *payload = line + 6;
|
|
|
|
char *c = NULL, *r = NULL;
|
|
tool_call_t *tc = NULL;
|
|
int ntc = 0;
|
|
if (!parse_gradio_stream(payload, &c, &r, &tc, &ntc)) {
|
|
free(c);
|
|
free(r);
|
|
free_toolcalls(tc, ntc);
|
|
return;
|
|
}
|
|
|
|
if (ctx->streaming) {
|
|
handle_stream_delta(ctx, c, r, tc, ntc);
|
|
free(c);
|
|
free(r);
|
|
free_toolcalls(tc, ntc);
|
|
} else {
|
|
free(ctx->final_content);
|
|
ctx->final_content = c ? c : strdup("");
|
|
free(ctx->final_reasoning);
|
|
ctx->final_reasoning = r ? r : strdup("");
|
|
free_toolcalls(ctx->final_tc, ctx->final_n_tc);
|
|
if (ntc > 0) {
|
|
ctx->final_tc = tc;
|
|
ctx->final_n_tc = ntc;
|
|
} else {
|
|
ctx->final_tc = NULL;
|
|
ctx->final_n_tc = 0;
|
|
free(tc);
|
|
}
|
|
}
|
|
}
|
|
|
|
static void process_complete_lines(sse_ctx_t *ctx) {
|
|
char *buf = ctx->line.buf;
|
|
while (1) {
|
|
char *nl = strchr(buf, '\n');
|
|
if (!nl) break;
|
|
*nl = 0;
|
|
char *line = buf;
|
|
size_t ll = strlen(line);
|
|
if (ll > 0 && line[ll - 1] == '\r') line[ll - 1] = 0;
|
|
process_one_line(ctx, line);
|
|
size_t consumed = (size_t)(nl - buf) + 1;
|
|
size_t rest = ctx->line.len - consumed;
|
|
memmove(buf, nl + 1, rest + 1);
|
|
ctx->line.len = rest;
|
|
}
|
|
}
|
|
|
|
static void flush_remaining(sse_ctx_t *ctx) {
|
|
if (ctx->line.len == 0) return;
|
|
char *line = ctx->line.buf;
|
|
size_t ll = ctx->line.len;
|
|
if (ll > 0 && line[ll - 1] == '\r') line[ll - 1] = 0;
|
|
if (ll > 0 && line[ll - 1] == '\n') line[ll - 1] = 0;
|
|
process_one_line(ctx, line);
|
|
sb_reset(&ctx->line);
|
|
}
|
|
|
|
static void handle_stream_delta(sse_ctx_t *ctx, char *c, char *r,
|
|
tool_call_t *tc, int ntc) {
|
|
if (strlen(r) > ctx->emitted_reasoning.len) {
|
|
const char *delta = r + ctx->emitted_reasoning.len;
|
|
sb_append(&ctx->emitted_reasoning, delta, strlen(delta));
|
|
emit_chunk(ctx, NULL, NULL, delta, NULL, NULL);
|
|
}
|
|
if (strlen(c) > ctx->emitted_content.len) {
|
|
const char *delta = c + ctx->emitted_content.len;
|
|
sb_append(&ctx->emitted_content, delta, strlen(delta));
|
|
emit_chunk(ctx, NULL, delta, NULL, NULL, NULL);
|
|
}
|
|
for (int i = 0; i < ntc; i++) {
|
|
char keybuf[64];
|
|
const char *key;
|
|
if (tc[i].id && tc[i].id[0]) key = tc[i].id;
|
|
else { snprintf(keybuf, sizeof keybuf, "idx_%d", i); key = keybuf; }
|
|
struct tcst *st = NULL;
|
|
for (int j = 0; j < ctx->tc_state_n; j++)
|
|
if (strcmp(ctx->tc_state[j].key, key) == 0) { st = &ctx->tc_state[j]; break; }
|
|
if (!st) {
|
|
if (ctx->tc_state_n >= ctx->tc_state_cap) {
|
|
ctx->tc_state_cap = ctx->tc_state_cap ? ctx->tc_state_cap * 2 : 8;
|
|
ctx->tc_state = (struct tcst *)realloc(ctx->tc_state,
|
|
(size_t)ctx->tc_state_cap * sizeof(*ctx->tc_state));
|
|
}
|
|
st = &ctx->tc_state[ctx->tc_state_n++];
|
|
st->key = strdup(key);
|
|
st->prev_args = strdup("");
|
|
st->emitted = false;
|
|
}
|
|
const char *cur = tc[i].arguments ? tc[i].arguments : "";
|
|
if (!st->emitted) {
|
|
st->emitted = true;
|
|
free(st->prev_args);
|
|
st->prev_args = strdup(cur);
|
|
ctx->any_toolcall = true;
|
|
char *js = toolcall_delta_json(&tc[i], i);
|
|
emit_chunk(ctx, NULL, NULL, NULL, js, NULL);
|
|
free(js);
|
|
} else {
|
|
size_t prev = strlen(st->prev_args);
|
|
size_t curlen = strlen(cur);
|
|
if (curlen > prev) {
|
|
const char *delta = cur + prev;
|
|
free(st->prev_args);
|
|
st->prev_args = strdup(cur);
|
|
char *js = toolcall_delta_argdelta(i, delta);
|
|
emit_chunk(ctx, NULL, NULL, NULL, js, NULL);
|
|
free(js);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ */
|
|
/* libcurl callbacks */
|
|
/* ------------------------------------------------------------------ */
|
|
static size_t write_mem_cb(void *ptr, size_t sz, size_t nmemb, void *data) {
|
|
strbuf *b = (strbuf *)data;
|
|
sb_append(b, ptr, sz * nmemb);
|
|
return sz * nmemb;
|
|
}
|
|
|
|
static size_t header_cb(char *buf, size_t sz, size_t nmemb, void *data) {
|
|
sse_ctx_t *ctx = (sse_ctx_t *)data;
|
|
size_t len = sz * nmemb;
|
|
if (len >= 5 && strncmp(buf, "HTTP/", 5) == 0) {
|
|
char *p = buf;
|
|
while (*p && *p != ' ') p++;
|
|
if (*p == ' ') ctx->status = atoi(p + 1);
|
|
if (ctx->streaming && ctx->status == 200 && !ctx->headers_sent) {
|
|
send_sse_headers(ctx->client_fd);
|
|
ctx->headers_sent = true;
|
|
emit_chunk(ctx, "assistant", NULL, NULL, NULL, NULL);
|
|
}
|
|
}
|
|
return len;
|
|
}
|
|
|
|
static size_t write_sse_cb(void *ptr, size_t sz, size_t nmemb, void *data) {
|
|
sse_ctx_t *ctx = (sse_ctx_t *)data;
|
|
sb_append(&ctx->line, ptr, sz * nmemb);
|
|
process_complete_lines(ctx);
|
|
return sz * nmemb;
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ */
|
|
/* Upstream calls (with Fibonacci backoff retry) */
|
|
/* ------------------------------------------------------------------ */
|
|
static int upstream_post(const char *url, const char *body, const char *ua,
|
|
strbuf *resp, long *http_code) {
|
|
for (int attempt = 1; attempt <= MAX_RETRIES; attempt++) {
|
|
CURL *c = curl_easy_init();
|
|
if (!c) return -1;
|
|
struct curl_slist *hdrs = NULL;
|
|
hdrs = curl_slist_append(hdrs, "Content-Type: application/json");
|
|
char uah[2048];
|
|
snprintf(uah, sizeof uah, "User-Agent: %s", ua);
|
|
hdrs = curl_slist_append(hdrs, uah);
|
|
curl_easy_setopt(c, CURLOPT_URL, url);
|
|
curl_easy_setopt(c, CURLOPT_POST, 1L);
|
|
curl_easy_setopt(c, CURLOPT_POSTFIELDS, body);
|
|
curl_easy_setopt(c, CURLOPT_POSTFIELDSIZE, (long)strlen(body));
|
|
curl_easy_setopt(c, CURLOPT_HTTPHEADER, hdrs);
|
|
curl_easy_setopt(c, CURLOPT_WRITEFUNCTION, write_mem_cb);
|
|
curl_easy_setopt(c, CURLOPT_WRITEDATA, resp);
|
|
curl_easy_setopt(c, CURLOPT_TIMEOUT, UPSTREAM_TIMEOUT);
|
|
CURLcode rc = curl_easy_perform(c);
|
|
long code = 0;
|
|
curl_easy_getinfo(c, CURLINFO_RESPONSE_CODE, &code);
|
|
curl_slist_free_all(hdrs);
|
|
curl_easy_cleanup(c);
|
|
if (rc == CURLE_OK && code == 200) {
|
|
if (http_code) *http_code = code;
|
|
return 0;
|
|
}
|
|
if (attempt < MAX_RETRIES) sleep((unsigned int)fib_delay(attempt));
|
|
}
|
|
return -1;
|
|
}
|
|
|
|
static int upstream_get_stream(const char *url, const char *ua, sse_ctx_t *ctx) {
|
|
for (int attempt = 1; attempt <= MAX_RETRIES; attempt++) {
|
|
CURL *c = curl_easy_init();
|
|
if (!c) return -1;
|
|
struct curl_slist *hdrs = NULL;
|
|
hdrs = curl_slist_append(hdrs, "Accept: text/event-stream");
|
|
char uah[2048];
|
|
snprintf(uah, sizeof uah, "User-Agent: %s", ua);
|
|
hdrs = curl_slist_append(hdrs, uah);
|
|
curl_easy_setopt(c, CURLOPT_URL, url);
|
|
curl_easy_setopt(c, CURLOPT_HTTPGET, 1L);
|
|
curl_easy_setopt(c, CURLOPT_HTTPHEADER, hdrs);
|
|
curl_easy_setopt(c, CURLOPT_HEADERFUNCTION, header_cb);
|
|
curl_easy_setopt(c, CURLOPT_HEADERDATA, ctx);
|
|
curl_easy_setopt(c, CURLOPT_WRITEFUNCTION, write_sse_cb);
|
|
curl_easy_setopt(c, CURLOPT_WRITEDATA, ctx);
|
|
curl_easy_setopt(c, CURLOPT_TIMEOUT, UPSTREAM_TIMEOUT);
|
|
CURLcode rc = curl_easy_perform(c);
|
|
long code = 0;
|
|
curl_easy_getinfo(c, CURLINFO_RESPONSE_CODE, &code);
|
|
curl_slist_free_all(hdrs);
|
|
curl_easy_cleanup(c);
|
|
if (rc == CURLE_OK && code == 200) return 0;
|
|
if (attempt < MAX_RETRIES) sleep((unsigned int)fib_delay(attempt));
|
|
}
|
|
return -1;
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ */
|
|
/* Endpoint handlers */
|
|
/* ------------------------------------------------------------------ */
|
|
static void handle_models(int fd, const config_t *cfg) {
|
|
json_object *root = json_object_new_object();
|
|
json_object_object_add(root, "object", json_object_new_string("list"));
|
|
json_object *data = json_object_new_array();
|
|
json_object *m = json_object_new_object();
|
|
json_object_object_add(m, "id", json_object_new_string(cfg->model));
|
|
json_object_object_add(m, "object", json_object_new_string("model"));
|
|
json_object_object_add(m, "created", json_object_new_int64((long)time(NULL)));
|
|
json_object_object_add(m, "owned_by", json_object_new_string("hunyuan"));
|
|
json_object_array_add(data, m);
|
|
json_object_object_add(root, "data", data);
|
|
const char *s = json_object_to_json_string(root);
|
|
if (s) send_json(fd, s);
|
|
json_object_put(root);
|
|
}
|
|
|
|
static void handle_chat(int fd, http_req_t *req, const config_t *cfg) {
|
|
json_object *jreq = json_tokener_parse(req->body);
|
|
if (!jreq || json_object_get_type(jreq) != json_type_object) {
|
|
send_simple(fd, 400, "Bad Request", "Invalid JSON payload");
|
|
if (jreq) json_object_put(jreq);
|
|
return;
|
|
}
|
|
bool stream = false;
|
|
json_object *o = NULL;
|
|
if (json_object_object_get_ex(jreq, "stream", &o) &&
|
|
json_object_get_type(o) == json_type_boolean)
|
|
stream = json_object_get_boolean(o);
|
|
char model_buf[256];
|
|
const char *model = cfg->model;
|
|
if (json_object_object_get_ex(jreq, "model", &o) &&
|
|
json_object_get_type(o) == json_type_string &&
|
|
json_object_get_string(o)[0]) {
|
|
snprintf(model_buf, sizeof model_buf, "%s", json_object_get_string(o));
|
|
model = model_buf;
|
|
}
|
|
|
|
char *payload = build_gradio_payload(jreq, model);
|
|
|
|
int prompt_tokens = count_messages_tokens(jreq);
|
|
bool include_usage = false;
|
|
json_object *so = NULL;
|
|
if (json_object_object_get_ex(jreq, "stream_options", &so) &&
|
|
json_object_get_type(so) == json_type_object) {
|
|
json_object *iu = NULL;
|
|
if (json_object_object_get_ex(so, "include_usage", &iu) &&
|
|
json_object_get_type(iu) == json_type_boolean)
|
|
include_usage = json_object_get_boolean(iu);
|
|
}
|
|
|
|
json_object_put(jreq);
|
|
if (!payload) {
|
|
send_simple(fd, 400, "Bad Request", "Invalid request: missing messages");
|
|
return;
|
|
}
|
|
|
|
const char *ua = req->has_x_ua ? req->ua_override
|
|
: (cfg->user_agent && cfg->user_agent[0] ? cfg->user_agent : DEFAULT_UA);
|
|
|
|
/* endpoint without trailing slash (mirrors TrimRight(endpoint, "/")) */
|
|
const char *ep = cfg->endpoint;
|
|
size_t eplen = strlen(ep);
|
|
char epclean[2048];
|
|
if (eplen > 0 && ep[eplen - 1] == '/') {
|
|
memcpy(epclean, ep, eplen - 1);
|
|
epclean[eplen - 1] = 0;
|
|
} else {
|
|
strncpy(epclean, ep, sizeof epclean - 1);
|
|
epclean[sizeof epclean - 1] = 0;
|
|
}
|
|
|
|
char call_url[4096];
|
|
snprintf(call_url, sizeof call_url, "%s/gradio_api/call/chat", epclean);
|
|
|
|
strbuf post_resp;
|
|
sb_init(&post_resp);
|
|
long code = 0;
|
|
if (upstream_post(call_url, payload, ua, &post_resp, &code) != 0) {
|
|
send_simple(fd, 502, "Bad Gateway",
|
|
"Upstream error: failed to reach Gradio call endpoint");
|
|
sb_free(&post_resp);
|
|
free(payload);
|
|
return;
|
|
}
|
|
|
|
char event_id[256];
|
|
event_id[0] = 0;
|
|
json_object *jr = json_tokener_parse(post_resp.buf);
|
|
if (jr) {
|
|
json_object *ev = NULL;
|
|
if (json_object_object_get_ex(jr, "event_id", &ev) &&
|
|
json_object_get_type(ev) == json_type_string)
|
|
strncpy(event_id, json_object_get_string(ev), sizeof event_id - 1);
|
|
json_object_put(jr);
|
|
}
|
|
sb_free(&post_resp);
|
|
if (event_id[0] == 0) {
|
|
send_simple(fd, 502, "Bad Gateway",
|
|
"Upstream error: failed to parse Gradio event ID");
|
|
free(payload);
|
|
return;
|
|
}
|
|
|
|
char stream_url[8192];
|
|
snprintf(stream_url, sizeof stream_url, "%s/gradio_api/call/chat/%s", epclean, event_id);
|
|
|
|
sse_ctx_t ctx;
|
|
memset(&ctx, 0, sizeof ctx);
|
|
ctx.client_fd = fd;
|
|
ctx.streaming = stream;
|
|
ctx.model = model;
|
|
ctx.include_usage = include_usage;
|
|
ctx.created = (long)time(NULL);
|
|
char uuid[37];
|
|
generate_uuid(uuid);
|
|
snprintf(ctx.completion_id, sizeof ctx.completion_id, "chatcmpl-%s", uuid);
|
|
sb_init(&ctx.line);
|
|
sb_init(&ctx.emitted_content);
|
|
sb_init(&ctx.emitted_reasoning);
|
|
ctx.final_content = strdup("");
|
|
ctx.final_reasoning = strdup("");
|
|
|
|
int rc = upstream_get_stream(stream_url, ua, &ctx);
|
|
|
|
if (rc != 0 || ctx.status != 200) {
|
|
if (!ctx.headers_sent)
|
|
send_simple(fd, 502, "Bad Gateway", "Upstream stream error");
|
|
goto cleanup;
|
|
}
|
|
|
|
flush_remaining(&ctx);
|
|
|
|
if (stream) {
|
|
emit_chunk(&ctx, NULL, NULL, NULL, NULL, ctx.any_toolcall ? "tool_calls" : "stop");
|
|
if (ctx.include_usage) {
|
|
int completion_tokens = tok_count_str(ctx.emitted_content.buf);
|
|
completion_tokens += tok_count_str(ctx.emitted_reasoning.buf);
|
|
for (int i = 0; i < ctx.tc_state_n; i++)
|
|
completion_tokens += tok_count_str(ctx.tc_state[i].prev_args);
|
|
emit_usage_chunk(&ctx, prompt_tokens, completion_tokens);
|
|
}
|
|
const char *done = "data: [DONE]\n\n";
|
|
write_all(fd, done, strlen(done));
|
|
} else {
|
|
char *resp = build_completion_response(ctx.completion_id, ctx.created, ctx.model,
|
|
ctx.final_content, ctx.final_reasoning,
|
|
ctx.final_tc, ctx.final_n_tc, prompt_tokens);
|
|
if (resp) {
|
|
send_json(fd, resp);
|
|
free(resp);
|
|
} else {
|
|
send_simple(fd, 500, "Internal Server Error", "Failed to build response");
|
|
}
|
|
}
|
|
|
|
cleanup:
|
|
sb_free(&ctx.line);
|
|
sb_free(&ctx.emitted_content);
|
|
sb_free(&ctx.emitted_reasoning);
|
|
for (int i = 0; i < ctx.tc_state_n; i++) {
|
|
free(ctx.tc_state[i].key);
|
|
free(ctx.tc_state[i].prev_args);
|
|
}
|
|
free(ctx.tc_state);
|
|
free(ctx.final_content);
|
|
free(ctx.final_reasoning);
|
|
free_toolcalls(ctx.final_tc, ctx.final_n_tc);
|
|
free(payload);
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ */
|
|
/* Request parsing / connection dispatch */
|
|
/* ------------------------------------------------------------------ */
|
|
int read_headers(int fd, strbuf *buf) {
|
|
char tmp[4096];
|
|
while (1) {
|
|
if (strstr(buf->buf, "\r\n\r\n")) break;
|
|
ssize_t n = read(fd, tmp, sizeof tmp);
|
|
if (n <= 0) return -1;
|
|
if (sb_append(buf, tmp, (size_t)n) != 0) return -1;
|
|
if (buf->len > (1 << 20)) return -1;
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
int parse_http_request(strbuf *buf, int fd, http_req_t *req, long max_body_bytes) {
|
|
char *p = buf->buf;
|
|
char *line_end = strstr(p, "\r\n");
|
|
if (!line_end) return -1;
|
|
*line_end = 0;
|
|
char *sp1 = strchr(p, ' ');
|
|
if (!sp1) return -1;
|
|
*sp1 = 0;
|
|
strncpy(req->method, p, sizeof req->method - 1);
|
|
req->method[sizeof req->method - 1] = 0;
|
|
char *rest = sp1 + 1;
|
|
char *sp2 = strchr(rest, ' ');
|
|
if (sp2) *sp2 = 0;
|
|
strncpy(req->path, rest, sizeof req->path - 1);
|
|
req->path[sizeof req->path - 1] = 0;
|
|
|
|
char *h = line_end + 2;
|
|
/* buf->buf was NUL-terminated at the first CRLF above, so searching from
|
|
buf->buf would only see the request line. Search from h (the intact
|
|
header block) for the end-of-headers marker. */
|
|
char *hend = strstr(h, "\r\n\r\n");
|
|
if (!hend) return -1;
|
|
char *hlimit = hend;
|
|
while (h < hlimit) {
|
|
char *le = strstr(h, "\r\n");
|
|
if (!le) break;
|
|
*le = 0;
|
|
char *colon = strchr(h, ':');
|
|
if (colon) {
|
|
*colon = 0;
|
|
char *k = h;
|
|
char *v = colon + 1;
|
|
if (*v == ' ') v++;
|
|
if (iequals(k, "content-length")) req->content_length = atol(v);
|
|
else if (iequals(k, "content-type")) {
|
|
strncpy(req->content_type, v, sizeof req->content_type - 1);
|
|
req->content_type[sizeof req->content_type - 1] = 0;
|
|
} else if (iequals(k, "x-user-agent")) {
|
|
strncpy(req->ua_override, v, sizeof req->ua_override - 1);
|
|
req->ua_override[sizeof req->ua_override - 1] = 0;
|
|
req->has_x_ua = true;
|
|
}
|
|
}
|
|
h = le + 2;
|
|
}
|
|
|
|
char *body_start = hlimit + 4;
|
|
size_t body_avail = buf->len - (size_t)(body_start - buf->buf);
|
|
if (max_body_bytes > 0 && req->content_length > max_body_bytes) return -2;
|
|
if (req->content_length > 0) {
|
|
req->body = (char *)malloc((size_t)req->content_length + 1);
|
|
if (!req->body) return -1;
|
|
size_t got = body_avail < (size_t)req->content_length
|
|
? body_avail
|
|
: (size_t)req->content_length;
|
|
if (got > 0) memcpy(req->body, body_start, got);
|
|
while (got < (size_t)req->content_length) {
|
|
ssize_t n = read(fd, req->body + got, (size_t)req->content_length - got);
|
|
if (n <= 0) break;
|
|
got += (size_t)n;
|
|
}
|
|
req->body[got] = 0;
|
|
req->body_len = got;
|
|
} else {
|
|
req->body = strdup("");
|
|
req->body_len = 0;
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
void http_req_free(http_req_t *r) {
|
|
free(r->body);
|
|
r->body = NULL;
|
|
}
|
|
|
|
void handle_connection(int fd, const config_t *cfg) {
|
|
strbuf reqbuf;
|
|
sb_init(&reqbuf);
|
|
if (read_headers(fd, &reqbuf) != 0) {
|
|
sb_free(&reqbuf);
|
|
return;
|
|
}
|
|
http_req_t req;
|
|
memset(&req, 0, sizeof req);
|
|
int prc = parse_http_request(&reqbuf, fd, &req, cfg->max_body_bytes);
|
|
if (prc != 0) {
|
|
if (prc == -2)
|
|
send_simple(fd, 413, "Payload Too Large",
|
|
"Request body exceeds the maximum allowed size");
|
|
http_req_free(&req);
|
|
sb_free(&reqbuf);
|
|
return;
|
|
}
|
|
|
|
char *q = strchr(req.path, '?');
|
|
if (q) *q = 0;
|
|
|
|
bool is_models = (strcmp(req.path, "/models") == 0 ||
|
|
strcmp(req.path, "/v1/models") == 0);
|
|
bool is_chat = (strcmp(req.path, "/chat/completions") == 0 ||
|
|
strcmp(req.path, "/v1/chat/completions") == 0);
|
|
|
|
if (strcmp(req.method, "OPTIONS") == 0) {
|
|
send_options(fd);
|
|
} else if (is_models) {
|
|
if (strcmp(req.method, "GET") == 0) handle_models(fd, cfg);
|
|
else send_simple(fd, 405, "Method Not Allowed", "Method not allowed");
|
|
} else if (is_chat) {
|
|
if (strcmp(req.method, "POST") == 0) handle_chat(fd, &req, cfg);
|
|
else send_simple(fd, 405, "Method Not Allowed", "Method not allowed");
|
|
} else {
|
|
send_simple(fd, 404, "Not Found", "Not Found");
|
|
}
|
|
|
|
http_req_free(&req);
|
|
sb_free(&reqbuf);
|
|
}
|