Cherry-pick three high-value upstream fixes

backend/src/agents/memory/updater.py

@@ -1,7 +1,9 @@
 """Memory updater for reading, writing, and updating memory data."""
 
+import copy
 import json
 import re
+import threading
 import uuid
 from datetime import datetime, timezone
 from pathlib import Path
@@ -59,6 +61,9 @@ def _create_empty_memory() -> dict[str, Any]:
 # Per-agent memory cache: keyed by agent_name (None = global)
 # Value: (memory_data, file_mtime)
 _memory_cache: dict[str | None, tuple[dict[str, Any], float | None]] = {}
+# Guards every read and write of _memory_cache. The background memory-update
+# timer thread and HTTP reload calls can race on this dict otherwise.
+_cache_lock = threading.Lock()
 
 
 def get_memory_data(agent_name: str | None = None) -> dict[str, Any]:
@@ -81,15 +86,17 @@ def get_memory_data(agent_name: str | None = None) -> dict[str, Any]:
     except OSError:
         current_mtime = None
 
-    cached = _memory_cache.get(agent_name)
+    with _cache_lock:
+        cached = _memory_cache.get(agent_name)
+        if cached is not None and cached[1] == current_mtime:
+            return cached[0]
 
-    # Invalidate cache if file has been modified or doesn't exist
-    if cached is None or cached[1] != current_mtime:
-        memory_data = _load_memory_from_file(agent_name)
+    memory_data = _load_memory_from_file(agent_name)
+
+    with _cache_lock:
         _memory_cache[agent_name] = (memory_data, current_mtime)
-        return memory_data
 
-    return cached[0]
+    return memory_data
 
 
 def reload_memory_data(agent_name: str | None = None) -> dict[str, Any]:
@@ -109,7 +116,8 @@ def reload_memory_data(agent_name: str | None = None) -> dict[str, Any]:
     except OSError:
         mtime = None
 
-    _memory_cache[agent_name] = (memory_data, mtime)
+    with _cache_lock:
+        _memory_cache[agent_name] = (memory_data, mtime)
     return memory_data
 
 
