fix: Agent infrastructure critical issues (C3, C4, H1-H5)

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<LlmProvider> llm;
+        std::shared_ptr<ToolRegistry> tools;
+        std::shared_ptr<MemoryStore> memory;
+        std::shared_ptr<Compactor> compactor;
+        std::shared_ptr<worldbuilding::WorldbuildingService> wb_service;
+        std::shared_ptr<skills::SkillRegistry> skill_registry;
+        std::shared_ptr<std::atomic<bool>> plan_mode;
+        std::shared_ptr<ContextAssembler> 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);
 

libs/app/src/application.cpp

@@ -19,6 +19,7 @@
 #include <merak/worldbuilding/condition_evaluator.hpp>
 #include <merak/worldbuilding/ids.hpp>
 #include <merak/skills/skill_registry.hpp>
+#include <merak/runtime_service.hpp>
 #include <merak/storage/image_service.hpp>
 #include <merak/storage/local_file_image_store.hpp>
 #include <merak/tool_registry.hpp>
@@ -341,65 +342,81 @@ 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<ToolRegistry>();
-    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<TokenCounter>();
-    auto sub_compactor = std::make_shared<Compactor>(llm_, counter);
+    auto sub_compactor = std::make_shared<Compactor>(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<AgentLoop>(
-        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);
 
-    return loop->run(task, control).get();
+    return loop->run(task, control);
 }
 
 // ─────────────────────────────────────────────────────
 // init_runtime
 // ─────────────────────────────────────────────────────
 
 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

libs/core/include/merak/json_rpc_client.hpp

@@ -47,6 +47,7 @@ class JsonRpcClient {
 
     std::atomic<int> next_id_{1};
     std::mutex io_mutex_;
+    std::mutex read_mutex_;  // serializes concurrent read_message calls
 };
 
 } // namespace merak

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<void()> on_escalate;
+  std::function<void()> on_escalate = []{};
 };
 
 struct BoundContext {

libs/core/src/json_rpc_client.cpp

@@ -6,6 +6,7 @@
 #include <sys/wait.h>
 #include <signal.h>
 
+#include <chrono>
 #include <cstdio>
 #include <cstring>
 #include <memory>
@@ -109,6 +110,8 @@ void JsonRpcClient::write_message(const nlohmann::json& msg) {
 }
 
 nlohmann::json JsonRpcClient::read_message() {
+    std::lock_guard<std::mutex> 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<nlohmann::json> 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();
     });
 }
 

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<std::string()> 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<AgentResponse> run(
+    AgentResponse run(
         const std::string& user_message,
         RunControl& control);
 
     // Resume the ReAct loop without appending a new user message.
-    std::future<AgentResponse> resume(RunControl& control);
+    AgentResponse resume(RunControl& control);
 
     TurnState current_state() const { return state_; }
     std::shared_ptr<ToolRegistry> 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<std::string> restricted_tools_;
 
     void transition_to(TurnState next, RunControl& control);

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<AgentResponse> 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<AgentResponse> 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;

libs/loop/src/sub_agent_runner.cpp

@@ -62,7 +62,7 @@ std::future<AgentResponse> 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);
     });
 }