recent changes
This commit is contained in:
@@ -4,7 +4,6 @@
|
||||
#include <string.h>
|
||||
#include <unistd.h>
|
||||
#include <time.h>
|
||||
#include <errno.h>
|
||||
#include <json-c/json.h>
|
||||
#include <curl/curl.h>
|
||||
|
||||
@@ -13,6 +12,7 @@
|
||||
/* ------------------------------------------------------------------ */
|
||||
struct tcst {
|
||||
char *key;
|
||||
char *name;
|
||||
char *prev_args;
|
||||
bool emitted;
|
||||
};
|
||||
@@ -126,14 +126,17 @@ static void emit_chunk(sse_ctx_t *ctx, const char *role, const char *content,
|
||||
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);
|
||||
if (buf) {
|
||||
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) {
|
||||
static void emit_usage_chunk(sse_ctx_t *ctx, int prompt_tokens,
|
||||
int completion_tokens, int reasoning_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"));
|
||||
@@ -146,14 +149,19 @@ static void emit_usage_chunk(sse_ctx_t *ctx, int prompt_tokens, int completion_t
|
||||
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 *details = json_object_new_object();
|
||||
json_object_object_add(details, "reasoning_tokens", json_object_new_int(reasoning_tokens));
|
||||
json_object_object_add(usage, "completion_tokens_details", details);
|
||||
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);
|
||||
if (buf) {
|
||||
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);
|
||||
}
|
||||
@@ -175,12 +183,7 @@ static void process_one_line(sse_ctx_t *ctx, char *line) {
|
||||
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 (!parse_gradio_stream(payload, &c, &r, &tc, &ntc)) return;
|
||||
|
||||
if (ctx->streaming) {
|
||||
handle_stream_delta(ctx, c, r, tc, ntc);
|
||||
@@ -188,32 +191,36 @@ static void process_one_line(sse_ctx_t *ctx, char *line) {
|
||||
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 (!ctx->final_content || c[0] != 0) {
|
||||
free(ctx->final_content);
|
||||
ctx->final_content = c;
|
||||
} else {
|
||||
free(c);
|
||||
}
|
||||
if (!ctx->final_reasoning || r[0] != 0) {
|
||||
free(ctx->final_reasoning);
|
||||
ctx->final_reasoning = r;
|
||||
} else {
|
||||
free(r);
|
||||
}
|
||||
if (ntc > 0) {
|
||||
free_toolcalls(ctx->final_tc, ctx->final_n_tc);
|
||||
ctx->final_tc = tc;
|
||||
ctx->final_n_tc = ntc;
|
||||
} else {
|
||||
ctx->final_tc = NULL;
|
||||
ctx->final_n_tc = 0;
|
||||
free(tc);
|
||||
free_toolcalls(tc, ntc);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
static void process_complete_lines(sse_ctx_t *ctx) {
|
||||
char *buf = ctx->line.buf;
|
||||
if (!buf) return;
|
||||
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);
|
||||
process_one_line(ctx, buf);
|
||||
size_t consumed = (size_t)(nl - buf) + 1;
|
||||
size_t rest = ctx->line.len - consumed;
|
||||
memmove(buf, nl + 1, rest + 1);
|
||||
@@ -222,12 +229,8 @@ static void process_complete_lines(sse_ctx_t *ctx) {
|
||||
}
|
||||
|
||||
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);
|
||||
if (ctx->line.len == 0 || !ctx->line.buf) return;
|
||||
process_one_line(ctx, ctx->line.buf);
|
||||
sb_reset(&ctx->line);
|
||||
}
|
||||
|
||||
@@ -253,34 +256,61 @@ static void handle_stream_delta(sse_ctx_t *ctx, char *c, char *r,
|
||||
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));
|
||||
int ncap = ctx->tc_state_cap ? ctx->tc_state_cap * 2 : 8;
|
||||
struct tcst *nt = (struct tcst *)realloc(ctx->tc_state,
|
||||
(size_t)ncap * sizeof(*ctx->tc_state));
|
||||
if (!nt) continue;
|
||||
ctx->tc_state = nt;
|
||||
ctx->tc_state_cap = ncap;
|
||||
}
|
||||
char *kcopy = strdup(key);
|
||||
char *ncopy = strdup(tc[i].name ? tc[i].name : "");
|
||||
char *acopy = strdup("");
|
||||
if (!kcopy || !ncopy || !acopy) {
|
||||
free(kcopy);
|
||||
free(ncopy);
|
||||
free(acopy);
|
||||
continue;
|
||||
}
|
||||
st = &ctx->tc_state[ctx->tc_state_n++];
|
||||
st->key = strdup(key);
|
||||
st->prev_args = strdup("");
|
||||
st->key = kcopy;
|
||||
st->name = ncopy;
|
||||
st->prev_args = acopy;
|
||||
st->emitted = false;
|
||||
} else if (tc[i].name && tc[i].name[0] && st->name[0] == 0) {
|
||||
char *ncopy = strdup(tc[i].name);
|
||||
if (ncopy) {
|
||||
free(st->name);
|
||||
st->name = ncopy;
|
||||
}
|
||||
}
|
||||
const char *cur = tc[i].arguments ? tc[i].arguments : "";
|
||||
if (!st->emitted) {
|
||||
char *ncur = strdup(cur);
|
||||
if (!ncur) continue;
|
||||
st->emitted = true;
|
||||
free(st->prev_args);
|
||||
st->prev_args = strdup(cur);
|
||||
st->prev_args = ncur;
|
||||
ctx->any_toolcall = true;
|
||||
char *js = toolcall_delta_json(&tc[i], i);
|
||||
emit_chunk(ctx, NULL, NULL, NULL, js, NULL);
|
||||
free(js);
|
||||
if (js) {
|
||||
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;
|
||||
char *ncur = strdup(cur);
|
||||
if (!ncur) continue;
|
||||
const char *delta = ncur + prev;
|
||||
free(st->prev_args);
|
||||
st->prev_args = strdup(cur);
|
||||
st->prev_args = ncur;
|
||||
char *js = toolcall_delta_argdelta(i, delta);
|
||||
emit_chunk(ctx, NULL, NULL, NULL, js, NULL);
|
||||
free(js);
|
||||
if (js) {
|
||||
emit_chunk(ctx, NULL, NULL, NULL, js, NULL);
|
||||
free(js);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -322,8 +352,9 @@ static size_t write_sse_cb(void *ptr, size_t sz, size_t nmemb, void *data) {
|
||||
/* Upstream calls (with Fibonacci backoff retry) */
|
||||
/* ------------------------------------------------------------------ */
|
||||
static int upstream_post(const char *url, const char *body, const char *ua,
|
||||
strbuf *resp, long *http_code) {
|
||||
strbuf *resp) {
|
||||
for (int attempt = 1; attempt <= MAX_RETRIES; attempt++) {
|
||||
sb_reset(resp);
|
||||
CURL *c = curl_easy_init();
|
||||
if (!c) return -1;
|
||||
struct curl_slist *hdrs = NULL;
|
||||
@@ -344,10 +375,7 @@ static int upstream_post(const char *url, const char *body, const char *ua,
|
||||
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 (rc == CURLE_OK && code == 200) return 0;
|
||||
if (attempt < MAX_RETRIES) sleep((unsigned int)fib_delay(attempt));
|
||||
}
|
||||
return -1;
|
||||
@@ -355,6 +383,15 @@ static int upstream_post(const char *url, const char *body, const char *ua,
|
||||
|
||||
static int upstream_get_stream(const char *url, const char *ua, sse_ctx_t *ctx) {
|
||||
for (int attempt = 1; attempt <= MAX_RETRIES; attempt++) {
|
||||
sb_reset(&ctx->line);
|
||||
ctx->status = 0;
|
||||
free(ctx->final_content);
|
||||
ctx->final_content = NULL;
|
||||
free(ctx->final_reasoning);
|
||||
ctx->final_reasoning = NULL;
|
||||
free_toolcalls(ctx->final_tc, ctx->final_n_tc);
|
||||
ctx->final_tc = NULL;
|
||||
ctx->final_n_tc = 0;
|
||||
CURL *c = curl_easy_init();
|
||||
if (!c) return -1;
|
||||
struct curl_slist *hdrs = NULL;
|
||||
@@ -376,6 +413,7 @@ static int upstream_get_stream(const char *url, const char *ua, sse_ctx_t *ctx)
|
||||
curl_slist_free_all(hdrs);
|
||||
curl_easy_cleanup(c);
|
||||
if (rc == CURLE_OK && code == 200) return 0;
|
||||
if (ctx->headers_sent) break;
|
||||
if (attempt < MAX_RETRIES) sleep((unsigned int)fib_delay(attempt));
|
||||
}
|
||||
return -1;
|
||||
@@ -421,7 +459,7 @@ static void handle_chat(int fd, http_req_t *req, const config_t *cfg) {
|
||||
model = model_buf;
|
||||
}
|
||||
|
||||
char *payload = build_gradio_payload(jreq, model);
|
||||
char *payload = build_gradio_payload(jreq);
|
||||
|
||||
int prompt_tokens = count_messages_tokens(jreq);
|
||||
bool include_usage = false;
|
||||
@@ -448,8 +486,9 @@ static void handle_chat(int fd, http_req_t *req, const config_t *cfg) {
|
||||
size_t eplen = strlen(ep);
|
||||
char epclean[2048];
|
||||
if (eplen > 0 && ep[eplen - 1] == '/') {
|
||||
memcpy(epclean, ep, eplen - 1);
|
||||
epclean[eplen - 1] = 0;
|
||||
size_t clen = eplen - 1 < sizeof epclean - 1 ? eplen - 1 : sizeof epclean - 1;
|
||||
memcpy(epclean, ep, clen);
|
||||
epclean[clen] = 0;
|
||||
} else {
|
||||
strncpy(epclean, ep, sizeof epclean - 1);
|
||||
epclean[sizeof epclean - 1] = 0;
|
||||
@@ -460,8 +499,7 @@ static void handle_chat(int fd, http_req_t *req, const config_t *cfg) {
|
||||
|
||||
strbuf post_resp;
|
||||
sb_init(&post_resp);
|
||||
long code = 0;
|
||||
if (upstream_post(call_url, payload, ua, &post_resp, &code) != 0) {
|
||||
if (!post_resp.buf || upstream_post(call_url, payload, ua, &post_resp) != 0) {
|
||||
send_simple(fd, 502, "Bad Gateway",
|
||||
"Upstream error: failed to reach Gradio call endpoint");
|
||||
sb_free(&post_resp);
|
||||
@@ -475,8 +513,10 @@ static void handle_chat(int fd, http_req_t *req, const config_t *cfg) {
|
||||
if (jr) {
|
||||
json_object *ev = NULL;
|
||||
if (json_object_object_get_ex(jr, "event_id", &ev) &&
|
||||
json_object_get_type(ev) == json_type_string)
|
||||
json_object_get_type(ev) == json_type_string) {
|
||||
strncpy(event_id, json_object_get_string(ev), sizeof event_id - 1);
|
||||
event_id[sizeof event_id - 1] = 0;
|
||||
}
|
||||
json_object_put(jr);
|
||||
}
|
||||
sb_free(&post_resp);
|
||||
@@ -501,10 +541,10 @@ static void handle_chat(int fd, http_req_t *req, const config_t *cfg) {
|
||||
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("");
|
||||
if (stream) {
|
||||
sb_init(&ctx.emitted_content);
|
||||
sb_init(&ctx.emitted_reasoning);
|
||||
}
|
||||
|
||||
int rc = upstream_get_stream(stream_url, ua, &ctx);
|
||||
|
||||
@@ -519,11 +559,13 @@ static void handle_chat(int fd, http_req_t *req, const config_t *cfg) {
|
||||
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++)
|
||||
int reasoning_tokens = tok_count_str(ctx.emitted_reasoning.buf);
|
||||
int completion_tokens = tok_count_str(ctx.emitted_content.buf) + reasoning_tokens;
|
||||
for (int i = 0; i < ctx.tc_state_n; i++) {
|
||||
completion_tokens += tok_count_str(ctx.tc_state[i].name);
|
||||
completion_tokens += tok_count_str(ctx.tc_state[i].prev_args);
|
||||
emit_usage_chunk(&ctx, prompt_tokens, completion_tokens);
|
||||
}
|
||||
emit_usage_chunk(&ctx, prompt_tokens, completion_tokens, reasoning_tokens);
|
||||
}
|
||||
const char *done = "data: [DONE]\n\n";
|
||||
write_all(fd, done, strlen(done));
|
||||
@@ -545,6 +587,7 @@ cleanup:
|
||||
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].name);
|
||||
free(ctx.tc_state[i].prev_args);
|
||||
}
|
||||
free(ctx.tc_state);
|
||||
@@ -558,6 +601,7 @@ cleanup:
|
||||
/* Request parsing / connection dispatch */
|
||||
/* ------------------------------------------------------------------ */
|
||||
int read_headers(int fd, strbuf *buf) {
|
||||
if (!buf->buf) return -1;
|
||||
char tmp[4096];
|
||||
while (1) {
|
||||
if (strstr(buf->buf, "\r\n\r\n")) break;
|
||||
@@ -570,6 +614,7 @@ int read_headers(int fd, strbuf *buf) {
|
||||
}
|
||||
|
||||
int parse_http_request(strbuf *buf, int fd, http_req_t *req, long max_body_bytes) {
|
||||
if (!buf->buf) return -1;
|
||||
char *p = buf->buf;
|
||||
char *line_end = strstr(p, "\r\n");
|
||||
if (!line_end) return -1;
|
||||
@@ -617,6 +662,7 @@ int parse_http_request(strbuf *buf, int fd, http_req_t *req, long max_body_bytes
|
||||
|
||||
char *body_start = hlimit + 4;
|
||||
size_t body_avail = buf->len - (size_t)(body_start - buf->buf);
|
||||
if (req->content_length < 0) return -1;
|
||||
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);
|
||||
@@ -634,6 +680,7 @@ int parse_http_request(strbuf *buf, int fd, http_req_t *req, long max_body_bytes
|
||||
req->body_len = got;
|
||||
} else {
|
||||
req->body = strdup("");
|
||||
if (!req->body) return -1;
|
||||
req->body_len = 0;
|
||||
}
|
||||
return 0;
|
||||
|
||||
Reference in New Issue
Block a user