response_generator.cpp

Go to the documentation of this file.

// SPDX-License-Identifier: Apache-2.0
#include <entropic/core/response_generator.h>
 
#include "prompt_log_util.h"  // gh#152: log the transcript, not the re-emission
#include <entropic/mcp/utf8_sanitize.h>
#include <entropic/types/error.h>
#include <entropic/types/logging.h>
#include <entropic/types/run_scope.h>
 
#include <nlohmann/json.hpp>
 
#include <atomic>
#include <chrono>
#include <cstring>
#include <functional>
#include <thread>
#include <unordered_map>
 
static auto logger = entropic::log::get("core.response_generator");
 
namespace entropic {
 
static bool generation_cancelled(const GenerationEvents& events) {
    return (events.interrupt != nullptr
            && events.interrupt->load(std::memory_order_acquire))
        || current_run_cancelled();
}
 
static std::unordered_map<std::string, size_t> s_tier_system_hash;
static std::unordered_map<std::string, entropic::PromptLogDedup>
    s_tier_prompt_dedup;
 
static void log_system_message(size_t i, const Message& msg,
                               const std::string& tier) {
    size_t h = std::hash<std::string>{}(msg.content);
    size_t prev = s_tier_system_hash[tier];
    s_tier_system_hash[tier] = h;
    if (h != prev || prev == 0) {
        logger->info("[{}] role=system hash={:016x} prev={:016x}\n{}",
                     i, h, prev, msg.content);
    } else {
        logger->info("[{}] role=system [unchanged, {} chars, hash={:016x}]",
                     i, msg.content.size(), h);
    }
}
 
static void log_other_message(size_t i, const Message& msg,
                              entropic::PromptLogDedup& dedup) {
    const size_t h = std::hash<std::string>{}(msg.content);
    if (dedup.note(h) || entropic::prompt_log_unabridged()) {
        logger->info("[{}] role={}\n{}", i, msg.role, msg.content);
        return;
    }
    logger->info("{}", entropic::prompt_log_reference(
        i, msg.role, msg.content.size(), h));
}
 
static void log_prompt(const std::vector<Message>& messages,
                       const std::string& tier) {
    logger->info("─── Prompt ({} messages, tier={}) ───",
                 messages.size(), tier);
    auto& dedup = s_tier_prompt_dedup[tier];
    for (size_t i = 0; i < messages.size(); ++i) {
        if (messages[i].role == "system") {
            log_system_message(i, messages[i], tier);
        } else {
            log_other_message(i, messages[i], dedup);
        }
    }
    logger->info("─── End prompt ───");
}
 
ResponseGenerator::ResponseGenerator(
    const InferenceInterface& inference,
    const LoopConfig& loop_config,
    EngineCallbacks& callbacks,
    GenerationEvents events)
    : inference_(inference),
      loop_config_(loop_config),
      callbacks_(callbacks),
      events_(events) {}
 
GenerateResult ResponseGenerator::generate_response(LoopContext& ctx) {
    lock_tier_if_needed(ctx);
 
    if (loop_config_.stream_output) {
        return generate_streaming(ctx);
    }
    return generate_batch(ctx);
}
 
bool ResponseGenerator::is_response_complete(
    const std::string& content,
    const std::string& tool_calls_json) {
    if (inference_.is_response_complete == nullptr) {
        return !content.empty();
    }
    return inference_.is_response_complete(
        content.c_str(), tool_calls_json.c_str(),
        inference_.adapter_data) != 0;
}
 
void ResponseGenerator::lock_tier_if_needed(LoopContext& ctx) {
    if (!ctx.locked_tier.empty()) {
        if (callbacks_.on_tier_selected != nullptr) {
            callbacks_.on_tier_selected(ctx.locked_tier.c_str(),
                                        callbacks_.user_data);
        }
        return;
    }
 
    if (inference_.route == nullptr) {
        ctx.locked_tier = "default";
        return;
    }
 
    auto msgs_json = serialize_messages(ctx.messages);
    char* result_json = nullptr;
    int rc = inference_.route(msgs_json.c_str(), &result_json,
                              inference_.orchestrator_data);
    if (rc == 0 && result_json != nullptr) {
        ctx.locked_tier = result_json;
        if (inference_.free_fn != nullptr) {
            inference_.free_fn(result_json);
        }
    } else {
        ctx.locked_tier = "default";
        logger->warn("Routing failed (rc={}), using default tier", rc);
    }
 
    logger->info("Locked tier: {}", ctx.locked_tier);
    if (callbacks_.on_tier_selected != nullptr) {
        callbacks_.on_tier_selected(ctx.locked_tier.c_str(),
                                    callbacks_.user_data);
    }
}
 
// ── Streaming token accumulator context ──────────────────
 
struct StreamAccumulator {
    std::string content;              
    EngineCallbacks* callbacks;       
    GenerationEvents* events;         
    const HookInterface* hooks;       
    int token_index = 0;              
    bool interrupted = false;         
    int* cancel_flag = nullptr;
    void (*observer)(const char*, size_t, void*) = nullptr;
    void* observer_data = nullptr;    
};
 
static void stream_token_callback(
    const char* token,
    size_t len,
    void* user_data) {
    auto* acc = static_cast<StreamAccumulator*>(user_data);
 
    // gh#20 (v2.1.5): two coupled bugs lived here.
    //
    // (A) The previous implementation set `acc->interrupted = true`
    //     and returned early WITHOUT propagating the interrupt to the
    //     backend. The backend's cancel_flag stayed 0, so llama_cpp
    //     ran to natural EOS — up to 60s of wasted decode after the
    //     user pressed Ctrl-C.
    //
    // (B) The early return also dropped every post-interrupt token
    //     from `acc->content`. When the backend finally finished
    //     cleanly, the response_generator built the iter result from
    //     this truncated buffer (e.g. 7 chars instead of the 107
    //     decoded tokens forming a valid tool call), throwing away
    //     fully-formed output.
    //
    // The fix raises the cancel flag for the backend AND keeps
    // appending the token so the content buffer is complete up to
    // the cancel point. The backend stops on its next loop iteration
    // (<= 1 token wall-time); whatever made it through is preserved.
    bool just_interrupted = acc->events != nullptr
        && generation_cancelled(*acc->events)
        && !acc->interrupted;
    if (just_interrupted) {
        acc->interrupted = true;
        // gh#49 (v2.1.12): log the cancel-flag raise so a session
        // log can confirm the per-token interrupt poll observed the
        // engine-level flag. Pre-v2.1.12 the bissell-llm-studio
        // repro saw the "Engine interrupted" line on 0->1 transition
        // but no evidence the per-token poll ever fired — this log
        // is the first observable receipt of the propagation.
        logger->info("Stream interrupt observed at token {}; "
                     "raising backend cancel_flag",
                     acc->token_index);
        if (acc->cancel_flag != nullptr) {
            *acc->cancel_flag = 1;
        }
    }
 
    acc->content.append(token, len);
    // gh#208 (v2.14.2): a streaming run's own sink first — published to
    // this (the run's) thread. The handle-wide slot is only for a caller
    // that registered one outside any streaming run.
    if (!emit_to_current_run_stream(token, len)
        && acc->callbacks->on_stream_chunk != nullptr) {
        acc->callbacks->on_stream_chunk(token, len,
                                         acc->callbacks->user_data);
    }
 
    // Global observer — fires on every token regardless of whether
    // the caller registered on_stream_chunk. (2.0.6-rc16)
    if (acc->observer != nullptr) {
        acc->observer(token, len, acc->observer_data);
    }
 
    // Hook: ON_STREAM_TOKEN (v1.9.1)
    if (acc->hooks != nullptr && acc->hooks->fire_info != nullptr) {
        std::string json = "{\"token_index\":"
            + std::to_string(acc->token_index++) + "}";
        acc->hooks->fire_info(acc->hooks->registry,
            ENTROPIC_HOOK_ON_STREAM_TOKEN, json.c_str());
    }
}
 
static std::string resolve_stream_finish_reason(int rc,
                                                size_t content_size) {
    std::string reason;
    if (rc == ENTROPIC_ERROR_CANCELLED) {
        logger->info("Stream cancelled by interrupt after {} chars",
                     content_size);
        reason = "interrupted";
    } else if (rc == ENTROPIC_ERROR_EVAL_CONTEXT_FULL) {
        logger->error("Stream refused: the prompt does not fit this tier's "
                      "context — see the inference log for the token "
                      "breakdown (rc={})", rc);
        reason = "context_overflow";
    } else if (rc != 0 && content_size > 0) {
        logger->warn("Stream failed (rc={}) after {} chars — "
                     "preserving partial", rc, content_size);
        reason = "partial";
    } else if (rc != 0) {
        logger->error("Stream failed (rc={}) with no partial content", rc);
        reason = "error";
    } else {
        reason = "stop";
    }
    return reason;
}
 
std::pair<std::string, std::string> ResponseGenerator::prepare_prompts(
    LoopContext& ctx, const char* mode) {
    // gh#87 (v2.7.0): tool defs no longer string-injected into the system
    // message — they flow as structured JSON via build_params_json →
    // params.tools, and common_chat renders them in the model's native
    // format. Only the runtime engine-state reminder is injected here.
    auto messages = inject_engine_state_reminder(ctx.messages, ctx);
    logger->info("Generate ({}): tier={}, {} messages",
                 mode, ctx.locked_tier, messages.size());
    log_prompt(messages, ctx.locked_tier);
    return {serialize_messages(messages),
            build_params_json(ctx.locked_tier, ctx.session_key)};
}
 
GenerateResult ResponseGenerator::generate_streaming(LoopContext& ctx) {
    if (inference_.generate_stream == nullptr) {
        logger->warn("No streaming function, falling back to batch");
        return generate_batch(ctx);
    }
 
    auto [msgs_json, params_json] = prepare_prompts(ctx, "stream");
 
    int cancel_flag = 0;
    StreamAccumulator acc;
    acc.callbacks = &callbacks_;
    acc.events = &events_;
    acc.hooks = &hooks_;
    // gh#20 (v2.1.5): give the token callback a path to raise the
    // backend cancel flag when an interrupt is observed. Without
    // this, the previous implementation would early-return out of
    // the token callback without ever telling the backend to stop.
    acc.cancel_flag = &cancel_flag;
    // Wire the persistent stream observer so every token — including
    // batch entropic_run and delegate child-loop generations — reaches
    // any registered observer. (P0-1, 2.0.6-rc16)
    acc.observer = stream_observer_;
    acc.observer_data = stream_observer_data_;
 
    int rc = inference_.generate_stream(
        msgs_json.c_str(), params_json.c_str(),
        stream_token_callback, &acc,
        &cancel_flag, inference_.backend_data);
 
    GenerateResult result;
    result.finish_reason = resolve_stream_finish_reason(rc,
                                                        acc.content.size());
    // Issue #3 (v2.1.1): inbound boundary from llama_cpp. Models can emit
    // malformed UTF-8 mid-stream (partial multi-byte runs under XML-tool-call
    // pressure, decoder desyncs). Sanitize ONCE at message-finalization,
    // never per-token — a multi-byte codepoint may split across token
    // boundaries and per-token sanitize would corrupt valid output.
    // See include/entropic/mcp/utf8_sanitize.h for the boundary policy.
    result.content = mcp::sanitize_utf8(acc.content);
    result.tool_calls_json = "[]";
    logger->info("Generate complete (stream): finish={}, {} chars",
                 result.finish_reason, result.content.size());
    return result;
}
 
static bool flag_raised(const std::atomic<bool>* flag) {
    return flag != nullptr && flag->load(std::memory_order_acquire);
}
 
std::thread ResponseGenerator::spawn_cancel_poller(
    int& cancel_int, std::atomic<bool>& done) const {
    const std::atomic<bool>* handle_flag = events_.interrupt;
    const std::atomic<bool>* run_token = current_run_cancel();
    if (handle_flag == nullptr && run_token == nullptr) {
        return {};
    }
    return std::thread(
        [&cancel_int, &done, handle_flag, run_token]() {
            while (!done.load(std::memory_order_acquire)) {
                if (flag_raised(handle_flag) || flag_raised(run_token)) {
                    cancel_int = 1;
                    return;
                }
                std::this_thread::sleep_for(
                    std::chrono::milliseconds(10));
            }
        });
}
 
int ResponseGenerator::dispatch_batch_generate(
    const std::string& msgs_json,
    const std::string& params_json,
    char** result_json) {
    // Fall back to the no-cancel entry for backends that predate the
    // generate_cancellable ABI field.
    //
    // gh#110 (v2.9.6): also prefer the no-cancel entry whenever
    // speculative decoding is enabled. `generate_cancellable`'s
    // backing orchestrator call deliberately bypasses
    // run_generate_dispatch (speculative/MTP routing) — batch-with-
    // cancel only ever calls plain decode. The plain `generate` entry
    // point runs run_generate_dispatch and is therefore the only
    // batch path that can reach MTP. v1 tradeoff: a speculative batch
    // turn is not cancellable mid-decode (documented, not silent).
    if (inference_.generate_cancellable == nullptr
        || loop_config_.speculative_enabled) {
        return inference_.generate(
            msgs_json.c_str(), params_json.c_str(),
            result_json, inference_.backend_data);
    }
 
    // The C-ABI side bridges this int → atomic<bool> for the backend
    // via its own poller; this side mirrors the engine's atomic
    // interrupt_flag_ → the int. Two cheap 10ms-poll hops, but it
    // keeps the C ABI int*-only (no atomic across the .so boundary).
    int cancel_int = generation_cancelled(events_) ? 1 : 0;
 
    std::atomic<bool> observer_done(false);
    std::thread observer = spawn_cancel_poller(cancel_int, observer_done);
 
    int rc = inference_.generate_cancellable(
        msgs_json.c_str(), params_json.c_str(),
        result_json, &cancel_int, inference_.backend_data);
 
    observer_done.store(true, std::memory_order_release);
    if (observer.joinable()) { observer.join(); }
    return rc;
}
 
static bool apply_terminal_batch_rc(int rc, const char* result_json,
                                    GenerateResult& result) {
    if (rc == ENTROPIC_ERROR_CANCELLED) {
        result.finish_reason = "interrupted";
        if (result_json != nullptr) {
            result.content = mcp::sanitize_utf8(result_json);
        }
        result.tool_calls_json = "[]";
        logger->info("Generate cancelled (batch) after {} chars",
                     result.content.size());
    } else if (rc == ENTROPIC_ERROR_EVAL_CONTEXT_FULL) {
        result.finish_reason = "context_overflow";
        result.tool_calls_json = "[]";
        logger->error("Generate refused (batch): the prompt does not fit "
                      "this tier's context — see the inference log for the "
                      "token breakdown (rc={})", rc);
    } else {
        return false;
    }
    return true;
}
 
GenerateResult ResponseGenerator::generate_batch(LoopContext& ctx) {
    if (inference_.generate == nullptr
        && inference_.generate_cancellable == nullptr) {
        logger->error("No generate function available");
        return {"", "[]", "error"};
    }
 
    auto [msgs_json, params_json] = prepare_prompts(ctx, "batch");
    char* result_json = nullptr;
 
    // gh#81 (v2.4.2): prefer the cancellable dispatch so an interrupt
    // is honored mid-decode (was ~60s lag pre-fix).
    int rc = dispatch_batch_generate(msgs_json, params_json, &result_json);
 
    GenerateResult result;
    // Cancellation and the v2.13.0 context refusal are both terminal and
    // both shaped unlike a retryable failure — see apply_terminal_batch_rc.
    if (!apply_terminal_batch_rc(rc, result_json, result)) {
        if (rc == 0 && result_json != nullptr) {
            // Issue #3 (v2.1.1): inbound boundary, batch path. See the
            // streaming branch above for rationale; same policy applies.
            result.content = mcp::sanitize_utf8(result_json);
            result.finish_reason = "stop";
            result.tool_calls_json = "[]";
            // Fire observer once with full content so the non-streaming
            // fallback still reaches registered observers. (2.0.6-rc16)
            if (stream_observer_ != nullptr && !result.content.empty()) {
                stream_observer_(result.content.data(),
                                 result.content.size(),
                                 stream_observer_data_);
            }
        } else {
            result.finish_reason = "error";
            logger->error("Generate failed (rc={})", rc);
        }
    }
    if (result_json != nullptr && inference_.free_fn != nullptr) {
        inference_.free_fn(result_json);
    }
    logger->info("Generate complete (batch): finish={}, {} chars",
                 result.finish_reason, result.content.size());
    return result;
}
 
std::string ResponseGenerator::handle_pause(
    LoopContext& ctx,
    const std::string& partial) {
    ctx.state = AgentState::PAUSED;
    if (callbacks_.on_state_change != nullptr) {
        callbacks_.on_state_change(
            static_cast<int>(AgentState::PAUSED),
            callbacks_.user_data);
    }
    // gh#40 fallout (v2.1.10): persistent slot fires alongside the
    // legacy on_state_change so consumers see PAUSED during
    // streaming runs where the legacy callbacks_ struct has been
    // overwritten by run_streaming's set_callbacks() shuffle.
    if (state_observer_ != nullptr) {
        state_observer_(static_cast<int>(AgentState::PAUSED),
                        state_observer_data_);
    }
 
    char* injection = nullptr;
    if (callbacks_.on_pause_prompt != nullptr) {
        callbacks_.on_pause_prompt(partial.c_str(), &injection,
                                    callbacks_.user_data);
    }
 
    if (injection == nullptr) {
        // gh#158: "the consumer declined to inject" means abandon THIS turn,
        // so raise this run's own token. Falling through to the handle-wide
        // flag (the pre-2.13.0 behaviour) would abort every concurrent
        // session's turn on one session's pause prompt.
        if (!cancel_current_run() && events_.interrupt != nullptr) {
            events_.interrupt->store(true);
        }
        return partial;
    }
 
    std::string inj(injection);
    if (inj.empty()) {
        ctx.state = AgentState::EXECUTING;
        return partial;
    }
 
    // Injection provided: append partial + injection to messages
    if (!partial.empty()) {
        Message partial_msg;
        partial_msg.role = "assistant";
        partial_msg.content = partial + "\n\n[Generation paused by user]";
        ctx.messages.push_back(std::move(partial_msg));
    }
    Message inject_msg;
    inject_msg.role = "user";
    inject_msg.content = "[User interjection]: " + inj
        + "\n\nPlease continue with this in mind.";
    ctx.messages.push_back(std::move(inject_msg));
 
    ctx.state = AgentState::EXECUTING;
    return "";
}
 
static void json_escape_into(const std::string& s, std::string& out) {
    for (char c : s) {
        switch (c) {
        case '"':  out += "\\\""; break;
        case '\\': out += "\\\\"; break;
        case '\n': out += "\\n";  break;
        case '\r': out += "\\r";  break;
        case '\t': out += "\\t";  break;
        default:   out += c;      break;
        }
    }
}
 
static void serialize_content_parts(
    const std::vector<ContentPart>& parts, std::string& out) {
    out += '[';
    for (size_t i = 0; i < parts.size(); ++i) {
        if (i > 0) { out += ','; }
        if (parts[i].type == ContentPartType::IMAGE) {
            out += R"({"type":"image","path":")";
            json_escape_into(parts[i].image_path, out);
            out += R"(","url":")";
            json_escape_into(parts[i].image_url, out);
            out += R"("})";
        } else {
            out += R"({"type":"text","text":")";
            json_escape_into(parts[i].text, out);
            out += R"("})";
        }
    }
    out += ']';
}
 
