feat: runtime state checkpointing, event system, and executor refactor

- Pass RuntimeState through the event bus and enable entity auto-registration
- Introduce checkpointing API:
  - .checkpoint(), .from_checkpoint(), and async checkpoint support
  - Provider-based storage with BaseProvider and JsonProvider
  - Mid-task resume and kickoff() integration
- Add EventRecord tracking and full event serialization with subtype preservation
- Enable checkpoint fidelity via llm_type and executor_type discriminators

- Refactor executor architecture:
  - Convert executors, tools, prompts, and TokenProcess to BaseModel
  - Introduce proper base classes with typed fields (CrewAgentExecutorMixin, BaseAgentExecutor)
  - Add generic from_checkpoint with full LLM serialization
  - Support executor back-references and resume-safe initialization

- Refactor runtime state system:
  - Move RuntimeState into state/ module with async checkpoint support
  - Add entity serialization improvements and JSON-safe round-tripping
  - Implement event scope tracking and replay for accurate resume behavior

- Improve tool and schema handling:
  - Make BaseTool fully serializable with JSON round-trip support
  - Serialize args_schema via JSON schema and dynamically reconstruct models
  - Add automatic subclass restoration via tool_type discriminator

- Enhance Flow checkpointing:
  - Support restoring execution state and subclass-aware deserialization

- Performance improvements:
  - Cache handler signature inspection
  - Optimize event emission and metadata preparation

- General cleanup:
  - Remove dead checkpoint payload structures
  - Simplify entity registration and serialization logic

Author: greysonlalonde

lib/crewai-tools/tests/test_generate_tool_specs.py

@@ -97,6 +97,7 @@ def test_extract_init_params_schema(mock_tool_extractor):
     assert init_params_schema.keys() == {
         "$defs",
         "properties",
+        "required",
         "title",
         "type",
     }

lib/crewai/pyproject.toml

@@ -43,6 +43,7 @@ dependencies = [
     "uv~=0.9.13",
     "aiosqlite~=0.21.0",
     "pyyaml~=6.0",
+    "aiofiles~=24.1.0",
     "lancedb>=0.29.2,<0.30.1",
 ]
 

lib/crewai/src/crewai/__init__.py

@@ -16,7 +16,6 @@
 from crewai.llm import LLM
 from crewai.llms.base_llm import BaseLLM
 from crewai.process import Process
-from crewai.runtime_state import _entity_discriminator
 from crewai.task import Task
 from crewai.tasks.llm_guardrail import LLMGuardrail
 from crewai.tasks.task_output import TaskOutput
@@ -99,8 +98,8 @@ def __getattr__(name: str) -> Any:
 
 try:
     from crewai.agents.agent_builder.base_agent import BaseAgent as _BaseAgent
-    from crewai.agents.agent_builder.base_agent_executor_mixin import (
-        CrewAgentExecutorMixin as _CrewAgentExecutorMixin,
+    from crewai.agents.agent_builder.base_agent_executor import (
+        BaseAgentExecutor as _BaseAgentExecutor,
     )
     from crewai.agents.tools_handler import ToolsHandler as _ToolsHandler
     from crewai.experimental.agent_executor import AgentExecutor as _AgentExecutor
@@ -118,10 +117,18 @@ def __getattr__(name: str) -> Any:
         "Flow": Flow,
         "BaseLLM": BaseLLM,
         "Task": Task,
-        "CrewAgentExecutorMixin": _CrewAgentExecutorMixin,
+        "BaseAgentExecutor": _BaseAgentExecutor,
         "ExecutionContext": ExecutionContext,
+        "StandardPromptResult": _StandardPromptResult,
+        "SystemPromptResult": _SystemPromptResult,
     }
 
+    from crewai.tools.base_tool import BaseTool as _BaseTool
+    from crewai.tools.structured_tool import CrewStructuredTool as _CrewStructuredTool
+
+    _base_namespace["BaseTool"] = _BaseTool
+    _base_namespace["CrewStructuredTool"] = _CrewStructuredTool
+
     try:
         from crewai.a2a.config import (
             A2AClientConfig as _A2AClientConfig,
@@ -155,36 +162,49 @@ def __getattr__(name: str) -> Any:
         **sys.modules[_BaseAgent.__module__].__dict__,
     }
 
