WHC/CST-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
9bf7707912cf0e4d585c52d346e0256a711ca8cd
CST-agent/src/llm_openai.c
T
WHC c7b32c3fd6 v0.7.5: arrow keys, image request fix, deepseek-flash vision, musl CI 
TUI keys: register both CSI and SS3 forms of the cursor/Home/End keys and
the xterm modifier sequences when terminfo lacks them, so an arrow sent in
the "other" form no longer turns into Esc + a typed letter. Up/Down move by
visual (soft-wrapped) row, and the draft survives a trip through history.
scripts/tui_keys drives the TUI in a pty to check all of this.

Images: text inside a multimodal content array is now a {"type":"text"}
part; DeepSeek rejected the bare string with HTTP 422. deepseek-flash
(and its deepseek-chat / deepseek-reasoner aliases) read images, so the
registry marks them vision; deepseek-v4-pro stays text only.

CI: models.c lost _DEFAULT_SOURCE in 0.7.0 and gmtime_r broke the musl
build; tests sleep with poll() instead of usleep for the same reason.
2026-10-08 22:35:34 +08:00

1280 lines
41 KiB
C
Raw Blame History

#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
}
#include "cst_attach.h"
#include "cst_models.h"
/* --- 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;
}
/* --- multimodal content --------------------------------------------------
* OpenAI-compatible servers take a parts array where a plain string is also
* accepted, so text-only messages keep the string form and only those with
* images switch to parts. Images live in the session store and are read (and
* base64'd) here, once per request. */
/* The message's "content" as a parts array, converting a string in place. */
static cJSON *content_parts(cJSON *msg) {
cJSON *c = cJSON_GetObjectItemCaseSensitive(msg, "content");
if (cJSON_IsArray(c))
return c;
cJSON *arr = cJSON_CreateArray();
if (cJSON_IsString(c)) {
cJSON *t = cJSON_CreateObject();
cJSON_AddStringToObject(t, "type", "text");
cJSON_AddStringToObject(t, "text", c->valuestring);
cJSON_AddItemToArray(arr, t);
}
cJSON_ReplaceItemInObjectCaseSensitive(msg, "content", arr);
return arr;
}
/* Append text, joining it onto a trailing text part when there is one.
* Takes ownership of txt. */
static void parts_add_text(cJSON *parts, cJSON *txt) {
if (!txt)
txt = cJSON_CreateString("");
cJSON *last = cJSON_GetArrayItem(parts, cJSON_GetArraySize(parts) - 1);
const cJSON *lt = last ? cJSON_GetObjectItemCaseSensitive(last, "type") : NULL;
if (cJSON_IsString(lt) && strcmp(lt->valuestring, "text") == 0) {
const cJSON *old = cJSON_GetObjectItemCaseSensitive(last, "text");
cst_buf_t b = {0};
if (cJSON_IsString(old))
cst_buf_puts(&b, old->valuestring);
if (b.len)
cst_buf_puts(&b, "\n\n");
cst_buf_puts(&b, txt->valuestring ? txt->valuestring : "");
cJSON_ReplaceItemInObjectCaseSensitive(
last, "text", cJSON_CreateString(b.buf ? b.buf : ""));
cst_buf_free(&b);
cJSON_Delete(txt);
return;
}
if (!txt->valuestring || !txt->valuestring[0]) {
cJSON_Delete(txt);
return;
}
cJSON *t = cJSON_CreateObject();
cJSON_AddStringToObject(t, "type", "text");
cJSON_AddItemToObject(t, "text", txt);
cJSON_AddItemToArray(parts, t);
}
/* One stored image as an image_url part, or NULL with why filled in. */
static cJSON *image_part(const cst_llm_request_t *req,
const cst_attachment_t *a, size_t *budget, char *why,
size_t why_sz) {
why[0] = 0;
if (!req->session_path || !req->session_path[0]) {
snprintf(why, why_sz, "there is no session store to read it from");
return NULL;
}
char err[256];
size_t n = 0;
char *data = cst_attach_read(req->session_path, a, &n, err, sizeof(err));
if (!data) {
snprintf(why, why_sz, "%s", err);
return NULL;
}
if (n > *budget) {
snprintf(why, why_sz, "%d KB left in the %d MB image budget",
(int)(*budget / 1024), (int)(CST_ATTACH_MAX_TOTAL >> 20));
free(data);
return NULL;
}
*budget -= n;
char mime[32];
cst_attach_sniff(data, n, mime, sizeof(mime)); /* read verified it */
char *b64 = cst_base64_encode(data, n);
free(data);
if (!b64) {
snprintf(why, why_sz, "out of memory");
return NULL;
}
cst_buf_t b = {0};
cst_buf_puts(&b, "data:");
cst_buf_puts(&b, mime);
cst_buf_puts(&b, ";base64,");
cst_buf_puts(&b, b64);
free(b64);
if (!b.buf) {
snprintf(why, why_sz, "out of memory");
return NULL;
}
cJSON *p = cJSON_CreateObject();
cJSON_AddStringToObject(p, "type", "image_url");
cJSON *iu = cJSON_AddObjectToObject(p, "image_url");
cJSON_AddStringToObject(iu, "url", b.buf);
if (a->detail[0] && strcmp(a->detail, "auto") != 0)
cJSON_AddStringToObject(iu, "detail", a->detail);
cst_buf_free(&b);
return p;
}
/* Append every attachment of m as image parts (or a note saying why not). */
static void parts_add_images(const cst_llm_request_t *req, cJSON *parts,
const cst_message_t *m, size_t *budget) {
int n = m->nattach;
if (n > CST_ATTACH_MAX_COUNT) {
char note[160];
snprintf(note, sizeof(note), "[%d of %d images sent]", CST_ATTACH_MAX_COUNT,
n);
parts_add_text(parts, cJSON_CreateString(note));
n = CST_ATTACH_MAX_COUNT;
}
for (int i = 0; i < n; i++) {
const cst_attachment_t *a = &m->attach[i];
char why[256];
cJSON *p = image_part(req, a, budget, why, sizeof(why));
if (p) {
cJSON_AddItemToArray(parts, p);
continue;
}
char note[512];
snprintf(note, sizeof(note), "[image \"%.120s\" not sent: %.200s]",
a->name, why);
parts_add_text(parts, cJSON_CreateString(note));
}
}
/* Images a tool result produced are deferred: an assistant tool_call must be
* answered by tool messages only, so they ride in one extra user message
* after the run of tool messages ends. *pending collects it. */
static void defer_images(const cst_llm_request_t *req, const cst_message_t *m,
cJSON **pending, size_t *budget) {
cJSON *p = *pending;
if (!p) {
p = cJSON_CreateObject();
cJSON_AddStringToObject(p, "role", "user");
cJSON_AddArrayToObject(p, "content");
*pending = p;
}
parts_add_images(req, cJSON_GetObjectItemCaseSensitive(p, "content"), m,
budget);
}
/* m's text plus one line per image it would have carried (malloc'd). */
static char *image_notes(const cst_message_t *m, const char *why) {
cst_buf_t b = {0};
cst_buf_puts(&b, m->content ? m->content : "");
for (int i = 0; i < m->nattach; i++) {
char d[128];
cst_attach_describe(&m->attach[i], d, sizeof(d));
cst_buf_printf(&b, "%s[image \"%.120s\" (%s) not sent: %s]",
b.len ? "\n" : "", m->attach[i].name, d, why);
}
return cst_buf_steal(&b);
}
/* --- cache breakpoints (Anthropic-style) ----------------------------------
* At most 4 per request. The system prompt one also covers tools[]; the
* newest message one moves forward every step; the newest real user message
* keeps a hit when a long tool run pushes the previous one out of the
* provider's lookback. */
static void mark_message(cJSON *msg, const char *ttl) {
const cJSON *c = cJSON_GetObjectItemCaseSensitive(msg, "content");
if (cJSON_IsString(c) ? !c->valuestring[0] : !cJSON_IsArray(c))
return;
cJSON *parts = content_parts(msg);
cJSON *last = cJSON_GetArrayItem(parts, cJSON_GetArraySize(parts) - 1);
if (!cJSON_IsObject(last) ||
cJSON_GetObjectItemCaseSensitive(last, "cache_control"))
return;
cJSON *cc = cJSON_AddObjectToObject(last, "cache_control");
cJSON_AddStringToObject(cc, "type", "ephemeral");
if (ttl && ttl[0])
cJSON_AddStringToObject(cc, "ttl", ttl);
}
static int msg_role_is(const cJSON *msg, const char *role) {
const cJSON *r = cJSON_GetObjectItemCaseSensitive(msg, "role");
return cJSON_IsString(r) && strcmp(r->valuestring, role) == 0;
}
static int msg_has_content(const cJSON *msg) {
const cJSON *c = cJSON_GetObjectItemCaseSensitive(msg, "content");
return cJSON_IsString(c) ? c->valuestring[0] != 0 : cJSON_IsArray(c);
}
static void cache_marks(cJSON *msgs, const char *ttl) {
int n = cJSON_GetArraySize(msgs);
cJSON *first = cJSON_GetArrayItem(msgs, 0);
if (first && msg_role_is(first, "system"))
mark_message(first, ttl);
int last = -1;
for (int i = n - 1; i > 0 && last < 0; i--)
if (msg_has_content(cJSON_GetArrayItem(msgs, i)))
last = i;
if (last < 0)
return;
mark_message(cJSON_GetArrayItem(msgs, last), ttl);
for (int i = last - 1; i > 0; i--) {
cJSON *m = cJSON_GetArrayItem(msgs, i);
if (msg_role_is(m, "user") && msg_has_content(m)) {
mark_message(m, ttl);
break;
}
}
}
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 */
cJSON *pending = NULL; /* images from tool results, waiting for a break */
size_t img_budget = CST_ATTACH_MAX_TOTAL;
const char *noimg = getenv("CST_NO_IMAGES");
const char *img_off =
noimg && noimg[0] && noimg[0] != '0'
? "image sending is disabled (CST_NO_IMAGES)"
: req->images_off ? "this model does not take images"
: NULL;
for (int i = 0; i < req->nmsgs; i++) {
cst_message_t mm = req->msgs[i];
const cst_message_t *m = &mm;
cst_role_t role = m->role;
if (role == CST_ROLE_TOOL && !m->tool_call_id[0])
role = CST_ROLE_USER;
int has_img = m->attach && m->nattach > 0;
char *noted = NULL;
if (has_img && img_off) {
/* plain text: text-only APIs may refuse content arrays */
noted = image_notes(m, img_off);
if (noted)
mm.content = noted;
has_img = 0;
}
/* Every tool_call must be answered by tool messages, so tool-produced
* images only leave the queue when the run of tool results is over. */
if (role != CST_ROLE_TOOL && pending) {
cJSON_AddItemToArray(msgs, pending);
prev_user = pending;
pending = NULL;
}
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 *pc = prev_user
? cJSON_GetObjectItemCaseSensitive(prev_user, "content")
: NULL;
if (!has_img && !cJSON_IsArray(pc)) {
cJSON *c = clean_string(m->content, 0);
if (prev_user) {
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);
free(noted);
continue;
}
cJSON *o = cJSON_CreateObject();
cJSON_AddStringToObject(o, "role", "user");
cJSON_AddItemToObject(o, "content", c);
cJSON_AddItemToArray(msgs, o);
prev_user = o;
free(noted);
continue;
}
cJSON *o = prev_user;
if (!o) {
o = cJSON_CreateObject();
cJSON_AddStringToObject(o, "role", "user");
cJSON_AddArrayToObject(o, "content");
cJSON_AddItemToArray(msgs, o);
prev_user = o;
}
cJSON *parts = content_parts(o);
parts_add_text(parts, clean_string(m->content, 0));
if (has_img)
parts_add_images(req, parts, m, &img_budget);
free(noted);
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));
}
if (role == CST_ROLE_ASSISTANT) {
if (req->send_reasoning && m->reasoning && m->reasoning[0])
cJSON_AddItemToObject(o, "reasoning_content",
clean_string(m->reasoning, 0));
if (m->prefix)
cJSON_AddTrueToObject(o, "prefix");
}
cJSON_AddItemToArray(msgs, o);
if (role == CST_ROLE_TOOL && has_img)
defer_images(req, m, &pending, &img_budget);
free(noted);
}
if (pending)
cJSON_AddItemToArray(msgs, pending);
if (req->enable_tools) {
cJSON *tools = req->tools_json ? cJSON_Parse(req->tools_json)
: 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");
}
}
if (req->cache_style == CST_CACHE_OPENAI && req->cache_key &&
req->cache_key[0]) {
cJSON_AddStringToObject(root, "prompt_cache_key", req->cache_key);
const char *ret = getenv("CST_CACHE_RETENTION");
if (ret && ret[0])
cJSON_AddStringToObject(root, "prompt_cache_retention", ret);
} else if (req->cache_style == CST_CACHE_ANTHROPIC) {
cache_marks(msgs, req->cache_ttl);
}
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");
/* DeepSeek: prompt_cache_hit/miss_tokens. OpenAI and routers:
* prompt_tokens_details.cached_tokens (+ cache_write_tokens).
* Anthropic fields passed through: cache_read/creation_input_tokens. */
const cJSON *hit =
cJSON_GetObjectItemCaseSensitive(usage, "prompt_cache_hit_tokens");
const cJSON *miss =
cJSON_GetObjectItemCaseSensitive(usage, "prompt_cache_miss_tokens");
const cJSON *det =
cJSON_GetObjectItemCaseSensitive(usage, "prompt_tokens_details");
const cJSON *cached = cJSON_GetObjectItemCaseSensitive(det, "cached_tokens");
const cJSON *dwrite =
cJSON_GetObjectItemCaseSensitive(det, "cache_write_tokens");
const cJSON *aread =
cJSON_GetObjectItemCaseSensitive(usage, "cache_read_input_tokens");
const cJSON *awrite =
cJSON_GetObjectItemCaseSensitive(usage, "cache_creation_input_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;
else if (cJSON_IsNumber(aread))
s->st->cached_tokens = (long)aread->valuedouble;
if (cJSON_IsNumber(dwrite))
s->st->cache_write_tokens = (long)dwrite->valuedouble;
else if (cJSON_IsNumber(awrite))
s->st->cache_write_tokens = (long)awrite->valuedouble;
if (cJSON_IsNumber(miss)) {
s->st->cache_miss_tokens = (long)miss->valuedouble;
} else {
long m = s->st->prompt_tokens - s->st->cached_tokens -
s->st->cache_write_tokens;
s->st->cache_miss_tokens = m > 0 ? m : 0;
}
}
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 void format_http_error(sse_t *s, long http, char *err, size_t err_sz);
/* NNNN.sse is fed as the stream; with an NNNN.status other than 200 it is
* the error body of that HTTP status instead. */
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, "status", path, sizeof(path));
FILE *sf = fopen(path, "rb");
long status = 200;
if (sf) {
if (fscanf(sf, "%ld", &status) != 1)
status = 200;
fclose(sf);
}
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;
}
if (status != 200) {
char eb[4096];
size_t got = fread(eb, 1, sizeof(eb), fp);
fclose(fp);
cst_buf_append(&s->errbody, eb, got);
if (s->st)
s->st->http_status = status;
format_http_error(s, status, err, err_sz);
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 (8 * 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;
}
/* A number, or a numeric string (OpenRouter prices are strings). */
static int json_num(const cJSON *v, double *out) {
if (cJSON_IsNumber(v)) {
*out = v->valuedouble;
return 0;
}
if (cJSON_IsString(v) && v->valuestring[0]) {
char *end;
double d = strtod(v->valuestring, &end);
if (end != v->valuestring && !*end) {
*out = d;
return 0;
}
}
return -1;
}
static int has_image(const cJSON *arr) {
const cJSON *x;
cJSON_ArrayForEach(x, arr) {
if (cJSON_IsString(x) && strcmp(x->valuestring, "image") == 0)
return 1;
}
return 0;
}
/* Optional OpenRouter-style metadata: context_length (or context_window /
* max_model_len), architecture.input_modalities, modalities, and pricing in
* USD per token. */
static void model_meta(const cJSON *it, cst_model_meta_t *m) {
static const char *const ctx_keys[] = {"context_length", "context_window",
"max_model_len"};
for (size_t k = 0; k < sizeof(ctx_keys) / sizeof(ctx_keys[0]); k++) {
double d;
if (json_num(cJSON_GetObjectItemCaseSensitive(it, ctx_keys[k]), &d) == 0 &&
d > 0) {
m->context = (long)d;
break;
}
}
const cJSON *arch = cJSON_GetObjectItemCaseSensitive(it, "architecture");
const cJSON *in = cJSON_GetObjectItemCaseSensitive(arch, "input_modalities");
const cJSON *mods = cJSON_GetObjectItemCaseSensitive(it, "modalities");
if (cJSON_IsObject(mods))
mods = cJSON_GetObjectItemCaseSensitive(mods, "input");
const cJSON *modality = cJSON_GetObjectItemCaseSensitive(arch, "modality");
if (cJSON_IsArray(in))
m->vision = has_image(in);
else if (cJSON_IsArray(mods))
m->vision = has_image(mods);
else if (cJSON_IsString(modality)) {
const char *arrow = strstr(modality->valuestring, "->");
size_t l = arrow ? (size_t)(arrow - modality->valuestring)
: strlen(modality->valuestring);
char left[64];
snprintf(left, sizeof(left), "%.*s", (int)l, modality->valuestring);
m->vision = strstr(left, "image") != NULL;
}
const cJSON *pr = cJSON_GetObjectItemCaseSensitive(it, "pricing");
double miss, out;
if (json_num(cJSON_GetObjectItemCaseSensitive(pr, "prompt"), &miss) == 0 &&
json_num(cJSON_GetObjectItemCaseSensitive(pr, "completion"), &out) == 0 &&
miss >= 0 && out >= 0) {
double hit = miss, write = miss;
json_num(cJSON_GetObjectItemCaseSensitive(pr, "input_cache_read"), &hit);
json_num(cJSON_GetObjectItemCaseSensitive(pr, "input_cache_write"), &write);
m->has_price = 1;
m->miss = miss * 1e6;
m->out = out * 1e6;
m->hit = (hit >= 0 ? hit : miss) * 1e6;
m->write = (write >= 0 ? write : miss) * 1e6;
}
}
static int add_model(cst_model_meta_t *out, int n, int max_out,
const char *id, const cJSON *it) {
if (!id || !id[0] || n >= max_out || strlen(id) >= sizeof(out->id))
return n;
for (int i = 0; i < n; i++)
if (strcmp(out[i].id, id) == 0)
return n;
memset(&out[n], 0, sizeof(out[n]));
out[n].vision = -1;
snprintf(out[n].id, sizeof(out[n].id), "%s", id);
if (cJSON_IsObject(it))
model_meta(it, &out[n]);
return n + 1;
}
/* {"data":[{"id":..}]}, {"models":[{"id"|"name":..}]} or a bare array. */
static int parse_models(const char *json, cst_model_meta_t *out, 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, it);
else if (cJSON_IsString(it))
n = add_model(out, n, max_out, it->valuestring, NULL);
}
}
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) {
if (!out || max_out <= 0)
return cst_llm_list_models_meta(api_base, api_key, NULL, 0, err, err_sz);
cst_model_meta_t *m = calloc((size_t)max_out, sizeof(*m));
if (!m) {
if (err && err_sz)
snprintf(err, err_sz, "out of memory");
return -1;
}
int n = cst_llm_list_models_meta(api_base, api_key, m, max_out, err, err_sz);
for (int i = 0; i < n; i++)
snprintf(out[i], 128, "%s", m[i].id);
free(m);
return n;
}
int cst_llm_list_models_meta(const char *api_base, const char *api_key,
cst_model_meta_t *out, 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;
}
Reference in New Issue View Git Blame Copy Permalink