[AAASM-1183] ✨ (crewai): Add team_id, delegation_reason, and hierarchical Crew.kickoff patch

agent_assembly/adapters/crewai/patch.py

@@ -3,19 +3,23 @@
 from __future__ import annotations
 
 import importlib
+from collections.abc import Mapping
 from dataclasses import dataclass
 from functools import wraps
 from threading import local
-from typing import Any, Literal, Mapping
+from typing import Any, Literal
 
 from agent_assembly.core.spawn import _SPAWN_CTX, SpawnContext, spawn_context_scope
 
 _TOOLS_PATCHED_FLAG = "_agent_assembly_crewai_tools_patched"
 _TASK_PATCHED_FLAG = "_agent_assembly_crewai_task_patched"
 _ORIGINAL_TOOL_RUN = "_agent_assembly_original_crewai_tool_run"
 _ORIGINAL_TASK_EXECUTE_SYNC = "_agent_assembly_original_crewai_task_execute_sync"
+_ORIGINAL_CREW_KICKOFF = "_agent_assembly_original_crewai_crew_kickoff"
+_CREW_KICKOFF_PATCHED_FLAG = "_agent_assembly_crewai_crew_kickoff_patched"
 _AGENT_CONTEXT = local()
 _DEFAULT_PENDING_APPROVAL_TIMEOUT_SECONDS = 300
+_MAX_DELEGATION_REASON_CHARS = 256
 
 
 @dataclass(slots=True)
@@ -34,14 +38,19 @@ def apply(self) -> bool:
         task_cls = _load_crewai_task_class()
         if task_cls is not None:
             _apply_task_execute_sync_patch(task_cls, self.callback_handler)
+        crew_cls = _load_crewai_crew_class()
+        if crew_cls is not None:
+            _apply_crew_kickoff_patch(crew_cls)
         return True
 
     def revert(self) -> None:
         """Revert CrewAI runtime monkey patches when available."""
+        crew_cls = _load_crewai_crew_class()
+        if crew_cls is not None:
+            _revert_crew_kickoff_patch(crew_cls)
         base_tool_cls = _load_crewai_basetool_class()
         if base_tool_cls is not None:
             _revert_basetool_run_patch(base_tool_cls)
-
         task_cls = _load_crewai_task_class()
         if task_cls is not None:
             _revert_task_execute_sync_patch(task_cls)
@@ -72,6 +81,49 @@ def _load_crewai_task_class() -> type[Any] | None:
     return None
 
 
+def _load_crewai_crew_class() -> type[Any] | None:
+    try:
+        module = importlib.import_module("crewai")
+    except ImportError:
+        return None
+
+    crew_cls = getattr(module, "Crew", None)
+    if isinstance(crew_cls, type):
+        return crew_cls
+    return None
+
+
+def _extract_crew_team_id(crew: Any) -> str | None:
+    if crew is None:
+        return None
+    crew_id = getattr(crew, "id", None)
+    if crew_id is not None:
+        return str(crew_id)
+    return None
+
+
+def _extract_manager_agent_id(crew: Any) -> str | None:
+    manager = getattr(crew, "manager_agent", None)
+    if manager is None:
+        return None
+    agent_id = getattr(manager, "id", None)
+    if isinstance(agent_id, str) and agent_id:
+        return agent_id
+    return None
+
+
+def _is_hierarchical_process(crew: Any) -> bool:
+    try:
+        module = importlib.import_module("crewai")
+        process_cls = getattr(module, "Process", None)
+        if process_cls is None:
+            return False
+        hierarchical = getattr(process_cls, "hierarchical", None)
+        return getattr(crew, "process", None) == hierarchical
+    except Exception:
+        return False
+
+
 def _set_thread_local_agent_id(agent_id: str | None) -> None:
     _AGENT_CONTEXT.agent_id = agent_id
 
@@ -290,7 +342,7 @@ def patched_run(self: Any, *args: Any, **kwargs: Any) -> Any:
         return result
 
     setattr(base_tool_cls, _ORIGINAL_TOOL_RUN, original_run)
-    setattr(base_tool_cls, "run", patched_run)
+    base_tool_cls.run = patched_run
     setattr(base_tool_cls, _TOOLS_PATCHED_FLAG, True)
 
 
