#include "agent_loop.hpp"

#include "agent_loop_doom_guard.hpp"

#include "agent_loop_shell_guard.hpp"

#include "prompt/context_usage_breakdown.hpp"

#include "prompt/system_prompt.hpp"

#include "gitinfo/git_context_collector.hpp"

#include "utils/encoding.hpp"

#include "utils/logger.hpp"

#include "utils/stream_processing.hpp"

#include "utils/text_file_buffer.hpp"

#include "utils/uuid.hpp"

#include "commands/compact.hpp"

#include "session/compact_checkpoint.hpp"

#include "session/compact_notice.hpp"

#include "session/session_history_recovery.hpp"

#include "session/tool_metadata_codec.hpp"

#include "session/tool_result_storage.hpp"

#include "session/output_attachments.hpp"

#include "session/session_rewind.hpp"

#include "session/session_serializer.hpp"

#include "session/session_storage.hpp"

#include "session/thread_goal_store.hpp"

#include "session/thread_repair.hpp"

#include "session/todo_state.hpp"

#include "session/turn_timing.hpp"

#include "session/turn_net_diff.hpp"

#include "skills/skill_activation.hpp"

#include "tool/ask_user_question_tool.hpp"

#include "tool/mtime_tracker.hpp"

#include "tool/tool_protocol_names.hpp"

#include "web/message_payload.hpp"

#include "web/tool_event_payload.hpp"

#include "hooks/hook_config.hpp"

#include "hooks/hook_manager.hpp"

#include "hooks/hook_payload.hpp"

#include "headless/headless_mode.hpp"

#include <nlohmann/json.hpp>

#include <chrono>

#include <mutex>

#include <future>

#include <algorithm>

#include <thread>

#include <sstream>

#include <deque>

#include <cstdint>

#include <limits>

#include <cctype>

