92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
"""Closed-loop self-evolution engine."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
import time
|
|
import uuid
|
|
from typing import Any, Callable
|
|
|
|
|
|
class EvolutionStage(str, Enum):
|
|
DETECT = "detect"
|
|
PLAN = "plan"
|
|
SCHEDULE = "schedule"
|
|
EXECUTE = "execute"
|
|
VALIDATE = "validate"
|
|
VOTE = "vote"
|
|
PATCH = "patch"
|
|
LEARN = "learn"
|
|
|
|
|
|
@dataclass
|
|
class EvolutionRun:
|
|
run_id: str = field(default_factory=lambda: str(uuid.uuid4()))
|
|
stage: EvolutionStage = EvolutionStage.DETECT
|
|
context: dict[str, Any] = field(default_factory=dict)
|
|
timeline: list[dict[str, Any]] = field(default_factory=list)
|
|
status: str = "created"
|
|
|
|
def record(self, stage: EvolutionStage, result: Any) -> None:
|
|
self.stage = stage
|
|
self.timeline.append({"stage": stage.value, "result": result, "ts": time.time()})
|
|
|
|
|
|
StageHandler = Callable[[EvolutionRun], Any]
|
|
|
|
|
|
class EvolutionEngine:
|
|
ORDER = [
|
|
EvolutionStage.DETECT,
|
|
EvolutionStage.PLAN,
|
|
EvolutionStage.SCHEDULE,
|
|
EvolutionStage.EXECUTE,
|
|
EvolutionStage.VALIDATE,
|
|
EvolutionStage.VOTE,
|
|
EvolutionStage.PATCH,
|
|
EvolutionStage.LEARN,
|
|
]
|
|
|
|
def __init__(self) -> None:
|
|
self._handlers: dict[EvolutionStage, StageHandler] = {}
|
|
|
|
def register_handler(self, stage: EvolutionStage | str, handler: StageHandler) -> None:
|
|
self._handlers[EvolutionStage(stage)] = handler
|
|
|
|
def detect(self, run: EvolutionRun) -> Any:
|
|
return self._handle(EvolutionStage.DETECT, run, {"detected": True, "signals": run.context.get("signals", [])})
|
|
|
|
def plan(self, run: EvolutionRun) -> Any:
|
|
return self._handle(EvolutionStage.PLAN, run, {"plan": run.context.get("plan", [])})
|
|
|
|
def schedule(self, run: EvolutionRun) -> Any:
|
|
return self._handle(EvolutionStage.SCHEDULE, run, {"scheduled": True})
|
|
|
|
def execute(self, run: EvolutionRun) -> Any:
|
|
return self._handle(EvolutionStage.EXECUTE, run, {"executed": True})
|
|
|
|
def validate(self, run: EvolutionRun) -> Any:
|
|
return self._handle(EvolutionStage.VALIDATE, run, {"valid": True})
|
|
|
|
def vote(self, run: EvolutionRun) -> Any:
|
|
return self._handle(EvolutionStage.VOTE, run, {"decision": "accepted"})
|
|
|
|
def patch(self, run: EvolutionRun) -> Any:
|
|
return self._handle(EvolutionStage.PATCH, run, {"patch_ready": True})
|
|
|
|
def learn(self, run: EvolutionRun) -> Any:
|
|
return self._handle(EvolutionStage.LEARN, run, {"learned": True})
|
|
|
|
def run_cycle(self, context: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
run = EvolutionRun(context=dict(context or {}), status="running")
|
|
for stage in self.ORDER:
|
|
result = getattr(self, stage.value)(run)
|
|
run.record(stage, result)
|
|
run.status = "completed"
|
|
return {"run_id": run.run_id, "status": run.status, "timeline": run.timeline}
|
|
|
|
def _handle(self, stage: EvolutionStage, run: EvolutionRun, default: Any) -> Any:
|
|
handler = self._handlers.get(stage)
|
|
return handler(run) if handler else default
|