Improve subagent and planexit tool.

Author: beacoder

python_agent_harness/agent.py

@@ -162,6 +162,13 @@ def compact(self) -> bool:
     # tool execution
     # ------------------------------------------------------------------
     def _execute_tool_call(self, call: ToolCall) -> str:
+        if not self.top_level and call.name in config.SUBAGENT_EXCLUDED_TOOLS:
+            # defense in depth: a hallucinated call must never reach the
+            # registry — the spec was filtered, so refuse it here too
+            return (
+                f"Error: {call.name} is not available to sub-agents — "
+                "one-shot/interactive tools are parent-only"
+            )
         args = call.arguments
         if isinstance(args, str):
             try:
@@ -256,9 +263,16 @@ def safe_delta(text: str) -> None:
                     session.on_delta(text)
 
             try:
+                # sub-agents are one-shot tasks: they must not see (or
+                # call) parent-only tools — Agent (no nesting), Question
+                # and PlanExit (interactive/handoff) — filtered from the
+                # specs before sending
+                tools = session.tool_specs(
+                    exclude=config.SUBAGENT_EXCLUDED_TOOLS if not self.top_level else ()
+                )
                 assistant, usage = session.client.chat(
                     self.messages,
-                    tools=session.tool_specs() if session.tools_enabled else None,
+                    tools=tools if session.tools_enabled else None,
                     system=self.system,
                     temperature=session.temperature,
                     max_tokens=session.max_tokens,

python_agent_harness/agent_session.py

@@ -164,8 +164,13 @@ def ask_questions(self, questions: list[dict]) -> str:
     # ------------------------------------------------------------------
     # tools
     # ------------------------------------------------------------------
-    def tool_specs(self) -> list:
-        return self.registry.specs()
+    def tool_specs(self, exclude: tuple[str, ...] = ()) -> list:
+        """Tool specs exposed to the model; ``exclude`` drops tools by
+        name (e.g. one-shot/interactive tools for sub-agent runs)."""
+        return [
+            spec for spec in self.registry.specs()
+            if spec.name not in exclude
+        ]
 
     def execute_tool(
         self, name: str, args: dict[str, Any], call_id: str | None = None
@@ -401,7 +406,13 @@ def run_subagent(self, subagent_type: str, description: str, prompt: str) -> str
             self.pop_todo_scope()
 
     def plan_exit(self) -> str:
-        """PlanExit tool implementation."""
+        """PlanExit tool implementation.
+
+        Asks the user to approve the plan→build switch through a y/n
+        confirmation UI (rendered like the Question tool's choice list,
+        but keyed with y/n instead of numbers).  The TUI hook decides
+        the exact look; the session only interprets the boolean answer.
+        """
         if not self.plan_mode.is_plan:
             return "Not in plan mode; PlanExit has no effect.  Continue as normal."
         approved = self.confirm(

python_agent_harness/config.py

@@ -135,6 +135,13 @@
     "The plan at %s has been approved, you can now edit files. Execute the plan"
 )
 
+# PlanExit asks the user with the same choice UI as the Question tool:
+# option[0] approves the switch to build mode, anything else rejects it.
+PLAN_EXIT_OPTIONS = (
+    "Yes, switch to build agent",
+    "No, keep refining the plan",
+)
+
 # ---- tools -------------------------------------------------------------------
 DEFAULT_TOOLS: list[str] = [
     "Agent", "TodoWrite", "Glob", "Grep", "Read", "Insert", "Edit",
@@ -149,6 +156,11 @@
 
 # ---- sub-agents ---------------------------------------------------------------
 SUBAGENT_MAX_ROUNDS = 60
+# Tools a sub-agent must NOT see or call: it runs autonomously as a
+# one-shot task inside the parent's tool round, so it cannot spawn
+# further sub-agents (Agent), ask the user questions (Question), nor
+# end in a plan/build handoff (PlanExit).
+SUBAGENT_EXCLUDED_TOOLS = ("Agent", "Question", "PlanExit")
 
 # ---- TUI preview limits -------------------------------------------------------
 TOOL_RESULT_PREVIEW_LINES = 5    # max lines of a tool result shown in the TUI

python_agent_harness/prompts/subagent.txt

@@ -15,6 +15,9 @@ You are an autonomous subagent. Your role is to independently complete well-defi
 - If you lack information needed to proceed, make reasonable assumptions based on context
 
 # Tool usage policy
+- You do NOT have access to the `Agent`, `Question`, and `PlanExit`
+  tools — they are parent-only (one-shot/interactive). Run autonomously
+  to completion; never delegate work to further sub-agents.
 - You have the capability to call multiple tools in a single response. When multiple independent pieces of information are requested, batch your tool calls together for optimal performance. When making multiple bash tool calls, you MUST send a single message with multiple tools calls to run the calls in parallel. For example, if you need to run "git status" and "git diff", send a single message with two tool calls to run the calls in parallel.
 
 You MUST answer concisely with fewer than 4 lines of text (not including tool use or code generation), unless user asks for detail.

python_agent_harness/tui.py

@@ -234,15 +234,38 @@ def get_completions(self, document: Any, complete_event: Any):
 class UiQuestion:
     def __init__(self, prompt: str, multiple: bool = False,
                  options: list[str] | None = None,
-                 custom: bool = True) -> None:
+                 custom: bool = True,
+                 keys: list[str] | None = None) -> None:
         self.prompt = prompt
         self.multiple = multiple
         self.options = options or []
         self.custom = custom
+        # keyed choices (e.g. ["y", "n"] for a confirm): render the
+        # options as a keyed list and resolve typed keys to labels,
+        # instead of the numbered-list style of the Question tool
+        self.keys = keys or []
         self.answer: str | None = None
         self.event = threading.Event()
 
 
+def _resolve_keyed_choice(answer: str, options: list[str], keys: list[str]) -> str:
+    """Map bare keys in ANSWER to the matching option label.
+
+    Comma-separated keys pick several options (multiple select);
+    non-key tokens pass through unchanged as free-text answers.
+    """
+    if not options or not keys or not answer.strip():
+        return answer
+    resolved: list[str] = []
+    for part in answer.split(","):
+        part = part.strip()
+        if part.lower() in keys:
+            resolved.append(options[keys.index(part.lower())])
+            continue
+        resolved.append(part)
+    return ", ".join(resolved)
+
+
 def _resolve_numbered_choice(answer: str, options: list[str]) -> str:
     """Map bare numbers in ANSWER (1-based) to the matching option label.
 
@@ -329,8 +352,20 @@ def _on_log(self, msg: str) -> None:
         self.status = f" {msg[:60]}"
 
     def _ui_confirm(self, prompt: str) -> bool:
-        q = UiQuestion(prompt)
-        return self._ask_sync(q) in ("y", "yes", "true", "1", "a")
+        """PlanExit confirmation: same look as the Question tool, but a
+        y/n keyed choice list instead of numbers (two choices only)."""
+        q = UiQuestion(
+            prompt,
+            options=list(config.PLAN_EXIT_OPTIONS),
+            keys=["y", "n"],
+            custom=False,
+        )
+        answer = self._ask_sync(q).strip().lower()
+        # resolved answers arrive as the option label; legacy free-text
+        # (y/yes/a/1/true) keeps working for muscle memory
+        return answer == config.PLAN_EXIT_OPTIONS[0].lower() or answer in (
+            "y", "yes", "a", "true", "1",
+        )
 
     def _ui_ask(self, questions: list[dict]) -> str:
         lines = []
@@ -608,7 +643,24 @@ def _ask_question_blocking(self) -> None:
         self.console.print()
         self._flush()
         options = q.options or []
-        if options and any(len(o) > 1 for o in options):
+        keys = q.keys or []
+        if keys and options and len(keys) == len(options):
+            # keyed choices (e.g. y/n confirm): type the key to pick —
+            # same list look as the Question tool, keys instead of numbers
+            self.console.print(Text(q.prompt))
+            for key, opt in zip(keys, options):
+                line = Text(f"  {key}) ", style="cyan")
+                line.append(opt)
+                self.console.print(line)
+            if q.multiple:
+                hint = "Enter keys, comma-separated"
+            else:
+                hint = "Enter a key"
+            if q.custom:
+                hint += ", or type your own answer"
+            self.console.print(f"[dim]{hint}[/dim]")
+            prompt = "> "
+        elif options and any(len(o) > 1 for o in options):
             # long option labels get a numbered list: type the number to pick
             self.console.print(Text(q.prompt))
             for i, opt in enumerate(options, 1):
@@ -632,7 +684,10 @@ def _ask_question_blocking(self) -> None:
                 answer = self.prompt_session.prompt(prompt, multiline=False)
         except (EOFError, KeyboardInterrupt):
             answer = ""
-        q.answer = _resolve_numbered_choice(answer, options)
+        if keys:
+            q.answer = _resolve_keyed_choice(answer, options, keys)
+        else:
+            q.answer = _resolve_numbered_choice(answer, options)
         q.event.set()
         self.question = None
         self._data_event.set()  # re-render promptly after the answer

tests/test_planmode.py

@@ -65,5 +65,63 @@ def test_plan_reminder(self):
         self.assertIn(pm.plan_file, reminder)
 
 
+class TestPlanExitConfirm(unittest.TestCase):
+    """plan_exit goes through the session confirm hook (a y/n choice UI
+    in the TUI) — not through the Question tool's ask_questions."""
+
+    class FakeClient:
+        def chat(self, *a, **k):
+            return None
+
+        def chat_sync(self, *a, **k):
+            return None
+
+        def close(self):
+            pass
+
+    def make_session(self):
+        from python_agent_harness.agent_session import AgentSession
+        from python_agent_harness.tools import default_registry
+
+        s = AgentSession(
+            project_dir="/tmp/proj", client=self.FakeClient(), model="m",
+            registry=default_registry(),
+        )
+        s.switch_to_plan()
+        return s
+
+    def test_plan_exit_uses_confirm_hook_not_ask_questions(self):
+        s = self.make_session()
+        seen = {}
+        s.confirm_fn = lambda prompt: seen.setdefault("prompt", prompt) or True
+        asked = []
+        s.ask_fn = lambda questions: asked.append(questions) or "Unanswered"
+        result = s.plan_exit()
+        self.assertIn("approved", result)
+        self.assertFalse(s.plan_mode.is_plan)
+        self.assertIn("Plan at", seen["prompt"])
+        self.assertIn("Switch to build agent", seen["prompt"])
+        # the Question path must NOT be used for the plan approval
+        self.assertEqual(asked, [])
+        s.close()
+
+    def test_plan_exit_rejected_stays_in_plan(self):
+        s = self.make_session()
+        s.confirm_fn = lambda prompt: False
+        result = s.plan_exit()
+        self.assertIn("rejected", result)
+        self.assertTrue(s.plan_mode.is_plan)
+        s.close()
+
+    def test_plan_exit_noop_outside_plan(self):
+        s = self.make_session()
+        s.switch_to_build()
+        s.confirm_fn = lambda prompt: True
+        result = s.plan_exit()
+        self.assertIn("Not in plan mode", result)
+        self.assertFalse(s.plan_mode.is_plan)
+        s.close()
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_subagent_isolation.py

@@ -20,11 +20,13 @@ def __init__(self, script):
         self.n = 0
         self.streamed = []      # turn indices that streamed
         self.sent = []
+        self.sent_tools = []    # tool names sent per chat call
 
     def chat(self, messages, tools=None, system=None, temperature=None,
              max_tokens=None, reasoning_effort=None, on_delta=None):
         self.n += 1
         self.sent.append([m.to_api() for m in messages])
+        self.sent_tools.append([t.name for t in tools] if tools else None)
         if on_delta:
             on_delta(f"stream-{self.n} ")
             self.streamed.append(self.n)
@@ -164,6 +166,62 @@ def test_subagent_does_not_touch_shared_context_accounting(self):
         self.assertEqual(s.calibrator.factor, 1.0)
         self.assertIsNone(s.calibrator.last_raw_estimate)
 
+    def test_subagent_does_not_get_parent_only_specs(self):
+        """Parent-only tools (Agent, Question, PlanExit) are excluded
+        from the sub-agent's request specs, while the parent keeps them."""
+        from python_agent_harness.tools import PlanExit
+
+        client = RecClient([
+            ("", [ToolCall(id="p1", name="Agent", arguments=AGENT_CALL)]),
+            "sub done",
+            "parent done",
+        ])
+        s = make_session(client)
+        s.registry.register(PlanExit())  # plan mode registers it too
+        run_agent_loop(
+            s, messages=[Message(role="user", content="delegate")], top_level=True
+        )
+        # turn 1 = parent's Agent call, turn 2 = the sub-agent, turn 3 = parent
+        parent_tools, sub_tools = client.sent_tools[0], client.sent_tools[1]
+        self.assertIn("Agent", parent_tools)
+        self.assertIn("Question", parent_tools)
+        self.assertIn("PlanExit", parent_tools)
+        self.assertNotIn("Agent", sub_tools)
+        self.assertNotIn("Question", sub_tools)
+        self.assertNotIn("PlanExit", sub_tools)
+        # the sub-agent keeps its working tools
+        self.assertIn("Read", sub_tools)
+        self.assertIn("Bash", sub_tools)
+
+    def test_subagent_parent_only_call_refused_at_execution(self):
+        """Defense in depth: even a hallucinated parent-only call (Agent,
+        Question, PlanExit) from a sub-agent must be refused at execution
+        time, not silently run."""
+        from python_agent_harness.agent import AgentLoop
+        from python_agent_harness.tools import PlanExit
+
+        client = RecClient([
+            ("", [ToolCall(id="s1", name="Agent", arguments=AGENT_CALL)]),
+            "sub done",
+        ])
+        s = make_session(client)
+        s.registry.register(PlanExit())
+        loop = AgentLoop(
+            s,
+            messages=[Message(role="user", content="find stuff")],
+            top_level=False,
+            system="SUB",
+        )
+        loop.run()
+        tool_rows = [m for m in loop.messages if m.role == "tool"]
+        self.assertTrue(tool_rows)
+        self.assertIn("not available to sub-agents", tool_rows[0].text())
+        # the refused call never executed the Agent tool, so no nested
+        # sub-agent loop ran: exactly 2 chat calls (refused call turn +
+        # the terminal "sub done" reply) — a real delegation would have
+        # spawned an additional nested loop's chat call
+        self.assertEqual(len(client.sent), 2)
+
 
 if __name__ == "__main__":
     unittest.main()