std::string ResponseGenerator::serialize_messages(
    const std::vector<Message>& messages) {
    std::string json = "[";
    for (size_t i = 0; i < messages.size(); ++i) {
        if (i > 0) { json += ','; }
        json += "{\"role\":\"" + messages[i].role + "\",\"content\":";
        if (messages[i].content_parts.empty()) {
            json += '"';
            json_escape_into(messages[i].content, json);
            json += '"';
        } else {
            serialize_content_parts(messages[i].content_parts, json);
        }
        json += '}';
    }
    json += ']';
    return json;
}
 
std::string ResponseGenerator::build_params_json(
    const std::string& tier, const std::string& session_key) {
    nlohmann::json j = nlohmann::json::object();
    if (!tier.empty()) { j["tier"] = tier; }
    // gh#144 (v2.12.0): only emitted when non-empty, so a handle that never
    // names a session sends a byte-identical params payload.
    if (!session_key.empty()) { j["session"] = session_key; }
 
    if (inference_.get_tool_prompt != nullptr) {
        char* tools = nullptr;
        int rc = inference_.get_tool_prompt(
            tier.c_str(), &tools, inference_.tool_prompt_data);
        if (rc == 0 && tools != nullptr) {
            j["tools"] = std::string(tools);
            if (inference_.free_fn) { inference_.free_fn(tools); }
        }
    }
    return j.dump();
}
 
