summaryrefslogtreecommitdiffstats
path: root/src/provider.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/provider.cpp')
-rw-r--r--src/provider.cpp941
1 files changed, 941 insertions, 0 deletions
diff --git a/src/provider.cpp b/src/provider.cpp
new file mode 100644
index 0000000..8ac95db
--- /dev/null
+++ b/src/provider.cpp
@@ -0,0 +1,941 @@
+/* provider.c — the session-bound provider layer for llshc.
+ *
+ * Implements the `llsh_model` vtable on top of llm.h's protocol layer:
+ * wires the session (config, preamble, allowed tools, .agents snapshot),
+ * the built-in + MCP tool registry, and the session emit callback into the
+ * one-shot `complete` and the streaming agent `stream`. */
+#include "provider.h"
+
+#include <atomic>
+#include <cmath>
+#include <limits>
+#include <stdlib.h>
+#include <string.h>
+
+#include "compact.h"
+#include "config.h"
+#include "dotagents.h"
+#include "goal.h"
+#include "message.h"
+#include "limits.h"
+#include <cjson/cJSON.h>
+#include "tools.h"
+#include "util.h"
+
+#define PCAST(m) ((provider_model *)(m))
+#define jstr(o, k) ({ const cJSON *_v = (o) ? cJSON_GetObjectItem((o), (k)) : NULL; \
+ (cJSON_IsString(_v) && _v->valuestring) ? _v->valuestring : ""; })
+
+typedef struct provider_model {
+ llsh_model base; /* must be first */
+ session *sess; /* non-owning back-pointer */
+ mcp_registry *mcp;
+ provider_role role;
+ /* Cached `"tools":[...]` request fragment. Valid while (provider,
+ * allowed list, mcp tools) is unchanged; invalidated on key mismatch. */
+ char *tools_json;
+ unsigned long tools_key;
+ CURL *curl; /* persistent easy handle, created on first post */
+} provider_model;
+
+/* ------------------------------------------------------------------ */
+/* Session -> llm.h adapters */
+/* ------------------------------------------------------------------ */
+
+static llm_conn conn_of(const session *s) {
+ const config *cfg = session_active_config((session *)s);
+ llm_conn c;
+ switch (cfg->provider) {
+ case PROV_ANTHROPIC:
+ c.provider = LLM_ANTHROPIC;
+ break;
+ case PROV_OLLAMA:
+ c.provider = LLM_OLLAMA;
+ break;
+ default:
+ c.provider = LLM_OPENAI;
+ break;
+ }
+ c.model = cfg->model;
+ c.base_url = cfg->base_url;
+ c.api_key = cfg->api_key;
+ c.ca_bundle = llsh_ca_bundle();
+ return c;
+}
+
+/* Join the tool-result content parts of a user_tool_result_part. Caller
+ * frees. */
+static char *trc_join(const user_tool_result_part &r) {
+ xbuf b;
+ xbuf_init(&b);
+ bool first = true;
+ for (const auto &c : r.content) {
+ if (!first)
+ xbuf_push(&b, '\n');
+ if (const auto *t = std::get_if<trc_text>(&c))
+ xbuf_put(&b, t->text.c_str());
+ else if (const auto *s = std::get_if<trc_json>(&c))
+ xbuf_put(&b, s->value.c_str());
+ first = false;
+ }
+ char *s = xbuf_steal(&b);
+ xbuf_free(&b);
+ return s;
+}
+
+/* App messages -> borrowed llm_message array (tool-result parts joined). */
+typedef struct {
+ llm_message *msgs;
+ size_t n;
+ char **joined; /* owned joined trc strings */
+ size_t joined_n;
+} app_llm_msgs;
+
+static app_llm_msgs app_to_llm(const message *const *msgs, size_t n) {
+ app_llm_msgs a;
+ memset(&a, 0, sizeof a);
+ a.msgs = (llm_message *)xmalloc(n ? n * sizeof *a.msgs : 1);
+ a.n = n;
+ for (size_t i = 0; i < n; i++) {
+ const message *m = msgs[i];
+ llm_message *o = &a.msgs[i];
+ size_t cnt = 0;
+ if (m->is_system()) {
+ o->role = LLM_ROLE_SYSTEM;
+ cnt = 1;
+ } else if (m->is_user()) {
+ o->role = LLM_ROLE_USER;
+ cnt = m->user().parts.size();
+ } else {
+ o->role = LLM_ROLE_ASSISTANT;
+ cnt = m->assistant().parts.size();
+ }
+ if (cnt == 0)
+ continue;
+ o->parts = (llm_part *)xmalloc(cnt * sizeof *o->parts);
+ size_t k = 0;
+ if (m->is_system()) {
+ o->parts[k].kind = LLM_TEXT;
+ o->parts[k].text = m->system().text.c_str();
+ k++;
+ } else if (m->is_user()) {
+ for (size_t j = 0; j < m->user().parts.size(); j++) {
+ const auto &p = m->user().parts[j];
+ if (const auto *t = std::get_if<user_text_part>(&p)) {
+ o->parts[k].kind = LLM_TEXT;
+ o->parts[k].text = t->text.c_str();
+ } else {
+ const auto &r = std::get<user_tool_result_part>(p);
+ char *joined = trc_join(r);
+ a.joined =
+ (char **)xrealloc(a.joined,
+ (a.joined_n + 1) * sizeof *a.joined);
+ a.joined[a.joined_n++] = joined;
+ o->parts[k].kind = LLM_TOOL_RESULT;
+ o->parts[k].text = joined;
+ o->parts[k].name = r.name.c_str();
+ o->parts[k].call = r.call.c_str();
+ }
+ k++;
+ }
+ } else {
+ for (size_t j = 0; j < m->assistant().parts.size(); j++) {
+ const auto &p = m->assistant().parts[j];
+ if (const auto *t = std::get_if<asst_text_part>(&p)) {
+ o->parts[k].kind = LLM_TEXT;
+ o->parts[k].text = t->text.c_str();
+ } else if (const auto *r = std::get_if<asst_reasoning_part>(&p)) {
+ o->parts[k].kind = LLM_REASONING;
+ o->parts[k].text = r->text.c_str();
+ } else if (const auto *c = std::get_if<asst_tool_call_part>(&p)) {
+ o->parts[k].kind = LLM_TOOL_CALL;
+ o->parts[k].name = c->name.c_str();
+ o->parts[k].args = c->args.c_str();
+ }
+ k++;
+ }
+ }
+ o->part_count = k;
+ }
+ return a;
+}
+
+static void app_llm_msgs_free(app_llm_msgs *a) {
+ for (size_t i = 0; i < a->n; i++)
+ free(a->msgs[i].parts);
+ free(a->msgs);
+ for (size_t i = 0; i < a->joined_n; i++)
+ free(a->joined[i]);
+ free(a->joined);
+ memset(a, 0, sizeof *a);
+}
+
+/* Bridge the session's emit callback onto llm.h's primitive emitter. */
+typedef struct {
+ sess_emit_cb cb;
+ void *user;
+} emit_user;
+
+static int emit_bridge(llm_provider prov, llm_emit_kind kind,
+ const char *delta, void *ud) {
+ (void)prov;
+ emit_user *eu = (emit_user *) ud;
+ sess_emit ev;
+ memset(&ev, 0, sizeof ev);
+ ev.kind = (kind == LLM_EMIT_TEXT) ? SESS_EMIT_TEXT : SESS_EMIT_REASONING;
+ ev.text = xstrdup(delta);
+ int rc = eu->cb ? eu->cb(&ev, eu->user) : 0;
+ sess_emit_free(&ev); /* creator owns the payload; cb only reads it */
+ return rc;
+}
+
+/* ------------------------------------------------------------------ */
+/* Tool execution */
+/* ------------------------------------------------------------------ */
+
+static bool tool_allowed(const char *name, const strlist *allowed);
+
+static const char *required_string(const cJSON *args, const char *key) {
+ const cJSON *v = args ? cJSON_GetObjectItemCaseSensitive(args, key) : NULL;
+ return cJSON_IsString(v) && v->valuestring ? v->valuestring : NULL;
+}
+
+static char *bad_tool_arg(const char *tool, const char *arg, char **err) {
+ if (err)
+ *err = xstrfmt("%s requires string `%s`", tool, arg);
+ return NULL;
+}
+
+static char *exec_builtin(const session *s, const char *name,
+ const char *args_json, unsigned long shell_timeout,
+ char **err) {
+ const dotagents *dot = s ? session_dot(s) : NULL;
+ const config *cfg = s ? session_config(s) : NULL;
+ if (err)
+ *err = NULL;
+ cJSON *a = cJSON_Parse(args_json ? args_json : "{}");
+ if (!cJSON_IsObject(a)) {
+ cJSON_Delete(a);
+ if (err)
+ *err = xstrdup("tool arguments must be a JSON object");
+ return NULL;
+ }
+ char *out = NULL;
+ if (xstreq(name, "read_file")) {
+ if (!required_string(a, "path")) {
+ out = bad_tool_arg(name, "path", err);
+ goto done;
+ }
+ unsigned long off = 0, lim = 0;
+ const cJSON *ov = cJSON_GetObjectItem(a, "offset");
+ const cJSON *lv = cJSON_GetObjectItem(a, "limit");
+ if (ov && (!cJSON_IsNumber(ov) || !std::isfinite(ov->valuedouble) ||
+ ov->valuedouble < 1 || std::floor(ov->valuedouble) != ov->valuedouble ||
+ ov->valuedouble >= (double)std::numeric_limits<unsigned long>::max())) {
+ out = bad_tool_arg(name, "offset (positive integer)", err);
+ goto done;
+ }
+ if (ov)
+ off = (unsigned long)ov->valuedouble;
+ if (lv && (!cJSON_IsNumber(lv) || !std::isfinite(lv->valuedouble) ||
+ lv->valuedouble < 1 || std::floor(lv->valuedouble) != lv->valuedouble ||
+ lv->valuedouble >= (double)std::numeric_limits<unsigned long>::max())) {
+ out = bad_tool_arg(name, "limit (positive integer)", err);
+ goto done;
+ }
+ if (lv)
+ lim = (unsigned long)lv->valuedouble;
+ out = tool_read_file(jstr(a, "path"), off, lim, err);
+ } else if (xstreq(name, "write_file")) {
+ if (!required_string(a, "path") || !required_string(a, "content")) {
+ out = bad_tool_arg(name,
+ required_string(a, "path") ? "content" : "path",
+ err);
+ goto done;
+ }
+ out = tool_write_file(required_string(a, "path"),
+ required_string(a, "content"), err);
+ } else if (xstreq(name, "run_shell")) {
+ const char *command = required_string(a, "command");
+ if (!command || !command[0]) {
+ out = bad_tool_arg(name, "command", err);
+ goto done;
+ }
+ const tool_limits *lim = tool_limits_current();
+ if (shell_gate(command, lim, err) == 0)
+ out = tool_run_shell_timed(command, shell_timeout, err);
+ } else if (xstreq(name, "use_skill")) {
+ if (!required_string(a, "id")) {
+ out = bad_tool_arg(name, "id", err);
+ goto done;
+ }
+ out = tool_use_skill(dot, required_string(a, "id"), err);
+ } else if (xstreq(name, "search_memories")) {
+ if (!required_string(a, "query")) {
+ out = bad_tool_arg(name, "query", err);
+ goto done;
+ }
+ out = tool_search_memories(dot, required_string(a, "query"), err);
+ } else if (xstreq(name, "embed")) {
+ if (!required_string(a, "text")) {
+ out = bad_tool_arg(name, "text", err);
+ goto done;
+ }
+ out = tool_embed(required_string(a, "text"), cfg, dot, err);
+ } else if (xstreq(name, "llm")) {
+ llm_tool_call call;
+ memset(&call, 0, sizeof call);
+ call.temperature = -1.0;
+ call.prompt = required_string(a, "prompt");
+ call.model = jstr(a, "model");
+ const cJSON *tv = cJSON_GetObjectItem(a, "temperature");
+ if (tv && (!cJSON_IsNumber(tv) || !std::isfinite(tv->valuedouble) ||
+ tv->valuedouble < 0 || tv->valuedouble > 2)) {
+ out = bad_tool_arg(name, "temperature (0..2)", err);
+ goto done;
+ }
+ if (tv)
+ call.temperature = tv->valuedouble;
+ const cJSON *mv = cJSON_GetObjectItem(a, "max_tokens");
+ if (mv && (!cJSON_IsNumber(mv) || !std::isfinite(mv->valuedouble) ||
+ mv->valuedouble < 1 || std::floor(mv->valuedouble) != mv->valuedouble ||
+ mv->valuedouble >= (double)std::numeric_limits<unsigned long>::max())) {
+ out = bad_tool_arg(name, "max_tokens (positive integer)", err);
+ goto done;
+ }
+ if (mv)
+ call.max_tokens = (unsigned long)mv->valuedouble;
+ call.reasoning_effort = jstr(a, "reasoning_effort");
+ const cJSON *bv = cJSON_GetObjectItem(a, "thinking_budget");
+ if (bv && (!cJSON_IsNumber(bv) || !std::isfinite(bv->valuedouble) ||
+ bv->valuedouble < 1 || std::floor(bv->valuedouble) != bv->valuedouble ||
+ bv->valuedouble >= (double)std::numeric_limits<unsigned long>::max())) {
+ out = bad_tool_arg(name, "thinking_budget (positive integer)", err);
+ goto done;
+ }
+ if (bv)
+ call.thinking_budget = (unsigned long)bv->valuedouble;
+ const cJSON *sv = cJSON_GetObjectItem(a, "response_schema");
+ char *schema_ser = NULL;
+ if (sv) {
+ if (cJSON_IsString(sv) && sv->valuestring)
+ call.response_schema = sv->valuestring;
+ else {
+ schema_ser = cJSON_PrintUnformatted(sv);
+ call.response_schema = schema_ser;
+ }
+ }
+ const cJSON *ev = cJSON_GetObjectItem(a, "embed");
+ if (ev && cJSON_IsBool(ev))
+ call.embed = cJSON_IsTrue(ev);
+ if (!cfg) {
+ if (err)
+ *err = xstrdup("llm: no session config");
+ } else {
+ out = tool_llm(&call, cfg, dot, err);
+ }
+ free(schema_ser);
+ } else if (xstreq(name, "compact")) {
+ compact_report rep;
+ memset(&rep, 0, sizeof rep);
+ int rc = session_force_compact((session *)s, &rep, err);
+ if (rc > 0) {
+ session_print_compact(&rep);
+ out = xstrfmt("compacted: %lu -> %lu tokens", rep.tokens_before,
+ rep.tokens_after);
+ } else if (rc == 0) {
+ out = xstrdup("history too short to compact");
+ }
+ } else {
+ if (err)
+ *err = xstrfmt("unknown tool `%s`", name);
+ }
+done:
+ cJSON_Delete(a);
+ return out;
+}
+
+static char *exec_tool(const provider_model *p, const char *name,
+ const char *args_json, unsigned long shell_timeout,
+ char **err) {
+ const strlist *allowed = session_allowed_tools(p->sess);
+ if (!tool_allowed(name, allowed)) {
+ if (err)
+ *err = xstrfmt("tool `%s` is not allowed by the active profile", name);
+ return NULL;
+ }
+ if (p->mcp) {
+ for (size_t i = 0; i < p->mcp->tool_count; i++)
+ if (xstreq(p->mcp->tools[i].qualified_name, name))
+ return mcp_tool_invoke(&p->mcp->tools[i], args_json, err);
+ }
+ return exec_builtin(p->sess, name, args_json, shell_timeout, err);
+}
+
+/* ------------------------------------------------------------------ */
+/* Tool schemas (registry -> `"tools":[...]` fragment) */
+/* ------------------------------------------------------------------ */
+
+static bool tool_allowed(const char *name, const strlist *allowed) {
+ if (!allowed || allowed->len == 0)
+ return true;
+ for (size_t i = 0; i < allowed->len; i++)
+ if (xstreq(allowed->items[i], name))
+ return true;
+ return false;
+}
+
+/* FNV-1a over (provider, allowed tools, mcp tool names) -- the cache key
+ * for the tool-schema fragment. */
+static unsigned long tfv(unsigned long h, const char *s) {
+ for (const char *c = s; *c; c++)
+ h = (h ^ (unsigned char)*c) * 1099511628211UL;
+ return (h ^ 0xff) * 1099511628211UL;
+}
+
+static unsigned long tools_fingerprint(const config *cfg,
+ const strlist *allowed,
+ mcp_registry *mcp) {
+ unsigned long h = 1469598103934665603UL;
+ h = (h ^ (unsigned long)cfg->provider) * 1099511628211UL;
+ if (allowed)
+ for (size_t i = 0; i < allowed->len; i++)
+ h = tfv(h, allowed->items[i]);
+ if (mcp)
+ for (size_t i = 0; i < mcp->tool_count; i++)
+ h = tfv(h, mcp->tools[i].qualified_name);
+ return h;
+}
+
+/* The provider-shaped `"tools":[...]` fragment, cached on the model.
+ * Rebuilt only when the fingerprint (provider, allowed list, mcp tools)
+ * changes. Borrowed from `p`; valid until provider_destroy. */
+static const char *tools_fragment(provider_model *p, const config *cfg,
+ const strlist *allowed) {
+ unsigned long key = tools_fingerprint(cfg, allowed, p->mcp);
+ if (p->tools_json && p->tools_key == key)
+ return p->tools_json;
+ free(p->tools_json);
+ p->tools_json = NULL;
+ p->tools_key = key;
+
+ xbuf frag;
+ xbuf_init(&frag);
+ xbuf_put(&frag, "\"tools\":[");
+ bool first = true;
+ for (size_t i = 0; i < tool_builtin_count(); i++) {
+ const char *name = tool_builtin_name(i);
+ if (!tool_allowed(name, allowed))
+ continue;
+ cJSON *t;
+ if (cfg->provider == PROV_ANTHROPIC) {
+ t = cJSON_CreateObject();
+ cJSON_AddStringToObject(t, "name", name);
+ cJSON_AddStringToObject(t, "description", tool_desc(name));
+ cJSON_AddItemToObject(t, "input_schema", tool_params_json(name));
+ } else {
+ t = cJSON_CreateObject();
+ cJSON_AddStringToObject(t, "type", "function");
+ cJSON *f = cJSON_AddObjectToObject(t, "function");
+ cJSON_AddStringToObject(f, "name", name);
+ cJSON_AddStringToObject(f, "description", tool_desc(name));
+ cJSON_AddItemToObject(f, "parameters", tool_params_json(name));
+ }
+ char *ts = cJSON_PrintUnformatted(t);
+ cJSON_Delete(t);
+ if (!first)
+ xbuf_push(&frag, ',');
+ xbuf_put(&frag, ts ? ts : "{}");
+ free(ts);
+ first = false;
+ }
+ if (p->mcp) {
+ for (size_t i = 0; i < p->mcp->tool_count; i++) {
+ const mcp_tool *mt = &p->mcp->tools[i];
+ if (!tool_allowed(mt->qualified_name, allowed))
+ continue;
+ cJSON *t = cJSON_CreateObject();
+ cJSON *schema = mt->input_schema ? cJSON_Parse(mt->input_schema)
+ : NULL;
+ if (!schema) {
+ schema = cJSON_CreateObject();
+ cJSON_AddStringToObject(schema, "type", "object");
+ }
+ if (cfg->provider == PROV_ANTHROPIC) {
+ cJSON_AddStringToObject(t, "name", mt->qualified_name);
+ if (mt->description)
+ cJSON_AddStringToObject(t, "description", mt->description);
+ cJSON_AddItemToObject(t, "input_schema", schema);
+ } else {
+ cJSON_AddStringToObject(t, "type", "function");
+ cJSON *fn = cJSON_AddObjectToObject(t, "function");
+ cJSON_AddStringToObject(fn, "name", mt->qualified_name);
+ if (mt->description)
+ cJSON_AddStringToObject(fn, "description", mt->description);
+ cJSON_AddItemToObject(fn, "parameters", schema);
+ }
+ char *ts = cJSON_PrintUnformatted(t);
+ cJSON_Delete(t);
+ if (!first)
+ xbuf_push(&frag, ',');
+ xbuf_put(&frag, ts ? ts : "{}");
+ free(ts);
+ first = false;
+ }
+ }
+ xbuf_push(&frag, ']');
+ p->tools_json = xbuf_steal(&frag);
+ xbuf_free(&frag);
+ return p->tools_json;
+}
+
+/* ------------------------------------------------------------------ */
+/* complete (summarizer / evaluator) */
+/* ------------------------------------------------------------------ */
+
+static char *provider_complete(const llsh_model *mm, const char *prompt,
+ sess_error *err) {
+ provider_model *p = PCAST(mm);
+ if (err)
+ *err = (sess_error){0};
+ if (!p->sess) {
+ if (err)
+ err->message = xstrdup("provider model not attached to a session");
+ return NULL;
+ }
+ const char *preamble;
+ if (p->role == PROV_ROLE_SUMMARIZER)
+ preamble = compact_summarizer_preamble();
+ else if (p->role == PROV_ROLE_EVALUATOR)
+ preamble = goal_evaluator_preamble();
+ else if (p->role == PROV_ROLE_RUBRIC)
+ preamble = compact_rubric_preamble();
+ else
+ preamble = session_current_preamble(p->sess);
+
+ /* single user message carrying the prompt */
+ llm_part part;
+ memset(&part, 0, sizeof part);
+ part.kind = LLM_TEXT;
+ part.text = prompt ? prompt : "";
+ llm_message one;
+ one.role = LLM_ROLE_USER;
+ one.parts = &part;
+ one.part_count = 1;
+
+ llm_conn conn = conn_of(p->sess);
+ char *req = llm_build_request(&conn, preamble, &one, 1, NULL, false);
+ char *url = llm_url(&conn);
+ long status = 0;
+ char *body = NULL, *headers = NULL, *herr = NULL;
+ int hrc =
+ llm_post(&conn, url, req, &status, &body, &headers, &herr, NULL,
+ &p->curl);
+ free(req);
+ free(url);
+
+ if (hrc != 0) {
+ if (err) {
+ err->message = herr ? herr : xstrdup("transport error");
+ err->http_status = 0;
+ err->transport = 1;
+ }
+ free(body);
+ free(headers);
+ return NULL;
+ }
+ char *result = NULL;
+ llm_completion c;
+ memset(&c, 0, sizeof c);
+ if (llm_parse_body(&conn, body, &c) == 0) {
+ if (c.text && c.text[0])
+ result = c.text;
+ else
+ result = xstrdup("");
+ c.text = NULL;
+ llm_completion_free(&c);
+ }
+ if (status < 200 || status >= 300 || !result) {
+ char *em = llm_parse_error(body ? body : "", status);
+ if (err) {
+ err->message = em;
+ err->http_status = (int)status;
+ err->transport = 0;
+ err->retry_after_secs = llm_header_retry_after(headers);
+ }
+ free(result);
+ result = NULL;
+ }
+ free(body);
+ free(headers);
+ return result;
+}
+
+/* ------------------------------------------------------------------ */
+/* stream (agent turn) */
+/* ------------------------------------------------------------------ */
+
+void provider_trim_dangling_calls(message *msgs, size_t n) {
+ for (size_t i = 0; i < n; i++) {
+ message *m = &msgs[i];
+ if (!m->is_assistant() || m->assistant().parts.empty())
+ continue;
+ size_t calls = 0;
+ for (const auto &p : m->assistant().parts)
+ if (std::holds_alternative<asst_tool_call_part>(p))
+ calls++;
+ if (calls == 0)
+ continue;
+ size_t results = 0;
+ for (size_t k = i + 1; k < n; k++) {
+ message *u = &msgs[k];
+ if (u->is_user() && !u->user().parts.empty() &&
+ std::holds_alternative<user_tool_result_part>(
+ u->user().parts[0]))
+ results++;
+ else
+ break;
+ }
+ if (results >= calls)
+ continue;
+ size_t kept = 0;
+ auto &parts = m->assistant().parts;
+ for (size_t j = 0; j < parts.size(); j++) {
+ if (std::holds_alternative<asst_tool_call_part>(parts[j])) {
+ if (kept >= results) {
+ parts.erase(parts.begin() + static_cast<long>(j));
+ j--;
+ } else {
+ kept++;
+ }
+ }
+ }
+ }
+}
+
+static int provider_stream(const llsh_model *mm,
+ const message *history, size_t nhistory,
+ const char *prompt, sess_emit_cb cb, void *user,
+ std::atomic<bool> *stop, char **assistant_text,
+ unsigned long *input_tokens, message **new_messages,
+ size_t *n_new, sess_error *err) {
+ provider_model *p = PCAST(mm);
+ if (assistant_text)
+ *assistant_text = NULL;
+ if (input_tokens)
+ *input_tokens = 0;
+ if (new_messages)
+ *new_messages = NULL;
+ if (n_new)
+ *n_new = 0;
+ if (err)
+ *err = (sess_error){0};
+
+ if (!p->sess) {
+ if (err)
+ err->message = xstrdup("provider model not attached to a session");
+ return SESS_STREAM_FAILED;
+ }
+
+ const config *cfg = session_active_config(p->sess);
+ const char *preamble = session_current_preamble(p->sess);
+ const strlist *allowed = session_allowed_tools(p->sess);
+ const unsigned long max_turns = cfg->max_turns; /* 0 = unbounded */
+ llm_conn conn = conn_of(p->sess);
+ emit_user eu = {cb, user};
+
+ /* local working messages for this turn (assistant tool calls + results) */
+ std::vector<message> local;
+ int rc = SESS_STREAM_OK;
+ unsigned long last_in = 0;
+ char *final_text = NULL;
+ char *final_reasoning = NULL;
+ size_t call_counter = 0;
+ bool stopped = false;
+
+ for (unsigned long turn = 0; max_turns == 0 || turn < max_turns; turn++) {
+ if (stop && stop->load()) {
+ stopped = true;
+ break;
+ }
+ /* combined = history + [prompt] + local */
+ message pm = msg_user_text(prompt ? prompt : "");
+ size_t total = nhistory + 1 + local.size();
+ const message **all = (const message **)xmalloc(total * sizeof *all);
+ for (size_t i = 0; i < nhistory; i++)
+ all[i] = &history[i];
+ all[nhistory] = &pm;
+ for (size_t i = 0; i < local.size(); i++)
+ all[nhistory + 1 + i] = &local[i];
+
+ app_llm_msgs am = app_to_llm(all, total);
+ free(all);
+ const char *tools = tools_fragment(p, cfg, allowed);
+ char *req =
+ llm_build_request(&conn, preamble, am.msgs, am.n, tools, true);
+ app_llm_msgs_free(&am);
+ char *url = llm_url(&conn);
+ long status = 0;
+ char *body = NULL, *headers = NULL, *herr = NULL;
+ bool stopped_post = false;
+ llm_sse_acc acc;
+ llm_sse_acc_init(&acc, conn.provider);
+ int hrc = llm_post_stream(&conn, url, req, &status, &body, &headers,
+ &acc, emit_bridge, &eu, stop, &stopped_post,
+ &herr, &p->curl);
+ free(req);
+ free(url);
+
+ if (hrc != 0) {
+ /* A mid-stream transport drop is retryable and the partial
+ * response is discarded -- but the user has already seen the
+ * streamed text, so a retry re-shows it. Accepted (AUDIT). */
+ if (err) {
+ err->message = herr ? herr : xstrdup("transport error");
+ err->http_status = 0;
+ err->transport = 1;
+ }
+ free(body);
+ free(headers);
+ llm_sse_acc_free(&acc);
+ if (stopped_post)
+ stopped = true; /* user abort: not a retryable failure */
+ else
+ rc = SESS_STREAM_RETRYABLE;
+ break;
+ }
+ if (status < 200 || status >= 300) {
+ char *em = llm_parse_error(body ? body : "", status);
+ bool transient = sess_is_transient_status(status);
+ rc = transient ? SESS_STREAM_RETRYABLE : SESS_STREAM_FAILED;
+ if (err) {
+ err->message = em;
+ err->http_status = (int)status;
+ err->transport = 0;
+ err->retry_after_secs = llm_header_retry_after(headers);
+ }
+ free(body);
+ free(headers);
+ llm_sse_acc_free(&acc);
+ break;
+ }
+
+ llm_completion c;
+ memset(&c, 0, sizeof c);
+ llm_sse_acc_into(&acc, &c);
+ bool live = llm_sse_acc_had_content(&acc);
+ /* Fallback for endpoints that ignore `stream:true` and answer with
+ * one complete JSON body: parse it the non-streaming way; the text
+ * was not emitted live, so it is emitted below. */
+ if (!live && body && body[0] == '{')
+ llm_parse_body(&conn, body, &c);
+ free(body);
+ free(headers);
+ llm_sse_acc_free(&acc);
+
+ if (c.input_tokens)
+ last_in = c.input_tokens;
+
+ /* emit reasoning (chain of thought), if the model returned one and
+ * it did not already stream live */
+ if (!live && c.reasoning && c.reasoning[0]) {
+ sess_emit ev;
+ memset(&ev, 0, sizeof ev);
+ if (sess_assistant_emit("reasoning", c.reasoning, NULL, &ev) == 0) {
+ if (cb && cb(&ev, user)) {
+ sess_emit_free(&ev);
+ stopped = true;
+ llm_completion_free(&c);
+ break;
+ }
+ sess_emit_free(&ev);
+ }
+ }
+
+ /* emit text, unless it already streamed live */
+ if (!live && c.text && c.text[0]) {
+ sess_emit ev;
+ memset(&ev, 0, sizeof ev);
+ if (sess_assistant_emit("text", c.text, NULL, &ev) == 0) {
+ if (cb && cb(&ev, user)) {
+ sess_emit_free(&ev);
+ stopped = true;
+ llm_completion_free(&c);
+ break;
+ }
+ sess_emit_free(&ev);
+ }
+ }
+
+ if (c.tc_len == 0) {
+ /* final answer: take ownership of c.text and reasoning */
+ final_text = c.text ? c.text : xstrdup("");
+ c.text = NULL;
+ final_reasoning = c.reasoning ? c.reasoning : NULL;
+ c.reasoning = NULL;
+ llm_completion_free(&c);
+ break;
+ }
+
+ /* tool calls: build the assistant message + execute + tool results */
+ message amsg;
+ if (c.reasoning && c.reasoning[0])
+ msg_push_assistant_reasoning(amsg, c.reasoning);
+ if (c.text && c.text[0])
+ msg_push_assistant_text(amsg, c.text);
+ for (size_t t = 0; t < c.tc_len; t++)
+ msg_push_assistant_tool_call(amsg, c.tc_name[t], c.tc_args[t]);
+ local.push_back(std::move(amsg));
+
+ for (size_t t = 0; t < c.tc_len; t++) {
+ if (stop && stop->load()) {
+ stopped = true;
+ llm_completion_free(&c);
+ goto end_loop;
+ }
+ /* emit the tool call */
+ sess_emit ev;
+ memset(&ev, 0, sizeof ev);
+ if (sess_assistant_emit("tool_call", c.tc_name[t], c.tc_args[t],
+ &ev) == 0) {
+ if (cb && cb(&ev, user)) {
+ sess_emit_free(&ev);
+ stopped = true;
+ llm_completion_free(&c);
+ goto end_loop;
+ }
+ sess_emit_free(&ev);
+ }
+ /* execute */
+ char *tresult = NULL, *terr = NULL;
+ tresult = exec_tool(p, c.tc_name[t], c.tc_args[t],
+ cfg->shell_timeout_secs, &terr);
+ if (!tresult) {
+ tresult = terr ? terr : xstrdup("tool failed");
+ } else if (terr) {
+ free(terr); /* exec_tool: result and error are exclusive */
+ }
+ char *bounded = tool_truncate_output(tresult);
+ free(tresult);
+ tresult = bounded;
+ /* emit the tool result */
+ {
+ sess_emit ev2;
+ memset(&ev2, 0, sizeof ev2);
+ if (sess_tool_result_emit(c.tc_name[t], tresult, &ev2) == 0) {
+ if (cb && cb(&ev2, user)) {
+ sess_emit_free(&ev2);
+ stopped = true;
+ llm_completion_free(&c);
+ goto end_loop;
+ }
+ sess_emit_free(&ev2);
+ }
+ }
+ /* tool result message */
+ call_counter++;
+ char *id = c.tc_id[t] ? xstrdup(c.tc_id[t])
+ : xstrfmt("call-%d", (int)call_counter);
+ local.push_back(msg_user_tool_result(id, c.tc_name[t], tresult));
+ free(id);
+ free(tresult);
+ }
+ /* One tool round completed: count it for the rubric gates. */
+ if (p->sess)
+ session_note_tool_round(p->sess);
+ llm_completion_free(&c);
+ }
+end_loop:
+ if (input_tokens)
+ *input_tokens = last_in;
+
+ if (stopped && !local.empty())
+ provider_trim_dangling_calls(local.data(), local.size());
+
+ /* build new_messages = [user_text(prompt)] + local */
+ {
+ std::vector<message> nm;
+ nm.push_back(msg_user_text(prompt ? prompt : ""));
+ for (size_t i = 0; i < local.size(); i++)
+ nm.push_back(local[i]);
+ if ((final_text && final_text[0]) ||
+ (final_reasoning && final_reasoning[0])) {
+ message fm;
+ if (final_reasoning && final_reasoning[0])
+ msg_push_assistant_reasoning(fm, final_reasoning);
+ if (final_text && final_text[0])
+ msg_push_assistant_text(fm, final_text);
+ nm.push_back(std::move(fm));
+ }
+ if (new_messages) {
+ *new_messages = new message[nm.size()];
+ for (size_t i = 0; i < nm.size(); i++)
+ (*new_messages)[i] = std::move(nm[i]);
+ }
+ if (n_new)
+ *n_new = nm.size();
+ }
+ if (assistant_text)
+ *assistant_text = xstrdup(final_text ? final_text : "");
+
+ free(final_text);
+ free(final_reasoning);
+
+ if (stopped)
+ rc = SESS_STREAM_STOPPED;
+
+ return rc;
+}
+
+/* ------------------------------------------------------------------ */
+/* Lifecycle */
+/* ------------------------------------------------------------------ */
+
+static void provider_destroy(llsh_model *base) {
+ provider_model *p = PCAST(base);
+ free(p->tools_json);
+ if (p->curl)
+ curl_easy_cleanup(p->curl);
+ free(p);
+}
+
+llsh_model *provider_model_new(session *sess, mcp_registry *mcp,
+ provider_role role) {
+ provider_model *p = (provider_model *) xmalloc(sizeof *p);
+ memset(p, 0, sizeof *p);
+ p->sess = sess;
+ p->mcp = mcp;
+ p->role = role;
+ p->base.complete = provider_complete;
+ p->base.stream = provider_stream;
+ p->base.destroy = provider_destroy;
+ return &p->base;
+}
+
+void provider_model_attach(llsh_model *m, session *sess) {
+ PCAST(m)->sess = sess;
+}
+
+char *provider_preview_request(session *sess, const char *prompt,
+ bool with_tools) {
+ llsh_model *mm = provider_model_new(sess, NULL, PROV_ROLE_MAIN);
+ provider_model *p = PCAST(mm);
+ const config *cfg = session_active_config(sess);
+ const char *preamble = session_current_preamble(sess);
+ const strlist *allowed = session_allowed_tools(sess);
+
+ llm_part part;
+ memset(&part, 0, sizeof part);
+ part.kind = LLM_TEXT;
+ part.text = prompt ? prompt : "";
+ llm_message one;
+ one.role = LLM_ROLE_USER;
+ one.parts = &part;
+ one.part_count = 1;
+
+ llm_conn conn = conn_of(sess);
+ const char *tools = with_tools ? tools_fragment(p, cfg, allowed) : NULL;
+ char *req = llm_build_request(&conn, preamble, &one, 1, tools, false);
+ llsh_model_free(mm);
+ return req;
+}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-12 16:49:03 +0000