FEAT: tool to launch provided sub-task managers

Author: mdw771

packages/eaa-core/src/eaa_core/task_manager/base.py

@@ -114,6 +114,9 @@ class BaseTaskManager:
     ----------
     llm_config : LLMConfig, optional
         Configuration used to build the chat model.
+    name : str, optional
+        Human-readable task manager name used when registering or launching
+        task managers from other agents.
     memory_config : Optional[MemoryManagerConfig], optional
         Configuration for the long-term memory store.
     tools : list[BaseTool], optional
@@ -142,6 +145,7 @@ class BaseTaskManager:
     def __init__(
         self,
         llm_config: LLMConfig = None,
+        name: Optional[str] = None,
         memory_config: Optional[MemoryManagerConfig] = None,
         tools: list[BaseTool] = (),
         skill_dirs: Optional[Sequence[str]] = None,
@@ -168,6 +172,7 @@ def __init__(
                 "`transcript_db_path` for WebUI transcript persistence and "
                 "`checkpoint_db_path` for LangGraph checkpoints."
             )
+        self.name = self.resolve_name(name)
         self.chat_state = ChatGraphState()
         self.task_state = TaskManagerState()
         self.active_state: TaskManagerState = self.task_state
@@ -239,6 +244,19 @@ def __init__(
         if build:
             self.build()
 
+    @classmethod
+    def get_default_name(cls) -> str:
+        """Return the default task manager name."""
+        return BaseTool.camel_to_snake(cls.__name__)
+
+    @classmethod
+    def resolve_name(cls, name: Optional[str]) -> str:
+        """Return a validated task manager name."""
+        resolved_name = cls.get_default_name() if name is None else str(name).strip()
+        if not resolved_name:
+            raise ValueError("`name` must be a non-empty string when provided.")
+        return resolved_name
+
     def get_default_system_prompt(self) -> str:
         """Return the default system prompt for the task manager."""
         prompt = render_prompt_template(

packages/eaa-core/src/eaa_core/tool/subagent.py

@@ -1,16 +1,19 @@
 from __future__ import annotations
 
+import inspect
 import re
-from typing import Annotated, Any
+from typing import Annotated, Any, ClassVar
 
 from eaa_core.message_proc import extract_message_text
+from eaa_core.task_manager.persistence import SQLiteTranscriptStore
 from eaa_core.tool.base import BaseTool, tool
 
 
 class SubagentTool(BaseTool):
     """Launch subordinate task managers for delegated agent work."""
 
     name: str = "subagent"
+    registered_task_managers: ClassVar[dict[str, Any]] = {}
 
     def __init__(self, task_manager: Any, *args: Any, **kwargs: Any) -> None:
         """Initialize the subagent launcher.
@@ -27,6 +30,52 @@ def __init__(self, task_manager: Any, *args: Any, **kwargs: Any) -> None:
         self.task_manager = task_manager
         super().__init__(*args, **kwargs)
 
+    @classmethod
+    def add_task_managers(cls, task_managers: Any | list[Any]) -> None:
+        """Register one or more task manager objects for subtask launches.
+
+        Parameters
+        ----------
+        task_managers : Any or list[Any]
+            Task manager object or objects with a unique ``name`` and callable
+            ``run`` method.
+        """
+        if isinstance(task_managers, (list, tuple)):
+            normalized_task_managers = list(task_managers)
+        else:
+            normalized_task_managers = [task_managers]
+        for task_manager in normalized_task_managers:
+            name = cls._get_task_manager_name(task_manager)
+            run_method = getattr(task_manager, "run", None)
+            if not callable(run_method):
+                raise ValueError(
+                    f"Registered task manager {name!r} must define a callable "
+                    "`run` method."
+                )
+            existing_task_manager = cls.registered_task_managers.get(name)
+            if (
+                existing_task_manager is not None
+                and existing_task_manager is not task_manager
+            ):
+                raise ValueError(
+                    f"A task manager named {name!r} is already registered."
+                )
+            cls.registered_task_managers[name] = task_manager
+
+    @staticmethod
+    def _get_task_manager_name(task_manager: Any) -> str:
+        """Return a validated task manager registry name."""
+        name = str(getattr(task_manager, "name", "")).strip()
+        if not name:
+            raise ValueError("Registered task managers must have a non-empty `name`.")
+        return name
+
+    @staticmethod
+    def _transcript_table_name_for_conversation(conversation_id: str) -> str:
+        """Return the transcript table name for a subordinate conversation."""
+        table_suffix = re.sub(r"[^A-Za-z0-9_]", "_", conversation_id)
+        return f"transcript_messages_{table_suffix}"
+
     @tool(name="subagent_tool.launch_subagent")
     def launch_subagent(
         self,
@@ -56,8 +105,9 @@ def launch_subagent(
             conversation_id = str(conversation["id"])
         transcript_table_name = "transcript_messages"
         if conversation_id is not None:
-            table_suffix = re.sub(r"[^A-Za-z0-9_]", "_", conversation_id)
-            transcript_table_name = f"transcript_messages_{table_suffix}"
+            transcript_table_name = self._transcript_table_name_for_conversation(
+                conversation_id
+            )
         inherited_tools = [
             tool_object
             for tool_object in self.task_manager.tool_manager
@@ -93,3 +143,125 @@ def launch_subagent(
                     conversation_id,
                     message="Subagent terminated",
                 )
+
+    @tool(name="subagent_tool.list_registered_task_managers")
+    def list_registered_task_managers(self) -> list[dict[str, str]]:
+        """Return task managers available for ``launch_subtask_manager``.
+
+        Returns
+        -------
+        list[dict[str, str]]
+            Registered task manager specs.
+        """
+        specs: list[dict[str, str]] = []
+        for name, task_manager in self.registered_task_managers.items():
+            run_method = getattr(task_manager, "run", None)
+            specs.append(
+                {
+                    "name": name,
+                    "task_class_name": type(task_manager).__name__,
+                    "run_method_docstring": inspect.getdoc(run_method) or "",
+                }
+            )
+        return specs
+
+    @tool(name="subagent_tool.launch_subtask_manager")
+    def launch_subtask_manager(
+        self,
+        task_manager_name: Annotated[
+            str,
+            "Name of the registered task manager to launch.",
+        ],
+        task_manager_kwargs: Annotated[
+            dict,
+            "Keyword arguments to pass to the selected task manager's run method.",
+        ],
+    ) -> dict[str, Any]:
+        """Run a registered task manager and return its run result.
+
+        Parameters
+        ----------
+        task_manager_name : str
+            Name of the registered task manager to launch.
+        task_manager_kwargs : dict
+            Keyword arguments passed to ``matched_task_manager.run``.
+
+        Returns
+        -------
+        dict[str, Any]
+            The selected task manager's ``run`` return value.
+        """
+        if not isinstance(task_manager_kwargs, dict):
+            raise ValueError("`task_manager_kwargs` must be a dictionary.")
+        task_manager_name = str(task_manager_name).strip()
+        matched_task_manager = self.registered_task_managers.get(task_manager_name)
+        if matched_task_manager is None:
+            available_names = ", ".join(self.registered_task_managers) or "none"
+            raise ValueError(
+                f"No registered task manager named {task_manager_name!r}. "
+                f"Available task managers: {available_names}."
+            )
+
+        runtime_controller = getattr(self.task_manager, "runtime_controller", None)
+        conversation_id = self._get_task_manager_name(matched_task_manager)
+        if runtime_controller is not None:
+            conversation = runtime_controller.create_conversation(
+                label=conversation_id,
+                kind="subagent",
+            )
+            conversation_id = str(conversation["id"])
+        transcript_table_name = self._transcript_table_name_for_conversation(
+            conversation_id
+        )
+
+        saved_state = {
+            "checkpoint_db_path": getattr(
+                matched_task_manager,
+                "checkpoint_db_path",
+                None,
+            ),
+            "transcript_db_path": getattr(
+                matched_task_manager,
+                "transcript_db_path",
+                None,
+            ),
+            "transcript_table_name": getattr(
+                matched_task_manager,
+                "transcript_table_name",
+                None,
+            ),
+            "transcript_store": getattr(matched_task_manager, "transcript_store", None),
+            "use_webui": getattr(matched_task_manager, "use_webui", None),
+            "runtime_controller": getattr(
+                matched_task_manager,
+                "runtime_controller",
+                None,
+            ),
+            "runtime_conversation_id": getattr(
+                matched_task_manager,
+                "runtime_conversation_id",
+                None,
+            ),
+        }
+        matched_task_manager.checkpoint_db_path = None
+        matched_task_manager.transcript_db_path = self.task_manager.transcript_db_path
+        matched_task_manager.transcript_table_name = transcript_table_name
+        matched_task_manager.transcript_store = SQLiteTranscriptStore(
+            self.task_manager.transcript_db_path,
+            table_name=transcript_table_name,
+        )
+        matched_task_manager.use_webui = self.task_manager.use_webui
+        matched_task_manager.runtime_controller = runtime_controller
+        matched_task_manager.runtime_conversation_id = conversation_id
+
+        try:
+            result = matched_task_manager.run(**task_manager_kwargs)
+            return {"result": result}
+        finally:
+            for attribute, value in saved_state.items():
+                setattr(matched_task_manager, attribute, value)
+            if runtime_controller is not None:
+                runtime_controller.terminate_conversation(
+                    conversation_id,
+                    message="Subtask manager terminated",
+                )