@@ -300,7 +352,7 @@ def _revert_basetool_run_patch(base_tool_cls: type[Any]) -> None:
 
     original_run = getattr(base_tool_cls, _ORIGINAL_TOOL_RUN, None)
     if callable(original_run):
-        setattr(base_tool_cls, "run", original_run)
+        base_tool_cls.run = original_run
 
     if hasattr(base_tool_cls, _ORIGINAL_TOOL_RUN):
         delattr(base_tool_cls, _ORIGINAL_TOOL_RUN)
@@ -352,10 +404,14 @@ def patched_execute_sync(self: Any, *args: Any, **kwargs: Any) -> Any:
 
         spawn_ctx: SpawnContext | None = None
         if worker_id:
+            crew = getattr(getattr(self, "agent", None), "crew", None)
+            raw_reason = str(getattr(self, "description", "") or "")[:_MAX_DELEGATION_REASON_CHARS]
             spawn_ctx = SpawnContext(
                 parent_agent_id=worker_id,
                 depth=_current_spawn_depth(),
                 spawned_by_tool="crewai_task",
+                team_id=_extract_crew_team_id(crew),
+                delegation_reason=raw_reason or None,
             )
 
         try:
@@ -371,7 +427,7 @@ def patched_execute_sync(self: Any, *args: Any, **kwargs: Any) -> Any:
         return result
 
     setattr(task_cls, _ORIGINAL_TASK_EXECUTE_SYNC, original_execute_sync)
-    setattr(task_cls, "execute_sync", patched_execute_sync)
+    task_cls.execute_sync = patched_execute_sync
     setattr(task_cls, _TASK_PATCHED_FLAG, True)
 
 
@@ -381,10 +437,48 @@ def _revert_task_execute_sync_patch(task_cls: type[Any]) -> None:
 
     original_execute_sync = getattr(task_cls, _ORIGINAL_TASK_EXECUTE_SYNC, None)
     if callable(original_execute_sync):
-        setattr(task_cls, "execute_sync", original_execute_sync)
+        task_cls.execute_sync = original_execute_sync
 
     if hasattr(task_cls, _ORIGINAL_TASK_EXECUTE_SYNC):
         delattr(task_cls, _ORIGINAL_TASK_EXECUTE_SYNC)
     if hasattr(task_cls, _TASK_PATCHED_FLAG):
         delattr(task_cls, _TASK_PATCHED_FLAG)
     return None
+
+
+def _apply_crew_kickoff_patch(crew_cls: type[Any]) -> None:
+    if getattr(crew_cls, _CREW_KICKOFF_PATCHED_FLAG, False):
+        return None
+
+    original_kickoff = crew_cls.kickoff
+
+    @wraps(original_kickoff)
+    def patched_kickoff(self: Any, *args: Any, **kwargs: Any) -> Any:
+        if not _is_hierarchical_process(self):
+            return original_kickoff(self, *args, **kwargs)
+
+        spawn_ctx = SpawnContext(
+            parent_agent_id=_extract_manager_agent_id(self) or "",
+            depth=_current_spawn_depth(),
+            spawned_by_tool="crewai_kickoff_hierarchical",
+            team_id=_extract_crew_team_id(self),
+        )
+        with spawn_context_scope(spawn_ctx):
+            return original_kickoff(self, *args, **kwargs)
+
+    setattr(crew_cls, _ORIGINAL_CREW_KICKOFF, original_kickoff)
+    crew_cls.kickoff = patched_kickoff
+    setattr(crew_cls, _CREW_KICKOFF_PATCHED_FLAG, True)
+    return None
+
+
+def _revert_crew_kickoff_patch(crew_cls: type[Any]) -> None:
+    if not getattr(crew_cls, _CREW_KICKOFF_PATCHED_FLAG, False):
+        return None
+    original_kickoff = getattr(crew_cls, _ORIGINAL_CREW_KICKOFF, None)
+    if callable(original_kickoff):
+        crew_cls.kickoff = original_kickoff
+    for attr in (_ORIGINAL_CREW_KICKOFF, _CREW_KICKOFF_PATCHED_FLAG):
+        if hasattr(crew_cls, attr):
+            delattr(crew_cls, attr)
+    return None

