compactor_registry.cpp

Go to the documentation of this file.

// SPDX-License-Identifier: Apache-2.0
#include <entropic/core/compactor_registry.h>
#include <entropic/types/logging.h>
 
#include <sstream>
 
static auto logger = entropic::log::get("core.compactor_registry");
 
namespace entropic {
 
// ── JSON helpers (internal to this TU) ──────────────────
 
static const char* json_escape_char(char c) {
    const char* esc = nullptr;
    switch (c) {
    case '"':  esc = "\\\""; break;
    case '\\': esc = "\\\\"; break;
    case '\n': esc = "\\n";  break;
    case '\r': esc = "\\r";  break;
    case '\t': esc = "\\t";  break;
    default: break;
    }
    return esc;
}
 
static std::string json_escape(const std::string& input) {
    std::ostringstream oss;
    for (char c : input) {
        const char* esc = json_escape_char(c);
        if (esc != nullptr) { oss << esc; }
        else { oss << c; }
    }
    return oss.str();
}
 
static std::string serialize_messages(
    const std::vector<Message>& messages) {
    std::ostringstream oss;
    oss << '[';
    for (size_t i = 0; i < messages.size(); ++i) {
        if (i > 0) oss << ',';
        oss << "{\"role\":\"" << messages[i].role
            << "\",\"content\":\""
            << json_escape(messages[i].content) << "\"}";
    }
    oss << ']';
    return oss.str();
}
 
static std::string serialize_config(
    const CompactionConfig& config,
    const std::string& identity,
    int token_count) {
    std::ostringstream oss;
    oss << "{\"identity\":\"" << json_escape(identity) << "\""
        << ",\"token_count\":" << token_count
        << ",\"max_tokens\":0"
        << ",\"threshold_percent\":" << config.threshold_percent
        << ",\"force\":true}";
    return oss.str();
}
 
// ── Minimal JSON parser for message arrays ──────────────
 
struct JsonCursor {
    const std::string& data; 
    size_t pos = 0;          
};
 
static void json_skip_ws(JsonCursor& c) {
    while (c.pos < c.data.size() && isspace(c.data[c.pos])) {
        ++c.pos;
    }
}
 
static char json_unescape_char(char ch) {
    char result = ch;
    switch (ch) {
    case '"':  result = '"';  break;
    case '\\': result = '\\'; break;
    case 'n':  result = '\n'; break;
    case 'r':  result = '\r'; break;
    case 't':  result = '\t'; break;
    default: break;
    }
    return result;
}
 
static bool json_read_string(JsonCursor& c, std::string& out) {
    json_skip_ws(c);
    if (c.pos >= c.data.size() || c.data[c.pos] != '"') {
        return false;
    }
    ++c.pos;
    out.clear();
    while (c.pos < c.data.size() && c.data[c.pos] != '"') {
        if (c.data[c.pos] == '\\' && c.pos + 1 < c.data.size()) {
            out += json_unescape_char(c.data[c.pos + 1]);
            c.pos += 2;
        } else {
            out += c.data[c.pos++];
        }
    }
    if (c.pos >= c.data.size()) return false;
    ++c.pos; // skip closing '"'
    return true;
}
 
static void json_skip_composite(JsonCursor& c) {
    char open = c.data[c.pos];
    char close = (open == '[') ? ']' : '}';
    int depth = 1;
    ++c.pos;
    while (c.pos < c.data.size() && depth > 0) {
        char ch = c.data[c.pos];
        if (ch == '"') {
            std::string dummy;
            json_read_string(c, dummy);
            continue;
        }
        if (ch == open) ++depth;
        else if (ch == close) --depth;
        ++c.pos;
    }
}
 
static void json_skip_value(JsonCursor& c) {
    json_skip_ws(c);
    if (c.pos >= c.data.size()) return;
    char ch = c.data[c.pos];
    if (ch == '"') {
        std::string dummy;
        json_read_string(c, dummy);
        return;
    }
    if (ch == '[' || ch == '{') {
        json_skip_composite(c);
        return;
    }
    while (c.pos < c.data.size()
           && c.data[c.pos] != ',' && c.data[c.pos] != '}'
           && c.data[c.pos] != ']') {
        ++c.pos;
    }
}
 
static bool json_read_content_part_field(JsonCursor& c,
                                         std::string& type_val,
                                         std::string& text_val) {
    std::string key;
    if (!json_read_string(c, key)) return false;
    json_skip_ws(c);
    if (c.pos >= c.data.size() || c.data[c.pos] != ':') return false;
    ++c.pos;
    if (key == "type") { json_read_string(c, type_val); }
    else if (key == "text") { json_read_string(c, text_val); }
    else { json_skip_value(c); }
    return true;
}
 
static bool json_parse_content_part(JsonCursor& c,
                                    std::string& type_val,
                                    std::string& text_val) {
    while (c.pos < c.data.size() && c.data[c.pos] != '}') {
        json_skip_ws(c);
        if (c.data[c.pos] == ',') { ++c.pos; continue; }
        if (!json_read_content_part_field(c, type_val, text_val)) {
            return false;
        }
    }
    if (c.pos < c.data.size()) ++c.pos; // skip '}'
    return true;
}
 
static void append_text_part(const std::string& type_val,
                             const std::string& text_val,
                             std::string& text) {
    if (type_val != "text" || text_val.empty()) return;
    if (!text.empty()) text += ' ';
    text += text_val;
}
 
static char json_skip_to_element(JsonCursor& c) {
    while (c.pos < c.data.size()) {
        json_skip_ws(c);
        char ch = c.data[c.pos];
        if (ch != ',') return ch;
        ++c.pos;
    }
    return '\0';
}
 
static bool json_read_one_content_part(JsonCursor& c,
                                       std::string& text) {
    ++c.pos; // skip '{'
    std::string type_val;
    std::string text_val;
    bool ok = json_parse_content_part(c, type_val, text_val);
    if (ok) append_text_part(type_val, text_val, text);
    return ok;
}
 
static bool json_extract_text_from_array(JsonCursor& c,
                                         std::string& text) {
    if (c.pos >= c.data.size() || c.data[c.pos] != '[') {
        return false;
    }
    ++c.pos; // skip '['
    text.clear();
 
    bool ok = true;
    char ch = json_skip_to_element(c);
    while (ok && ch == '{') {
        ok = json_read_one_content_part(c, text);
        ch = ok ? json_skip_to_element(c) : '\0';
    }
    if (ok && ch == ']') { ++c.pos; }
    return ok && ch == ']';
}
 
static bool json_read_field(JsonCursor& c, Message& msg) {
    std::string key;
    bool ok = json_read_string(c, key);
    json_skip_ws(c);
    if (ok && c.pos < c.data.size() && c.data[c.pos] == ':') {
        ++c.pos;
    } else {
        ok = false;
    }
 
    if (!ok) return false;
 
    if (key == "role") {
        ok = json_read_string(c, msg.role);
    } else if (key == "content") {
        json_skip_ws(c);
        if (c.pos < c.data.size() && c.data[c.pos] == '[') {
            ok = json_extract_text_from_array(c, msg.content);
        } else {
            ok = json_read_string(c, msg.content);
        }
    } else {
        json_skip_value(c);
    }
    return ok;
}
 
static bool json_parse_message(JsonCursor& c, Message& msg) {
    while (c.pos < c.data.size()) {
        json_skip_ws(c);
        if (c.data[c.pos] == '}') { ++c.pos; return true; }
        if (c.data[c.pos] == ',') { ++c.pos; continue; }
        if (!json_read_field(c, msg)) return false;
    }
    return false;
}
 
static bool json_parse_array(JsonCursor& c,
                             std::vector<Message>& messages) {
    bool ok = true;
    while (ok && c.pos < c.data.size()) {
        json_skip_ws(c);
        if (c.data[c.pos] == ']') break;
        if (c.data[c.pos] == ',') { ++c.pos; continue; }
        ok = (c.data[c.pos] == '{');
        if (!ok) break;
        ++c.pos;
 
        Message msg;
        ok = json_parse_message(c, msg);
        if (ok) messages.push_back(std::move(msg));
    }
    return ok;
}
 
static bool parse_messages(const char* json,
                           std::vector<Message>& messages) {
    if (json == nullptr || json[0] != '[') return false;
    messages.clear();
    std::string input(json);
    JsonCursor c{input, 1};
    return json_parse_array(c, messages);
}
 
// ── CompactorRegistry ───────────────────────────────────
 
CompactorRegistry::CompactorRegistry(
    CompactionManager& default_manager)
    : default_manager_(default_manager) {
    default_compactor_ = [&mgr = default_manager_](
        const std::vector<Message>& messages,
        const CompactionConfig& /*config*/,
        const std::string& /*identity*/) {
        return mgr.compact_messages(messages);
    };
}
 
entropic_error_t CompactorRegistry::register_compactor(
    const std::string& identity,
    entropic_compactor_fn compactor,
    void* user_data) {
    if (compactor == nullptr) {
        logger->error("NULL compactor for identity '{}'", identity);
        return ENTROPIC_ERROR_INVALID_CONFIG;
    }
 
    CompactorEntry entry;
    entry.c_callback = compactor;
    entry.user_data = user_data;
    entry.cpp_fn = wrap_c_compactor(compactor, user_data);
 
    std::unique_lock lock(mutex_);
    compactors_[identity] = std::move(entry);
 
    logger->info("Registered compactor for identity '{}'",
                 identity.empty() ? "(global)" : identity);
    return ENTROPIC_OK;
}
 
entropic_error_t CompactorRegistry::deregister_compactor(
    const std::string& identity) {
    std::unique_lock lock(mutex_);
    compactors_.erase(identity);
 
    logger->info("Deregistered compactor for identity '{}'",
                 identity.empty() ? "(global)" : identity);
    return ENTROPIC_OK;
}
 
CompactionResult CompactorRegistry::compact(
    const std::string& identity,
    const std::vector<Message>& messages,
    const CompactionConfig& config) {
    CompactorFn selected;
    std::string source;
    bool is_custom = false;
 
    {   // Snapshot under read lock, release before calling
        std::shared_lock lock(mutex_);
        auto it = compactors_.find(identity);
        if (it != compactors_.end()) {
            selected = it->second.cpp_fn;
            source = identity;
            is_custom = true;
        } else {
            it = compactors_.find("");
            if (it != compactors_.end()) {
                selected = it->second.cpp_fn;
                source = "global_custom";
                is_custom = true;
            }
        }
    }
 
    if (!is_custom) {
        return run_default(identity, messages, config);
    }
    return run_custom(
        selected, source, identity, messages, config);
}
 
CompactionResult CompactorRegistry::run_default(
    const std::string& identity,
    const std::vector<Message>& messages,
    const CompactionConfig& config) {
    auto result = default_compactor_(messages, config, identity);
    result.identity = identity;
    result.compactor_source = "default";
    result.custom_compactor_used = false;
    return result;
}
 
CompactionResult CompactorRegistry::run_custom(
    const CompactorFn& selected,
    const std::string& source,
    const std::string& identity,
    const std::vector<Message>& messages,
    const CompactionConfig& config) {
    auto result = selected(messages, config, identity);
 
    if (result.compacted) {
        result.identity = identity;
        result.compactor_source = source;
        result.custom_compactor_used = true;
        return result;
    }
 
    logger->warn("Custom compactor '{}' failed for '{}', "
                 "falling back to default", source, identity);
    return run_default(identity, messages, config);
}
 
bool CompactorRegistry::has_custom_compactor(
    const std::string& identity) const {
    std::shared_lock lock(mutex_);
    if (compactors_.count(identity) > 0) {
        return true;
    }
    return compactors_.count("") > 0;
}
 
CompactorFn CompactorRegistry::wrap_c_compactor(
    entropic_compactor_fn compactor,
    void* user_data) {
    return [compactor, user_data](
        const std::vector<Message>& messages,
        const CompactionConfig& config,
        const std::string& identity) -> CompactionResult {
        auto msg_json = serialize_messages(messages);
        auto cfg_json = serialize_config(config, identity, 0);
 
        char* out_messages = nullptr;
        char* out_summary = nullptr;
        int rc = compactor(
            msg_json.c_str(), cfg_json.c_str(),
            &out_messages, &out_summary, user_data);
 
        CompactionResult result;
        if (rc != 0) {
            result.compacted = false;
            free(out_messages);
            free(out_summary);
            return result;
        }
 
        result.compacted = true;
        if (out_summary != nullptr) {
            result.summary = out_summary;
            free(out_summary);
        }
        if (out_messages != nullptr) {
            parse_messages(out_messages, result.messages);
            result.preserved_messages =
                static_cast<int>(result.messages.size());
            free(out_messages);
        }
        return result;
    };
}
 
} // namespace entropic