std::vector<Message> ResponseGenerator::inject_engine_state_reminder(
    const std::vector<Message>& messages,
    const LoopContext& ctx) {
    int max_iter = ctx.effective_max_iterations >= 0
        ? ctx.effective_max_iterations
        : loop_config_.max_iterations;
    std::string reminder = "[engine] iteration "
        + std::to_string(ctx.metrics.iterations)
        + "/" + std::to_string(max_iter)
        + ", tool calls so far: "
        + std::to_string(ctx.metrics.tool_calls) + ".";
 
    // Demo ask #2 (v2.1.0): if the previous turn was validator-rejected,
    // surface the reason so the model knows WHY it's being asked again.
    // Engine clears pending_validation_feedback after this turn — the
    // line is one-shot.
    if (!ctx.pending_validation_feedback.empty()) {
        reminder += "\n[engine] previous turn rejected: "
                  + ctx.pending_validation_feedback;
    }
    // Demo ask #5 (v2.1.0): anti-spiral primitive. ToolExecutor
    // populated this when consecutive_same_tool_calls hit
    // max_consecutive_same_tool. Same one-shot lifecycle as the
    // validation feedback above; engine clears after this turn.
    if (!ctx.pending_anti_spiral_warning.empty()) {
        reminder += "\n[engine] anti-spiral: "
                  + ctx.pending_anti_spiral_warning;
    }
 
    auto result = messages;
    Message reminder_msg;
    reminder_msg.role = "user";
    reminder_msg.content = std::move(reminder);
    result.push_back(std::move(reminder_msg));
    return result;
}
 
} // namespace entropic