agent_assembly/core/spawn.py

@@ -16,6 +16,7 @@ class SpawnContext:
     depth: int
     spawned_by_tool: str | None = None
     delegation_reason: str | None = None
+    team_id: str | None = None
 
 
 _SPAWN_CTX: ContextVar[SpawnContext | None] = ContextVar("_spawn_ctx", default=None)

test/integration/test_spawn_lineage_integration.py

@@ -247,3 +247,87 @@ async def __call__(self, *_args: object, **_kwargs: object) -> str:
     assert sc_c.delegation_reason == "Transfer to agent C"
 
     assert _SPAWN_CTX.get() is None
+
+
+# ---------------------------------------------------------------------------
+# Test 8: CrewAI hierarchical kickoff — team_id and spawn ctx at each task
+# ---------------------------------------------------------------------------
+
+
+def test_crewai_hierarchical_kickoff_sets_team_id_and_spawn_ctx() -> None:
+    """Crew.kickoff with hierarchical process sets _SPAWN_CTX with team_id."""
+    from unittest.mock import patch
+
+    from agent_assembly.adapters.crewai.patch import (
+        _apply_crew_kickoff_patch,
+        _apply_task_execute_sync_patch,
+        _revert_crew_kickoff_patch,
+        _revert_task_execute_sync_patch,
+    )
+
+    captured_kickoff: list[SpawnContext] = []
+    captured_tasks: list[SpawnContext] = []
+
+    class FakeCrew:
+        def __init__(self) -> None:
+            self.id = "crew-team-42"
+            self.manager_agent = type("Mgr", (), {"id": "manager-007"})()
+            self.process = "hierarchical"
+
+        def kickoff(self, *_args: object, **_kwargs: object) -> str:
+            sc = _SPAWN_CTX.get()
+            if sc is not None:
+                captured_kickoff.append(sc)
+            # Simulate manager delegating two tasks during hierarchical kickoff
+            task1 = FakeTask(description="Summarize reports", worker_id="worker-A", crew=self)
+            task2 = FakeTask(description="Write conclusions", worker_id="worker-B", crew=self)
+            task1.execute_sync()
+            task2.execute_sync()
+            return "crew-done"
+
+    class FakeTask:
+        def __init__(self, *, description: str, worker_id: str, crew: object) -> None:
+            self.description = description
+            self.agent = type("Agent", (), {"id": worker_id, "crew": crew})()
+
+        def execute_sync(self, *_args: object, **_kwargs: object) -> str:
+            sc = _SPAWN_CTX.get()
+            if sc is not None:
+                captured_tasks.append(sc)
+            return "task-done"
+
+    _apply_crew_kickoff_patch(FakeCrew)
+    _apply_task_execute_sync_patch(FakeTask, type("CB", (), {})())
+    try:
+        crew = FakeCrew()
+        with patch(
+            "agent_assembly.adapters.crewai.patch._is_hierarchical_process",
+            return_value=True,
+        ):
+            crew.kickoff()
+    finally:
+        _revert_task_execute_sync_patch(FakeTask)
+        _revert_crew_kickoff_patch(FakeCrew)
+
+    # kickoff body sees the outer hierarchical spawn context
+    assert len(captured_kickoff) == 1
+    ks = captured_kickoff[0]
+    assert ks.parent_agent_id == "manager-007"
+    assert ks.team_id == "crew-team-42"
+    assert ks.spawned_by_tool == "crewai_kickoff_hierarchical"
+    assert ks.depth == 1
+
+    # each delegated task sees its own spawn context (depth 2 inside the kickoff scope)
+    assert len(captured_tasks) == 2
+    ts1, ts2 = captured_tasks
+    assert ts1.parent_agent_id == "worker-A"
+    assert ts1.team_id == "crew-team-42"
+    assert ts1.spawned_by_tool == "crewai_task"
+    assert ts1.delegation_reason == "Summarize reports"
+    assert ts1.depth == 2  # incremented from kickoff's depth=1
+
+    assert ts2.parent_agent_id == "worker-B"
+    assert ts2.delegation_reason == "Write conclusions"
+    assert ts2.depth == 2
+
+    assert _SPAWN_CTX.get() is None