LLMConnection.cpp

Go to the documentation of this file.

#include "LLMConnection.h"
#include "NetworkManager.h"
#include "HTML.h"
#include "jsmn.h"
#include "StreamDecoders.h"
 
#include <cstdio>
#include <cstdlib>
#include <cstring>
 
namespace   cmlabs{
 
const char* LLMResultText(LLMResult result) {
    switch (result) {
        case LLM_OK:                         return "OK";
        case LLM_ERR_NOT_CONFIGURED:         return "Not configured";
        case LLM_ERR_NOT_CONNECTED:          return "Not connected";
        case LLM_ERR_NOT_IMPLEMENTED:        return "Not implemented";
        case LLM_ERR_CONFIG_EMPTY:           return "Config: empty <llm> element";
        case LLM_ERR_CONFIG_NOT_LLM:         return "Config: not an <llm> element";
        case LLM_ERR_CONFIG_PARSE:           return "Config: XML parse error";
        case LLM_ERR_CONFIG_NO_NAME:         return "Config: missing name attribute";
        case LLM_ERR_CONFIG_NO_VENDOR:       return "Config: missing <llmvendor>";
        case LLM_ERR_CONFIG_UNKNOWN_VENDOR:  return "Config: unknown vendor type";
        case LLM_ERR_CONFIG_NO_ENDPOINT:     return "Config: missing endpoint";
        case LLM_ERR_CONFIG_INLINE_SECRET:   return "Config: inline apikey not allowed, use apikeyref";
        case LLM_ERR_CONFIG_INVALID:         return "Config: invalid";
        case LLM_ERR_UNKNOWN_MODEL:          return "Unknown model";
        case LLM_ERR_SECRET_UNRESOLVED:      return "API key secret unresolved";
        case LLM_ERR_NO_INPUT:               return "Empty input";
        case LLM_ERR_HTTP_UNREACHABLE:       return "HTTP: endpoint unreachable";
        case LLM_ERR_HTTP:                   return "HTTP: non-200 reply";
        case LLM_ERR_REPLY_PARSE:            return "Reply parse error";
        case LLM_ERR_CONFIG_BAD_TRANSPORT:   return "Config: unknown stream transport";
        case LLM_ERR_STREAM_CANCELLED:       return "Stream cancelled by caller";
    }
    return "Unknown result";
}
 
const char* LLMVendorTypeText(LLMVendorType type) {
    switch (type) {
        case LLM_VENDOR_NONE:      return "none";
        case LLM_VENDOR_OPENAI:    return "openai";
        case LLM_VENDOR_ANTHROPIC: return "anthropic";
        case LLM_VENDOR_BEDROCK:   return "bedrock";
        case LLM_VENDOR_GOOGLE:    return "google";
        case LLM_VENDOR_CUSTOM:    return "custom";
    }
    return "none";
}
 
const char* LLMStreamTransportText(LLMStreamTransport transport) {
    switch (transport) {
        case LLM_STREAM_NONE:        return "none";
        case LLM_STREAM_SSE:         return "sse";
        case LLM_STREAM_EVENTSTREAM: return "eventstream";
        case LLM_STREAM_CHUNKED:     return "chunked";
    }
    return "none";
}
 
LLMConnection::LLMConnection() :
    configured(false), connected(false), lastError(LLM_OK),
    vendorType(LLM_VENDOR_NONE), streamTransport(LLM_STREAM_NONE),
    globalConnection(false), owner(NULL), ownerEndCallback(NULL),
    ownerNotified(false), cancelRequested(false), connectTimeMS(0),
    netManager(NULL) {
}
 
LLMConnection::~LLMConnection() {
    if (connected) {
        currentStats.uptimeMS = GetTimeNow() - connectTimeMS;
        currentStats.ended = true;
    }
    notifyOwnerEnded();
    delete netManager;
}
 
void LLMConnection::setOwner(void* newOwner, LLMOwnerEndCallback callback) {
    owner = newOwner;
    ownerEndCallback = callback;
    ownerNotified = false;
}
 
void LLMConnection::notifyOwnerEnded() {
    if (ownerEndCallback && !ownerNotified) {
        ownerNotified = true;
        ownerEndCallback(owner, this, currentStats);
    }
}
 
LLMResult LLMConnection::fail(LLMResult result) {
    configured = false;
    lastError = result;
    return result;
}
 
void LLMConnection::clearConfig() {
    configured = false;
    connected = false;
    lastError = LLM_OK;
    name.clear();
    vendorType = LLM_VENDOR_NONE;
    vendorTypeName.clear();
    endpoint.clear();
    streamTransport = LLM_STREAM_NONE;
    globalConnection = false;
    schemaVersion.clear();
    apiKeyRef.clear();
    secretsFile.clear();
    apiKeySecret.clear();
    requestTemplate.clear();
    responseTemplate.clear();
    headers.clear();
    constraints = LLMConstraints();
    models.clear();
    selectedModel.clear();
    currentStats = LLMStats();
    cancelRequested = false;
    connectTimeMS = 0;
    delete netManager;
    netManager = NULL;
}
 
static bool attrTrue(const XMLNode& node, const char* attr) {
    const char* v = node.getAttribute(attr);
    return (v && (!stricmp(v, "true") || !stricmp(v, "yes") || !stricmp(v, "1")));
}
 
bool LLMConnection::nodeHasInlineSecret(const XMLNode& node) {
    if (node.isAttributeSet("apikey")) return true;
    if (node.nChildNode("apikey") > 0) return true;
    return false;
}
 
void LLMConnection::resolveSecret() {
    apiKeySecret.clear();
    if (apiKeyRef.empty()) return;
    // 1.3a (Q-C): explicit indirect references.
    //   env:NAME    -> environment variable NAME (must be set and non-empty)
    //   file:/path  -> whole file contents, trailing whitespace/newlines trimmed
    //   NAME (bare) -> legacy: env var NAME, else key=value lookup in `secretsfile`
    // The secret value itself is never logged or included in any error text;
    // failure surfaces only as the typed LLM_ERR_SECRET_UNRESOLVED at connect().
    if (!strnicmp(apiKeyRef.c_str(), "env:", 4)) {
        const char* env = getenv(apiKeyRef.c_str() + 4);
        if (env && *env) apiKeySecret = env;
        return;
    }
    if (!strnicmp(apiKeyRef.c_str(), "file:", 5)) {
        FILE* f = fopen(apiKeyRef.c_str() + 5, "rb");
        if (!f) return;
        std::string content;
        char buf[4096];
        size_t n;
        while ((n = fread(buf, 1, sizeof(buf), f)) > 0) content.append(buf, n);
        fclose(f);
        while (!content.empty() && (content[content.size()-1] == '\n' ||
            content[content.size()-1] == '\r' || content[content.size()-1] == ' ' ||
            content[content.size()-1] == '\t'))
            content.erase(content.size()-1);
        apiKeySecret = content;
        return;
    }
    const char* env = getenv(apiKeyRef.c_str());
    if (env && *env) { apiKeySecret = env; return; }
    if (secretsFile.empty()) return;
    FILE* f = fopen(secretsFile.c_str(), "r");
    if (!f) return;
    char line[2048];
    while (fgets(line, sizeof(line), f)) {
        char* eq = strchr(line, '=');
        if (!eq) continue;
        *eq = 0;
        // trim trailing whitespace/newline from value
        char* val = eq + 1;
        size_t len = strlen(val);
        while (len && (val[len-1] == '\n' || val[len-1] == '\r' || val[len-1] == ' ' || val[len-1] == '\t'))
            val[--len] = 0;
        if (!stricmp(line, apiKeyRef.c_str()) && len) {
            apiKeySecret = val;
            break;
        }
    }
    fclose(f);
}
 
LLMResult LLMConnection::configureFromString(const char* xml) {
    if (!xml || !*xml) return fail(LLM_ERR_CONFIG_EMPTY);
    XMLResults results;
    XMLNode node = XMLNode::parseString(xml, "llm", &results);
    if (results.error != eXMLErrorNone || node.isEmpty())
        return fail(LLM_ERR_CONFIG_PARSE);
    return configure(node);
}
 
LLMResult LLMConnection::configure(const XMLNode& llmNode) {
    clearConfig();
    if (llmNode.isEmpty()) return fail(LLM_ERR_CONFIG_EMPTY);
    const char* nodeName = llmNode.getName();
    if (!nodeName || stricmp(nodeName, "llm")) return fail(LLM_ERR_CONFIG_NOT_LLM);
    if (nodeHasInlineSecret(llmNode)) return fail(LLM_ERR_CONFIG_INLINE_SECRET);
    const char* v = llmNode.getAttribute("name");
    if (!v || !*v) return fail(LLM_ERR_CONFIG_NO_NAME);
    name = v;
    globalConnection = attrTrue(llmNode, "global");
    if ((v = llmNode.getAttribute("schemaversion")) != NULL) schemaVersion = v;
    if ((v = llmNode.getAttribute("apikeyref")) != NULL) apiKeyRef = v;
    if ((v = llmNode.getAttribute("secretsfile")) != NULL) secretsFile = v;
    if ((v = llmNode.getAttribute("maxtokensperrequest")) != NULL) constraints.maxTokensPerRequest = atoi(v);
    if ((v = llmNode.getAttribute("maxrequestspermin")) != NULL) constraints.maxRequestsPerMin = atoi(v);
    if ((v = llmNode.getAttribute("maxconcurrent")) != NULL) constraints.maxConcurrent = atoi(v);
    if ((v = llmNode.getAttribute("costceiling")) != NULL) constraints.costCeiling = atof(v);
    if (llmNode.nChildNode("llmvendor") < 1) return fail(LLM_ERR_CONFIG_NO_VENDOR);
    LLMResult res = parseVendor(llmNode.getChildNode("llmvendor"));
    if (res != LLM_OK) return fail(res);
    resolveSecret();
    configured = true;
    lastError = LLM_OK;
    return LLM_OK;
}
 
LLMResult LLMConnection::parseVendor(const XMLNode& vendorNode) {
    if (vendorNode.isEmpty()) return LLM_ERR_CONFIG_NO_VENDOR;
    if (nodeHasInlineSecret(vendorNode)) return LLM_ERR_CONFIG_INLINE_SECRET;
    const char* v = vendorNode.getAttribute("type");
    if (!v || !*v) return LLM_ERR_CONFIG_UNKNOWN_VENDOR;
    vendorTypeName = v;
    if (!stricmp(v, "openai")) vendorType = LLM_VENDOR_OPENAI;
    else if (!stricmp(v, "anthropic")) vendorType = LLM_VENDOR_ANTHROPIC;
    else if (!stricmp(v, "bedrock")) vendorType = LLM_VENDOR_BEDROCK;
    else if (!stricmp(v, "google")) vendorType = LLM_VENDOR_GOOGLE;
    else if (!stricmp(v, "custom")) vendorType = LLM_VENDOR_CUSTOM;
    else return LLM_ERR_CONFIG_UNKNOWN_VENDOR;
    v = vendorNode.getAttribute("endpoint");
    if (!v || !*v) return LLM_ERR_CONFIG_NO_ENDPOINT;
    endpoint = v;
    LLMResult tres = resolveStreamTransport(vendorNode.getAttribute("transport"));
    if (tres != LLM_OK) return tres;
    int n = vendorNode.nChildNode("header");
    for (int i = 0; i < n; i++) {
        XMLNode h = vendorNode.getChildNode("header", i);
        LLMHeader header;
        const char* hn = h.getAttribute("name");
        const char* hv = h.getAttribute("value");
        if (!hn || !*hn) return LLM_ERR_CONFIG_INVALID;
        header.name = hn;
        if (hv) header.value = hv;
        headers.push_back(header);
    }
    XMLNode t = vendorNode.getChildNode("requesttemplate");
    if (!t.isEmpty() && t.getText()) requestTemplate = t.getText();
    t = vendorNode.getChildNode("responsetemplate");
    if (!t.isEmpty() && t.getText()) responseTemplate = t.getText();
    XMLNode modelsNode = vendorNode.getChildNode("models");
    if (!modelsNode.isEmpty()) {
        n = modelsNode.nChildNode("model");
        for (int i = 0; i < n; i++) {
            const char* mn = modelsNode.getChildNode("model", i).getAttribute("name");
            if (!mn || !*mn) return LLM_ERR_CONFIG_INVALID;
            models.push_back(mn);
        }
        const char* def = modelsNode.getAttribute("default");
        if (def && *def) {
            bool found = false;
            for (size_t i = 0; i < models.size(); i++)
                if (!stricmp(models[i].c_str(), def)) { selectedModel = models[i]; found = true; break; }
            if (!found) return LLM_ERR_UNKNOWN_MODEL;
        }
        else if (!models.empty()) selectedModel = models[0];
    }
    return LLM_OK;
}
 
std::vector<std::string> LLMConnection::listModels() const {
    return models;
}
 
LLMResult LLMConnection::selectModel(const char* modelName) {
    if (!configured) { lastError = LLM_ERR_NOT_CONFIGURED; return LLM_ERR_NOT_CONFIGURED; }
    if (modelName && *modelName) {
        for (size_t i = 0; i < models.size(); i++) {
            if (!stricmp(models[i].c_str(), modelName)) {
                selectedModel = models[i];
                lastError = LLM_OK;
                return LLM_OK;
            }
        }
    }
    lastError = LLM_ERR_UNKNOWN_MODEL;
    return LLM_ERR_UNKNOWN_MODEL;
}
 
// Escape a plain string for embedding inside a JSON string literal.
static std::string llmJSONEscape(const char* s) {
    std::string out;
    if (!s) return out;
    for (const char* p = s; *p; p++) {
        unsigned char c = (unsigned char)*p;
        switch (c) {
            case '"':  out += "\\\""; break;
            case '\\': out += "\\\\"; break;
            case '\b': out += "\\b"; break;
            case '\f': out += "\\f"; break;
            case '\n': out += "\\n"; break;
            case '\r': out += "\\r"; break;
            case '\t': out += "\\t"; break;
            default:
                if (c < 0x20) {
                    char buf[8];
                    snprintf(buf, sizeof(buf), "\\u%04x", c);
                    out += buf;
                }
                else out += (char)c;
        }
    }
    return out;
}
 
static void llmReplaceAll(std::string& text, const char* what, const std::string& with) {
    size_t wlen = strlen(what), pos = 0;
    while ((pos = text.find(what, pos)) != std::string::npos) {
        text.replace(pos, wlen, with);
        pos += with.length();
    }
}
 
// Substitute %apikey%, %model% and %input% (input JSON-escaped) in `text`.
std::string LLMConnection::applyTemplates(const std::string& text, const char* input) const {
    std::string out = text;
    llmReplaceAll(out, "%apikey%", apiKeySecret);
    llmReplaceAll(out, "%model%", selectedModel);
    if (input) llmReplaceAll(out, "%input%", llmJSONEscape(input));
    return out;
}
 
// Return the index just past token i's whole subtree (DFS order).
static int llmSkipToken(const jsmntok_t* t, int i, int count) {
    int end = t[i].end;
    i++;
    while (i < count && t[i].start >= 0 && t[i].start < end) i++;
    return i;
}
 
// Walk a dotted/indexed path (e.g. choices[0].message.content) through the
// jsmn token stream; `out` gets the raw token text (strings unescaped by jsmn
// bounds only — quotes excluded). Returns false when the path does not match.
bool LLMConnection::extractJSONPath(const char* json, uint32 size, const char* path,
    std::string& out) {
    if (!json || !size || !path || !*path) return false;
    jsmn_parser parser;
    jsmn_init(&parser);
    int count = jsmn_parse(&parser, json, size, NULL, 0);
    if (count <= 0) return false;
    std::vector<jsmntok_t> tokens((size_t)count);
    jsmn_init(&parser);
    if (jsmn_parse(&parser, json, size, &tokens[0], (unsigned int)count) != count)
        return false;
    jsmntok_t* t = &tokens[0];
    int cur = 0;
    const char* p = path;
    while (*p) {
        if (*p == '.') { p++; continue; }
        if (*p == '[') { // array index segment
            int idx = (int)strtol(p + 1, (char**)&p, 10);
            if (*p != ']') return false;
            p++;
            if (t[cur].type != JSMN_ARRAY || idx < 0 || idx >= t[cur].size) return false;
            int child = cur + 1;
            for (int n = 0; n < idx; n++) child = llmSkipToken(t, child, count);
            cur = child;
            continue;
        }
        const char* segStart = p; // object key segment
        while (*p && *p != '.' && *p != '[') p++;
        std::string key(segStart, (size_t)(p - segStart));
        if (t[cur].type != JSMN_OBJECT) return false;
        int child = cur + 1;
        bool found = false;
        for (int n = 0; n < t[cur].size && child < count; n++) {
            int keyLen = t[child].end - t[child].start;
            if (t[child].type == JSMN_STRING && (int)key.length() == keyLen &&
                strncmp(json + t[child].start, key.c_str(), (size_t)keyLen) == 0) {
                cur = child + 1;
                found = true;
                break;
            }
            child = llmSkipToken(t, child + 1, count); // skip key's value subtree
        }
        if (!found) return false;
    }
    out.assign(json + t[cur].start, (size_t)(t[cur].end - t[cur].start));
    return true;
}
 
LLMResult LLMConnection::connect() {
    if (!configured) return LLM_ERR_NOT_CONFIGURED;
    if (!apiKeyRef.empty() && apiKeySecret.empty()) {
        lastError = LLM_ERR_SECRET_UNRESOLVED;
        return LLM_ERR_SECRET_UNRESOLVED;
    }
    if (!netManager) netManager = new NetworkManager();
    connected = true;
    connectTimeMS = GetTimeNow();
    currentStats = LLMStats(); // fresh stats record per connect()
    ownerNotified = false;     // 1.2d-i: re-arm owner-end notification
    lastError = LLM_OK;
    return LLM_OK;
}
 
LLMResult LLMConnection::reset() {
    if (!configured) return LLM_ERR_NOT_CONFIGURED;
    cancelRequested = true; // aborts any in-flight interactStream() promptly
    if (connected) {
        currentStats.uptimeMS = GetTimeNow() - connectTimeMS;
        currentStats.ended = true; // stats record retained, marked ended
    }
    connected = false;
    connectTimeMS = 0;
    delete netManager;
    netManager = NULL;
    lastError = LLM_OK;
    notifyOwnerEnded(); // 1.2d-i: owning Node retains the ended stats snapshot
    return LLM_OK;
}
 
// Unescape common JSON string escapes in extracted reply text.
static std::string llmJSONUnescape(const std::string& in) {
    std::string out;
    out.reserve(in.length());
    for (size_t i = 0; i < in.length(); i++) {
        if (in[i] == '\\' && i + 1 < in.length()) {
            char c = in[++i];
            switch (c) {
                case 'n': out += '\n'; break;
                case 't': out += '\t'; break;
                case 'r': out += '\r'; break;
                case 'b': out += '\b'; break;
                case 'f': out += '\f'; break;
                case 'u':
                    if (i + 4 < in.length()) { // basic BMP codepoint -> UTF-8
                        unsigned int cp = (unsigned int)strtoul(in.substr(i + 1, 4).c_str(), NULL, 16);
                        i += 4;
                        if (cp < 0x80) out += (char)cp;
                        else if (cp < 0x800) {
                            out += (char)(0xC0 | (cp >> 6));
                            out += (char)(0x80 | (cp & 0x3F));
                        }
                        else {
                            out += (char)(0xE0 | (cp >> 12));
                            out += (char)(0x80 | ((cp >> 6) & 0x3F));
                            out += (char)(0x80 | (cp & 0x3F));
                        }
                    }
                    break;
                default: out += c; break; // covers \" \\ \/
            }
        }
        else out += in[i];
    }
    return out;
}
 
// Fold vendor usage token counts (OpenAI/Anthropic/Bedrock field names)
// from a reply or stream-event JSON body into currentStats. Events without
// usage fields are a silent no-op.
void LLMConnection::accountUsageJSON(const char* json, uint32 size) {
    if (!json || !size) return;
    std::string tok;
    if (extractJSONPath(json, size, "usage.prompt_tokens", tok) ||
        extractJSONPath(json, size, "usage.input_tokens", tok) ||
        extractJSONPath(json, size, "usage.inputTokens", tok))
        currentStats.tokensIn += (uint64)strtoull(tok.c_str(), NULL, 10);
    if (extractJSONPath(json, size, "usage.completion_tokens", tok) ||
        extractJSONPath(json, size, "usage.output_tokens", tok) ||
        extractJSONPath(json, size, "usage.outputTokens", tok))
        currentStats.tokensOut += (uint64)strtoull(tok.c_str(), NULL, 10);
}
 
// Record a runtime (non-configuration) error without unconfiguring.
LLMResult LLMConnection::interactFail(LLMResult result) {
    lastError = result;
    return result;
}
 
// Resolve the streaming wire format: explicit <llmvendor transport="...">
// wins; otherwise default by vendor (bedrock -> eventstream, else sse).
LLMResult LLMConnection::resolveStreamTransport(const char* transportAttr) {
    if (transportAttr && *transportAttr) {
        if (!stricmp(transportAttr, "sse")) streamTransport = LLM_STREAM_SSE;
        else if (!stricmp(transportAttr, "eventstream")) streamTransport = LLM_STREAM_EVENTSTREAM;
        else if (!stricmp(transportAttr, "chunked")) streamTransport = LLM_STREAM_CHUNKED;
        else return LLM_ERR_CONFIG_BAD_TRANSPORT;
        return LLM_OK;
    }
    streamTransport = (vendorType == LLM_VENDOR_BEDROCK)
        ? LLM_STREAM_EVENTSTREAM : LLM_STREAM_SSE;
    return LLM_OK;
}
 
// Byte source that yields a prefix buffer (bytes read past the HTTP header)
// before delegating to the live connection source, and turns cancellation
// (callback returned false) into an immediate SRC_ERROR so the bounded pump
// loops exit promptly.
namespace {
class LLMPrefixByteSource : public cmsdk::StreamByteSource {
public:
    LLMPrefixByteSource(const std::string& prefixBytes,
        cmsdk::StreamByteSource& nextSource, const bool& cancelledFlag,
        const volatile bool& resetFlag, uint64& bodyByteCounter)
        : prefix(prefixBytes), pos(0), next(nextSource), cancelled(cancelledFlag),
          resetRequested(resetFlag), bodyBytes(bodyByteCounter) {}
    cmsdk::StreamReadResult read(char* buf, size_t maxSize, size_t& got) {
        if (cancelled || resetRequested) return cmsdk::SRC_ERROR;
        if (pos < prefix.size()) {
            got = prefix.size() - pos;
            if (got > maxSize) got = maxSize;
            memcpy(buf, prefix.data() + pos, got);
            pos += got;
            return cmsdk::SRC_DATA;
        }
        cmsdk::StreamReadResult r = next.read(buf, maxSize, got);
        if (r == cmsdk::SRC_DATA)
            bodyBytes += (uint64)got; // streamed body bytes past the header block
        return r;
    }
private:
    std::string prefix;
    size_t pos;
    cmsdk::StreamByteSource& next;
    const bool& cancelled;
    const volatile bool& resetRequested;
    uint64& bodyBytes;
};
} // anonymous namespace
 
// Reads the HTTP reply status line + headers off the live connection, then
// pumps the remaining body bytes through the decoder selected at configure()
// time, delivering incremental token text to the callback. Per-call state
// only; the connection is owned (and closed) by interactStream().
LLMResult LLMConnection::pumpStreamReply(void* conPtr,
    LLMStreamCallback callback, void* userData, std::string* fullReply) {
    NetworkConnection* con = (NetworkConnection*)conPtr;
    cmsdk::ConnectionByteSource<NetworkConnection> raw(*con, 250);
 
    // 1. Accumulate the HTTP header block (status line + headers + CRLFCRLF).
    std::string head;
    size_t headerEnd = std::string::npos;
    char buf[4096];
    for (size_t iter = 0; iter < 1000 && headerEnd == std::string::npos; iter++) {
        if (cancelRequested) return LLM_ERR_STREAM_CANCELLED;
        size_t got = 0;
        cmsdk::StreamReadResult r = raw.read(buf, sizeof(buf), got);
        if (r == cmsdk::SRC_ERROR || r == cmsdk::SRC_CLOSED)
            return LLM_ERR_HTTP_UNREACHABLE;
        if (r == cmsdk::SRC_DATA) {
            head.append(buf, got);
            if (head.size() > 65536) return LLM_ERR_REPLY_PARSE;
            headerEnd = head.find("\r\n\r\n");
        }
    }
    if (headerEnd == std::string::npos) return LLM_ERR_HTTP_UNREACHABLE;
    currentStats.bytesReceived += (uint64)head.size();
 
    // 2. Status code from "HTTP/1.x NNN ...".
    size_t sp = head.find(' ');
    int status = (sp != std::string::npos) ? atoi(head.c_str() + sp + 1) : 0;
    if (status != 200) return LLM_ERR_HTTP;
 
    // 3. Body bytes already read past the header become the pump prefix.
    std::string leftover = head.substr(headerEnd + 4);
    bool cancelled = false;
    LLMPrefixByteSource src(leftover, raw, cancelled, cancelRequested,
        currentStats.bytesReceived);
 
    // Deliver one decoded token: append to fullReply, hand to the callback;
    // callback == false flips `cancelled`, which stops the byte source.
    auto deliver = [&](const char* text, size_t size) {
        if (cancelled || cancelRequested || !size) return;
        currentStats.replyChunks++;
        if (fullReply) fullReply->append(text, size);
        if (!callback(text, (uint32)size, userData)) cancelled = true;
    };
    // Token text from one SSE event / eventstream frame payload: extract via
    // the configured responsetemplate JSON path (mode-A convention); events
    // that do not match the path (metadata, usage, stops) are skipped —
    // except usage token counts, folded into stats exactly like mode A.
    auto deliverJSON = [&](const std::string& json) {
        accountUsageJSON(json.data(), (uint32)json.size());
        std::string extracted;
        if (extractJSONPath(json.data(), (uint32)json.size(),
                responseTemplate.c_str(), extracted)) {
            std::string text = llmJSONUnescape(extracted);
            deliver(text.data(), text.size());
        }
    };
 
    // 4. Drive the transport decoder (bounded pumps; local decoder state).
    cmsdk::StreamPumpResult pr;
    switch (streamTransport) {
        case LLM_STREAM_SSE: {
            cmsdk::SSEEventParser parser;
            pr = cmsdk::pumpSSEStream(src, parser,
                [&](const std::string& ev) { deliverJSON(ev); });
            break;
        }
        case LLM_STREAM_EVENTSTREAM: {
            cmsdk::EventStreamFrameParser parser;
            pr = cmsdk::pumpEventStream(src, parser,
                [&](const cmsdk::EventStreamFrameParser::Frame& f) {
                    deliverJSON(f.payload);
                });
            break;
        }
        case LLM_STREAM_CHUNKED: {
            cmsdk::HTTPChunkedDecoder dec;
            pr = cmsdk::pumpChunkedStream(src, dec,
                [&](const char* data, size_t size) { deliver(data, size); });
            break;
        }
        default:
            return LLM_ERR_CONFIG_BAD_TRANSPORT;
    }
 
    // 5. Map pump outcome onto LLMResult (cancellation wins).
    if (cancelled || cancelRequested) return LLM_ERR_STREAM_CANCELLED;
    if (pr == cmsdk::PUMP_COMPLETE) return LLM_OK;
    if (pr == cmsdk::PUMP_CLOSED)  // no [DONE]/0-chunk: normal for plain SSE
        return (streamTransport == LLM_STREAM_SSE) ? LLM_OK : LLM_ERR_REPLY_PARSE;
    return LLM_ERR_REPLY_PARSE;  // PUMP_ERROR / PUMP_LIMIT
}
 
// Mode-B streaming interact (T1.1 step 1.2c-ii): open a dedicated outbound
// connection to the configured endpoint, send the mode-A-shaped request, and
// drive the transport decoder selected at configure() time, delivering
// incremental token text to the callback. All per-call state (connection,
// decoders, buffers) is local; nothing streams through NetworkManager (its
// HTTP path is blocking whole-reply).
LLMResult LLMConnection::interactStream(const char* input,
    LLMStreamCallback callback, void* userData, std::string* fullReply) {
    if (fullReply) fullReply->clear();
    if (!configured) return interactFail(LLM_ERR_NOT_CONFIGURED);
    if (!connected) return interactFail(LLM_ERR_NOT_CONNECTED);
    if (!input || !*input) return interactFail(LLM_ERR_NO_INPUT);
    if (!callback) return interactFail(LLM_ERR_CONFIG_INVALID);
    if (streamTransport == LLM_STREAM_NONE)
        return interactFail(LLM_ERR_CONFIG_BAD_TRANSPORT);
    cancelRequested = false; // fresh stream; a reset() from here on cancels it
 
    // URL split + encryption choice, matching NetworkManager::makeHTTPRequest.
    std::string protocol = html::GetProtocolFromURL(endpoint);
    std::string host = html::GetHostFromURL(endpoint);
    uint16 port = html::GetPortFromURL(endpoint);
    bool useSSL = (stricmp(protocol.c_str(), "https") == 0);
    if (!useSSL && stricmp(protocol.c_str(), "http"))
        return interactFail(LLM_ERR_HTTP_UNREACHABLE);
    if (!host.length()) return interactFail(LLM_ERR_HTTP_UNREACHABLE);
    if (!port) port = useSSL ? 443 : 80;
    std::string uri = html::GetURIFromURL(endpoint);
    if (!uri.length()) uri = "/";
 
    // Shape request body and headers exactly like mode-A interact().
    std::string body = applyTemplates(requestTemplate, input);
    std::map<std::string, std::string> headerEntries;
    for (size_t i = 0; i < headers.size(); i++)
        headerEntries[headers[i].name] = applyTemplates(headers[i].value, NULL);
 
    currentStats.requestCount++;
    currentStats.bytesSent += (uint64)body.length();
 
    // Dedicated per-call socket (plain or SSL), not owned by netManager.
    SSLConnection sslCon;
    TCPConnection tcpCon;
    NetworkConnection* con = useSSL
        ? (NetworkConnection*)&sslCon : (NetworkConnection*)&tcpCon;
    uint64 location = 0;
    bool ok = useSSL
        ? sslCon.connect(host.c_str(), port, location, 30000)
        : tcpCon.connect(host.c_str(), port, location, 30000);
    if (!ok) return interactFail(LLM_ERR_HTTP_UNREACHABLE);
 
    HTTPRequest req;
    req.createRequest(HTTP_POST, host.c_str(), uri.c_str(), headerEntries,
        body.c_str(), "application/json", (uint32)body.length(), false, 0);
    if (!HTTPProtocol::SendHTTPRequest(con, &req)) {
        con->disconnect();
        return interactFail(LLM_ERR_HTTP_UNREACHABLE);
    }
 
    LLMResult res = pumpStreamReply(con, callback, userData, fullReply);
    con->disconnect();
    if (res == LLM_OK) lastError = LLM_OK;
    return (res == LLM_OK) ? LLM_OK : interactFail(res);
}
 
LLMResult LLMConnection::interact(const char* input, std::string& reply) {
    reply.clear();
    if (!configured) return interactFail(LLM_ERR_NOT_CONFIGURED);
    if (!connected || !netManager) return interactFail(LLM_ERR_NOT_CONNECTED);
    if (!input || !*input) return interactFail(LLM_ERR_NO_INPUT);
 
    // Shape request body and headers from the configured templates.
    std::string body = applyTemplates(requestTemplate, input);
    std::map<std::string, std::string> headerEntries;
    for (size_t i = 0; i < headers.size(); i++)
        headerEntries[headers[i].name] = applyTemplates(headers[i].value, NULL);
 
    currentStats.requestCount++;
    currentStats.bytesSent += (uint64)body.length();
 
    HTTPReply* httpReply = netManager->makeHTTPRequest(HTTP_POST, endpoint, 30000,
        headerEntries, body.c_str(), "application/json", (uint32)body.length());
    if (!httpReply) return interactFail(LLM_ERR_HTTP_UNREACHABLE);
 
    uint32 contentSize = 0;
    const char* content = httpReply->getContent(contentSize);
    currentStats.bytesReceived += contentSize;
    uint8 status = httpReply->type;
 
    if (status == HTTP_SERVER_UNAVAILABLE || status == HTTP_MALFORMED_URL) {
        delete httpReply;
        return interactFail(LLM_ERR_HTTP_UNREACHABLE);
    }
    if (status != HTTP_OK) {
#ifdef PSY_LLM_DEBUG_HTTP
        printf("LLM DEBUG non-200: status=%u content(%u)='%.300s'\n",
            (unsigned)status, contentSize, content ? content : "");
#endif
        delete httpReply;
        return interactFail(LLM_ERR_HTTP);
    }
    if (!content || !contentSize) {
        delete httpReply;
        return interactFail(LLM_ERR_REPLY_PARSE);
    }
 
    // Token accounting from vendor usage fields, when present.
    accountUsageJSON(content, contentSize);
 
    std::string extracted;
    if (!extractJSONPath(content, contentSize, responseTemplate.c_str(), extracted)) {
        delete httpReply;
        return interactFail(LLM_ERR_REPLY_PARSE);
    }
    delete httpReply;
    reply = llmJSONUnescape(extracted);
    lastError = LLM_OK;
    return LLM_OK;
}
 
LLMStats LLMConnection::stats() const {
    LLMStats s = currentStats;
    if (connected && connectTimeMS)
        s.uptimeMS = GetTimeNow() - connectTimeMS; // live uptime while connected
    return s;
}
 
} // namespace cmlabs