142 lines
4.0 KiB
Python
142 lines
4.0 KiB
Python
"""Failure-counting module for Tier 2 autonomous runs.
|
|
|
|
Tracks three orthogonal signals: red-phase failures, green-phase failures,
|
|
and no-progress wall-clock. When any signal hits its threshold, the run
|
|
gives up and writes a failure report.
|
|
|
|
Pure logic, no external dependencies. Fully unit-testable.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import tomllib
|
|
|
|
|
|
@dataclass
|
|
class FailcountConfig:
|
|
"""Threshold configuration. Loaded from failcount.toml or defaults."""
|
|
red_phase_threshold: int = 3
|
|
green_phase_threshold: int = 3
|
|
no_progress_minutes: int = 30
|
|
|
|
|
|
@dataclass
|
|
class FailcountState:
|
|
"""The three failure signals + the no-progress timer anchor."""
|
|
red_phase_failures: int = 0
|
|
green_phase_failures: int = 0
|
|
no_progress_started_at: datetime | None = None
|
|
|
|
|
|
def _load_config_from_toml(path: Path) -> FailcountConfig:
|
|
if not path.exists():
|
|
return FailcountConfig()
|
|
with path.open("rb") as f:
|
|
data = tomllib.load(f)
|
|
return FailcountConfig(
|
|
red_phase_threshold=data.get("red_phase_threshold", 3),
|
|
green_phase_threshold=data.get("green_phase_threshold", 3),
|
|
no_progress_minutes=data.get("no_progress_minutes", 30),
|
|
)
|
|
|
|
|
|
def load_config() -> FailcountConfig:
|
|
"""Load config from `scripts/tier2/failcount.toml`, or defaults."""
|
|
toml_path = Path(__file__).parent / "failcount.toml"
|
|
return _load_config_from_toml(toml_path)
|
|
|
|
|
|
def record_red_failure(state: FailcountState) -> FailcountState:
|
|
return FailcountState(
|
|
red_phase_failures=state.red_phase_failures + 1,
|
|
green_phase_failures=state.green_phase_failures,
|
|
no_progress_started_at=state.no_progress_started_at,
|
|
)
|
|
|
|
|
|
def record_green_failure(state: FailcountState) -> FailcountState:
|
|
return FailcountState(
|
|
red_phase_failures=state.red_phase_failures,
|
|
green_phase_failures=state.green_phase_failures + 1,
|
|
no_progress_started_at=state.no_progress_started_at,
|
|
)
|
|
|
|
|
|
def record_green_success(state: FailcountState, now: datetime) -> FailcountState:
|
|
return FailcountState(
|
|
red_phase_failures=0,
|
|
green_phase_failures=0,
|
|
no_progress_started_at=now,
|
|
)
|
|
|
|
|
|
def record_commit(state: FailcountState, now: datetime) -> FailcountState:
|
|
return FailcountState(
|
|
red_phase_failures=state.red_phase_failures,
|
|
green_phase_failures=state.green_phase_failures,
|
|
no_progress_started_at=now,
|
|
)
|
|
|
|
|
|
def should_give_up(state: FailcountState, config: FailcountConfig, now: datetime) -> bool:
|
|
if state.red_phase_failures >= config.red_phase_threshold:
|
|
return True
|
|
if state.green_phase_failures >= config.green_phase_threshold:
|
|
return True
|
|
if state.no_progress_started_at is not None:
|
|
elapsed_minutes = (now - state.no_progress_started_at).total_seconds() / 60.0
|
|
if elapsed_minutes >= config.no_progress_minutes:
|
|
return True
|
|
return False
|
|
|
|
|
|
def to_dict(state: FailcountState) -> dict[str, Any]:
|
|
d = asdict(state)
|
|
if state.no_progress_started_at is not None:
|
|
d["no_progress_started_at"] = state.no_progress_started_at.isoformat()
|
|
return d
|
|
|
|
|
|
def from_dict(d: dict[str, Any]) -> FailcountState:
|
|
no_progress = d.get("no_progress_started_at")
|
|
if isinstance(no_progress, str):
|
|
no_progress = datetime.fromisoformat(no_progress)
|
|
return FailcountState(
|
|
red_phase_failures=d.get("red_phase_failures", 0),
|
|
green_phase_failures=d.get("green_phase_failures", 0),
|
|
no_progress_started_at=no_progress,
|
|
)
|
|
|
|
|
|
def _state_dir(track_name: str) -> Path:
|
|
base = os.environ.get(
|
|
"TIER2_STATE_DIR",
|
|
r"C:\Users\Ed\AppData\Local\manual_slop\tier2",
|
|
)
|
|
return Path(base) / track_name
|
|
|
|
|
|
def load_state(track_name: str) -> FailcountState:
|
|
path = _state_dir(track_name) / "state.json"
|
|
if not path.exists():
|
|
return FailcountState()
|
|
import json
|
|
with path.open("r", encoding="utf-8") as f:
|
|
return from_dict(json.load(f))
|
|
|
|
|
|
def save_state(track_name: str, state: FailcountState) -> None:
|
|
import json
|
|
d = _state_dir(track_name)
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
target = d / "state.json"
|
|
tmp = d / "state.json.tmp"
|
|
with tmp.open("w", encoding="utf-8") as f:
|
|
json.dump(to_dict(state), f, indent=2)
|
|
os.replace(tmp, target)
|