40 lines
940 B
Python
40 lines
940 B
Python
"""Agent OS lifecycle runtime manager."""
|
|
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
from typing import Dict
|
|
|
|
|
|
class AgentState(str, Enum):
|
|
OFFLINE = "offline"
|
|
STARTING = "starting"
|
|
RUNNING = "running"
|
|
DEGRADED = "degraded"
|
|
STOPPED = "stopped"
|
|
|
|
|
|
@dataclass
|
|
class RuntimeAgent:
|
|
agent_id: str
|
|
state: AgentState = AgentState.OFFLINE
|
|
|
|
|
|
class AgentRuntimeManager:
|
|
def __init__(self):
|
|
self.agents: Dict[str, RuntimeAgent] = {}
|
|
|
|
def register(self, agent_id: str):
|
|
self.agents[agent_id] = RuntimeAgent(agent_id)
|
|
|
|
def start(self, agent_id: str):
|
|
self.agents[agent_id].state = AgentState.RUNNING
|
|
|
|
def stop(self, agent_id: str):
|
|
self.agents[agent_id].state = AgentState.STOPPED
|
|
|
|
def health_check(self, agent_id: str):
|
|
return self.agents.get(agent_id, RuntimeAgent(agent_id)).state
|
|
|
|
def recover(self, agent_id: str):
|
|
self.start(agent_id)
|