feat: run subagents as tasks with background mode, events and profiles

src/bub/builtin/agent.py

@@ -27,9 +27,11 @@ class BubAgent(core.Agent):
     """Bub's agent: the core loop configured by settings and the framework's hooks.
 
     Adds commands, skills, code mode and deferred tools to :class:`bub.core.Agent`, and
-    connects the framework's agent-loop hooks through :class:`AgentPlugins`. System prompt
-    sections, in order: code mode, deferred tools, skills, the framework's ``system_prompt``
-    hooks, then sections added with ``system_prompt.add``. The framework's hooks handle
+    connects the framework's agent-loop hooks through :class:`AgentPlugins`. A run that
+    started subagents does not end before their results arrive. System prompt
+    sections, in order: a subagent's profile instructions, code mode, deferred tools, skills,
+    subagent profiles, the framework's ``system_prompt`` hooks, then sections added with
+    ``system_prompt.add``. The framework's hooks handle
     events before handlers added with ``events.on``.
 
     One instance serves any number of sessions: per-turn choices (model, allowed tools
@@ -73,9 +75,11 @@ def __init__(
             store=framework.get_tape_store() if tape_store is None else tape_store,
             tools=() if tools is None else tools,
             system_prompt=[
+                self._subagent_instructions_prompt,
                 self._code_mode_prompt,
                 self._deferred_tools_prompt,
                 self._skills_prompt,
+                self._subagent_profiles_prompt,
             ],
             context=framework.build_tape_context(),
             sidecars=framework.get_tape_sidecars(),
@@ -84,6 +88,10 @@ def __init__(
             command_prefix=self.settings.command_prefix if command_prefix is None else command_prefix,
         )
         self.plugins.attach(self, tools=tools is None)
+        from bub.builtin.subagents import deliver_subagent_results
+
+        # After the plugins' steering handler, so steered messages are seen first.
+        self.events.on(core.MessagesPending, deliver_subagent_results)
 
     async def run_stream(
         self,
@@ -95,6 +103,7 @@ async def run_stream(
         allowed_skills: Collection[str] | None = None,
         allowed_tools: Collection[str] | None = None,
         reasoning_effort: str | None = None,
+        fork_from: Tape | None = None,
     ) -> AsyncStreamEvents:
         """Prepare a turn and return its stream; await this method before iterating.
 
@@ -114,6 +123,8 @@ async def run_stream(
                 None allows all instance tools and an empty collection allows none.
                 Command execution uses the instance's tools directly.
             reasoning_effort: Per-turn override of the value in state.
+            fork_from: An open tape, such as a running turn's, to start from instead of the
+                session's tape. The turn sees what that tape holds and its fork is discarded.
 
         Loads the turn state and starts the turn with :meth:`start` on the session's tape.
         Consume the stream to completion to finish execution and tape merging. A
@@ -142,9 +153,9 @@ async def run_stream(
             model=self.settings.chat_model(
                 state.get("model") if model is None else model, reasoning_effort=state.get("reasoning_effort")
             ),
-            tape=session_tape_name(session_id, workspace_from_state(state)),
+            tape=session_tape_name(session_id, workspace_from_state(state)) if fork_from is None else fork_from,
             state=state,
-            merge_back=not session_id.startswith("temp/"),
+            merge_back=fork_from is None and not session_id.startswith("temp/"),
             metadata=metadata,
         )
 
@@ -219,6 +230,21 @@ def _deferred_tools_prompt(self, context: core.PromptContext) -> str:
         tools, _ = self._split_code_mode(context.state)
         return render_deferred_tools_prompt(_deferred_tools(tools))
 
+    @staticmethod
+    def _subagent_instructions_prompt(context: core.PromptContext) -> str | None:
+        from bub.builtin.subagents import SUBAGENT_INSTRUCTIONS_STATE_KEY
+
+        return context.state.get(SUBAGENT_INSTRUCTIONS_STATE_KEY)
+
+    @staticmethod
+    def _subagent_profiles_prompt(context: core.PromptContext) -> str:
+        """List subagent profiles when the run may start subagents."""
+        from bub.builtin.subagents import SUBAGENT_TOOL_NAME, discover_subagent_profiles, render_profiles_prompt
+
+        if not any(tool.name == SUBAGENT_TOOL_NAME for tool in context.tools):
+            return ""
+        return render_profiles_prompt(list(discover_subagent_profiles(workspace_from_state(context.state)).values()))
+
     def _skills_prompt(self, context: core.PromptContext) -> str:
         allowed = context.state.get(ALLOWED_SKILLS_STATE_KEY)
         return self._load_skills_prompt(

src/bub/builtin/hook_impl.py

@@ -263,8 +263,10 @@ def render_outbound(
     @hookimpl
     async def provide_lifespan(self) -> AsyncIterator[None]:
         from bub.builtin.shell_manager import shell_manager
+        from bub.builtin.subagents import subagent_manager
 
-        async with shell_manager.lifespan():
+        # Subagents stop before the shells they may have started.
+        async with shell_manager.lifespan(), subagent_manager.lifespan():
             yield
 
     @hookimpl

src/bub/builtin/settings.py

@@ -83,6 +83,12 @@ class AgentSettings(Settings):
     providers: dict[str, CustomProvider] = Field(default_factory=dict)
     max_steps: int = Field(default=sys.maxsize, gt=0)
     max_tokens: int = DEFAULT_MAX_TOKENS
+    subagent_max_concurrency: int = Field(
+        default=4, gt=0, description="Maximum subagents running at once; later ones wait in a queue."
+    )
+    subagent_timeout_seconds: int | None = Field(
+        default=None, gt=0, description="Default time limit of one subagent run; None means no limit."
+    )
     model_timeout_seconds: int | None = None
     client_args: dict[str, Any] = Field(default_factory=dict)
     completion_args: dict[str, Any] = Field(default_factory=dict)