diff --git a/src/praisonai-agents/praisonaiagents/agent/context_agent.py b/src/praisonai-agents/praisonaiagents/agent/context_agent.py
index d4a5194cf..c1ca03c00 100644
--- a/src/praisonai-agents/praisonaiagents/agent/context_agent.py
+++ b/src/praisonai-agents/praisonaiagents/agent/context_agent.py
@@ -213,7 +213,8 @@ class ContextAgent(Agent):
             self.output_dir / "agent_responses",
             self.output_dir / "markdown_outputs", 
             self.output_dir / "debug_logs",
-            self.output_dir / "final_results"
+            self.output_dir / "final_results",
+            self.output_dir / "PRPs",
         ]
         
         for directory in directories:
@@ -1270,11 +1271,12 @@ Every agent interaction has been saved for full audit trail and reproducibility.
             return self._generate_combined_analysis(feature_request)
         
         # Otherwise just save and return the PRP
+        from ..utils.atomic_io import atomic_write_text, safe_join
+        prp_dir = os.path.join(self.output_dir, "PRPs")
+        os.makedirs(prp_dir, exist_ok=True)
         safe_filename = feature_request.lower().replace(' ', '_').replace('-', '_')[:50]
-        prp_path = os.path.join(self.output_dir, "PRPs", f"{safe_filename}.md")
-        
-        with open(prp_path, "w", encoding="utf-8") as f:
-            f.write(prp)
+        prp_path = safe_join(prp_dir, safe_filename, ".md")
+        atomic_write_text(prp_path, prp)
         
         print(f"📄 Comprehensive PRP following PRD template saved to: {prp_path}")
         return prp
diff --git a/src/praisonai-agents/praisonaiagents/planning/storage.py b/src/praisonai-agents/praisonaiagents/planning/storage.py
index 859126a5c..e4b32c88f 100644
--- a/src/praisonai-agents/praisonaiagents/planning/storage.py
+++ b/src/praisonai-agents/praisonaiagents/planning/storage.py
@@ -104,7 +104,12 @@ class PlanStorage:
         if filename is None:
             filename = self._get_plan_filename(plan)
             
-        path = os.path.join(self.plans_dir, filename)
+        from ..utils.atomic_io import safe_join
+        # Keep ".md" as the suffix so listing/lookup still find the file.
+        stem, ext = os.path.splitext(filename)
+        if ext.lower() != ".md":
+            stem, ext = filename, ".md"
+        path = safe_join(self.plans_dir, stem, ext)
         
         # Convert plan to markdown
         markdown = plan.to_markdown()
@@ -215,6 +220,29 @@ class PlanStorage:
             
         return False
     
+    def _todo_path(self, name: str) -> str:
+        """Confined path for a todo name, reusing a pre-existing legacy file.
+
+        Legacy files (``<name>.json`` written before Issue #6238) are reused
+        only when ``name`` is a plain filename, so upgrades keep working
+        without ever allowing traversal out of ``todos_dir``.
+        """
+        from ..utils.atomic_io import safe_join
+        path = safe_join(self.todos_dir, name, ".json")
+        name = str(name)
+        if (
+            not os.path.exists(path)
+            and name not in ("", ".", "..")
+            and "\0" not in name
+            and "%" not in name
+            and os.path.basename(name) == name
+            and not (os.altsep and os.altsep in name)
+        ):
+            legacy = os.path.join(self.todos_dir, f"{name}.json")
+            if os.path.isfile(legacy):
+                return legacy
+        return path
+
     def save_todo(self, todo: TodoList, name: str = "current") -> str:
         """
         Save a todo list to storage.
@@ -226,11 +254,10 @@ class PlanStorage:
         Returns:
             Path to the saved file
         """
-        filename = f"{name}.json"
-        path = os.path.join(self.todos_dir, filename)
+        from ..utils.atomic_io import atomic_write_text
+        path = self._todo_path(name)
 
         # Atomic write so a crash mid-write never truncates the previous todos.
-        from ..utils.atomic_io import atomic_write_text
         atomic_write_text(path, todo.to_json())
 
         logger.debug(f"Saved todo list to {path}")
@@ -246,8 +273,7 @@ class PlanStorage:
         Returns:
             TodoList instance, or None if not found
         """
-        filename = f"{name}.json"
-        path = os.path.join(self.todos_dir, filename)
+        path = self._todo_path(name)
         
         if not os.path.exists(path):
             return None
@@ -265,12 +291,13 @@ class PlanStorage:
         Returns:
             List of todo list names
         """
+        from urllib.parse import unquote
         todos = []
         
         for filename in os.listdir(self.todos_dir):
             if filename.endswith(".json"):
-                name = filename[:-5]  # Remove .json extension
-                todos.append(name)
+                # Decode so listed names round-trip through load/delete_todo.
+                todos.append(unquote(filename[:-5]))
                 
         return todos
     
@@ -284,8 +311,7 @@ class PlanStorage:
         Returns:
             True if deleted, False if not found
         """
-        filename = f"{name}.json"
-        path = os.path.join(self.todos_dir, filename)
+        path = self._todo_path(name)
         
         if os.path.exists(path):
             os.remove(path)
diff --git a/src/praisonai-agents/praisonaiagents/session/title.py b/src/praisonai-agents/praisonaiagents/session/title.py
index 483707a6d..ba5b93cf0 100644
--- a/src/praisonai-agents/praisonaiagents/session/title.py
+++ b/src/praisonai-agents/praisonaiagents/session/title.py
@@ -73,28 +73,15 @@ def generate_title(
     fallback_title = _create_fallback_title(user_msg, max_length)
     
     try:
-        # Check if we're already in an event loop
-        asyncio.get_running_loop()
-    except RuntimeError:
-        # No running loop - safe to use asyncio.run
-        try:
-            return asyncio.run(generate_title_async(
-                user_msg, assistant_msg, llm_model, timeout, max_length,
-                primary_model=primary_model,
-            ))
-        except Exception:
-            import logging
-            logger = logging.getLogger(__name__)
-            logger.debug("Title generation failed in sync context", exc_info=True)
-            return fallback_title
-    else:
-        # Already in event loop - cannot use asyncio.run
+        # Works with or without a running event loop (Issue #6238).
+        from ..utils.async_bridge import run_coroutine_from_any_context
+        return run_coroutine_from_any_context(generate_title_async(
+            user_msg, assistant_msg, llm_model, timeout, max_length,
+            primary_model=primary_model,
+        ), timeout=None)
+    except Exception:
         import logging
-        logger = logging.getLogger(__name__)
-        logger.warning(
-            "generate_title() called from running event loop; "
-            "use generate_title_async() instead. Returning fallback title."
-        )
+        logging.getLogger(__name__).debug("Title generation failed", exc_info=True)
         return fallback_title
 
 
diff --git a/src/praisonai-agents/praisonaiagents/tools/crawl4ai_tools.py b/src/praisonai-agents/praisonaiagents/tools/crawl4ai_tools.py
index 6e820cb86..2366d733b 100644
--- a/src/praisonai-agents/praisonaiagents/tools/crawl4ai_tools.py
+++ b/src/praisonai-agents/praisonaiagents/tools/crawl4ai_tools.py
@@ -645,9 +645,8 @@ def crawl4ai_sync(
     Returns:
         Dict with crawl results
     """
