{"record":{"id":"443b29a7d9f0b018","repo":"microsoft/autogen","slug":"saving-state-is-not-yet-implemented","errorCode":null,"errorMessage":"Saving state is not yet implemented.","messagePattern":"Saving 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":493,"sourceCode":"                runtime_message = agent_worker_pb2.Message(\n                    cloudEvent=cloudevent_pb2.CloudEvent(\n                        id=message_id,\n                        spec_version=\"1.0\",\n                        type=topic_id.type,\n                        source=topic_id.source,\n                        attributes=attributes,\n                        proto_data=any_proto,\n                    )\n                )\n\n            telemetry_metadata = get_telemetry_grpc_metadata()\n            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","sourceCodeStart":475,"sourceCodeEnd":511,"githubUrl":"https://github.com/microsoft/autogen/blob/027ecf0a379bcc1d09956d46d12d44a3ad9cee14/python/packages/autogen-ext/src/autogen_ext/runtimes/grpc/_worker_runtime.py#L475-L511","documentation":"GrpcWorkerAgentRuntime does not implement global state persistence: save_state() unconditionally raises NotImplementedError. In the distributed gRPC design, runtime state lives with agents, so snapshotting the whole runtime is intentionally unimplemented.","triggerScenarios":"Calling `await runtime.save_state()` on a GrpcWorkerAgentRuntime, directly or via generic framework code (e.g. checkpointing helpers) that calls save_state on any AgentRuntime.","commonSituations":"Reusing checkpoint/restore code written for SingleThreadedAgentRuntime with the gRPC runtime; migrating an app to distributed workers while keeping whole-runtime state snapshots.","solutions":["Remove runtime-level save_state usage with GrpcWorkerAgentRuntime; persist agent state through your own storage in each agent's handlers","If you need checkpointing, serialize the relevant application state yourself before stopping workers","Use SingleThreadedAgentRuntime for local scenarios that require save_state/load_state"],"exampleFix":"# before\nstate = await runtime.save_state()  # NotImplementedError\n\n# after (persist within your agents / app layer)\nawait my_store.put(\"checkpoint\", my_app_state)","handlingStrategy":"type-guard","validationCode":"from autogen_ext.runtimes.grpc import GrpcWorkerAgentRuntime\n\nif isinstance(runtime, GrpcWorkerAgentRuntime):\n    raise NotImplementedError(\"use app-level persistence for gRPC runtimes\")\nstate = await runtime.save_state()","typeGuard":"def runtime_supports_state(runtime: object) -> bool:\n    return not isinstance(runtime, GrpcWorkerAgentRuntime)","tryCatchPattern":"try:\n    state = await runtime.save_state()\nexcept NotImplementedError:\n    state = None  # gRPC runtime: persist via app-level storage instead","preventionTips":["Feature-check runtime capabilities before checkpointing","Persist state inside agents or an application store for distributed deployments","Keep local (SingleThreadedAgentRuntime) tests for state save/restore flows"],"tags":["python","autogen","grpc","not-implemented","state-management"],"backgroundTag":null,"analyzedSha":"027ecf0a379bcc1d09956d46d12d44a3ad9cee14","analyzedAt":"2026-08-15T03:38:00.719Z","schemaVersion":2},"datasetVersion":"2026-08-15T17:31:12.345Z"}