forked from WHC/Hermit
An OpenSSL-based libcurl keeps the CA path it was built with. MSYS2's is "/mingw64/etc/ssl/certs/ca-bundle.crt", which resolves to nothing once the binary runs outside an MSYS2 shell, and that build ignores CURL_CA_BUNDLE, SSL_CERT_FILE and SSL_CERT_DIR. Every HTTPS request then failed with CURLE_SSL_CACERT_BADFILE (77). llm_apply_ca_bundle() sets CURLOPT_CAINFO from $CST_CA_BUNDLE, falling back to ca-bundle.crt beside the executable. If neither exists nothing is set, so a libcurl linked against Schannel (the Windows certificate store) behaves exactly as before.
855 lines
25 KiB
C
855 lines
25 KiB
C
#ifndef _DEFAULT_SOURCE
|
|
#define _DEFAULT_SOURCE
|
|
#endif
|
|
|
|
#include "cJSON.h"
|
|
#include "cst_agent.h"
|
|
#include "cst_tools.h"
|
|
#include "cst_util.h"
|
|
#include "os_compat.h"
|
|
|
|
#include <curl/curl.h>
|
|
#include <stdatomic.h>
|
|
#include <stdio.h>
|
|
#include <stdlib.h>
|
|
#include <string.h>
|
|
#ifdef _WIN32
|
|
#include <windows.h>
|
|
#else
|
|
#include <unistd.h>
|
|
#endif
|
|
|
|
#define ERRBODY_MAX 4096
|
|
#define LLM_MAX_RETRIES 3
|
|
|
|
/* Point libcurl at a CA bundle on Windows.
|
|
*
|
|
* An OpenSSL-based libcurl keeps the path it was built with -- MSYS2's is
|
|
* "/mingw64/etc/ssl/certs/ca-bundle.crt", which resolves to nothing once the
|
|
* binary runs outside an MSYS2 shell, and that build ignores CURL_CA_BUNDLE.
|
|
* Every HTTPS request then dies with CURLE_SSL_CACERT_BADFILE (77).
|
|
*
|
|
* Order: $CST_CA_BUNDLE, then ca-bundle.crt beside the executable. If neither
|
|
* exists nothing is set, so a libcurl linked against Schannel (the Windows
|
|
* certificate store) behaves exactly as before. */
|
|
static void llm_apply_ca_bundle(CURL *curl) {
|
|
#if defined(_WIN32)
|
|
static char ca[OS_PATH_MAX];
|
|
static int checked;
|
|
if (!checked) {
|
|
checked = 1;
|
|
const char *env = getenv("CST_CA_BUNDLE");
|
|
if (env && env[0]) {
|
|
snprintf(ca, sizeof(ca), "%s", env);
|
|
} else {
|
|
char dir[OS_PATH_MAX];
|
|
if (os_exe_dir(dir, sizeof(dir)) == 0)
|
|
os_path_join(ca, sizeof(ca), dir, "ca-bundle.crt");
|
|
}
|
|
if (ca[0]) {
|
|
FILE *f = fopen(ca, "rb");
|
|
if (f)
|
|
fclose(f);
|
|
else
|
|
ca[0] = '\0';
|
|
}
|
|
}
|
|
if (ca[0])
|
|
curl_easy_setopt(curl, CURLOPT_CAINFO, ca);
|
|
#else
|
|
(void)curl;
|
|
#endif
|
|
}
|
|
|
|
/* --- wire body ----------------------------------------------------------- */
|
|
|
|
static size_t wire_tool_cap(const cst_llm_request_t *req) {
|
|
if (req->tool_out_cap)
|
|
return req->tool_out_cap;
|
|
const char *e = getenv("CST_TOOL_OUT_CAP");
|
|
if (e && atol(e) > 256)
|
|
return (size_t)atol(e);
|
|
return CST_DEFAULT_TOOL_OUT_CAP;
|
|
}
|
|
|
|
/* Keep head and tail (errors usually sit at the end of build/test logs). */
|
|
static char *cap_middle(const char *s, size_t n, size_t cap, size_t *out_n) {
|
|
if (n <= cap) {
|
|
*out_n = n;
|
|
return NULL;
|
|
}
|
|
size_t head = cst_utf8_floor(s, cap * 3 / 5);
|
|
size_t tail_start = n - (cap - head);
|
|
while (tail_start < n && ((unsigned char)s[tail_start] & 0xC0) == 0x80)
|
|
tail_start++;
|
|
char note[96];
|
|
int nl = snprintf(note, sizeof(note),
|
|
"\n…(%zu bytes omitted for the model)…\n",
|
|
tail_start - head);
|
|
size_t total = head + (size_t)nl + (n - tail_start);
|
|
char *out = malloc(total + 1);
|
|
if (!out) {
|
|
*out_n = n;
|
|
return NULL;
|
|
}
|
|
memcpy(out, s, head);
|
|
memcpy(out + head, note, (size_t)nl);
|
|
memcpy(out + head + nl, s + tail_start, n - tail_start);
|
|
out[total] = 0;
|
|
*out_n = total;
|
|
return out;
|
|
}
|
|
|
|
static cJSON *clean_string(const char *s, size_t cap) {
|
|
if (!s)
|
|
s = "";
|
|
size_t n = strlen(s);
|
|
size_t cn = n;
|
|
char *capped = cap ? cap_middle(s, n, cap, &cn) : NULL;
|
|
size_t sl = 0;
|
|
char *san = cst_utf8_sanitize(capped ? capped : s, cn, &sl);
|
|
free(capped);
|
|
if (!san)
|
|
return cJSON_CreateString("");
|
|
cJSON *j = cJSON_CreateString(san);
|
|
free(san);
|
|
return j;
|
|
}
|
|
|
|
static const char *role_api(cst_role_t r) {
|
|
switch (r) {
|
|
case CST_ROLE_SYSTEM:
|
|
return "system";
|
|
case CST_ROLE_USER:
|
|
return "user";
|
|
case CST_ROLE_ASSISTANT:
|
|
return "assistant";
|
|
case CST_ROLE_TOOL:
|
|
return "tool";
|
|
}
|
|
return "user";
|
|
}
|
|
|
|
char *cst_llm_build_body(const cst_llm_request_t *req) {
|
|
if (!req || !req->model)
|
|
return NULL;
|
|
cJSON *root = cJSON_CreateObject();
|
|
cJSON_AddStringToObject(root, "model", req->model);
|
|
if (req->reasoning_effort && req->reasoning_effort[0])
|
|
cJSON_AddStringToObject(root, "reasoning_effort", req->reasoning_effort);
|
|
cJSON_AddTrueToObject(root, "stream");
|
|
const char *nousage = getenv("CST_NO_STREAM_USAGE");
|
|
if (!nousage || !nousage[0] || nousage[0] == '0') {
|
|
cJSON *so = cJSON_AddObjectToObject(root, "stream_options");
|
|
cJSON_AddTrueToObject(so, "include_usage");
|
|
}
|
|
cJSON *msgs = cJSON_AddArrayToObject(root, "messages");
|
|
size_t tcap = wire_tool_cap(req);
|
|
cJSON *prev_user = NULL; /* merge consecutive user messages */
|
|
|
|
for (int i = 0; i < req->nmsgs; i++) {
|
|
const cst_message_t *m = &req->msgs[i];
|
|
cst_role_t role = m->role;
|
|
if (role == CST_ROLE_TOOL && !m->tool_call_id[0])
|
|
role = CST_ROLE_USER;
|
|
|
|
if (role == CST_ROLE_USER) {
|
|
/* Text-protocol result frames are capped by the agent before framing
|
|
* (their byte counts must stay exact), so never cap user text here. */
|
|
cJSON *c = clean_string(m->content, 0);
|
|
if (prev_user) {
|
|
cJSON *pc = cJSON_GetObjectItemCaseSensitive(prev_user, "content");
|
|
cst_buf_t b = {0};
|
|
cst_buf_puts(&b, pc->valuestring);
|
|
cst_buf_puts(&b, "\n\n");
|
|
cst_buf_puts(&b, c->valuestring);
|
|
cJSON_ReplaceItemInObjectCaseSensitive(prev_user, "content",
|
|
cJSON_CreateString(b.buf));
|
|
cst_buf_free(&b);
|
|
cJSON_Delete(c);
|
|
continue;
|
|
}
|
|
cJSON *o = cJSON_CreateObject();
|
|
cJSON_AddStringToObject(o, "role", "user");
|
|
cJSON_AddItemToObject(o, "content", c);
|
|
cJSON_AddItemToArray(msgs, o);
|
|
prev_user = o;
|
|
continue;
|
|
}
|
|
prev_user = NULL;
|
|
|
|
cJSON *o = cJSON_CreateObject();
|
|
cJSON_AddStringToObject(o, "role", role_api(role));
|
|
if (role == CST_ROLE_ASSISTANT && m->tool_calls_json &&
|
|
m->tool_calls_json[0]) {
|
|
cJSON *tc = cJSON_Parse(m->tool_calls_json);
|
|
int has_tc = cJSON_IsArray(tc) && cJSON_GetArraySize(tc) > 0;
|
|
if (has_tc && (!m->content || !m->content[0]))
|
|
cJSON_AddNullToObject(o, "content");
|
|
else
|
|
cJSON_AddItemToObject(o, "content", clean_string(m->content, 0));
|
|
if (has_tc)
|
|
cJSON_AddItemToObject(o, "tool_calls", tc);
|
|
else
|
|
cJSON_Delete(tc);
|
|
} else if (role == CST_ROLE_TOOL) {
|
|
cJSON_AddStringToObject(o, "tool_call_id", m->tool_call_id);
|
|
cJSON_AddItemToObject(o, "content", clean_string(m->content, tcap));
|
|
} else {
|
|
cJSON_AddItemToObject(o, "content", clean_string(m->content, 0));
|
|
}
|
|
cJSON_AddItemToArray(msgs, o);
|
|
}
|
|
|
|
if (req->enable_tools) {
|
|
cJSON *tools = cst_tools_schema_cfg(req->tool_cfg);
|
|
if (tools) {
|
|
cJSON_AddItemToObject(root, "tools", tools);
|
|
cJSON_AddStringToObject(root, "tool_choice",
|
|
req->tool_choice_none ? "none" : "auto");
|
|
}
|
|
}
|
|
char *out = cJSON_PrintUnformatted(root);
|
|
cJSON_Delete(root);
|
|
return out;
|
|
}
|
|
|
|
/* --- SSE decoding -------------------------------------------------------- */
|
|
|
|
typedef struct {
|
|
int index;
|
|
int emitted;
|
|
int overflow;
|
|
int opened;
|
|
char id[CTL_MAX_ID];
|
|
char name[CTL_MAX_NAME];
|
|
cst_buf_t args;
|
|
} slot_t;
|
|
|
|
typedef struct {
|
|
const cst_llm_request_t *req;
|
|
cst_llm_stats_t *st;
|
|
cst_buf_t line;
|
|
slot_t *slots;
|
|
int nslots;
|
|
int cap;
|
|
int last_index;
|
|
int in_reason;
|
|
int emitted_any;
|
|
int stopped;
|
|
double t0;
|
|
CURL *curl;
|
|
int http_checked;
|
|
int error_body;
|
|
cst_buf_t errbody;
|
|
FILE *rec;
|
|
} sse_t;
|
|
|
|
static void emit_text(sse_t *s, const char *t, size_t n) {
|
|
if (n && s->req->on_delta) {
|
|
s->req->on_delta(t, n, s->req->userdata);
|
|
s->emitted_any = 1;
|
|
}
|
|
}
|
|
|
|
static void slot_emit(sse_t *s, slot_t *sl) {
|
|
if (sl->emitted)
|
|
return;
|
|
sl->emitted = 1;
|
|
if (!sl->name[0] || !s->req->on_native_tool)
|
|
return;
|
|
const char *args = sl->args.buf ? sl->args.buf : "";
|
|
int valid = !sl->overflow;
|
|
if (valid && sl->args.len) {
|
|
cJSON *j = cJSON_Parse(args);
|
|
valid = j != NULL;
|
|
cJSON_Delete(j);
|
|
}
|
|
s->emitted_any = 1;
|
|
s->req->on_native_tool(sl->id, sl->name, valid ? args : NULL,
|
|
s->req->userdata);
|
|
}
|
|
|
|
static void slots_emit_all(sse_t *s, int except_index) {
|
|
for (int i = 0; i < s->nslots; i++) {
|
|
if (s->slots[i].index != except_index)
|
|
slot_emit(s, &s->slots[i]);
|
|
}
|
|
}
|
|
|
|
static slot_t *slot_get(sse_t *s, int index) {
|
|
for (int i = 0; i < s->nslots; i++) {
|
|
if (s->slots[i].index == index)
|
|
return &s->slots[i];
|
|
}
|
|
if (s->nslots == s->cap) {
|
|
int nc = s->cap ? s->cap * 2 : 4;
|
|
slot_t *nv = realloc(s->slots, (size_t)nc * sizeof(slot_t));
|
|
if (!nv)
|
|
return NULL;
|
|
s->slots = nv;
|
|
s->cap = nc;
|
|
}
|
|
slot_t *sl = &s->slots[s->nslots++];
|
|
memset(sl, 0, sizeof(*sl));
|
|
sl->index = index;
|
|
return sl;
|
|
}
|
|
|
|
static int max_index(const sse_t *s) {
|
|
int m = -1;
|
|
for (int i = 0; i < s->nslots; i++)
|
|
if (s->slots[i].index > m)
|
|
m = s->slots[i].index;
|
|
return m;
|
|
}
|
|
|
|
static void handle_tool_delta(sse_t *s, const cJSON *tc) {
|
|
const cJSON *jidx = cJSON_GetObjectItemCaseSensitive(tc, "index");
|
|
const cJSON *jid = cJSON_GetObjectItemCaseSensitive(tc, "id");
|
|
const cJSON *fn = cJSON_GetObjectItemCaseSensitive(tc, "function");
|
|
const cJSON *jname = cJSON_GetObjectItemCaseSensitive(fn, "name");
|
|
const cJSON *jargs = cJSON_GetObjectItemCaseSensitive(fn, "arguments");
|
|
const char *id = cJSON_IsString(jid) && jid->valuestring[0] ? jid->valuestring
|
|
: NULL;
|
|
int index = cJSON_IsNumber(jidx) ? jidx->valueint
|
|
: (s->last_index < 0 ? 0 : s->last_index);
|
|
slot_t *cur = slot_get(s, index);
|
|
if (!cur)
|
|
return;
|
|
/* Same index but a different id: gateway without indices. */
|
|
if (id && cur->id[0] && strcmp(cur->id, id) != 0) {
|
|
index = max_index(s) + 1;
|
|
cur = slot_get(s, index);
|
|
if (!cur)
|
|
return;
|
|
}
|
|
if (index != s->last_index) {
|
|
slots_emit_all(s, index); /* a new call started: earlier ones are done */
|
|
s->last_index = index;
|
|
}
|
|
if (id && !cur->id[0])
|
|
snprintf(cur->id, sizeof(cur->id), "%s", id);
|
|
if (cJSON_IsString(jname) && jname->valuestring[0] && !cur->name[0])
|
|
snprintf(cur->name, sizeof(cur->name), "%s", jname->valuestring);
|
|
if (cur->name[0] && !cur->opened) {
|
|
cur->opened = 1;
|
|
if (s->req->on_native_open)
|
|
s->req->on_native_open(index, cur->id, cur->name, s->req->userdata);
|
|
}
|
|
if (cJSON_IsString(jargs) && jargs->valuestring[0] && !cur->overflow) {
|
|
size_t n = strlen(jargs->valuestring);
|
|
if (cur->args.len + n > CTL_MAX_ARG_BYTES)
|
|
cur->overflow = 1;
|
|
else
|
|
cst_buf_append(&cur->args, jargs->valuestring, n);
|
|
}
|
|
}
|
|
|
|
static void handle_payload(sse_t *s, const char *payload) {
|
|
if (strncmp(payload, "[DONE]", 6) == 0) {
|
|
if (s->in_reason) {
|
|
emit_text(s, "</think>", 8);
|
|
s->in_reason = 0;
|
|
}
|
|
slots_emit_all(s, -1);
|
|
return;
|
|
}
|
|
cJSON *root = cJSON_Parse(payload);
|
|
if (!root)
|
|
return;
|
|
const cJSON *err = cJSON_GetObjectItemCaseSensitive(root, "error");
|
|
if (err && s->st) {
|
|
const cJSON *msg = cJSON_GetObjectItemCaseSensitive(err, "message");
|
|
if (cJSON_IsString(msg))
|
|
snprintf(s->st->error, sizeof(s->st->error), "%s", msg->valuestring);
|
|
else if (cJSON_IsString(err))
|
|
snprintf(s->st->error, sizeof(s->st->error), "%s", err->valuestring);
|
|
}
|
|
const cJSON *usage = cJSON_GetObjectItemCaseSensitive(root, "usage");
|
|
if (cJSON_IsObject(usage) && s->st) {
|
|
const cJSON *pt = cJSON_GetObjectItemCaseSensitive(usage, "prompt_tokens");
|
|
const cJSON *ct =
|
|
cJSON_GetObjectItemCaseSensitive(usage, "completion_tokens");
|
|
const cJSON *hit =
|
|
cJSON_GetObjectItemCaseSensitive(usage, "prompt_cache_hit_tokens");
|
|
const cJSON *det =
|
|
cJSON_GetObjectItemCaseSensitive(usage, "prompt_tokens_details");
|
|
const cJSON *cached = cJSON_GetObjectItemCaseSensitive(det, "cached_tokens");
|
|
if (cJSON_IsNumber(pt))
|
|
s->st->prompt_tokens = (long)pt->valuedouble;
|
|
if (cJSON_IsNumber(ct))
|
|
s->st->completion_tokens = (long)ct->valuedouble;
|
|
if (cJSON_IsNumber(hit))
|
|
s->st->cached_tokens = (long)hit->valuedouble;
|
|
else if (cJSON_IsNumber(cached))
|
|
s->st->cached_tokens = (long)cached->valuedouble;
|
|
}
|
|
const cJSON *choices = cJSON_GetObjectItemCaseSensitive(root, "choices");
|
|
const cJSON *ch = cJSON_IsArray(choices) ? cJSON_GetArrayItem(choices, 0)
|
|
: NULL;
|
|
if (ch) {
|
|
const cJSON *delta = cJSON_GetObjectItemCaseSensitive(ch, "delta");
|
|
const cJSON *rc =
|
|
cJSON_GetObjectItemCaseSensitive(delta, "reasoning_content");
|
|
if (cJSON_IsString(rc) && rc->valuestring[0]) {
|
|
if (!s->in_reason) {
|
|
emit_text(s, "<think>", 7);
|
|
s->in_reason = 1;
|
|
}
|
|
emit_text(s, rc->valuestring, strlen(rc->valuestring));
|
|
}
|
|
const cJSON *c = cJSON_GetObjectItemCaseSensitive(delta, "content");
|
|
if (cJSON_IsString(c) && c->valuestring[0]) {
|
|
if (s->in_reason) {
|
|
emit_text(s, "</think>", 8);
|
|
s->in_reason = 0;
|
|
}
|
|
emit_text(s, c->valuestring, strlen(c->valuestring));
|
|
}
|
|
const cJSON *tcs = cJSON_GetObjectItemCaseSensitive(delta, "tool_calls");
|
|
if (cJSON_IsArray(tcs)) {
|
|
if (s->in_reason) {
|
|
emit_text(s, "</think>", 8);
|
|
s->in_reason = 0;
|
|
}
|
|
const cJSON *tc = NULL;
|
|
cJSON_ArrayForEach(tc, tcs) handle_tool_delta(s, tc);
|
|
}
|
|
const cJSON *fr = cJSON_GetObjectItemCaseSensitive(ch, "finish_reason");
|
|
if (cJSON_IsString(fr) && fr->valuestring[0]) {
|
|
if (s->st)
|
|
snprintf(s->st->finish_reason, sizeof(s->st->finish_reason), "%s",
|
|
fr->valuestring);
|
|
if (s->in_reason) {
|
|
emit_text(s, "</think>", 8);
|
|
s->in_reason = 0;
|
|
}
|
|
slots_emit_all(s, -1);
|
|
}
|
|
}
|
|
cJSON_Delete(root);
|
|
}
|
|
|
|
static int flag_set(volatile int *f) { return f && *f; }
|
|
|
|
/* Feed raw response bytes. Returns -1 when the caller asked to stop. */
|
|
static int sse_feed(sse_t *s, const char *data, size_t n) {
|
|
if (s->st && s->st->t_first_byte_ms < 0)
|
|
s->st->t_first_byte_ms = os_now_ms() - s->t0;
|
|
if (s->rec)
|
|
fwrite(data, 1, n, s->rec);
|
|
if (s->error_body) {
|
|
if (s->errbody.len < ERRBODY_MAX)
|
|
cst_buf_append(&s->errbody, data,
|
|
n < ERRBODY_MAX - s->errbody.len ? n
|
|
: ERRBODY_MAX - s->errbody.len);
|
|
return 0;
|
|
}
|
|
for (size_t i = 0; i < n; i++) {
|
|
if (data[i] != '\n') {
|
|
cst_buf_append(&s->line, &data[i], 1);
|
|
continue;
|
|
}
|
|
char *l = s->line.buf ? s->line.buf : (char *)"";
|
|
size_t ll = s->line.len;
|
|
if (ll && l[ll - 1] == '\r')
|
|
l[--ll] = 0;
|
|
if (ll >= 5 && strncmp(l, "data:", 5) == 0) {
|
|
const char *p = l + 5;
|
|
while (*p == ' ')
|
|
p++;
|
|
handle_payload(s, p);
|
|
}
|
|
s->line.len = 0;
|
|
if (s->line.buf)
|
|
s->line.buf[0] = 0;
|
|
if (flag_set(s->req->stop_flag) || flag_set(s->req->cancel_flag)) {
|
|
if (flag_set(s->req->stop_flag) && s->st)
|
|
s->st->client_stopped = 1;
|
|
s->stopped = 1;
|
|
return -1;
|
|
}
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
static size_t write_cb(char *ptr, size_t size, size_t nmemb, void *userdata) {
|
|
sse_t *s = userdata;
|
|
size_t n = size * nmemb;
|
|
if (!s->http_checked) {
|
|
long code = 0;
|
|
curl_easy_getinfo(s->curl, CURLINFO_RESPONSE_CODE, &code);
|
|
s->http_checked = 1;
|
|
if (code >= 400)
|
|
s->error_body = 1;
|
|
}
|
|
if (sse_feed(s, ptr, n) != 0)
|
|
return 0; /* abort transfer */
|
|
return n;
|
|
}
|
|
|
|
static int xferinfo_cb(void *clientp, curl_off_t dltotal, curl_off_t dlnow,
|
|
curl_off_t ultotal, curl_off_t ulnow) {
|
|
(void)dltotal;
|
|
(void)dlnow;
|
|
(void)ultotal;
|
|
(void)ulnow;
|
|
sse_t *s = clientp;
|
|
return (flag_set(s->req->cancel_flag) || flag_set(s->req->stop_flag)) ? 1 : 0;
|
|
}
|
|
|
|
static void sse_reset(sse_t *s) {
|
|
for (int i = 0; i < s->nslots; i++)
|
|
cst_buf_free(&s->slots[i].args);
|
|
free(s->slots);
|
|
s->slots = NULL;
|
|
s->nslots = s->cap = 0;
|
|
cst_buf_free(&s->line);
|
|
cst_buf_free(&s->errbody);
|
|
s->last_index = -1;
|
|
s->in_reason = 0;
|
|
s->emitted_any = 0;
|
|
s->stopped = 0;
|
|
s->http_checked = 0;
|
|
s->error_body = 0;
|
|
}
|
|
|
|
/* --- record / replay ----------------------------------------------------- */
|
|
|
|
static atomic_int g_trace_seq;
|
|
|
|
void cst_llm_trace_reset(void) { atomic_store(&g_trace_seq, 0); }
|
|
|
|
static void trace_path(const char *dir, int seq, const char *ext, char *out,
|
|
size_t n) {
|
|
snprintf(out, n, "%s/%04d.%s", dir, seq, ext);
|
|
}
|
|
|
|
static int replay_run(sse_t *s, const char *dir, int seq, char *err,
|
|
size_t err_sz) {
|
|
char path[OS_PATH_MAX];
|
|
trace_path(dir, seq, "sse", path, sizeof(path));
|
|
FILE *fp = fopen(path, "rb");
|
|
if (!fp) {
|
|
snprintf(err, err_sz, "replay: missing %s", path);
|
|
return -1;
|
|
}
|
|
size_t chunk = 16;
|
|
const char *e = getenv("CST_REPLAY_CHUNK");
|
|
if (e && atoi(e) > 0)
|
|
chunk = (size_t)atoi(e);
|
|
char buf[4096];
|
|
if (chunk > sizeof(buf))
|
|
chunk = sizeof(buf);
|
|
size_t n;
|
|
if (s->st)
|
|
s->st->http_status = 200;
|
|
while ((n = fread(buf, 1, chunk, fp)) > 0) {
|
|
if (sse_feed(s, buf, n) != 0)
|
|
break;
|
|
}
|
|
fclose(fp);
|
|
if (flag_set(s->req->cancel_flag)) {
|
|
snprintf(err, err_sz, "cancelled");
|
|
return -1;
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
static void sleep_ms_cancellable(int ms, volatile int *cancel) {
|
|
for (int t = 0; t < ms; t += 100) {
|
|
if (flag_set(cancel))
|
|
return;
|
|
#ifdef _WIN32
|
|
Sleep(100);
|
|
#else
|
|
usleep(100000);
|
|
#endif
|
|
}
|
|
}
|
|
|
|
static int retryable(CURLcode rc, long http) {
|
|
if (rc == CURLE_OK)
|
|
return http == 429 || http >= 500;
|
|
return rc == CURLE_COULDNT_CONNECT || rc == CURLE_COULDNT_RESOLVE_HOST ||
|
|
rc == CURLE_OPERATION_TIMEDOUT || rc == CURLE_SEND_ERROR ||
|
|
rc == CURLE_RECV_ERROR || rc == CURLE_GOT_NOTHING ||
|
|
rc == CURLE_SSL_CONNECT_ERROR || rc == CURLE_PARTIAL_FILE;
|
|
}
|
|
|
|
static void format_http_error(sse_t *s, long http, char *err, size_t err_sz) {
|
|
const char *body = s->errbody.buf ? s->errbody.buf : "";
|
|
cJSON *j = cJSON_Parse(body);
|
|
const cJSON *e = cJSON_GetObjectItemCaseSensitive(j, "error");
|
|
const cJSON *m = cJSON_GetObjectItemCaseSensitive(e, "message");
|
|
if (cJSON_IsString(m))
|
|
snprintf(err, err_sz, "http %ld: %s", http, m->valuestring);
|
|
else if (body[0])
|
|
snprintf(err, err_sz, "http %ld: %.400s", http, body);
|
|
else
|
|
snprintf(err, err_sz, "http %ld", http);
|
|
cJSON_Delete(j);
|
|
}
|
|
|
|
int cst_llm_chat_stream(const cst_llm_request_t *req, char *err,
|
|
size_t err_sz) {
|
|
char dummy[8];
|
|
if (!err || !err_sz) {
|
|
err = dummy;
|
|
err_sz = sizeof(dummy);
|
|
}
|
|
err[0] = 0;
|
|
if (!req || !req->api_base || !req->model) {
|
|
snprintf(err, err_sz, "bad request");
|
|
return -1;
|
|
}
|
|
cst_llm_stats_t local_st;
|
|
cst_llm_stats_t *st = req->stats ? req->stats : &local_st;
|
|
memset(st, 0, sizeof(*st));
|
|
st->t_first_byte_ms = -1;
|
|
|
|
char *body = cst_llm_build_body(req);
|
|
if (!body) {
|
|
snprintf(err, err_sz, "cannot build request body");
|
|
return -1;
|
|
}
|
|
|
|
int seq = atomic_fetch_add(&g_trace_seq, 1);
|
|
const char *rec_dir = getenv("CST_RECORD_SSE");
|
|
const char *rep_dir = getenv("CST_REPLAY_SSE");
|
|
if (rec_dir && rec_dir[0]) {
|
|
os_mkdir_p(rec_dir, 0755);
|
|
char p[OS_PATH_MAX];
|
|
trace_path(rec_dir, seq, "req.json", p, sizeof(p));
|
|
FILE *fp = fopen(p, "wb");
|
|
if (fp) {
|
|
fputs(body, fp);
|
|
fclose(fp);
|
|
}
|
|
}
|
|
|
|
sse_t s;
|
|
memset(&s, 0, sizeof(s));
|
|
s.req = req;
|
|
s.st = st;
|
|
s.last_index = -1;
|
|
s.t0 = os_now_ms();
|
|
int ret = 0;
|
|
|
|
if (rep_dir && rep_dir[0]) {
|
|
ret = replay_run(&s, rep_dir, seq, err, err_sz);
|
|
goto finish;
|
|
}
|
|
|
|
char url[1024];
|
|
snprintf(url, sizeof(url), "%s/chat/completions", req->api_base);
|
|
struct curl_slist *hdrs = NULL;
|
|
hdrs = curl_slist_append(hdrs, "Content-Type: application/json");
|
|
hdrs = curl_slist_append(hdrs, "Accept: text/event-stream");
|
|
char auth[700];
|
|
if (req->api_key && req->api_key[0]) {
|
|
snprintf(auth, sizeof(auth), "Authorization: Bearer %s", req->api_key);
|
|
hdrs = curl_slist_append(hdrs, auth);
|
|
}
|
|
|
|
for (int attempt = 0;; attempt++) {
|
|
CURL *curl = curl_easy_init();
|
|
if (!curl) {
|
|
snprintf(err, err_sz, "curl init failed");
|
|
ret = -1;
|
|
break;
|
|
}
|
|
llm_apply_ca_bundle(curl);
|
|
sse_reset(&s);
|
|
s.curl = curl;
|
|
if (rec_dir && rec_dir[0]) {
|
|
char p[OS_PATH_MAX];
|
|
trace_path(rec_dir, seq, "sse", p, sizeof(p));
|
|
s.rec = fopen(p, "wb");
|
|
}
|
|
curl_easy_setopt(curl, CURLOPT_URL, url);
|
|
curl_easy_setopt(curl, CURLOPT_HTTPHEADER, hdrs);
|
|
curl_easy_setopt(curl, CURLOPT_POSTFIELDS, body);
|
|
curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, write_cb);
|
|
curl_easy_setopt(curl, CURLOPT_WRITEDATA, &s);
|
|
curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, 30L);
|
|
curl_easy_setopt(curl, CURLOPT_LOW_SPEED_LIMIT, 1L);
|
|
curl_easy_setopt(curl, CURLOPT_LOW_SPEED_TIME, 300L);
|
|
curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1L);
|
|
curl_easy_setopt(curl, CURLOPT_NOPROGRESS, 0L);
|
|
curl_easy_setopt(curl, CURLOPT_XFERINFOFUNCTION, xferinfo_cb);
|
|
curl_easy_setopt(curl, CURLOPT_XFERINFODATA, &s);
|
|
|
|
CURLcode rc = curl_easy_perform(curl);
|
|
long http = 0;
|
|
curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &http);
|
|
st->http_status = http;
|
|
curl_easy_cleanup(curl);
|
|
s.curl = NULL;
|
|
if (s.rec) {
|
|
fclose(s.rec);
|
|
s.rec = NULL;
|
|
}
|
|
|
|
if (s.stopped && !flag_set(req->cancel_flag)) {
|
|
ret = 0; /* client-requested stop */
|
|
break;
|
|
}
|
|
if (flag_set(req->cancel_flag)) {
|
|
snprintf(err, err_sz, "cancelled");
|
|
ret = -1;
|
|
break;
|
|
}
|
|
if (!s.emitted_any && attempt < LLM_MAX_RETRIES && retryable(rc, http)) {
|
|
st->retries++;
|
|
sleep_ms_cancellable(1000 << attempt, req->cancel_flag);
|
|
continue;
|
|
}
|
|
if (rc != CURLE_OK) {
|
|
snprintf(err, err_sz, "curl: %s", curl_easy_strerror(rc));
|
|
ret = -1;
|
|
} else if (http >= 400) {
|
|
format_http_error(&s, http, err, err_sz);
|
|
ret = -1;
|
|
}
|
|
break;
|
|
}
|
|
curl_slist_free_all(hdrs);
|
|
|
|
finish:
|
|
if (ret == 0) {
|
|
if (s.line.len && !s.stopped) { /* last line without trailing newline */
|
|
sse_feed(&s, "\n", 1);
|
|
}
|
|
if (s.in_reason) {
|
|
emit_text(&s, "</think>", 8);
|
|
s.in_reason = 0;
|
|
}
|
|
slots_emit_all(&s, -1);
|
|
if (st->error[0] && !s.emitted_any) {
|
|
snprintf(err, err_sz, "api: %s", st->error);
|
|
ret = -1;
|
|
}
|
|
}
|
|
if (ret != 0 && !st->error[0])
|
|
snprintf(st->error, sizeof(st->error), "%s", err);
|
|
st->t_end_ms = os_now_ms() - s.t0;
|
|
sse_reset(&s);
|
|
free(body);
|
|
return ret;
|
|
}
|
|
|
|
/* --- model listing -------------------------------------------------------- */
|
|
|
|
#define MODELS_BODY_MAX (2 * 1024 * 1024)
|
|
|
|
static size_t models_write_cb(char *ptr, size_t size, size_t nmemb, void *ud) {
|
|
cst_buf_t *b = ud;
|
|
size_t n = size * nmemb;
|
|
if (b->len + n > MODELS_BODY_MAX || cst_buf_append(b, ptr, n) != 0)
|
|
return 0;
|
|
return n;
|
|
}
|
|
|
|
static int add_model(char out[][128], int n, int max_out, const char *id) {
|
|
if (!id || !id[0] || n >= max_out || strlen(id) >= 128)
|
|
return n;
|
|
for (int i = 0; i < n; i++)
|
|
if (strcmp(out[i], id) == 0)
|
|
return n;
|
|
snprintf(out[n], 128, "%s", id);
|
|
return n + 1;
|
|
}
|
|
|
|
/* {"data":[{"id":..}]}, {"models":[{"id"|"name":..}]} or a bare array. */
|
|
static int parse_models(const char *json, char out[][128], int max_out) {
|
|
cJSON *root = cJSON_Parse(json);
|
|
if (!root)
|
|
return -1;
|
|
const cJSON *arr = root;
|
|
if (!cJSON_IsArray(arr))
|
|
arr = cJSON_GetObjectItemCaseSensitive(root, "data");
|
|
if (!cJSON_IsArray(arr))
|
|
arr = cJSON_GetObjectItemCaseSensitive(root, "models");
|
|
int n = -1;
|
|
if (cJSON_IsArray(arr)) {
|
|
n = 0;
|
|
const cJSON *it;
|
|
cJSON_ArrayForEach(it, arr) {
|
|
const cJSON *id = cJSON_GetObjectItemCaseSensitive(it, "id");
|
|
if (!cJSON_IsString(id))
|
|
id = cJSON_GetObjectItemCaseSensitive(it, "name");
|
|
if (cJSON_IsString(id))
|
|
n = add_model(out, n, max_out, id->valuestring);
|
|
else if (cJSON_IsString(it))
|
|
n = add_model(out, n, max_out, it->valuestring);
|
|
}
|
|
}
|
|
cJSON_Delete(root);
|
|
return n;
|
|
}
|
|
|
|
int cst_llm_list_models(const char *api_base, const char *api_key,
|
|
char out[][128], int max_out, char *err,
|
|
size_t err_sz) {
|
|
char dummy[8];
|
|
if (!err || !err_sz) {
|
|
err = dummy;
|
|
err_sz = sizeof(dummy);
|
|
}
|
|
err[0] = 0;
|
|
if (!api_base || !api_base[0] || !out || max_out <= 0) {
|
|
snprintf(err, err_sz, "bad request");
|
|
return -1;
|
|
}
|
|
CURL *curl = curl_easy_init();
|
|
if (!curl) {
|
|
snprintf(err, err_sz, "curl init failed");
|
|
return -1;
|
|
}
|
|
llm_apply_ca_bundle(curl);
|
|
char url[1024];
|
|
size_t bl = strlen(api_base);
|
|
while (bl && api_base[bl - 1] == '/')
|
|
bl--;
|
|
snprintf(url, sizeof(url), "%.*s/models", (int)bl, api_base);
|
|
struct curl_slist *hdrs = curl_slist_append(NULL, "Accept: application/json");
|
|
char auth[700];
|
|
if (api_key && api_key[0]) {
|
|
snprintf(auth, sizeof(auth), "Authorization: Bearer %s", api_key);
|
|
hdrs = curl_slist_append(hdrs, auth);
|
|
}
|
|
cst_buf_t body = {0};
|
|
curl_easy_setopt(curl, CURLOPT_URL, url);
|
|
curl_easy_setopt(curl, CURLOPT_HTTPHEADER, hdrs);
|
|
curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, models_write_cb);
|
|
curl_easy_setopt(curl, CURLOPT_WRITEDATA, &body);
|
|
curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, 10L);
|
|
curl_easy_setopt(curl, CURLOPT_TIMEOUT, 10L);
|
|
curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1L);
|
|
curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L);
|
|
CURLcode rc = curl_easy_perform(curl);
|
|
long http = 0;
|
|
curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &http);
|
|
curl_slist_free_all(hdrs);
|
|
curl_easy_cleanup(curl);
|
|
memset(auth, 0, sizeof(auth));
|
|
|
|
int n = -1;
|
|
if (rc != CURLE_OK) {
|
|
snprintf(err, err_sz, "curl: %s", curl_easy_strerror(rc));
|
|
} else if (http == 401 || http == 403) {
|
|
snprintf(err, err_sz, "HTTP %ld: the API key was rejected", http);
|
|
n = CST_MODELS_AUTH;
|
|
} else if (http >= 400) {
|
|
snprintf(err, err_sz, "HTTP %ld from %s", http, url);
|
|
} else {
|
|
n = parse_models(body.buf ? body.buf : "", out, max_out);
|
|
if (n < 0)
|
|
snprintf(err, err_sz, "unexpected /models response");
|
|
}
|
|
cst_buf_free(&body);
|
|
return n;
|
|
}
|