Restore a specific version of an agent from history Args: agent_name: Name of the agent version: Version string to restore auto_initialize: Whether to automatically initialize the restored agent Returns: AgentConfi
(
self, agent_name: str, version: str, auto_initialize: bool = True
)
| 983 | return False |
| 984 | |
| 985 | async def restore( |
| 986 | self, agent_name: str, version: str, auto_initialize: bool = True |
| 987 | ) -> Optional[AgentConfig]: |
| 988 | """Restore a specific version of an agent from history |
| 989 | |
| 990 | Args: |
| 991 | agent_name: Name of the agent |
| 992 | version: Version string to restore |
| 993 | auto_initialize: Whether to automatically initialize the restored agent |
| 994 | |
| 995 | Returns: |
| 996 | AgentConfig of the restored version, or None if not found |
| 997 | """ |
| 998 | # Look up version from dict-based history (O(1) lookup) |
| 999 | version_config = None |
| 1000 | if agent_name in self._agent_history_versions: |
| 1001 | version_config = self._agent_history_versions[agent_name].get(version) |
| 1002 | |
| 1003 | if version_config is None: |
| 1004 | logger.warning(f"| ⚠️ Version {version} not found for agent {agent_name}") |
| 1005 | return None |
| 1006 | |
| 1007 | # Create a copy to avoid modifying the history |
| 1008 | restored_config = AgentConfig(**version_config.model_dump()) |
| 1009 | |
| 1010 | # Set as current active config |
| 1011 | self._agent_configs[agent_name] = restored_config |
| 1012 | |
| 1013 | # Update version manager current version |
| 1014 | version_history = await version_manager.get_version_history("agent", agent_name) |
| 1015 | if version_history: |
| 1016 | # Check if version exists in version history, if not register it |
| 1017 | if version not in version_history.versions: |
| 1018 | await version_manager.register_version("agent", agent_name, version) |
| 1019 | version_history.current_version = version |
| 1020 | else: |
| 1021 | # If version history doesn't exist, register the version first |
| 1022 | await version_manager.register_version("agent", agent_name, version) |
| 1023 | |
| 1024 | # Initialize if requested |
| 1025 | if auto_initialize and restored_config.cls is not None: |
| 1026 | await self.build(restored_config) |
| 1027 | |
| 1028 | # Persist to JSON (current_version changes) |
| 1029 | await self.save_to_json() |
| 1030 | |
| 1031 | logger.info(f"| 🔄 Restored agent {agent_name} to version {version}") |
| 1032 | return restored_config |
| 1033 | |
| 1034 | async def save_contract(self, agent_names: Optional[List[str]] = None): |
| 1035 | """Save the contract for an agent""" |
nothing calls this directly
no test coverage detected