namespace acecode {

namespace {

constexpr const char* kDefaultNoModelConfiguredPrompt =

u8"请先配置大模型服务。";

std::vector<ChatMessage> recovered_provider_messages(

const std::vector<ChatMessage>& messages,

const char* boundary) {

auto recovery = recover_provider_history(provider_relevant_messages(messages));

if (recovery.stats.changed()) {

const auto& stats = recovery.stats;

LOG_WARN(std::string{"[session-recovery] boundary="} + boundary +

" malformed_calls=" + std::to_string(stats.malformed_tool_calls) +

" duplicate_calls=" + std::to_string(stats.duplicate_tool_calls) +

" synthesized_results=" +

std::to_string(stats.synthesized_tool_results) +

" standalone_results=" +

std::to_string(stats.standalone_tool_results) +

" unexpected_results=" +

std::to_string(stats.unexpected_tool_results) +

" duplicate_results=" +

std::to_string(stats.duplicate_tool_results) +

" empty_assistants=" +

std::to_string(stats.empty_assistant_messages));

}

return std::move(recovery.messages);

}

bool has_meaningful_user_input(const UserInput& input) {

if (input.has_content_parts()) return true;

return std::any_of(input.text.begin(), input.text.end(), [](unsigned char ch) {

return std::isspace(ch) == 0;

});

}

std::string trim_ascii_copy(const std::string& raw) {

std::size_t first = 0;

while (first < raw.size() &&

std::isspace(static_cast<unsigned char>(raw[first])) != 0) {

++first;

}

std::size_t last = raw.size();

while (last > first &&

std::isspace(static_cast<unsigned char>(raw[last - 1])) != 0) {

--last;

}

return raw.substr(first, last - first);

}

ChatMessage build_side_question_message(const std::string& question) {

ChatMessage message;

message.role = "user";

message.content =

"[SYSTEM NOTE] Answer the side question below using the conversation "

"context above. This is a separate, read-only, one-turn question. "

"Do not call tools, do not continue the main task, and do not claim "

"that you changed files or session state. Answer directly and "

"concisely.\n\nSide question:\n" + question;

return message;

}

std::int64_t now_epoch_ms() {

return std::chrono::duration_cast<std::chrono::milliseconds>(

std::chrono::system_clock::now().time_since_epoch()).count();

}

nlohmann::json build_agent_progress_payload(

const std::string& phase,

const std::string& label,

const std::string& detail,

const std::string& tool,

const std::string& tool_call_id,

int tool_index,

std::int64_t started_at_ms) {

nlohmann::json payload;

payload["phase"] = phase;

payload["label"] = label;

if (!detail.empty()) payload["detail"] = detail;

if (!tool.empty()) payload["tool"] = tool;

if (!tool_call_id.empty()) payload["tool_call_id"] = tool_call_id;

if (tool_index >= 0) payload["tool_index"] = tool_index;

if (started_at_ms > 0) payload["started_at_ms"] = started_at_ms;

return payload;

}

std::string build_session_scratch_dir(const std::string& cwd,

SessionManager* session_manager) {

if (cwd.empty() || !session_manager) return {};

const std::string session_id = session_manager->ensure_active_session_id();

if (session_id.empty()) return {};

return path_to_utf8(path_from_utf8(cwd) / ".acecode" / "tmp" /

("session-" + session_id));

}

std::string provider_error_kind_to_json_string(ProviderErrorKind kind) {

switch (kind) {

case ProviderErrorKind::None: return "none";

case ProviderErrorKind::UserCancelled: return "user_cancelled";

case ProviderErrorKind::Timeout: return "timeout";

case ProviderErrorKind::Network: return "network";

case ProviderErrorKind::Http: return "http";

case ProviderErrorKind::MalformedSse: return "malformed_sse";

case ProviderErrorKind::MalformedJson: return "malformed_json";

case ProviderErrorKind::Unknown: return "unknown";

}

return "unknown";

}

nlohmann::json provider_error_to_json(const ProviderErrorInfo& info) {

nlohmann::json j = {

{"kind", provider_error_kind_to_json_string(info.kind)},

{"status_code", info.status_code},

{"provider", info.provider},

{"model", info.model},

{"request_id", info.request_id},

{"display_message", info.display_message},

{"raw_body", info.raw_body},

{"body_is_json", info.body_is_json},

{"pretty_json", info.pretty_json},

{"retryable", info.retryable},

{"retry_attempt", info.retry_attempt},

{"retry_max_attempts", info.retry_max_attempts},

{"retry_delay_ms", info.retry_delay_ms},

{"server_retry_after_ms", info.server_retry_after_ms},

};

return j;

}

nlohmann::json model_step_usage_to_json(const TokenUsage& usage) {

nlohmann::json value = {

{"prompt_tokens", usage.prompt_tokens},

{"completion_tokens", usage.completion_tokens},

{"total_tokens", usage.total_tokens},

{"cache_read_tokens", usage.cache_read_tokens},

{"cache_write_tokens", usage.cache_write_tokens},

{"reasoning_tokens", usage.reasoning_tokens},

{"has_data", usage.has_data},

};

if (usage.context_breakdown.has_data) {

value["context_breakdown"] =

context_usage_breakdown_to_json(usage.context_breakdown);

}

return value;

}

std::string provider_error_summary_for_log(const ProviderErrorInfo& info) {

std::string message = info.display_message;

if (message.empty()) message = info.pretty_json;

if (message.empty()) message = info.raw_body;

std::ostringstream oss;

oss << "kind=" << provider_error_kind_to_json_string(info.kind)

<< " status=" << info.status_code

<< " provider=" << info.provider

<< " model=" << info.model

<< " request_id=" << info.request_id

<< " retryable=" << (info.retryable ? "true" : "false")

<< " retry_attempt=" << info.retry_attempt

<< " retry_max_attempts=" << info.retry_max_attempts

<< " retry_delay_ms=" << info.retry_delay_ms

<< " raw_body_bytes=" << info.raw_body.size()

<< " pretty_json_bytes=" << info.pretty_json.size()

<< " message=" << log_truncate(message, 300);

return oss.str();

}

// 人类可读的字节量:< 1KB 显示原始字节,否则进位到 KB / MB(保留一位小数)。

// 进度文案里直接打印原始字节数(如 "8641 字节")观感上会显得异常地大,统一走这里。

std::string human_bytes(std::size_t bytes) {

if (bytes < 1024) return std::to_string(bytes) + " 字节";

std::ostringstream oss;

oss.setf(std::ios::fixed);

oss.precision(1);

double kb = static_cast<double>(bytes) / 1024.0;

if (kb < 1024.0) oss << kb << " KB";

else oss << (kb / 1024.0) << " MB";

return oss.str();

}

std::string format_bytes_detail(std::size_t bytes) {

return "参数 " + human_bytes(bytes);

}

std::string ascii_lower(std::string value) {

std::transform(value.begin(), value.end(), value.begin(), [](unsigned char c) {

return static_cast<char>(std::tolower(c));

});

return value;

}

nlohmann::json parse_tool_args_for_permission_payload(const std::string& args_json) {

if (args_json.empty()) return nlohmann::json::object();

try {

auto parsed = nlohmann::json::parse(args_json);

return parsed.is_object() ? parsed : nlohmann::json{{"raw", args_json}};

} catch (...) {

return nlohmann::json{{"raw", args_json}};

}

}

std::string build_plan_permission_args(const std::string& tool_name,

const std::string& args_json,

SessionManager* session_manager) {

nlohmann::json payload;

payload["tool_args"] = parse_tool_args_for_permission_payload(args_json);

if (tool_name == "EnterPlanMode") {

payload["kind"] = "enter_plan_mode";

if (session_manager) {

payload["plan_file_path"] = session_manager->current_plan_file_path();

}

return payload.dump();

}

if (tool_name == "ExitPlanMode") {

payload["kind"] = "plan_approval";

if (session_manager) {

payload["plan_file_path"] = session_manager->ensure_plan_file_path();

payload["plan"] = session_manager->read_plan_file();

}

return payload.dump();

}

return args_json;

}

std::string build_plan_mode_context_prompt(SessionManager* session_manager,

bool ask_user_allowed,

bool exit_plan_mode_allowed) {

if (!session_manager) return {};

const std::string plan_file = session_manager->ensure_plan_file_path();

if (plan_file.empty()) return {};

const std::string existing_plan = session_manager->read_plan_file();

MtimeTracker::instance().record_read(plan_file, existing_plan, false);

std::ostringstream oss;

oss << "<plan_mode>\n"

<< "Plan mode is active. You MUST NOT make any edits except to the plan file.\n\n"

<< "Plan file path: " << plan_file << "\n"

<< "Plan exists: " << (existing_plan.empty() ? "false" : "true") << "\n\n"

<< "Workflow:\n"

<< "1. Explore the codebase with read-only tools until the approach is clear.\n"

<< "2. Keep the implementation plan in the plan file. Update that file as your plan changes.\n";

int workflow_step = 3;

if (ask_user_allowed) {

oss << workflow_step++

<< ". Use AskUserQuestion only for unresolved requirements or approach choices.\n";

}

if (exit_plan_mode_allowed) {

oss << workflow_step++

<< ". When the plan is complete and unambiguous, call ExitPlanMode for user approval.\n\n";

if (ask_user_allowed) {

oss << "Do not ask the user whether the plan is OK with AskUserQuestion; ExitPlanMode is the approval request.\n";

}

} else {

oss << workflow_step

<< ". When the plan is complete, present the result in your final reply.\n";

}

oss << "</plan_mode>";

return oss.str();

}

void append_plan_mode_context_for_api(std::vector<ChatMessage>& messages,

const std::string& context) {

if (context.empty()) return;

ChatMessage msg;

msg.role = "user";

msg.content = context;

msg.metadata = nlohmann::json{{"hidden_plan_mode_context", true}};

messages.push_back(std::move(msg));

}

void append_todo_context_for_api(std::vector<ChatMessage>& messages,

const std::vector<TodoItem>& todos) {

std::string context = format_todo_injection(todos);

if (context.empty()) return;

ChatMessage msg;

msg.role = "user";

msg.content = std::move(context);

msg.metadata = nlohmann::json{{"hidden_todo_context", true}};

messages.push_back(std::move(msg));

}

bool is_hidden_goal_context_message(const ChatMessage& msg) {

return msg.metadata.is_object() &&

msg.metadata.value("hidden_goal_context", false);

}

std::string escape_xml_text(const std::string& input) {

std::string out;

out.reserve(input.size());

for (char c : input) {

switch (c) {

case '&': out += "&amp;"; break;

case '<': out += "&lt;"; break;

case '>': out += "&gt;"; break;

default: out.push_back(c); break;

}

}

return out;

}

std::string format_goal_status_chip(const ThreadGoal& goal) {

std::ostringstream oss;

oss << "goal: " << to_string(goal.status) << " "

<< TokenTracker::format_tokens(static_cast<int>(std::min<std::int64_t>(

goal.tokens_used,

static_cast<std::int64_t>(std::numeric_limits<int>::max()))));

if (goal.token_budget.has_value()) {

oss << "/" << TokenTracker::format_tokens(static_cast<int>(std::min<std::int64_t>(

*goal.token_budget,

static_cast<std::int64_t>(std::numeric_limits<int>::max()))));

}

return oss.str();

}

nlohmann::json build_transcript_replace_payload(

const std::vector<ChatMessage>& messages,

const CompactResult& result) {

nlohmann::json arr = nlohmann::json::array();

for (const auto& msg : messages) {

if (is_file_checkpoint_message(msg)) continue;

if (is_compact_checkpoint_message(msg)) continue;

if (is_content_replacement_message(msg)) continue;

if (is_turn_timing_message(msg)) continue;

if (web::is_hidden_goal_context_message(msg)) continue;

arr.push_back(web::chat_message_to_payload_json(msg));

}

return nlohmann::json{

{"messages", std::move(arr)},

{"messages_compressed", result.messages_compressed},

{"estimated_tokens_saved", result.estimated_tokens_saved},

};

}

void append_request_context_for_api(std::vector<ChatMessage>& messages,

const std::string& context) {

if (context.empty()) return;

ChatMessage msg;

msg.role = "user";

msg.content = context;

messages.push_back(std::move(msg));

}

bool should_persist_trajectory_event(const SessionEvent& event) {

switch (event.kind) {

case SessionEventKind::Token:

case SessionEventKind::Reasoning:

case SessionEventKind::ToolUpdate:

case SessionEventKind::ToolEnd:

case SessionEventKind::TurnDiff:

case SessionEventKind::TranscriptReplace:

case SessionEventKind::GoalUpdated:

case SessionEventKind::GoalCleared:

case SessionEventKind::TodoUpdated:

case SessionEventKind::SessionUpdated:

case SessionEventKind::Done:

case SessionEventKind::BusyChanged:

return false;

case SessionEventKind::AgentProgress: {

const std::string phase = event.payload.value("phase", std::string{});

return phase == "model_retry" || phase == "compacting";

}

case SessionEventKind::Message: {

const std::string role = event.payload.value("role", std::string{});

return role != "tool_call" && role != "tool_result";

}

default:

return true;

}

}

std::string cached_context_for_api(const PromptContextBlock& block,

std::string& cached_key,

std::string& cached_content) {

if (block.cache_key != cached_key) {

cached_key = block.cache_key;

cached_content = block.content;

}

return cached_content;

}

} // namespace

AgentLoop::AgentLoop(ProviderAccessor provider_accessor, ToolExecutor& tools,

AgentCallbacks callbacks, const std::string& cwd,

PermissionManager& permissions)

: provider_accessor_(std::move(provider_accessor))

, tools_(tools)

, callbacks_(std::move(callbacks))

, cwd_(cwd)

, permissions_(permissions)

, path_validator_(cwd, permissions.is_dangerous())

, no_model_config_prompt_(kDefaultNoModelConfiguredPrompt)

{

worker_thread_ = std::thread(&AgentLoop::worker_main, this);

}

AgentLoop::~AgentLoop() {

shutdown();

}

void AgentLoop::set_session_manager(SessionManager* sm) {

session_manager_ = sm;

if (!sm) {

events_.set_observer({});

return;

}

events_.set_observer([sm](const SessionEvent& event) {

if (!should_persist_trajectory_event(event)) return;

sm->record_trajectory_event(

to_string(event.kind), event.payload, event.timestamp_ms);

});

}

void AgentLoop::record_terminal_trajectory_events(

nlohmann::json busy_payload,

nlohmann::json done_payload) {

if (!session_manager_) return;

const std::int64_t timestamp_ms = now_epoch_ms();

session_manager_->record_trajectory_event(

"busy_changed", std::move(busy_payload), timestamp_ms);

session_manager_->record_trajectory_event(

"done", std::move(done_payload), timestamp_ms);

}

void AgentLoop::set_cwd(const std::string& new_cwd) {

cwd_ = new_cwd;

path_validator_ = PathValidator(new_cwd, permissions_.is_dangerous());

// cwd 变了(EnterWorktree/ExitWorktree),旧 gitStatus 快照作废,

// 下一次模型请求按新 cwd 重采(openspec add-git-context)。

git_snapshot_cache_.reset();

}

ResolvedQuestionPolicy AgentLoop::resolved_question_policy() const {

const bool has_cli = !loop_cfg_.question_policy_cli.empty();

const std::string& configured =

has_cli ? loop_cfg_.question_policy_cli : loop_cfg_.question_policy;

const bool explicit_choice = has_cli || loop_cfg_.question_policy_explicit;

const int timeout_seconds =

(has_cli && loop_cfg_.question_timeout_seconds_cli > 0)

? loop_cfg_.question_timeout_seconds_cli

: loop_cfg_.question_timeout_seconds;

return resolve_question_policy(configured, explicit_choice, timeout_seconds);

}

void AgentLoop::dispatch_message(const std::string& role,

const std::string& content,

bool is_tool,

nlohmann::json metadata,

nlohmann::json content_parts) {

if (callbacks_.on_message) {

callbacks_.on_message(role, content, is_tool);

}

// Web 协议给每条 message 带稳定 id:user 走持久 uuid(走另一路径

// 直接 emit,见 run_agent),其它角色 lazy sha1(role + " " + content

// + " " + timestamp)。这里 timestamp 默认空字符串,跟磁盘上 JSONL

// 重读时算出来的 ID 保持一致(JSONL 里 assistant 消息也没 timestamp)。

ChatMessage tmp;

tmp.role = role;

tmp.content = content;

if (content_parts.is_array() && !content_parts.empty()) {

tmp.content_parts = content_parts;

}

nlohmann::json payload = {

{"role", role}, {"content", content}, {"is_tool", is_tool},

{"id", web::compute_message_id(tmp)}};

if (content_parts.is_array() && !content_parts.empty()) {

payload["content_parts"] = std::move(content_parts);

}

if (metadata.is_object() && !metadata.empty()) {

payload["metadata"] = std::move(metadata);

}

events_.emit(SessionEventKind::Message, std::move(payload));

}

void AgentLoop::append_turn_timing_record(const std::string& user_message_uuid,

std::int64_t started_at_ms,

std::int64_t completed_at_ms,

const std::string& status) {

if (user_message_uuid.empty()) return;

TurnTimingRecord timing;

timing.user_message_uuid = user_message_uuid;

timing.started_at_ms = started_at_ms;

timing.completed_at_ms = completed_at_ms;

timing.duration_ms = std::max<std::int64_t>(0, completed_at_ms - started_at_ms);

timing.status = status;

ChatMessage msg = make_turn_timing_message(timing, SessionStorage::now_iso8601());

messages_.push_back(msg);

if (session_manager_) {

session_manager_->on_message(msg);

session_manager_->record_trajectory_event(

"turn_end",

{{"turn_id", timing.user_message_uuid},

{"user_message_id", timing.user_message_uuid},

{"started_at_ms", timing.started_at_ms},

{"completed_at_ms", timing.completed_at_ms},

{"duration_ms", timing.duration_ms},

{"outcome", timing.status}},

timing.completed_at_ms);

}

}

void AgentLoop::append_tool_user_prompt(const std::string& content,

const std::string& display_text,

const std::string& source_tool) {

if (content.empty()) return;

ChatMessage msg;

msg.role = "user";

msg.content = content;

msg.metadata = nlohmann::json::object();

msg.metadata["display_text"] = display_text.empty()

? "[Tool prompt loaded]"

: display_text;

msg.metadata["synthetic_user_prompt"] = true;

if (!source_tool.empty()) msg.metadata["source_tool"] = source_tool;

ensure_user_message_identity(msg);

messages_.push_back(msg);

if (session_manager_) {

session_manager_->on_message(msg);

}

if (callbacks_.on_message) {

callbacks_.on_message("user", msg.metadata.value("display_text", msg.content), false);

}

nlohmann::json event = {

{"role", "user"},

{"content", msg.content},

{"is_tool", false},

{"id", msg.uuid},

{"metadata", msg.metadata},

};

events_.emit(SessionEventKind::Message, std::move(event));

}

void AgentLoop::dispatch_assistant_completed_hook(

const ChatMessage& assistant_msg,

const std::shared_ptr<LlmProvider>& provider_snapshot) {

if (!hook_manager_ || assistant_msg.role != "assistant") return;

std::string session_id;

if (session_manager_) {

session_id = session_manager_->current_session_id();

}

std::string provider_name;

std::string model_name;

if (provider_snapshot) {

provider_name = provider_snapshot->name();

model_name = provider_snapshot->model();

}

auto payload = build_assistant_message_completed_payload(

cwd_,

session_id,

provider_name,

model_name,

assistant_msg);

hook_manager_->dispatch(kHookEventAssistantMessageCompleted, payload, cwd_);

}

HookCommonPayloadFields AgentLoop::build_hook_common_fields(

const std::string& event_name) const {

HookCommonPayloadFields fields;

fields.cwd = cwd_;

fields.hook_event_name = event_name;

fields.permission_mode = PermissionManager::mode_name(permissions_.mode());

if (session_manager_) {

fields.session_id = session_manager_->current_session_id();

if (!fields.session_id.empty()) {

fields.transcript_path = SessionStorage::session_path(

SessionStorage::get_project_dir(cwd_), fields.session_id);

}

}

if (provider_accessor_) {

auto provider = provider_accessor_();

if (provider) fields.model = provider->model();

}

return fields;

}

void AgentLoop::apply_hook_side_effects(const HookAggregateOutcome& outcome,

bool include_additional_context) {

for (const auto& message : outcome.system_messages) {

if (!message.empty()) dispatch_message("system", "[Hook] " + message, false);

}

if (include_additional_context) {

for (const auto& context : outcome.additional_context) {

if (!context.empty()) hook_request_context_.push_back(context);

}

}

for (const auto& diagnostic : outcome.diagnostics) {

if (diagnostic.severity == HookDiagnosticSeverity::Error ||

diagnostic.severity == HookDiagnosticSeverity::Warning) {

LOG_WARN("[hooks] " + diagnostic.code + " " + diagnostic.message);

}

}

}

std::string AgentLoop::drain_hook_request_context() {

if (hook_request_context_.empty()) return {};

std::ostringstream oss;

oss << "<hook_context>\n";

for (const auto& context : hook_request_context_) {

if (!context.empty()) oss << context << "\n";

}

oss << "</hook_context>";

hook_request_context_.clear();

return oss.str();

}

HookAggregateOutcome AgentLoop::dispatch_codex_hook(

const std::string& event_name,

const std::string& matcher_value,

const nlohmann::json& payload) {

if (!hook_manager_) return {};

HookDispatchRequest request;

request.event_name = event_name;

request.matcher_value = matcher_value;

request.cwd = cwd_;

request.payload = payload.is_object() ? payload : nlohmann::json::object();

return hook_manager_->dispatch_codex(request);

}

void AgentLoop::dispatch_session_start_hook(const std::string& source) {

if (!hook_manager_) return;

auto fields = build_hook_common_fields(kCodexHookEventSessionStart);

auto payload = build_session_start_hook_payload(fields, source);

auto outcome = dispatch_codex_hook(kCodexHookEventSessionStart, source, payload);

apply_hook_side_effects(outcome);

}

void AgentLoop::dispatch_session_title_changed_hook(

const std::string& title,

const std::string& source,

const std::string& title_source) {

if (!hook_manager_) return;

auto fields = build_hook_common_fields(kCodexHookEventSessionTitleChanged);

auto payload = build_session_title_changed_hook_payload(

fields, title, source, title_source);

(void)dispatch_codex_hook(

kCodexHookEventSessionTitleChanged, source, payload);

}

void AgentLoop::abort() {

abort_requested_ = true;

wake_active_provider_retry();

}

void AgentLoop::clear_stale_abort_request() {

if (!busy_.load()) {

abort_requested_ = false;

}

}

void AgentLoop::shutdown() {

side_question_shutdown_.store(true);

{

std::lock_guard<std::mutex> lk(queue_mu_);

shutdown_requested_ = true;

}

abort_requested_ = true;

wake_active_provider_retry();

queue_cv_.notify_one();

if (worker_thread_.joinable()) {

worker_thread_.join();

}

join_side_question_threads();

}

void AgentLoop::set_active_provider_for_retry(

const std::shared_ptr<LlmProvider>& provider) {

std::lock_guard<std::mutex> lock(active_provider_mu_);

active_provider_ = provider;

}

void AgentLoop::clear_active_provider_for_retry(

const std::shared_ptr<LlmProvider>& provider) {

std::lock_guard<std::mutex> lock(active_provider_mu_);

auto active = active_provider_.lock();

if (!active || active == provider) {

active_provider_.reset();

}

}

void AgentLoop::wake_active_provider_retry() {

std::shared_ptr<LlmProvider> provider;

{

std::lock_guard<std::mutex> lock(active_provider_mu_);

provider = active_provider_.lock();

}

if (provider) provider->wake_retry_waiter();

}

void AgentLoop::join_side_question_threads() {

std::vector<std::thread> threads;

{

std::lock_guard<std::mutex> lk(side_question_threads_mu_);

threads.swap(side_question_threads_);

}

for (auto& thread : threads) {

if (thread.joinable()) thread.join();

}

}

void AgentLoop::set_callbacks(AgentCallbacks cb) {

callbacks_ = std::move(cb);

}

void AgentLoop::worker_main() {

while (true) {

WorkerTask task;

{

std::unique_lock<std::mutex> lk(queue_mu_);

queue_cv_.wait(lk, [this] {

return !priority_task_queue_.empty() ||

!task_queue_.empty() || shutdown_requested_;

});

if (shutdown_requested_) return;

if (!priority_task_queue_.empty()) {

task = std::move(priority_task_queue_.front());

priority_task_queue_.pop();

} else {

task = std::move(task_queue_.front());

task_queue_.pop();

}

worker_task_active_ = true;

worker_task_kind_ = task.kind;

}

switch (task.kind) {

case WorkerTask::Kind::Chat:

if (task.input.empty() && !task.payload.empty()) {

task.input.text = std::move(task.payload);

task.input.display_text = std::move(task.display_text);

}

run_agent_with_input(task.input, task.hidden_goal_context);

break;

case WorkerTask::Kind::Shell:

run_shell(task.payload);

break;

case WorkerTask::Kind::Compact:

run_compact();

break;

case WorkerTask::Kind::Control:

if (task.control) task.control();

break;

}

{

std::lock_guard<std::mutex> lk(queue_mu_);

worker_task_active_ = false;

worker_task_kind_ = WorkerTask::Kind::Control;

}

}

}

void AgentLoop::submit(const std::string& user_message) {

submit(user_message, std::string{});

}

void AgentLoop::submit(const std::string& prompt, const std::string& display_text) {

UserInput input;

input.text = prompt;

input.display_text = display_text;

submit(input);

}

void AgentLoop::submit(const UserInput& input) {

clear_stale_abort_request();

{

std::lock_guard<std::mutex> lk(queue_mu_);

WorkerTask task;

task.kind = WorkerTask::Kind::Chat;

task.input = input;

task.hidden_goal_context = false;

task_queue_.push(std::move(task));

}

queue_cv_.notify_one();

}

ControlEnqueueReceipt AgentLoop::enqueue_control(

std::function<bool()> control) {

ControlEnqueueReceipt receipt;

if (!control) return receipt;

auto execution = std::make_shared<ControlExecutionState>();

{

std::lock_guard<std::mutex> lk(queue_mu_);

if (shutdown_requested_) return receipt;

auto is_turn_task = [](WorkerTask::Kind kind) {

return kind == WorkerTask::Kind::Chat ||

kind == WorkerTask::Kind::Shell ||

kind == WorkerTask::Kind::Compact;

};

bool queued_behind_turn =

worker_task_active_ && is_turn_task(worker_task_kind_);

auto urgent = priority_task_queue_;

while (!queued_behind_turn && !urgent.empty()) {

queued_behind_turn = is_turn_task(urgent.front().kind);

urgent.pop();

}

auto ordinary = task_queue_;

while (!queued_behind_turn && !ordinary.empty()) {

queued_behind_turn = is_turn_task(ordinary.front().kind);

ordinary.pop();

}

WorkerTask task;

task.kind = WorkerTask::Kind::Control;

task.control = [control = std::move(control), execution]() mutable {

bool succeeded = false;

try {

succeeded = control();

} catch (const std::exception& e) {

LOG_ERROR(std::string("Control task failed: ") + e.what());

} catch (...) {

LOG_ERROR("Control task failed with unknown exception");

}

{

std::lock_guard<std::mutex> lock(execution->mu);

execution->succeeded = succeeded;

execution->completed = true;

}

execution->cv.notify_all();

};

task_queue_.push(std::move(task));

receipt.sequence = ++next_control_sequence_;

receipt.accepted = true;

receipt.queued_behind_turn = queued_behind_turn;

receipt.execution = std::move(execution);

}

queue_cv_.notify_one();

return receipt;

}

TurnSteerResult AgentLoop::steer_input(

const std::string& expected_turn_id,

const UserInput& input) {

if (expected_turn_id.empty()) {

return {

TurnSteerStatus::InvalidInput,

{},

"expected turn id is required",

};

}

if (!has_meaningful_user_input(input)) {

return {

TurnSteerStatus::InvalidInput,

{},

"steering input is empty",

};

}

std::lock_guard<std::mutex> lk(active_turn_mu_);

if (!active_turn_accepting_ || active_turn_id_.empty()) {

return {

busy_.load()

? TurnSteerStatus::NonSteerable

: TurnSteerStatus::NoActiveTurn,

{},

busy_.load()

? "the busy operation is not steerable"

: "no active turn",

};

}

if (expected_turn_id != active_turn_id_) {

return {

TurnSteerStatus::TurnMismatch,

active_turn_id_,

"expected turn does not match the active turn",

};

}

if (pending_turn_inputs_.size() >= kMaxPendingTurnSteers) {

return {

TurnSteerStatus::QueueFull,

active_turn_id_,

"active turn steering queue is full",

};

}

pending_turn_inputs_.push_back(input);

return {

TurnSteerStatus::Accepted,

active_turn_id_,

"accepted",

};

}

TurnSteerResult AgentLoop::interrupt_turn(

const std::string& expected_turn_id,

const UserInput& input) {

if (expected_turn_id.empty()) {

return {

TurnSteerStatus::InvalidInput,

{},

"expected turn id is required",

};

}

if (!has_meaningful_user_input(input)) {

return {

TurnSteerStatus::InvalidInput,

{},

"steering input is empty",

};

}

std::string interrupted_turn_id;

std::size_t promised_inputs = 0;

{

// Lock order is intentionally active_turn_mu_ -> queue_mu_. No worker

// path holds queue_mu_ while acquiring active_turn_mu_.

std::lock_guard<std::mutex> turn_lk(active_turn_mu_);

if (!active_turn_accepting_ || active_turn_id_.empty()) {

return {

busy_.load()

? TurnSteerStatus::NonSteerable

: TurnSteerStatus::NoActiveTurn,

{},

busy_.load()

? "the busy operation is not steerable"

: "no active turn",

};

}

if (expected_turn_id != active_turn_id_) {

return {

TurnSteerStatus::TurnMismatch,

active_turn_id_,

"expected turn does not match the active turn",

};

}

interrupted_turn_id = active_turn_id_;

const std::size_t input_count = pending_turn_inputs_.size() + 1;

std::lock_guard<std::mutex> queue_lk(queue_mu_);

if (priority_task_queue_.size() + input_count >

kMaxPendingTurnSteers) {

return {

TurnSteerStatus::QueueFull,

interrupted_turn_id,

"interrupting turn queue is full",

};

}

auto promise_follow_up = [&](UserInput follow_up) {

if (!follow_up.metadata.is_object()) {

follow_up.metadata = nlohmann::json::object();

}

follow_up.metadata["turn_interrupt"] = true;

follow_up.metadata["interrupted_turn_id"] = interrupted_turn_id;