From fc92e1aa74f2cee3ac98224681f9e85854b45b9b Mon Sep 17 00:00:00 2001
From: Ed_ <edwardgz@gmail.com>
Date: Tue, 16 Jun 2026 18:59:38 -0400
Subject: [PATCH] feat(failcount): add FailcountState + FailcountConfig
 dataclasses + all stub functions

---
 scripts/tier2/failcount.py | 141 +++++++++++++++++++++++++++++++++++++
 1 file changed, 141 insertions(+)
 create mode 100644 scripts/tier2/failcount.py

diff --git a/scripts/tier2/failcount.py b/scripts/tier2/failcount.py
new file mode 100644
index 00000000..563e0246
--- /dev/null
+++ b/scripts/tier2/failcount.py
@@ -0,0 +1,141 @@
+"""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)