Anton
052591281d
- Complete agent system with planning and execution - Bridge API with FastAPI server - WhatsApp-style mobile dashboard - Enhanced web search with query improvement - Security guards and policy checks - Comprehensive documentation
63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
"""Controller run state model and persistence."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import asdict, dataclass, field
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
@dataclass
|
|
class ControllerState:
|
|
controller_run_id: str
|
|
original_goal: str
|
|
decision_mode: str # "real" | "stub"
|
|
iteration_count: int = 0
|
|
decisions: list[dict[str, Any]] = field(default_factory=list)
|
|
linked_bridge_run_ids: list[str] = field(default_factory=list)
|
|
bridge_run_results: list[dict[str, Any]] = field(default_factory=list)
|
|
final_status: str = "running" # running | done | failed | stopped
|
|
recovery_attempted: bool = False # True after rewrite rule applied
|
|
recovery_reason: Optional[str] = None # rewrite_explain() text or skip reason
|
|
rewritten_goal: Optional[str] = None # goal after rewrite (if recovery applied)
|
|
created_at: str = field(default_factory=_now)
|
|
updated_at: str = field(default_factory=_now)
|
|
|
|
@property
|
|
def trace_id(self) -> str:
|
|
"""trace_id == controller_run_id by definition."""
|
|
return self.controller_run_id
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
def touch(self) -> None:
|
|
self.updated_at = _now()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Persistence helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def state_path(runs_dir: Path, ctrl_id: str) -> Path:
|
|
return runs_dir / ctrl_id / "controller_state.json"
|
|
|
|
|
|
def save(state: ControllerState, runs_dir: Path) -> None:
|
|
p = state_path(runs_dir, state.controller_run_id)
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
p.write_text(json.dumps(state.to_dict(), indent=2), encoding="utf-8")
|
|
|
|
|
|
def load(runs_dir: Path, ctrl_id: str) -> Optional[ControllerState]:
|
|
p = state_path(runs_dir, ctrl_id)
|
|
if not p.exists():
|
|
return None
|
|
data = json.loads(p.read_text(encoding="utf-8"))
|
|
return ControllerState(**data)
|