From 4d189596e354f9be2dc0e1a970980dbeff3c530c Mon Sep 17 00:00:00 2001 From: ULookup Date: Fri, 19 Jun 2026 18:14:03 +0000 Subject: [PATCH 1/5] fix: JSON-RPC reader race condition and 30s timeout Add read_mutex_ to serialize concurrent read_message calls from std::async tasks sharing the same stdout pipe. Add 30s timeout wrapper via wait_for to prevent indefinite blocking if the child process hangs. --- libs/core/include/merak/json_rpc_client.hpp | 1 + libs/core/src/json_rpc_client.cpp | 46 +++++++++++++-------- 2 files changed, 30 insertions(+), 17 deletions(-) diff --git a/libs/core/include/merak/json_rpc_client.hpp b/libs/core/include/merak/json_rpc_client.hpp index 530e148c..6076b4ab 100644 --- a/libs/core/include/merak/json_rpc_client.hpp +++ b/libs/core/include/merak/json_rpc_client.hpp @@ -47,6 +47,7 @@ class JsonRpcClient { std::atomic next_id_{1}; std::mutex io_mutex_; + std::mutex read_mutex_; // serializes concurrent read_message calls }; } // namespace merak diff --git a/libs/core/src/json_rpc_client.cpp b/libs/core/src/json_rpc_client.cpp index 14fcb44c..fc58439b 100644 --- a/libs/core/src/json_rpc_client.cpp +++ b/libs/core/src/json_rpc_client.cpp @@ -6,6 +6,7 @@ #include #include +#include #include #include #include @@ -109,6 +110,8 @@ void JsonRpcClient::write_message(const nlohmann::json& msg) { } nlohmann::json JsonRpcClient::read_message() { + std::lock_guard lock(read_mutex_); + // Use a FILE* wrapper around the fd so we can use getline() conveniently. // dup the fd first so closing the FILE* does not close the underlying fd. int fd = dup(stdout_fd_); @@ -169,25 +172,34 @@ std::future JsonRpcClient::send_request( write_message(req); return std::async(std::launch::async, [this, id]() -> nlohmann::json { - nlohmann::json resp = read_message(); - - // Wait for a response matching our id (retry a few times in case - // stray notifications arrive first). - constexpr int kMaxRetries = 10; - for (int i = 0; i < kMaxRetries; ++i) { - if (resp.contains("id") && resp["id"] == id) { - return resp; + // Inner async: the actual read+retry loop (now serialized via read_mutex_) + auto read_fut = std::async(std::launch::async, [this, id]() -> nlohmann::json { + nlohmann::json resp = read_message(); + + constexpr int kMaxRetries = 10; + for (int i = 0; i < kMaxRetries; ++i) { + if (resp.contains("id") && resp["id"] == id) { + return resp; + } + resp = read_message(); } - // Read another message (may be a notification without id, etc.) - resp = read_message(); - } - // If we still can't find our id, return an error object - nlohmann::json err; - err["jsonrpc"] = "2.0"; - err["id"] = id; - err["error"] = {{"code", -1}, {"message", "no matching response received"}}; - return err; + nlohmann::json err; + err["jsonrpc"] = "2.0"; + err["id"] = id; + err["error"] = {{"code", -1}, {"message", "no matching response received"}}; + return err; + }); + + // 30s timeout — prevent indefinite blocking on hung child process + if (read_fut.wait_for(std::chrono::seconds(30)) == std::future_status::timeout) { + nlohmann::json err; + err["jsonrpc"] = "2.0"; + err["id"] = id; + err["error"] = {{"code", -1}, {"message", "read timeout"}}; + return err; + } + return read_fut.get(); }); } From 05ee1193f204f9cf372ac87adf58c3246b59980c Mon Sep 17 00:00:00 2001 From: ULookup Date: Fri, 19 Jun 2026 18:17:51 +0000 Subject: [PATCH 2/5] fix: AgentLoop lifetime safety and checkpoint data serialization - Change run()/resume() return type from future to direct AgentResponse (all callers immediately .get(), async added no benefit) - Delete move constructor/assignment (AgentLoop holds this in async paths) - Track last compaction summary for checkpoint recovery - Serialize actual tool calls and pipeline stats in checkpoint instead of hardcoded empty strings --- libs/loop/include/merak/agent_loop.hpp | 12 ++++- libs/loop/src/agent_loop.cpp | 75 ++++++++++++++------------ 2 files changed, 51 insertions(+), 36 deletions(-) diff --git a/libs/loop/include/merak/agent_loop.hpp b/libs/loop/include/merak/agent_loop.hpp index df4ff2ea..73163298 100644 --- a/libs/loop/include/merak/agent_loop.hpp +++ b/libs/loop/include/merak/agent_loop.hpp @@ -57,14 +57,19 @@ class AgentLoop { // When not set (default), working_memory_text returns empty string. void set_working_memory_provider(std::function provider); + AgentLoop(const AgentLoop&) = delete; + AgentLoop& operator=(const AgentLoop&) = delete; + AgentLoop(AgentLoop&&) = delete; + AgentLoop& operator=(AgentLoop&&) = delete; + // Process a user message. Appends user msg to session_history_, // then enters the ReAct loop. Returns final response. - std::future run( + AgentResponse run( const std::string& user_message, RunControl& control); // Resume the ReAct loop without appending a new user message. - std::future resume(RunControl& control); + AgentResponse resume(RunControl& control); TurnState current_state() const { return state_; } std::shared_ptr tools() { return tools_; } @@ -110,6 +115,9 @@ class AgentLoop { int consecutive_content_avoidance_ = 0; int current_turn_ = 0; + // H2: last compaction summary text for checkpoint recovery + std::string last_compaction_text_; + std::vector restricted_tools_; void transition_to(TurnState next, RunControl& control); diff --git a/libs/loop/src/agent_loop.cpp b/libs/loop/src/agent_loop.cpp index ebb297d5..3ff82440 100644 --- a/libs/loop/src/agent_loop.cpp +++ b/libs/loop/src/agent_loop.cpp @@ -48,42 +48,34 @@ void AgentLoop::transition_to(TurnState next, RunControl& control) { spdlog::debug("Loop: {} → {}", state_name(prev), state_name(next)); } -std::future AgentLoop::run( +AgentResponse AgentLoop::run( const std::string& user_message, RunControl& control) { - return std::async(std::launch::async, - [this, user_message, &control]() -> AgentResponse { - - Message user_msg; - user_msg.role = "user"; - user_msg.content = user_message; - session_history_.push_back(user_msg); - memory_->append_message(user_msg); - control.append_message(user_msg); - - tool_failure_streak_.clear(); - turn_guard_.reset(); - stall_detector_.reset(); - consecutive_read_only_rounds_ = 0; - consecutive_world_query_rounds_ = 0; - consecutive_content_avoidance_ = 0; - return run_loop(control); - }); + Message user_msg; + user_msg.role = "user"; + user_msg.content = user_message; + session_history_.push_back(user_msg); + memory_->append_message(user_msg); + control.append_message(user_msg); + + tool_failure_streak_.clear(); + turn_guard_.reset(); + stall_detector_.reset(); + consecutive_read_only_rounds_ = 0; + consecutive_world_query_rounds_ = 0; + consecutive_content_avoidance_ = 0; + last_compaction_text_.clear(); + + return run_loop(control); } -std::future AgentLoop::resume(RunControl& control) { - return std::async(std::launch::async, - [this, &control]() -> AgentResponse { - // Detector state is cleared on resume (fresh turn starts). - // Counter fields (consecutive_read_only_rounds_ etc.) are - // intentionally preserved — they track behavior across the - // full session, not per-run. - tool_failure_streak_.clear(); - turn_guard_.reset(); - stall_detector_.reset(); - return run_loop(control); - }); +AgentResponse AgentLoop::resume(RunControl& control) { + tool_failure_streak_.clear(); + turn_guard_.reset(); + stall_detector_.reset(); + last_compaction_text_.clear(); + return run_loop(control); } AgentResponse AgentLoop::run_loop(RunControl& control) { @@ -319,14 +311,28 @@ AgentResponse AgentLoop::run_loop(RunControl& control) { nlohmann::json ts; ts["state"] = state_name(state_); ts["turn_count"] = turn_count; + + nlohmann::json pending = nlohmann::json::array(); + for (auto& tc : accumulated_tool_calls) { + pending.push_back({ + {"id", tc.id}, + {"name", tc.name}, + {"arguments", tc.arguments} + }); + } + + nlohmann::json pipeline_snapshot; + pipeline_snapshot["turn"] = turn_count; + pipeline_snapshot["cache_hit_ratio"] = pipeline_->stats().cache_hit_ratio(); + control.save_checkpoint( current_turn_, ts.dump(), llm_response.total_input_tokens, llm_response.total_output_tokens, - "[]", - "", - ""); + pending.dump(), + last_compaction_text_, + pipeline_snapshot.dump()); } // Set total tool output chars after knowing actual results @@ -653,6 +659,7 @@ void AgentLoop::maybe_compact(RunControl& control) { int keep_recent = config_.max_turns * 2; auto result = compactor_->compact_history(session_history_, keep_recent).get(); if (!result.summary.empty()) { + last_compaction_text_ = result.summary; Message summary_msg; summary_msg.role = "system"; summary_msg.content = "[Previous conversation summary]\n" + result.summary; From 1ef3bdbc5dace7801290e47862184e7279f38aeb Mon Sep 17 00:00:00 2001 From: ULookup Date: Fri, 19 Jun 2026 18:17:53 +0000 Subject: [PATCH 3/5] fix: remove .get() calls after AgentLoop::run/resume sync change --- libs/app/src/application.cpp | 2 +- libs/loop/src/sub_agent_runner.cpp | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/libs/app/src/application.cpp b/libs/app/src/application.cpp index 8e6859b6..fb0a1a15 100644 --- a/libs/app/src/application.cpp +++ b/libs/app/src/application.cpp @@ -384,7 +384,7 @@ AgentResponse Application::execute_sub_run( if (!context.scene_id.empty()) loop->set_active_scene_id(context.scene_id); if (!context.caller_agent_id.empty()) loop->set_caller_agent_id(context.caller_agent_id); - return loop->run(task, control).get(); + return loop->run(task, control); } // ───────────────────────────────────────────────────── diff --git a/libs/loop/src/sub_agent_runner.cpp b/libs/loop/src/sub_agent_runner.cpp index 701d3e17..16fc6ec3 100644 --- a/libs/loop/src/sub_agent_runner.cpp +++ b/libs/loop/src/sub_agent_runner.cpp @@ -62,7 +62,7 @@ std::future SubAgentRunner::delegate( spdlog::info("SubAgentRunner: delegating '{}' to '{}'", task.substr(0, 30), agent_id); NullRunControl control; - return sub->run(task, control).get(); + return sub->run(task, control); }); } From fe7c555eeab3828abf664a7878f28491aac7927e Mon Sep 17 00:00:00 2001 From: ULookup Date: Fri, 19 Jun 2026 18:17:57 +0000 Subject: [PATCH 4/5] fix: default-initialize PlanInput::on_escalate to no-op Prevents std::bad_function_call when callers invoke without checking. Existing callers already guard with operator bool(); this is defense-in-depth. --- libs/core/include/merak/pipeline_types.hpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libs/core/include/merak/pipeline_types.hpp b/libs/core/include/merak/pipeline_types.hpp index 11ce9b7b..0b0662c3 100644 --- a/libs/core/include/merak/pipeline_types.hpp +++ b/libs/core/include/merak/pipeline_types.hpp @@ -95,7 +95,7 @@ struct PlanInput { int model_max; int schema_count; double avg_schema_tokens; - std::function on_escalate; + std::function on_escalate = []{}; }; struct BoundContext { From 2fbd91a8fc693137ec66b7aa96e8bab3265c683e Mon Sep 17 00:00:00 2001 From: ULookup Date: Fri, 19 Jun 2026 18:19:04 +0000 Subject: [PATCH 5/5] fix: explicit dependency injection for sub-agent executor Replace [this] capture in init_runtime with explicit SubRunDeps struct, matching the PipelineManager::Dependencies DI pattern. execute_sub_run is now a static method taking explicit dependencies. This eliminates the hidden lifetime dependency on Application member declaration order. --- libs/app/include/merak/app/application.hpp | 18 ++++++- libs/app/src/application.cpp | 57 ++++++++++++++-------- 2 files changed, 54 insertions(+), 21 deletions(-) diff --git a/libs/app/include/merak/app/application.hpp b/libs/app/include/merak/app/application.hpp index df3a821e..982a37d5 100644 --- a/libs/app/include/merak/app/application.hpp +++ b/libs/app/include/merak/app/application.hpp @@ -28,6 +28,7 @@ class PortableNeo4j; class WorldbuildingHttpHandler; namespace kg { class KnowledgeGraphProvider; } namespace skills { class SkillRegistry; } +struct AgentRunContext; } namespace merak::app { @@ -80,8 +81,23 @@ class Application { void init_image_service(); // Factory helpers + + // Bundled dependencies for sub-agent execution (DI pattern, cf. PipelineManager::Dependencies) + struct SubRunDeps { + Config config; + std::shared_ptr llm; + std::shared_ptr tools; + std::shared_ptr memory; + std::shared_ptr compactor; + std::shared_ptr wb_service; + std::shared_ptr skill_registry; + std::shared_ptr> plan_mode; + std::shared_ptr context_assembler; + }; + AgentLoop::Config make_loop_config(const std::string& model) const; - AgentResponse execute_sub_run( + static AgentResponse execute_sub_run( + const SubRunDeps& deps, const SubAgentConfig& agent, const std::string& task, RunControl& control, const AgentRunContext& context); diff --git a/libs/app/src/application.cpp b/libs/app/src/application.cpp index fb0a1a15..e23c849a 100644 --- a/libs/app/src/application.cpp +++ b/libs/app/src/application.cpp @@ -19,6 +19,7 @@ #include #include #include +#include #include #include #include @@ -341,45 +342,50 @@ void Application::init_skills() { // make_loop_config / execute_sub_run — factory helpers // ───────────────────────────────────────────────────── -AgentLoop::Config Application::make_loop_config(const std::string& model) const { +static AgentLoop::Config make_loop_config_impl(const std::string& model, const Config& config) { AgentLoop::Config c; - c.system_prompt = config_.agent.system_prompt; - c.max_turns = config_.agent.max_tool_turns; - c.default_model = model.empty() ? config_.llm.default_model : model; - c.max_output_tokens = config_.llm.max_output_tokens; + c.system_prompt = config.agent.system_prompt; + c.max_turns = config.agent.max_tool_turns; + c.default_model = model.empty() ? config.llm.default_model : model; + c.max_output_tokens = config.llm.max_output_tokens; return c; } +AgentLoop::Config Application::make_loop_config(const std::string& model) const { + return make_loop_config_impl(model, config_); +} + AgentResponse Application::execute_sub_run( + const SubRunDeps& deps, const SubAgentConfig& agent, const std::string& task, RunControl& control, const AgentRunContext& context) { auto sub_tools = std::make_shared(); - sub_tools->set_permission_mode(config_.agent.permission_mode); + sub_tools->set_permission_mode(deps.config.agent.permission_mode); if (agent.tool_allowlist.empty()) { - for (auto& spec : tools_->all_tools()) { - if (auto* t = tools_->get_tool(spec.name)) + for (auto& spec : deps.tools->all_tools()) { + if (auto* t = deps.tools->get_tool(spec.name)) sub_tools->register_tool(t->clone()); } } else { for (auto& name : agent.tool_allowlist) { - if (auto* t = tools_->get_tool(name)) + if (auto* t = deps.tools->get_tool(name)) sub_tools->register_tool(t->clone()); } } auto counter = std::make_shared(); - auto sub_compactor = std::make_shared(llm_, counter); + auto sub_compactor = std::make_shared(deps.llm, counter); - auto cfg = make_loop_config(agent.model); + auto cfg = make_loop_config_impl(agent.model, deps.config); cfg.system_prompt = agent.system_prompt.empty() - ? config_.agent.system_prompt : agent.system_prompt; + ? deps.config.agent.system_prompt : agent.system_prompt; auto loop = std::make_unique( - cfg, llm_, sub_tools, memory_, sub_compactor, - wb_service_, skill_registry_); - loop->set_plan_mode_source(plan_mode_); + cfg, deps.llm, sub_tools, deps.memory, sub_compactor, + deps.wb_service, deps.skill_registry); + loop->set_plan_mode_source(deps.plan_mode); if (!context.world_id.empty()) loop->set_active_world_id(context.world_id); if (!context.scene_id.empty()) loop->set_active_scene_id(context.scene_id); if (!context.caller_agent_id.empty()) loop->set_caller_agent_id(context.caller_agent_id); @@ -392,14 +398,25 @@ AgentResponse Application::execute_sub_run( // ───────────────────────────────────────────────────── void Application::init_runtime() { - // Sub-executor delegates to the canonical execute_sub_run method. - // Capturing 'this' is safe because tools_ (which stores AgentTool holding - // this lambda) is destroyed before Application members during destruction. - auto sub_executor = [this](const SubAgentConfig& agent, const std::string& task, + // Bundled dependencies for sub-agent execution. + // Captured by value — no [this], safe regardless of destruction order. + SubRunDeps deps{ + config_, + llm_, + tools_, + memory_, + compactor_, + wb_service_, + skill_registry_, + plan_mode_, + context_assembler_ + }; + + auto sub_executor = [deps](const SubAgentConfig& agent, const std::string& task, RunControl& control, const AgentRunContext& context) -> AgentResponse { - return execute_sub_run(agent, task, control, context); + return execute_sub_run(deps, agent, task, control, context); }; // Register AgentTool before creating RuntimeService (SubExecutor is