fix(delegation): preserve child workspace hints and queue busy input while subagents run

cli.py

@@ -6238,6 +6238,18 @@ def _clear_current_input(self) -> None:
             except Exception:
                 pass
 
+    def _should_queue_busy_input(self) -> bool:
+        """Return True when busy input should be queued instead of interrupting.
+
+        Queue mode always queues. In interrupt mode, we still prefer queueing when
+        active delegated child agents are running, because interrupting the parent
+        will propagate to those children and discard in-flight delegated work.
+        """
+        if getattr(self, "busy_input_mode", "interrupt") == "queue":
+            return True
+        agent = getattr(self, "agent", None)
+        children = getattr(agent, "_active_children", None) or []
+        return bool(children)
 
     def chat(self, message, images: list = None) -> Optional[str]:
         """
@@ -6451,6 +6463,9 @@ def run_agent():
                             # But if it does (race condition), don't interrupt.
                             if self._clarify_state or self._clarify_freetext:
                                 continue
+                            if self._should_queue_busy_input():
+                                self._pending_input.put(interrupt_msg)
+                                continue
                             print("\n⚡ New message detected, interrupting...")
                             # Signal TTS to stop on interrupt
                             if stop_event is not None:
@@ -7074,7 +7089,7 @@ def handle_enter(event):
                 # Bundle text + images as a tuple when images are present
                 payload = (text, images) if images else text
                 if self._agent_running and not (text and _looks_like_slash_command(text)):
-                    if self.busy_input_mode == "queue":
+                    if self._should_queue_busy_input():
                         # Queue for the next turn instead of interrupting
                         self._pending_input.put(payload)
                         preview = text if text else f"[{len(images)} image{'s' if len(images) != 1 else ''} attached]"

tests/test_cli_init.py

@@ -148,6 +148,22 @@ def test_interrupt_mode_routes_busy_enter_to_interrupt(self):
         assert cli._interrupt_queue.get_nowait() == "redirect"
         assert cli._pending_input.empty()
 
+    def test_active_subagents_force_queue_even_in_interrupt_mode(self):
+        cli = _make_cli()
+        cli._agent_running = True
+        cli.agent = MagicMock()
+        cli.agent._active_children = [MagicMock()]
+        assert cli.busy_input_mode == "interrupt"
+        assert cli._should_queue_busy_input() is True
+
+    def test_no_active_subagents_keeps_interrupt_mode_behavior(self):
+        cli = _make_cli()
+        cli._agent_running = True
+        cli.agent = MagicMock()
+        cli.agent._active_children = []
+        assert cli.busy_input_mode == "interrupt"
+        assert cli._should_queue_busy_input() is False
+
 
 class TestSingleQueryState:
     def test_voice_and_interrupt_state_initialized_before_run(self):

tests/tools/test_delegate.py

@@ -74,13 +74,24 @@ def test_goal_only(self):
         self.assertIn("Fix the tests", prompt)
         self.assertIn("YOUR TASK", prompt)
         self.assertNotIn("CONTEXT", prompt)
+        self.assertIn("Never assume a repository lives at", prompt)
 
     def test_goal_with_context(self):
         prompt = _build_child_system_prompt("Fix the tests", "Error: assertion failed in test_foo.py line 42")
         self.assertIn("Fix the tests", prompt)
         self.assertIn("CONTEXT", prompt)
         self.assertIn("assertion failed", prompt)
 
+    def test_goal_with_workspace_hint(self):
+        prompt = _build_child_system_prompt(
+            "Inspect the repo",
+            "Need local validation",
+            workspace_path="/home/ubuntu/.hermes/hermes-agent",
+        )
+        self.assertIn("WORKSPACE PATH", prompt)
+        self.assertIn("/home/ubuntu/.hermes/hermes-agent", prompt)
+        self.assertIn("Use this exact path", prompt)
+
     def test_empty_context_ignored(self):
         prompt = _build_child_system_prompt("Do something", "  ")
         self.assertNotIn("CONTEXT", prompt)
@@ -127,6 +138,29 @@ def test_task_missing_goal(self):
         result = json.loads(delegate_task(tasks=[{"context": "no goal here"}], parent_agent=parent))
         self.assertIn("error", result)
 
+    @patch("run_agent.AIAgent")
+    def test_child_prompt_includes_workspace_hint_when_parent_cwd_is_repo(self, MockAgent):
+        parent = _make_mock_parent()
+        mock_child = MagicMock()
+        MockAgent.return_value = mock_child
+
+        with patch("tools.delegate_tool.os.getenv", return_value="/home/ubuntu/.hermes/hermes-agent"), \
+             patch("tools.delegate_tool.os.path.isdir", return_value=True):
+            _build_child_agent(
+                task_index=0,
+                goal="Inspect repo",
+                context="Need local git commands",
+                toolsets=["terminal"],
+                model=None,
+                max_iterations=5,
+                parent_agent=parent,
+            )
+
+        _, kwargs = MockAgent.call_args
+        prompt = kwargs["ephemeral_system_prompt"]
+        self.assertIn("WORKSPACE PATH", prompt)
+        self.assertIn("/home/ubuntu/.hermes/hermes-agent", prompt)
+
     @patch("tools.delegate_tool._run_single_child")
     def test_single_task_mode(self, mock_run):
         mock_run.return_value = {

tools/delegate_tool.py

@@ -45,7 +45,12 @@ def check_delegate_requirements() -> bool:
     return True
 
 
-def _build_child_system_prompt(goal: str, context: Optional[str] = None) -> str:
+def _build_child_system_prompt(
+    goal: str,
+    context: Optional[str] = None,
+    *,
+    workspace_path: Optional[str] = None,
+) -> str:
     """Build a focused system prompt for a child agent."""
     parts = [
         "You are a focused subagent working on a specific delegated task.",
@@ -54,13 +59,21 @@ def _build_child_system_prompt(goal: str, context: Optional[str] = None) -> str:
     ]
     if context and context.strip():
         parts.append(f"\nCONTEXT:\n{context}")
+    if workspace_path and str(workspace_path).strip():
+        parts.append(
+            "\nWORKSPACE PATH:\n"
+            f"{workspace_path}\n"
+            "Use this exact path for local repository/workdir operations unless the task explicitly says otherwise."
+        )
     parts.append(
         "\nComplete this task using the tools available to you. "
         "When finished, provide a clear, concise summary of:\n"
         "- What you did\n"
         "- What you found or accomplished\n"
         "- Any files you created or modified\n"
         "- Any issues encountered\n\n"
+        "Important workspace rule: Never assume a repository lives at /workspace/... or any other container-style path unless the task/context explicitly gives that path. "
+        "If no exact local path is provided, discover it first before issuing git/workdir-specific commands.\n\n"
         "Be thorough but concise -- your response is returned to the "
         "parent agent as a summary."
     )
@@ -75,6 +88,31 @@ def _strip_blocked_tools(toolsets: List[str]) -> List[str]:
     return [t for t in toolsets if t not in blocked_toolset_names]
 
 
+def _resolve_workspace_hint(parent_agent) -> Optional[str]:
+    """Best-effort local workspace hint for child prompts.
+
+    We only inject a path when we have a concrete absolute directory. This avoids
+    teaching subagents a fake container path while still helping them avoid
+    guessing `/workspace/...` for local repo tasks.
+    """
+    candidates = [
+        os.getenv("TERMINAL_CWD"),
+        getattr(getattr(parent_agent, "_subdirectory_hints", None), "working_dir", None),
+        getattr(parent_agent, "terminal_cwd", None),
+        getattr(parent_agent, "cwd", None),
+    ]
+    for candidate in candidates:
+        if not candidate:
+            continue
+        try:
+            text = os.path.abspath(os.path.expanduser(str(candidate)))
+        except Exception:
+            continue
+        if os.path.isabs(text) and os.path.isdir(text):
+            return text
+    return None
+
+
 def _build_child_progress_callback(task_index: int, parent_agent, task_count: int = 1) -> Optional[callable]:
     """Build a callback that relays child agent tool calls to the parent display.
 
@@ -194,7 +232,8 @@ def _build_child_agent(
     else:
         child_toolsets = _strip_blocked_tools(DEFAULT_TOOLSETS)
 
-    child_prompt = _build_child_system_prompt(goal, context)
+    workspace_hint = _resolve_workspace_hint(parent_agent)
+    child_prompt = _build_child_system_prompt(goal, context, workspace_path=workspace_hint)
     # Extract parent's API key so subagents inherit auth (e.g. Nous Portal).
     parent_api_key = getattr(parent_agent, "api_key", None)
     if (not parent_api_key) and hasattr(parent_agent, "_client_kwargs"):