+    import crewai.state.runtime as _runtime_state_mod
+
     for _mod_name in (
         _BaseAgent.__module__,
         Agent.__module__,
         Crew.__module__,
         Flow.__module__,
         Task.__module__,
+        "crewai.agents.crew_agent_executor",
+        _runtime_state_mod.__name__,
         _AgentExecutor.__module__,
     ):
         sys.modules[_mod_name].__dict__.update(_resolve_namespace)
 
+    from crewai.agents.crew_agent_executor import (
+        CrewAgentExecutor as _CrewAgentExecutor,
+    )
     from crewai.tasks.conditional_task import ConditionalTask as _ConditionalTask
 
+    _BaseAgentExecutor.model_rebuild(force=True, _types_namespace=_full_namespace)
     _BaseAgent.model_rebuild(force=True, _types_namespace=_full_namespace)
     Task.model_rebuild(force=True, _types_namespace=_full_namespace)
     _ConditionalTask.model_rebuild(force=True, _types_namespace=_full_namespace)
+    _CrewAgentExecutor.model_rebuild(force=True, _types_namespace=_full_namespace)
     Crew.model_rebuild(force=True, _types_namespace=_full_namespace)
     Flow.model_rebuild(force=True, _types_namespace=_full_namespace)
     _AgentExecutor.model_rebuild(force=True, _types_namespace=_full_namespace)
 
     from typing import Annotated
 
-    from pydantic import Discriminator, RootModel, Tag
+    from pydantic import Field
+
+    from crewai.state.runtime import RuntimeState
 
     Entity = Annotated[
-        Annotated[Flow, Tag("flow")]  # type: ignore[type-arg]
-        | Annotated[Crew, Tag("crew")]
-        | Annotated[Agent, Tag("agent")],
-        Discriminator(_entity_discriminator),
+        Flow | Crew | Agent,  # type: ignore[type-arg]
+        Field(discriminator="entity_type"),
     ]
-    RuntimeState = RootModel[list[Entity]]
+
+    RuntimeState.model_rebuild(
+        force=True,
+        _types_namespace={**_full_namespace, "Entity": Entity},
+    )
 
     try:
         Agent.model_rebuild(force=True, _types_namespace=_full_namespace)
@@ -205,6 +225,7 @@ def __getattr__(name: str) -> Any:
     "BaseLLM",
     "Crew",
     "CrewOutput",
+    "Entity",
     "ExecutionContext",
     "Flow",
     "Knowledge",

lib/crewai/src/crewai/agent/core.py

@@ -27,7 +27,6 @@
     BeforeValidator,
     ConfigDict,
     Field,
-    InstanceOf,
     PrivateAttr,
     model_validator,
 )
