diff --git a/backend/src/agents/memory/updater.py b/backend/src/agents/memory/updater.py index 21e7ee4e9ad..ec6300721a5 100644 --- a/backend/src/agents/memory/updater.py +++ b/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, diff --git a/backend/src/gateway/routers/uploads.py b/backend/src/gateway/routers/uploads.py index cf2a724841c..245024eb50b 100644 --- a/backend/src/gateway/routers/uploads.py +++ b/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,6 +74,33 @@ 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, @@ -80,8 +108,11 @@ async def upload_files( ) -> 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) diff --git a/backend/src/subagents/executor.py b/backend/src/subagents/executor.py index c0d182442e7..de8ed6778b9 100644 --- a/backend/src/subagents/executor.py +++ b/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") diff --git a/config.example.yaml b/config.example.yaml index ec77c6b2b5f..0fd71af48bc 100644 --- a/config.example.yaml +++ b/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