@@ -189,8 +197,10 @@ def _save_memory_to_file(memory_data: dict[str, Any], agent_name: str | None = N
         # Ensure directory exists
         file_path.parent.mkdir(parents=True, exist_ok=True)
 
-        # Update lastUpdated timestamp
-        memory_data["lastUpdated"] = datetime.now(timezone.utc).isoformat()
+        # Shallow-copy before adding lastUpdated so the caller's dict is not
+        # mutated as a side-effect, and the cache reference is not silently
+        # updated before the file write succeeds.
+        memory_data = {**memory_data, "lastUpdated": datetime.now(timezone.utc).isoformat()}
 
         # Write atomically using temp file
         temp_path = file_path.with_suffix(".tmp")
@@ -206,7 +216,8 @@ def _save_memory_to_file(memory_data: dict[str, Any], agent_name: str | None = N
         except OSError:
             mtime = None
 
-        _memory_cache[agent_name] = (memory_data, mtime)
+        with _cache_lock:
+            _memory_cache[agent_name] = (memory_data, mtime)
 
         print(f"Memory saved to {file_path}")
         return True
@@ -291,8 +302,9 @@ def update_memory(self, messages: list[Any], thread_id: str | None = None, agent
 
             update_data = json.loads(response_text)
 
-            # Apply updates
-            updated_memory = self._apply_updates(current_memory, update_data, thread_id)
+            # Deep-copy before in-place mutation so a subsequent save() failure
+            # cannot corrupt the still-cached original object reference.
+            updated_memory = self._apply_updates(copy.deepcopy(current_memory), update_data, thread_id)
 
             # Strip file-upload mentions from all summaries before saving.
             # Uploaded files are session-scoped and won't exist in future sessions,

backend/src/gateway/routers/uploads.py

@@ -6,6 +6,7 @@
 from fastapi import APIRouter, File, HTTPException, UploadFile
 from pydantic import BaseModel
 
+from src.config.app_config import get_app_config
 from src.config.paths import VIRTUAL_PATH_PREFIX, get_paths
 from src.sandbox.sandbox_provider import get_sandbox_provider
 
@@ -73,15 +74,45 @@ async def convert_file_to_markdown(file_path: Path) -> Path | None:
         return None
 
 
+def _get_uploads_config_value(key: str, default: object) -> object:
+    """Read a value from the uploads config, supporting dict and attribute access."""
+    cfg = get_app_config()
+    uploads_cfg = getattr(cfg, "uploads", None)
+    if isinstance(uploads_cfg, dict):
+        return uploads_cfg.get(key, default)
+    return getattr(uploads_cfg, key, default)
+
+
+def _auto_convert_documents_enabled() -> bool:
+    """Return whether automatic host-side document conversion is enabled.
+
+    The secure default is disabled unless an operator explicitly opts in via
+    ``uploads.auto_convert_documents`` in config.yaml. Without this gate, any
+    user-uploaded document is fed through markitdown on the host process —
+    that is a meaningful sandbox-escape surface (parser bugs in PDF/Office
+    handling can be exploited via a crafted upload).
+    """
+    try:
+        raw = _get_uploads_config_value("auto_convert_documents", False)
+        if isinstance(raw, str):
+            return raw.strip().lower() in {"1", "true", "yes", "on"}
+        return bool(raw)
+    except Exception:
+        return False
+
+
 @router.post("", response_model=UploadResponse)
 async def upload_files(
     thread_id: str,
     files: list[UploadFile] = File(...),
 ) -> UploadResponse:
     """Upload multiple files to a thread's uploads directory.
 
-    For PDF, PPT, Excel, and Word files, they will be converted to markdown using markitdown.
-    All files (original and converted) are saved to /mnt/user-data/uploads.
+    For PDF, PPT, Excel, and Word files, they may be converted to markdown using
+    markitdown — but only when ``uploads.auto_convert_documents: true`` is set in
+    config.yaml. The default is disabled (the host-side parser surface is a
+    sandbox-escape risk if exposed to untrusted uploads).
+    All files (original and any converted output) are saved to /mnt/user-data/uploads.
 
     Args:
         thread_id: The thread ID to upload files to.
@@ -100,6 +131,7 @@ async def upload_files(
     sandbox_provider = get_sandbox_provider()
     sandbox_id = sandbox_provider.acquire(thread_id)
     sandbox = sandbox_provider.get(sandbox_id)
+    auto_convert_documents = _auto_convert_documents_enabled()
 
     for file in files:
         if not file.filename:
@@ -135,9 +167,12 @@ async def upload_files(
 
             logger.info(f"Saved file: {safe_filename} ({len(content)} bytes) to {relative_path}")
 
-            # Check if file should be converted to markdown
+            # Check if file should be converted to markdown.
+            # Conversion is gated behind uploads.auto_convert_documents in config.yaml
+            # because feeding user-supplied PDFs/Office docs through markitdown on
+            # the host process is a sandbox-escape surface (CVE-class).
             file_ext = file_path.suffix.lower()
-            if file_ext in CONVERTIBLE_EXTENSIONS:
+            if auto_convert_documents and file_ext in CONVERTIBLE_EXTENSIONS:
                 md_path = await convert_file_to_markdown(file_path)
                 if md_path:
                     md_relative_path = str(paths.sandbox_uploads_dir(thread_id) / md_path.name)

backend/src/subagents/executor.py

@@ -70,6 +70,8 @@ def __post_init__(self):
 # Thread pools — lazily initialized on first use so config.yaml values are available
 _scheduler_pool: ThreadPoolExecutor | None = None
 _execution_pool: ThreadPoolExecutor | None = None
+# Dedicated pool for sync execute() calls made from an already-running event loop.
+_isolated_loop_pool: ThreadPoolExecutor | None = None
 _pool_lock = threading.Lock()
 
 
@@ -101,6 +103,29 @@ def get_execution_pool() -> ThreadPoolExecutor:
     return _execution_pool
 
 
+def get_isolated_loop_pool() -> ThreadPoolExecutor:
+    """Return the isolated-loop thread pool, creating it on first use.
+
+    Used when SubagentExecutor.execute() is called from inside an already-running
+    event loop (e.g. an async parent agent). Running asyncio.run() in that
+    situation creates a nested loop that conflicts with asyncio primitives
+    bound to the parent loop (httpx clients, etc.). Submitting the work to a
+    dedicated thread with its own fresh loop sidesteps the conflict.
+    """
+    global _isolated_loop_pool
+    if _isolated_loop_pool is None:
+        with _pool_lock:
+            if _isolated_loop_pool is None:
+                # Reuse the execution pool size; the workload pattern is identical
+                # (one subagent call per worker, blocking until the inner loop completes).
+                from src.config.subagents_config import get_subagents_app_config
+
+                size = get_subagents_app_config().execution_pool_size
+                _isolated_loop_pool = ThreadPoolExecutor(max_workers=size, thread_name_prefix="subagent-isolated-")
+                logger.info(f"Isolated-loop pool initialized with {size} workers")
+    return _isolated_loop_pool
+
+
 def _filter_tools(
     all_tools: list[BaseTool],
     allowed: list[str] | None,
@@ -351,12 +376,57 @@ async def _aexecute(self, task: str, result_holder: SubagentResult | None = None
 
         return result
 
+    def _execute_in_isolated_loop(
+        self, task: str, result_holder: SubagentResult | None = None
+    ) -> SubagentResult:
+        """Execute the subagent in a completely fresh event loop.
+
+        This method is designed to run in a separate thread to ensure complete
+        isolation from any parent event loop, preventing conflicts with asyncio
+        primitives that may be bound to the parent loop (e.g., httpx clients).
+        """
+        try:
+            previous_loop = asyncio.get_event_loop()
+        except RuntimeError:
+            previous_loop = None
+
+        loop = asyncio.new_event_loop()
+        try:
+            asyncio.set_event_loop(loop)
+            return loop.run_until_complete(self._aexecute(task, result_holder))
+        finally:
+            try:
+                pending = asyncio.all_tasks(loop)
+                if pending:
+                    for task_obj in pending:
+                        task_obj.cancel()
+                    loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
+
+                loop.run_until_complete(loop.shutdown_asyncgens())
+                loop.run_until_complete(loop.shutdown_default_executor())
+            except Exception:
+                logger.debug(
+                    f"[trace={self.trace_id}] Failed while cleaning up isolated event loop "
+                    f"for subagent {self.config.name}",
+                    exc_info=True,
+                )
+            finally:
+                try:
+                    loop.close()
+                finally:
+                    asyncio.set_event_loop(previous_loop)
+
     def execute(self, task: str, result_holder: SubagentResult | None = None) -> SubagentResult:
         """Execute a task synchronously (wrapper around async execution).
 
         This method runs the async execution in a new event loop, allowing
         asynchronous tools (like MCP tools) to be used within the thread pool.
 
+        When called from within an already-running event loop (e.g., when the
+        parent agent is async), this method isolates the subagent execution in
+        a separate thread to avoid event loop conflicts with shared async
+        primitives like httpx clients.
+
         Args:
             task: The task description for the subagent.
             result_holder: Optional pre-created result object to update during execution.
@@ -374,6 +444,21 @@ def execute(self, task: str, result_holder: SubagentResult | None = None) -> Sub
         # an async context where an event loop already exists). Subagent execution
         # errors are handled within _aexecute() and returned as FAILED status.
         try:
+            try:
+                running_loop = asyncio.get_running_loop()
+            except RuntimeError:
+                running_loop = None
+
+            if running_loop is not None and running_loop.is_running():
+                logger.debug(
+                    f"[trace={self.trace_id}] Subagent {self.config.name} "
+                    f"detected running event loop, using isolated thread"
+                )
+                future = get_isolated_loop_pool().submit(
+                    self._execute_in_isolated_loop, task, result_holder
+                )
+                return future.result()
+
             return asyncio.run(self._aexecute(task, result_holder))
         except Exception as e:
             logger.exception(f"[trace={self.trace_id}] Subagent {self.config.name} execution failed")

config.example.yaml

@@ -444,3 +444,14 @@ checkpointer:
 #           context:
 #             thinking_enabled: true
 #             subagent_enabled: true
+
+# Uploads
+# ----------------------------------------------------------------------
+# Controls how user-uploaded files are processed by the host process.
+# uploads:
+#   # Convert PDF/PPT/Excel/Word uploads to Markdown using markitdown on the host.
+#   # SECURE DEFAULT: false. Enabling this exposes the markitdown parser stack
+#   # (and its transitive PDF/Office libraries) to untrusted user input — bugs in
+#   # those libraries become potential sandbox-escape vectors. Only enable on
+#   # deployments where uploads come from trusted users.
+#   auto_convert_documents: false