@@ -195,12 +194,12 @@ class Agent(BaseAgent):
     llm: Annotated[
         str | BaseLLM | None,
         BeforeValidator(_validate_llm_ref),
-        PlainSerializer(_serialize_llm_ref, return_type=str | None, when_used="json"),
+        PlainSerializer(_serialize_llm_ref, return_type=dict | None, when_used="json"),
     ] = Field(description="Language model that will run the agent.", default=None)
     function_calling_llm: Annotated[
         str | BaseLLM | None,
         BeforeValidator(_validate_llm_ref),
-        PlainSerializer(_serialize_llm_ref, return_type=str | None, when_used="json"),
+        PlainSerializer(_serialize_llm_ref, return_type=dict | None, when_used="json"),
     ] = Field(description="Language model that will run the agent.", default=None)
     system_template: str | None = Field(
         default=None, description="System format for the agent."
@@ -297,8 +296,8 @@ class Agent(BaseAgent):
         Can be a single A2AConfig/A2AClientConfig/A2AServerConfig, or a list of any number of A2AConfig/A2AClientConfig with a single A2AServerConfig.
         """,
     )
-    agent_executor: InstanceOf[CrewAgentExecutor] | InstanceOf[AgentExecutor] | None = (
-        Field(default=None, description="An instance of the CrewAgentExecutor class.")
+    agent_executor: CrewAgentExecutor | AgentExecutor | None = Field(
+        default=None, description="An instance of the CrewAgentExecutor class."
     )
     executor_class: Annotated[
         type[CrewAgentExecutor] | type[AgentExecutor],
@@ -1011,10 +1010,10 @@ def create_agent_executor(
                 )
             self.agent_executor = self.executor_class(
                 llm=self.llm,
-                task=task,  # type: ignore[arg-type]
+                task=task,
                 i18n=self.i18n,
                 agent=self,
-                crew=self.crew,  # type: ignore[arg-type]
+                crew=self.crew,
                 tools=parsed_tools,
                 prompt=prompt,
                 original_tools=raw_tools,
@@ -1057,7 +1056,8 @@ def _update_executor_parameters(
         if self.agent_executor is None:
             raise RuntimeError("Agent executor is not initialized.")
 
-        self.agent_executor.task = task
+        if task is not None:
+            self.agent_executor.task = task
         self.agent_executor.tools = tools
         self.agent_executor.original_tools = raw_tools
         self.agent_executor.prompt = prompt
@@ -1076,7 +1076,7 @@ def _update_executor_parameters(
         self.agent_executor.tools_handler = self.tools_handler
         self.agent_executor.request_within_rpm_limit = rpm_limit_fn
 
-        if self.agent_executor.llm:
+        if isinstance(self.agent_executor.llm, BaseLLM):
             existing_stop = getattr(self.agent_executor.llm, "stop", [])
             self.agent_executor.llm.stop = list(
                 set(

lib/crewai/src/crewai/agents/agent_builder/base_agent.py

@@ -14,8 +14,8 @@
     BaseModel,
     BeforeValidator,
     Field,
-    InstanceOf,
     PrivateAttr,
+    SerializeAsAny,
     field_validator,
     model_validator,
 )
@@ -24,7 +24,7 @@
 from typing_extensions import Self
 
 from crewai.agent.internal.meta import AgentMeta
-from crewai.agents.agent_builder.base_agent_executor_mixin import CrewAgentExecutorMixin
+from crewai.agents.agent_builder.base_agent_executor import BaseAgentExecutor
 from crewai.agents.agent_builder.utilities.base_token_process import TokenProcess
 from crewai.agents.cache.cache_handler import CacheHandler
 from crewai.agents.tools_handler import ToolsHandler
@@ -51,6 +51,7 @@
 if TYPE_CHECKING:
     from crewai.context import ExecutionContext
     from crewai.crew import Crew
+    from crewai.state.provider.core import BaseProvider
 
 
 def _validate_crew_ref(value: Any) -> Any:
@@ -63,7 +64,31 @@ def _serialize_crew_ref(value: Any) -> str | None:
     return str(value.id) if hasattr(value, "id") else str(value)
 
 
+_LLM_TYPE_REGISTRY: dict[str, str] = {
+    "base": "crewai.llms.base_llm.BaseLLM",
+    "litellm": "crewai.llm.LLM",
+    "openai": "crewai.llms.providers.openai.completion.OpenAICompletion",
+    "anthropic": "crewai.llms.providers.anthropic.completion.AnthropicCompletion",
+    "azure": "crewai.llms.providers.azure.completion.AzureCompletion",
+    "bedrock": "crewai.llms.providers.bedrock.completion.BedrockCompletion",
+    "gemini": "crewai.llms.providers.gemini.completion.GeminiCompletion",
+}
+
+
 def _validate_llm_ref(value: Any) -> Any:
+    if isinstance(value, dict):
+        import importlib
+
+        llm_type = value.get("llm_type")
+        if not llm_type or llm_type not in _LLM_TYPE_REGISTRY:
+            raise ValueError(
+                f"Unknown or missing llm_type: {llm_type!r}. "
+                f"Expected one of {list(_LLM_TYPE_REGISTRY)}"
+            )
+        dotted = _LLM_TYPE_REGISTRY[llm_type]
+        mod_path, cls_name = dotted.rsplit(".", 1)
+        cls = getattr(importlib.import_module(mod_path), cls_name)
+        return cls(**value)
     return value
 
 
@@ -75,12 +100,37 @@ def _resolve_agent(value: Any, info: Any) -> Any:
     return Agent.model_validate(value, context=getattr(info, "context", None))
 
 
-def _serialize_llm_ref(value: Any) -> str | None:
+_EXECUTOR_TYPE_REGISTRY: dict[str, str] = {
+    "base": "crewai.agents.agent_builder.base_agent_executor.BaseAgentExecutor",
+    "crew": "crewai.agents.crew_agent_executor.CrewAgentExecutor",
+    "experimental": "crewai.experimental.agent_executor.AgentExecutor",
+}
+
+
+def _validate_executor_ref(value: Any) -> Any:
+    if isinstance(value, dict):
+        import importlib
+
+        executor_type = value.get("executor_type")
+        if not executor_type or executor_type not in _EXECUTOR_TYPE_REGISTRY:
+            raise ValueError(
+                f"Unknown or missing executor_type: {executor_type!r}. "
+                f"Expected one of {list(_EXECUTOR_TYPE_REGISTRY)}"
+            )
+        dotted = _EXECUTOR_TYPE_REGISTRY[executor_type]
+        mod_path, cls_name = dotted.rsplit(".", 1)
+        cls = getattr(importlib.import_module(mod_path), cls_name)
+        return cls.model_validate(value)
+    return value
+
+
+def _serialize_llm_ref(value: Any) -> dict[str, Any] | None:
     if value is None:
         return None
     if isinstance(value, str):
-        return value
-    return getattr(value, "model", str(value))
+        return {"model": value}
+    result: dict[str, Any] = value.model_dump()
+    return result
 
 
 _SLUG_RE: Final[re.Pattern[str]] = re.compile(
@@ -197,13 +247,19 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
     max_iter: int = Field(
         default=25, description="Maximum iterations for an agent to execute a task"
     )
-    agent_executor: InstanceOf[CrewAgentExecutorMixin] | None = Field(
+    agent_executor: SerializeAsAny[BaseAgentExecutor] | None = Field(
         default=None, description="An instance of the CrewAgentExecutor class."
     )
+
+    @field_validator("agent_executor", mode="before")
+    @classmethod
+    def _validate_agent_executor(cls, v: Any) -> Any:
+        return _validate_executor_ref(v)
+
     llm: Annotated[
         str | BaseLLM | None,
         BeforeValidator(_validate_llm_ref),
-        PlainSerializer(_serialize_llm_ref, return_type=str | None, when_used="json"),
+        PlainSerializer(_serialize_llm_ref, return_type=dict | None, when_used="json"),
     ] = Field(default=None, description="Language model that will run the agent.")
     crew: Annotated[
         Crew | str | None,
@@ -276,6 +332,30 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
     )
     execution_context: ExecutionContext | None = Field(default=None)
 
+    @classmethod
+    def from_checkpoint(
+        cls, path: str, *, provider: BaseProvider | None = None
+    ) -> Self:
+        """Restore an Agent from a checkpoint file."""
+        from crewai.context import apply_execution_context
+        from crewai.state.provider.json_provider import JsonProvider
+        from crewai.state.runtime import RuntimeState
+
+        state = RuntimeState.from_checkpoint(
+            path,
+            provider=provider or JsonProvider(),
+            context={"from_checkpoint": True},
+        )
+        for entity in state.root:
+            if isinstance(entity, cls):
+                if entity.execution_context is not None:
+                    apply_execution_context(entity.execution_context)
+                if entity.agent_executor is not None:
+                    entity.agent_executor.agent = entity
+                    entity.agent_executor._resuming = True
+                return entity
+        raise ValueError(f"No {cls.__name__} found in checkpoint: {path}")
+
     @model_validator(mode="before")
     @classmethod
     def process_model_config(cls, values: Any) -> dict[str, Any]: