{"record":{"id":"c83ce95274a1bd92","repo":"microsoft/autogen","slug":"agent-load-state-is-not-yet-implemented","errorCode":null,"errorMessage":"Agent load_state is not yet implemented.","messagePattern":"Agent load_state is not yet implemented\\.","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"python/packages/autogen-ext/src/autogen_ext/runtimes/grpc/_worker_runtime.py","lineNumber":505,"sourceCode":"            task = asyncio.create_task(self._send_message(runtime_message, \"publish\", topic_id, telemetry_metadata))\n            self._background_tasks.add(task)\n            task.add_done_callback(self._raise_on_exception)\n            task.add_done_callback(self._background_tasks.discard)\n\n    async def save_state(self) -> Mapping[str, Any]:\n        raise NotImplementedError(\"Saving state is not yet implemented.\")\n\n    async def load_state(self, state: Mapping[str, Any]) -> None:\n        raise NotImplementedError(\"Loading state is not yet implemented.\")\n\n    async def agent_metadata(self, agent: AgentId) -> AgentMetadata:\n        raise NotImplementedError(\"Agent metadata is not yet implemented.\")\n\n    async def agent_save_state(self, agent: AgentId) -> Mapping[str, Any]:\n        raise NotImplementedError(\"Agent save_state is not yet implemented.\")\n\n    async def agent_load_state(self, agent: AgentId, state: Mapping[str, Any]) -> None:\n        raise NotImplementedError(\"Agent load_state is not yet implemented.\")\n\n    async def _get_new_request_id(self) -> str:\n        async with self._pending_requests_lock:\n            self._next_request_id += 1\n            return str(self._next_request_id)\n\n    async def _process_request(self, request: agent_worker_pb2.RpcRequest) -> None:\n        assert self._host_connection is not None\n        recipient = AgentId(request.target.type, request.target.key)\n        sender: AgentId | None = None\n        if request.HasField(\"source\"):\n            sender = AgentId(request.source.type, request.source.key)\n            logging.info(f\"Processing request from {sender} to {recipient}\")\n        else:\n            logging.info(f\"Processing request from unknown source to {recipient}\")\n\n        # Deserialize the message.\n        message = self._serialization_registry.deserialize(","sourceCodeStart":487,"sourceCodeEnd":523,"githubUrl":"https://github.com/microsoft/autogen/blob/027ecf0a379bcc1d09956d46d12d44a3ad9cee14/python/packages/autogen-ext/src/autogen_ext/runtimes/grpc/_worker_runtime.py#L487-L523","documentation":"GrpcWorkerAgentRuntime.agent_load_state() is a stub that raises NotImplementedError: restoring an individual agent's state through the runtime is not supported for distributed workers, symmetric with agent_save_state.","triggerScenarios":"Calling `await runtime.agent_load_state(agent_id, state)` on GrpcWorkerAgentRuntime, usually from generic restore/checkpoint code.","commonSituations":"Restore logic written for SingleThreadedAgentRuntime reused against gRPC workers; resuming conversations by reloading agent state through the runtime API.","solutions":["Send a custom LoadState message to the agent and let it restore itself","Hydrate agent state at construction/registration time from your own store","Skip runtime-level restore for gRPC runtimes via a type check"],"exampleFix":"# before\nawait runtime.agent_load_state(agent_id, state)  # NotImplementedError\n\n# after\nawait runtime.send_message(LoadStateRequest(state), agent_id)  # agent applies it","handlingStrategy":"type-guard","validationCode":"from autogen_ext.runtimes.grpc import GrpcWorkerAgentRuntime\n\nif isinstance(runtime, GrpcWorkerAgentRuntime):\n    await runtime.send_message(LoadStateRequest(state), agent_id)\nelse:\n    await runtime.agent_load_state(agent_id, state)","typeGuard":"def runtime_has_per_agent_state(runtime: object) -> bool:\n    return not isinstance(runtime, GrpcWorkerAgentRuntime)","tryCatchPattern":"try:\n    await runtime.agent_load_state(agent_id, state)\nexcept NotImplementedError:\n    await runtime.send_message(LoadStateRequest(state), agent_id)","preventionTips":["Give agents a LoadState message handler for distributed deployments","Hydrate state at construction from your own store as a simpler alternative","Branch persistence code on runtime capabilities"],"tags":["python","autogen","grpc","not-implemented","state-management"],"backgroundTag":null,"analyzedSha":"027ecf0a379bcc1d09956d46d12d44a3ad9cee14","analyzedAt":"2026-08-15T03:38:00.719Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}