diff --git a/tests/tools/test_delegate.py b/tests/tools/test_delegate.py
index ebdf60d296b3..7fff8f7c52e9 100644
--- a/tests/tools/test_delegate.py
+++ b/tests/tools/test_delegate.py
@@ -11,8 +11,10 @@
 
 import json
 import os
+import re
 import sys
 import threading
+import time
 import unittest
 from unittest.mock import MagicMock, patch
 
@@ -1052,5 +1054,92 @@ def test_run_single_child_releases_lease_after_failure(self):
         child._credential_pool.release_lease.assert_called_once_with("cred-a")
 
 
+class TestBackgroundDelegation(unittest.TestCase):
+    """Tests for non-blocking background delegation (background=True)."""
+
+    def test_background_returns_immediately(self):
+        """background=True must return within ~1 second even if child takes longer."""
+        parent = _make_mock_parent(depth=0)
+
+        # Patch _run_delegation_background to simulate slow child
+        import tools.delegate_tool as dt
+
+        original_bg = dt._run_delegation_background
+
+        def slow_bg(*args, **kwargs):
+            # Simulate a 5-second task but this should NOT block the caller
+            time.sleep(5)
+            original_bg(*args, **kwargs)
+
+        with patch.object(dt, '_run_delegation_background', slow_bg):
+            start = time.monotonic()
+            result = json.loads(delegate_task(goal="Long task", background=True, parent_agent=parent))
+            elapsed = time.monotonic() - start
+
+        # Must return immediately (within 2 seconds)
+        self.assertLess(elapsed, 2.0, "background=True took too long to return")
+        self.assertTrue(result.get("background"))
+        self.assertIn("session_id", result)
+        self.assertEqual(len(result["session_id"]), 8)
+        # Must be valid hex
+        self.assertTrue(re.fullmatch(r'[a-f0-9]{8}', result["session_id"]))
+
+    def test_session_id_retrieval(self):
+        """Providing session_id alone retrieves previously saved results."""
+        import tools.delegate_tool as dt
+
+        parent = _make_mock_parent(depth=0)
+
+        # Manually save a result file
+        test_session_id = "deadbeef"
+        saved_data = {
+            "results": [{"task_index": 0, "status": "completed", "summary": "Done!"}],
+            "total_duration_seconds": 1.5,
+        }
+        dt._save_background_result(test_session_id, saved_data)
+
+        try:
+            result = json.loads(delegate_task(session_id=test_session_id, parent_agent=parent))
+            self.assertEqual(result["results"][0]["summary"], "Done!")
+            self.assertEqual(result["total_duration_seconds"], 1.5)
+        finally:
+            # Clean up
+            results_dir = dt._get_background_results_dir()
+            result_file = os.path.join(results_dir, f"{test_session_id}.json")
+            if os.path.exists(result_file):
+                os.unlink(result_file)
+
+    def test_invalid_session_id_raises(self):
+        """Invalid session_id format must raise ValueError."""
+        import tools.delegate_tool as dt
+
+        parent = _make_mock_parent(depth=0)
+
+        # Test various malicious formats
+        malicious_ids = [
+            "../../../etc/passwd",
+            "..",
+            "invalid!chars",
+            "TOOLONGSESSIONID",
+            "",
+            "g" * 8,  # non-hex chars
+        ]
+        for bad_id in malicious_ids:
+            with self.assertRaises(ValueError, msg=f"Should reject: {bad_id!r}"):
+                dt._load_background_result(bad_id)
+
+    def test_background_branch_prevents_child_construction_in_main_thread(self):
+        """When background=True, no AIAgent construction happens in the main thread."""
+        parent = _make_mock_parent(depth=0)
+
+        with patch("run_agent.AIAgent") as MockAgent:
+            result = json.loads(delegate_task(goal="Test", background=True, parent_agent=parent))
+
+        # AIAgent should NOT have been called (no blocking child construction)
+        MockAgent.assert_not_called()
+        self.assertTrue(result.get("background"))
+        self.assertIn("session_id", result)
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py
index a148a31f059e..50bfa13f371e 100644
--- a/tools/delegate_tool.py
+++ b/tools/delegate_tool.py
@@ -20,7 +20,11 @@
 import logging
 logger = logging.getLogger(__name__)
 import os
+import re
+import tempfile
 import time
+import uuid
+import threading
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from typing import Any, Dict, List, Optional
 
@@ -40,6 +44,50 @@
 DEFAULT_TOOLSETS = ["terminal", "file", "web"]
 
 
+def _get_background_results_dir() -> str:
+    """Get the directory for background delegation results."""
+    base = os.environ.get("HERMES_HOME", os.path.expanduser("~/.hermes"))
+    results_dir = os.path.join(base, "delegation_results")
+    os.makedirs(results_dir, exist_ok=True)
+    return results_dir
+
+
+def _save_background_result(session_id: str, data: dict) -> None:
+    """Save delegation result to a file for later retrieval."""
+    # FIX 2: Validate session_id format to prevent path traversal
+    if not re.fullmatch(r'[a-f0-9]{8}', session_id):
+        raise ValueError(f"Invalid session_id format: {session_id!r}. Must be 8 hex characters.")
+    results_dir = _get_background_results_dir()
+    result_file = os.path.join(results_dir, f"{session_id}.json")
+    # FIX 3: Atomic write via temp file + os.replace()
+    fd, tmp_path = tempfile.mkstemp(dir=results_dir, suffix='.json')
+    try:
+        with os.fdopen(fd, 'w') as f:
+            json.dump(data, f, ensure_ascii=False)
+        os.replace(tmp_path, result_file)
+    except Exception:
+        # Clean up temp file on failure
+        if os.path.exists(tmp_path):
+            os.unlink(tmp_path)
+        raise
+
+
+def _load_background_result(session_id: str) -> Optional[dict]:
+    """Load a previously saved delegation result."""
+    # FIX 2: Validate session_id format to prevent path traversal
+    if not re.fullmatch(r'[a-f0-9]{8}', session_id):
+        raise ValueError(f"Invalid session_id format: {session_id!r}. Must be 8 hex characters.")
+    results_dir = _get_background_results_dir()
+    result_file = os.path.join(results_dir, f"{session_id}.json")
+    if not os.path.exists(result_file):
+        return None
+    try:
+        with open(result_file, encoding="utf-8") as f:
+            return json.load(f)
+    except (json.JSONDecodeError, IOError):
+        return None
+
+
 def check_delegate_requirements() -> bool:
     """Delegation has no external requirements -- always available."""
     return True
@@ -507,6 +555,181 @@ def _run_single_child(
             except (ValueError, UnboundLocalError) as e:
                 logger.debug("Could not remove child from active_children: %s", e)
 
+
+def _run_delegation_background(
+    session_id: str,
+    goal: Optional[str],
+    context: Optional[str],
+    toolsets: Optional[List[str]],
+    tasks: Optional[List[Dict[str, Any]]],
+    max_iterations: Optional[int],
+    acp_command: Optional[str],
+    acp_args: Optional[List[str]],
+    parent_agent,
+) -> None:
+    """Execute full delegation in a background thread and save result to file.
+
+    FIX 1: This function receives all raw input parameters and does ALL the
+    work (validation, credential resolution, child building, execution, result
+    saving) so the main thread can return immediately when background=True.
+
+    FIX 4: We save _parent_tool_names at thread start for use within this
+    thread's own child building, but we do NOT restore the global when done.
+    The main thread's _last_resolved_tool_names is separate — each thread
+    operates on its own view of the global at the time it was spawned.
+    """
+    import model_tools as _model_tools
+    _parent_tool_names = list(_model_tools._last_resolved_tool_names)
+
+    # FIX 1: Depth limit (checked in background thread since we branched early)
+    depth = getattr(parent_agent, '_delegate_depth', 0)
+    if depth >= MAX_DEPTH:
+        _save_background_result(session_id, {
+            "error": (
+                f"Delegation depth limit reached ({MAX_DEPTH}). "
+                "Subagents cannot spawn further subagents."
+            )
+        })
+        return
+
+    # Load config
+    cfg = _load_config()
+    default_max_iter = cfg.get("max_iterations", DEFAULT_MAX_ITERATIONS)
+    effective_max_iter = max_iterations or default_max_iter
+
+    # Resolve delegation credentials
+    try:
+        creds = _resolve_delegation_credentials(cfg, parent_agent)
+    except ValueError as exc:
+        _save_background_result(session_id, {"error": str(exc)})
+        return
+
+    # Normalize to task list
+    if tasks and isinstance(tasks, list):
+        task_list = tasks[:MAX_CONCURRENT_CHILDREN]
+    elif goal and isinstance(goal, str) and goal.strip():
+        task_list = [{"goal": goal, "context": context, "toolsets": toolsets}]
+    else:
+        _save_background_result(session_id, {"error": "Provide either 'goal' (single task) or 'tasks' (batch)."})
+        return
+
+    if not task_list:
+        _save_background_result(session_id, {"error": "No tasks provided."})
+        return
+
+    # Validate each task has a goal
+    for i, task in enumerate(task_list):
+        if not task.get("goal", "").strip():
+            _save_background_result(session_id, {"error": f"Task {i} is missing a 'goal'."})
+            return
+
+    overall_start = time.monotonic()
+    results = []
+
+    n_tasks = len(task_list)
+    task_labels = [t["goal"][:40] for t in task_list]
+
+    # FIX 4: Save parent tool names BEFORE any child construction mutates the global
+    _parent_tool_names = list(_model_tools._last_resolved_tool_names)
+
+    children = []
+    try:
+        for i, t in enumerate(task_list):
+            child = _build_child_agent(
+                task_index=i, goal=t["goal"], context=t.get("context"),
+                toolsets=t.get("toolsets") or toolsets, model=creds["model"],
+                max_iterations=effective_max_iter, parent_agent=parent_agent,
+                override_provider=creds["provider"], override_base_url=creds["base_url"],
+                override_api_key=creds["api_key"],
+                override_api_mode=creds["api_mode"],
+                override_acp_command=t.get("acp_command") or acp_command,
+                override_acp_args=t.get("acp_args") or acp_args,
+            )
+            child._delegate_saved_tool_names = _parent_tool_names
+            children.append((i, t, child))
+    finally:
+        # Authoritative restore: reset global to parent's tool names after all children built
+        _model_tools._last_resolved_tool_names = _parent_tool_names
+
+    if n_tasks == 1:
+        _i, _t, child = children[0]
+        result = _run_single_child(0, _t["goal"], child, parent_agent)
+        results.append(result)
+    else:
+        completed_count = 0
+        spinner_ref = getattr(parent_agent, '_delegate_spinner', None)
+
+        with ThreadPoolExecutor(max_workers=MAX_CONCURRENT_CHILDREN) as executor:
+            futures = {}
+            for i, t, child in children:
+                future = executor.submit(
+                    _run_single_child,
+                    task_index=i,
+                    goal=t["goal"],
+                    child=child,
+                    parent_agent=parent_agent,
+                )
+                futures[future] = i
+
+            for future in as_completed(futures):
+                try:
+                    entry = future.result()
+                except Exception as exc:
+                    idx = futures[future]
+                    entry = {
+                        "task_index": idx,
+                        "status": "error",
+                        "summary": None,
+                        "error": str(exc),
+                        "api_calls": 0,
+                        "duration_seconds": 0,
+                    }
+                results.append(entry)
+                completed_count += 1
+
+                idx = entry["task_index"]
+                label = task_labels[idx] if idx < len(task_labels) else f"Task {idx}"
+                dur = entry.get("duration_seconds", 0)
+                status = entry.get("status", "?")
+                icon = "✓" if status == "completed" else "✗"
+                remaining = n_tasks - completed_count
+                completion_line = f"{icon} [{idx+1}/{n_tasks}] {label}  ({dur}s)"
+                if spinner_ref:
+                    try:
+                        spinner_ref.print_above(completion_line)
+                    except Exception:
+                        print(f"  {completion_line}")
+                else:
+                    print(f"  {completion_line}")
+
+                if spinner_ref and remaining > 0:
+                    try:
+                        spinner_ref.update_text(f"🔀 {remaining} task{'s' if remaining != 1 else ''} remaining")
+                    except Exception as e:
+                        logger.debug("Spinner update_text failed: %s", e)
+
+        results.sort(key=lambda r: r["task_index"])
+
+    if parent_agent and hasattr(parent_agent, '_memory_manager') and parent_agent._memory_manager:
+        for entry in results:
+            try:
+                _task_goal = task_list[entry["task_index"]]["goal"] if entry["task_index"] < len(task_list) else ""
+                parent_agent._memory_manager.on_delegation(
+                    task=_task_goal,
+                    result=entry.get("summary", "") or "",
+                    child_session_id=getattr(children[entry["task_index"]][2], "session_id", "") if entry["task_index"] < len(children) else "",
+                )
+            except Exception:
+                pass
+
+    total_duration = round(time.monotonic() - overall_start, 2)
+
+    _save_background_result(session_id, {
+        "results": results,
+        "total_duration_seconds": total_duration,
+    })
+
+
 def delegate_task(
     goal: Optional[str] = None,
     context: Optional[str] = None,
@@ -515,6 +738,8 @@ def delegate_task(
     max_iterations: Optional[int] = None,
     acp_command: Optional[str] = None,
     acp_args: Optional[List[str]] = None,
+    background: bool = False,
+    session_id: Optional[str] = None,
     parent_agent=None,
 ) -> str:
     """
@@ -524,11 +749,51 @@ def delegate_task(
       - Single: provide goal (+ optional context, toolsets)
       - Batch:  provide tasks array [{goal, context, toolsets}, ...]
 
-    Returns JSON with results array, one entry per task.
+    Non-blocking mode (background=True):
+      - Returns immediately with a session_id
+      - Subagent runs in a background thread
+      - Results saved to ~/.hermes/delegation_results/<session_id>.json
+      - Retrieve results with session_id parameter
+
+    Retrieve background results:
+      - Pass session_id to get previously saved results
+      - Returns the same JSON format as normal delegation
     """
     if parent_agent is None:
         return tool_error("delegate_task requires a parent agent context.")
 
+    # Handle result retrieval mode (non-blocking)
+    if session_id and not goal and not tasks:
+        result = _load_background_result(session_id)
+        if result is None:
+            return json.dumps({
+                "error": f"No delegation result found for session_id '{session_id}'. "
+                         "The delegation may still be running or the result was discarded."
+            })
+        return json.dumps(result)
+
+    # FIX 1: Branch on background=True BEFORE any child construction or execution.
+    # If background, spawn a thread that does ALL the work and return immediately.
+    if background:
+        bg_session_id = str(uuid.uuid4())[:8]
+        thread = threading.Thread(
+            target=_run_delegation_background,
+            args=(
+                bg_session_id, goal, context, toolsets, tasks,
+                max_iterations, acp_command, acp_args, parent_agent,
+            ),
+            daemon=True,
+        )
+        thread.start()
+        return json.dumps({
+            "background": True,
+            "session_id": bg_session_id,
+            "message": "Delegation started in background. Use session_id to retrieve results.",
+        })
+
+    # Blocking path: do all validation and credential resolution on main thread
+    # before building any children.
+
     # Depth limit
     depth = getattr(parent_agent, '_delegate_depth', 0)
     if depth >= MAX_DEPTH:
@@ -685,10 +950,12 @@ def delegate_task(
 
     total_duration = round(time.monotonic() - overall_start, 2)
 
-    return json.dumps({
+    result_data = {
         "results": results,
         "total_duration_seconds": total_duration,
-    }, ensure_ascii=False)
+    }
+
+    return json.dumps(result_data, ensure_ascii=False)
 
 
 def _resolve_child_credential_pool(effective_provider: Optional[str], parent_agent):
@@ -850,6 +1117,12 @@ def _load_config() -> dict:
         "1. Single task: provide 'goal' (+ optional context, toolsets)\n"
         "2. Batch (parallel): provide 'tasks' array with up to 3 items. "
         "All run concurrently and results are returned together.\n\n"
+        "NON-BLOCKING MODE (background=True):\n"
+        "- Set background=True to launch subagent without waiting\n"
+        "- Returns immediately with a session_id (UUID prefix)\n"
+        "- Subagent runs in a background thread\n"
+        "- Results saved to ~/.hermes/delegation_results/<session_id>.json\n"
+        "- Retrieve results later by calling delegate_task with session_id parameter\n\n"
         "WHEN TO USE delegate_task:\n"
         "- Reasoning-heavy subtasks (debugging, code review, research synthesis)\n"
         "- Tasks that would flood your context with intermediate data\n"
@@ -951,6 +1224,23 @@ def _load_config() -> dict:
                     "Only used when acp_command is set. Example: ['--acp', '--stdio', '--model', 'claude-opus-4-6']"
                 ),
             },
+            "background": {
+                "type": "boolean",
+                "description": (
+                    "If True, launch subagent in background and return immediately "
+                    "with a session_id. The subagent runs asynchronously. "
+                    "Use session_id to retrieve results when ready. "
+                    "Default: False (blocking)."
+                ),
+            },
+            "session_id": {
+                "type": "string",
+                "description": (
+                    "Pass a session_id to retrieve results from a previously launched "
+                    "background delegation. When provided without goal/tasks, returns the "
+                    "saved result. session_id is the UUID prefix returned by background=True."
+                ),
+            },
         },
         "required": [],
     },
@@ -972,6 +1262,8 @@ def _load_config() -> dict:
         max_iterations=args.get("max_iterations"),
         acp_command=args.get("acp_command"),
         acp_args=args.get("acp_args"),
+        background=args.get("background", False),
+        session_id=args.get("session_id"),
         parent_agent=kw.get("parent_agent")),
     check_fn=check_delegate_requirements,
     emoji="🔀",