-    return asyncio.run(
-        crawl4ai(url, css_selector, js_code, wait_for)
-    )
+    from ..utils.async_bridge import run_coroutine_from_any_context
+    return run_coroutine_from_any_context(crawl4ai(url, css_selector, js_code, wait_for), timeout=None)
 
 def crawl4ai_extract_sync(
     url: str,
@@ -664,9 +663,8 @@ def crawl4ai_extract_sync(
     Returns:
         Dict with extracted data
     """
-    return asyncio.run(
-        crawl4ai_extract(url, schema, js_code)
-    )
+    from ..utils.async_bridge import run_coroutine_from_any_context
+    return run_coroutine_from_any_context(crawl4ai_extract(url, schema, js_code), timeout=None)
 
 if __name__ == "__main__":
     # Example usage
diff --git a/src/praisonai-agents/praisonaiagents/utils/atomic_io.py b/src/praisonai-agents/praisonaiagents/utils/atomic_io.py
index bd7d2df2f..ec4e3f216 100644
--- a/src/praisonai-agents/praisonaiagents/utils/atomic_io.py
+++ b/src/praisonai-agents/praisonaiagents/utils/atomic_io.py
@@ -9,8 +9,10 @@ the previous good file: the old bytes stay intact and ``os.replace`` swaps in th
 new content atomically on both POSIX and Windows.
 """
 
+import hashlib
 import json
 import os
+import re
 import tempfile
 import threading
 import time
@@ -21,9 +23,52 @@ __all__ = [
     "atomic_write_json",
     "atomic_write_text",
     "quarantine_corrupt_file",
+    "safe_join",
     "update_json",
 ]
 
+# Max UTF-8 bytes for the final basename; leaves room under the common
+# 255-byte limit for ``atomic_write_text``'s ``<basename>.XXXXXXXX.tmp`` file.
+_MAX_NAME_BYTES = 200
+_HASHED = re.compile(r"(.*)%~([0-9a-f]{16})", re.DOTALL)
+_ESCAPE = re.compile(r'[%/\\:*?"<>|\x00-\x1f\x7f]')
+
+
+def _escape(name: str) -> str:
+    stem = _ESCAPE.sub(lambda m: "%%%02X" % ord(m.group()), name)
+    return "%2E" + stem[1:] if stem.startswith(".") else stem
+
+
+def safe_join(base: str, name: str, suffix: str = "") -> str:
+    """Join a caller-supplied ``name`` under ``base`` without escaping it.
+
+    Only ``%``, path separators, control and Windows-reserved characters (and
+    a leading ``.``) are percent-encoded, so ordinary names - including spaces
+    and non-ASCII - keep their pre-#6238 filename, ``..`` and separators can
+    never leave ``base``, and ``urllib.parse.unquote`` recovers the name.
+    Names whose filename would exceed ``_MAX_NAME_BYTES`` are truncated and
+    tagged ``%~<sha256[:16]>``; that listed (unquoted) form maps back to the
+    same file, while no ordinary name can produce a raw ``%~`` (Issue #6238).
+    """
+    name = str(name)
+    hashed = _HASHED.fullmatch(name)
+    if hashed:
+        stem = _escape(hashed.group(1)) + "%~" + hashed.group(2)
+    else:
+        stem = _escape(name)
+    budget = _MAX_NAME_BYTES - len(suffix.encode("utf-8"))
+    if len(stem.encode("utf-8", "surrogatepass")) > budget:
+        tag = "%~" + hashlib.sha256(name.encode("utf-8", "surrogatepass")).hexdigest()[:16]
+        kept, used = [], len(tag)
+        for tok in re.findall(r"%[0-9A-F]{2}|.", _escape(name), re.DOTALL):
+            used += len(tok.encode("utf-8", "surrogatepass"))
+            if used > budget:
+                break
+            kept.append(tok)
+        stem = "".join(kept) + tag
+    return os.path.join(base, stem + suffix)
+
+
 # Per-path in-process mutexes so threads in one process serialise before they
 # even reach the cross-process ``FileLock``. Keyed by absolute path.
 _path_mutexes: "defaultdict[str, threading.RLock]" = defaultdict(threading.RLock)
diff --git a/src/praisonai-agents/tests/unit/test_accepted_and_ignored_config.py b/src/praisonai-agents/tests/unit/test_accepted_and_ignored_config.py
index 4a9ab0503..a3d75a339 100644
--- a/src/praisonai-agents/tests/unit/test_accepted_and_ignored_config.py
+++ b/src/praisonai-agents/tests/unit/test_accepted_and_ignored_config.py
@@ -34,11 +34,11 @@ def _restore_approval_state():
     from praisonaiagents.approval import get_approval_registry, clear_approval_context
 
     registry = get_approval_registry()
-    previous = registry.get_backend()
+    previous = registry._global_backend  # not get_backend(): its ConsoleBackend fallback would be installed globally on restore
     try:
         yield
     finally:
-        registry.set_backend(previous)
+        registry._global_backend = previous
         clear_approval_context()
 
 
@@ -154,7 +154,7 @@ def test_code_mode_changes_what_the_tool_can_do():
 
     _probe_tool_name = "code_mode_probe"
     get_registry().register(code_mode_probe)
-    previous = get_approval_registry().get_backend()
+    previous = get_approval_registry()._global_backend  # not get_backend(): its ConsoleBackend fallback would be installed globally on restore
     get_approval_registry().set_backend(AutoApproveBackend())
     try:
         safe = Agent(
@@ -182,7 +182,7 @@ def test_code_mode_changes_what_the_tool_can_do():
         safe_result = safe_exec(code)
         unsafe_result = unsafe_exec(code)
     finally:
-        get_approval_registry().set_backend(previous)
+        get_approval_registry()._global_backend = previous
         clear_approval_context()
         # The tool registry is process-global. Leaving these probe tools in it
         # changed what later tests saw when they enumerated or dispatched
@@ -221,7 +221,7 @@ def test_unsafe_code_mode_cannot_reach_tools_the_agent_was_not_granted():
 
     _probe_tool_name = "ungranted_global_tool"
     get_registry().register(ungranted_global_tool)
-    previous = get_approval_registry().get_backend()
+    previous = get_approval_registry()._global_backend  # not get_backend(): its ConsoleBackend fallback would be installed globally on restore
     get_approval_registry().set_backend(AutoApproveBackend())
     try:
         agent = Agent(
@@ -240,7 +240,7 @@ def test_unsafe_code_mode_cannot_reach_tools_the_agent_was_not_granted():
         )
         result = execute("print(ungranted_global_tool())")
     finally:
-        get_approval_registry().set_backend(previous)
+        get_approval_registry()._global_backend = previous
         clear_approval_context()
         # The tool registry is process-global. Leaving these probe tools in it
         # changed what later tests saw when they enumerated or dispatched
@@ -279,13 +279,13 @@ def test_code_execution_runs_a_trailing_expression_exactly_once():
     from praisonaiagents.approval import clear_approval_context
 
     registry = get_approval_registry()
-    previous = registry.get_backend()
+    previous = registry._global_backend  # not get_backend(): its ConsoleBackend fallback would be installed globally on restore
     registry.set_backend(AutoApproveBackend())
     try:
         assert execute_code("print(7)")["stdout"] == "7\n"
         assert execute_code_with_tools("print(7)")["stdout"] == "7\n"
     finally:
-        registry.set_backend(previous)
+        registry._global_backend = previous
         clear_approval_context()
 
 
diff --git a/src/praisonai-agents/tests/unit/test_issue_6238_safe_paths_and_bridge.py b/src/praisonai-agents/tests/unit/test_issue_6238_safe_paths_and_bridge.py
new file mode 100644
index 000000000..d7e4946df
--- /dev/null
+++ b/src/praisonai-agents/tests/unit/test_issue_6238_safe_paths_and_bridge.py
@@ -0,0 +1,107 @@
+"""Regression tests for Issue #6238.
+
+* Caller-supplied names must never escape the storage directory.
+* Sync entry points must work from inside a running event loop.
+"""
+
+import asyncio
+import os
+
+from praisonaiagents.planning.storage import PlanStorage
+from praisonaiagents.planning.todo import TodoList
+from praisonaiagents.utils.atomic_io import safe_join
+
+
+def test_safe_join_keeps_legacy_names_and_confines_others(tmp_path):
+    base = str(tmp_path)
+    assert safe_join(base, "current", ".json") == os.path.join(base, "current.json")
+    for name in ("../../x", "/etc/passwd", "..", "", "a/b", "a\\b", ".hidden"):
+        path = safe_join(base, name, ".json")
+        assert os.path.dirname(path) == base, name
+    # Distinct inputs never collide on one file.
+    assert safe_join(base, "a/b") != safe_join(base, "a_b")
+
+
+def test_plan_storage_todo_name_cannot_traverse(tmp_path):
+    storage = PlanStorage(base_path=str(tmp_path / "store"))
+    path = storage.save_todo(TodoList(), name="../../escaped")
+    assert os.path.dirname(path) == storage.todos_dir
+    assert not (tmp_path / "escaped.json").exists()
+    assert storage.load_todo("../../escaped") is not None
+    assert storage.delete_todo("../../escaped") is True
+
+
+def test_generate_title_works_inside_running_loop(monkeypatch):
+    from praisonaiagents.session import title
+
+    async def fake(*args, **kwargs):
+        return "LLM Title"
+
+    monkeypatch.setattr(title, "generate_title_async", fake)
+
+    async def main():
+        return title.generate_title("hello", "world")
+
+    assert asyncio.run(main()) == "LLM Title"
+
+
+def test_listed_todo_names_round_trip(tmp_path):
+    storage = PlanStorage(base_path=str(tmp_path / "store"))
+    storage.save_todo(TodoList(), name="my tasks")
+    (listed,) = storage.list_todos()
+    assert storage.load_todo(listed) is not None
+    assert storage.load_todo("my tasks") is not None
+    assert storage.delete_todo(listed) is True
+    assert storage.list_todos() == []
+
+
+def test_legacy_todo_file_still_found(tmp_path):
+    storage = PlanStorage(base_path=str(tmp_path / "store"))
+    legacy = os.path.join(storage.todos_dir, "my tasks.json")
+    with open(legacy, "w", encoding="utf-8") as f:
+        f.write(TodoList().to_json())
+    assert storage.load_todo("my tasks") is not None
+    assert storage.save_todo(TodoList(), name="my tasks") == legacy
+    assert storage.list_todos() == ["my tasks"]
+    assert storage.delete_todo("my tasks") is True
+
+
+def test_custom_plan_filename_keeps_md_suffix(tmp_path):
+    from praisonaiagents.planning.plan import Plan
+
+    storage = PlanStorage(base_path=str(tmp_path / "store"))
+    plan = Plan(name="demo")
+    path = storage.save_plan(plan, filename="my plan.md")
+    assert path.endswith(".md") and os.path.dirname(path) == storage.plans_dir
+    assert len(storage.list_plans()) == 1
+
+
+def test_encoded_looking_names_do_not_share_a_file(tmp_path):
+    storage = PlanStorage(base_path=str(tmp_path / "store"))
+    names = ["a/b", "a_b~c14cddc033f6", "a%2Fb", "a b", "a%20b", "~x", ".."]
+    paths = {storage.save_todo(TodoList(name=n), name=n) for n in names}
+    assert len(paths) == len(names)
+    assert all(os.path.dirname(p) == storage.todos_dir for p in paths)
+    assert sorted(storage.list_todos()) == sorted(names)
+    for n in names:
+        assert storage.load_todo(n).name == n
+    assert storage.delete_todo("a/b") is True
+    assert storage.load_todo("a%2Fb") is not None
+
+
+def test_long_unicode_names_are_bounded_and_round_trip(tmp_path):
+    storage = PlanStorage(base_path=str(tmp_path / "store"))
+    long_a, long_b = "我的任务" * 40, "我的任务" * 40 + "x"
+    pa = storage.save_todo(TodoList(name=long_a), name=long_a)
+    pb = storage.save_todo(TodoList(name=long_b), name=long_b)
+    assert pa != pb
+    for p in (pa, pb):
+        assert os.path.dirname(p) == storage.todos_dir
+        assert len(os.path.basename(p).encode("utf-8")) <= 200
+    # Short non-ASCII names keep their plain (pre-#6238) filename.
+    assert storage.save_todo(TodoList(), name="我的任务").endswith("我的任务.json")
+    for listed in storage.list_todos():
+        assert storage.load_todo(listed) is not None
+        assert storage.delete_todo(listed) is True
+    assert storage.list_todos() == []
+    assert storage.load_todo(long_a) is None
diff --git a/src/praisonai-agents/tests/unit/trace/test_run_events.py b/src/praisonai-agents/tests/unit/trace/test_run_events.py
index 87502ae38..2a7a8d853 100644
--- a/src/praisonai-agents/tests/unit/trace/test_run_events.py
+++ b/src/praisonai-agents/tests/unit/trace/test_run_events.py
@@ -491,7 +491,7 @@ class TestTeamTasks:
                 return ApprovalDecision(approved=True, reason="looks good", approver="ana")
 
         registry = get_approval_registry()
-        previous = registry.get_backend()
+        previous = registry._global_backend  # not get_backend(): its ConsoleBackend fallback would be installed globally on restore
         registry.set_backend(Reviewer())
         try:
             writer = Agent(name="Writer", instructions="write")
@@ -500,7 +500,7 @@ class TestTeamTasks:
             out = TaskOutput(description="Draft", raw="api_key=sk-secret123456 text", agent="Writer")
             team._apply_human_review(task, 0, out)
         finally:
-            registry.set_backend(previous)
+            registry._global_backend = previous
         requested, decided = _of(recorder, "approval_requested"), _of(recorder, "approval_decided")
         assert len(requested) == len(decided) == 1
         assert requested[0].data["kind"] == "task_output_review"