Add Talos runtime evaluation contract

talos/__init__.py

@@ -1,5 +1,19 @@
 """Talos — deterministic WIP enforcement scoring."""
 
-from .scorer import compute_talos_level, load_cycles, save_cycles
+from .models import TalosEvaluation, TalosPolicy, TalosSnapshot, TalosState
+from .runtime import evaluate_snapshot, load_cycles, load_policy, load_state, save_cycles, save_state
+from .scorer import compute_talos_level
 
-__all__ = ["compute_talos_level", "load_cycles", "save_cycles"]
+__all__ = [
+    "TalosEvaluation",
+    "TalosPolicy",
+    "TalosSnapshot",
+    "TalosState",
+    "compute_talos_level",
+    "evaluate_snapshot",
+    "load_cycles",
+    "load_policy",
+    "load_state",
+    "save_cycles",
+    "save_state",
+]

talos/models.py

@@ -0,0 +1,100 @@
+"""Data models for Talos policy, snapshots, state, and evaluation results."""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass, field
+from typing import Any, Dict, List
+
+
+@dataclass
+class TalosPolicy:
+    """Configurable policy knobs for deterministic Talos scoring."""
+
+    friction_ratio: float = 0.8
+    write_block_cycles: int = 2
+
+    @classmethod
+    def from_dict(cls, payload: Dict[str, Any]) -> "TalosPolicy":
+        return cls(
+            friction_ratio=float(payload.get("friction_ratio", 0.8)),
+            write_block_cycles=int(payload.get("write_block_cycles", 2)),
+        )
+
+    def to_dict(self) -> Dict[str, Any]:
+        return asdict(self)
+
+
+@dataclass
+class TalosSnapshot:
+    """Input contract for a single Talos evaluation cycle."""
+
+    wip_total: int
+    global_max: int
+    at_cap_projects: List[str] = field(default_factory=list)
+    backlog_total: int = 0
+    backlog_delta: int = 0
+    projects: List[Dict[str, Any]] = field(default_factory=list)
+    source: str = "manual"
+    timestamp: str = ""
+
+    @classmethod
+    def from_dict(cls, payload: Dict[str, Any]) -> "TalosSnapshot":
+        at_cap_projects = payload.get("at_cap_projects", [])
+        if not isinstance(at_cap_projects, list):
+            raise TypeError(f"at_cap_projects must be a list, got {type(at_cap_projects)}")
+        projects = payload.get("projects", [])
+        if not isinstance(projects, list):
+            raise TypeError(f"projects must be a list, got {type(projects)}")
+        return cls(
+            wip_total=int(payload["wip_total"]),
+            global_max=int(payload["global_max"]),
+            at_cap_projects=at_cap_projects.copy(),
+            backlog_total=int(payload.get("backlog_total", 0)),
+            backlog_delta=int(payload.get("backlog_delta", 0)),
+            projects=projects.copy(),
+            source=str(payload.get("source", "manual")),
+            timestamp=str(payload.get("timestamp", "")),
+        )
+
+    def to_dict(self) -> Dict[str, Any]:
+        return asdict(self)
+
+
+@dataclass
+class TalosState:
+    """Persisted cross-cycle state for Talos."""
+
+    level: int = 0
+    count: int = 0
+    last_backlog: int = 0
+
+    @classmethod
+    def from_dict(cls, payload: Dict[str, Any]) -> "TalosState":
+        return cls(
+            level=int(payload.get("level", 0)),
+            count=int(payload.get("count", 0)),
+            last_backlog=int(payload.get("last_backlog", 0)),
+        )
+
+    def to_dict(self) -> Dict[str, Any]:
+        return asdict(self)
+
+
+@dataclass
+class TalosEvaluation:
+    """Structured output contract for a Talos evaluation cycle."""
+
+    level: int
+    previous_level: int
+    cycles_at_level: int
+    wip_total: int
+    global_max: int
+    backlog_total: int
+    backlog_delta: int
+    at_cap_projects: List[str]
+    reasons: List[str]
+    source: str = "manual"
+    timestamp: str = ""
+
+    def to_dict(self) -> Dict[str, Any]:
+        return asdict(self)

talos/runtime.py

@@ -0,0 +1,157 @@
+"""State, history, and evaluation helpers for Talos."""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+from pathlib import Path
+from typing import List, Optional
+
+from .models import TalosEvaluation, TalosPolicy, TalosSnapshot, TalosState
+from .scorer import compute_talos_level
+
+logger = logging.getLogger(__name__)
+
+
+def _default_data_dir() -> Path:
+    return Path.home() / ".config" / "talos"
+
+
+def _path_from_env(name: str, default_name: str) -> Path:
+    configured = os.environ.get(name)
+    if configured:
+        return Path(configured).expanduser()
+    return _default_data_dir() / default_name
+
+
+TALOS_CYCLES_FILE = _path_from_env("TALOS_CYCLES_FILE", "cycles.json")
+TALOS_STATUS_FILE = _path_from_env("TALOS_STATUS_FILE", "status.json")
+TALOS_HISTORY_FILE = _path_from_env("TALOS_HISTORY_FILE", "history.jsonl")
+TALOS_POLICY_FILE = _path_from_env("TALOS_POLICY_FILE", "policy.json")
+
+
+def _atomic_write_text(target: Path, payload: str) -> None:
+    target.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+    tmp_path = target.with_suffix(target.suffix + ".tmp")
+    tmp_path.write_text(payload, encoding="utf-8")
+    tmp_path.replace(target)
+
+
+def load_state(path: Optional[Path] = None) -> TalosState:
+    target = path or TALOS_CYCLES_FILE
+    try:
+        return TalosState.from_dict(json.loads(target.read_text()))
+    except Exception:
+        return TalosState()
+
+
+def save_state(level: int, count: int, last_backlog: int, path: Optional[Path] = None) -> None:
+    target = path or TALOS_CYCLES_FILE
+    try:
+        _atomic_write_text(
+            target,
+            json.dumps({"level": level, "count": count, "last_backlog": last_backlog}, sort_keys=True),
+        )
+    except Exception:
+        logger.warning("Failed to persist Talos state to %s", target, exc_info=True)
+
+
+def load_policy(path: Optional[Path] = None) -> TalosPolicy:
+    target = path or TALOS_POLICY_FILE
+    try:
+        return TalosPolicy.from_dict(json.loads(target.read_text()))
+    except Exception:
+        return TalosPolicy()
+
+
+def save_status(evaluation: TalosEvaluation, path: Optional[Path] = None) -> None:
+    target = path or TALOS_STATUS_FILE
+    try:
+        _atomic_write_text(target, json.dumps(evaluation.to_dict(), indent=2, sort_keys=True))
+    except Exception:
+        logger.warning("Failed to persist Talos status to %s", target, exc_info=True)
+
+
+def append_history(evaluation: TalosEvaluation, path: Optional[Path] = None) -> None:
+    target = path or TALOS_HISTORY_FILE
+    try:
+        target.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+        with target.open("a", encoding="utf-8") as handle:
+            handle.write(json.dumps(evaluation.to_dict(), sort_keys=True) + "\n")
+    except Exception:
+        logger.warning("Failed to append Talos history to %s", target, exc_info=True)
+
+
+def read_snapshot(path: Path) -> TalosSnapshot:
+    return TalosSnapshot.from_dict(json.loads(path.read_text()))
+
+
+def explain_level(snapshot: TalosSnapshot, level: int, policy: Optional[TalosPolicy] = None) -> List[str]:
+    active_policy = policy or TalosPolicy()
+    reasons: List[str] = []
+    ratio_threshold = int(snapshot.global_max * active_policy.friction_ratio) if snapshot.global_max > 0 else 0
+    if snapshot.global_max <= 0:
+        return ["global_max<=0 so Talos stayed at clean level 0"]
+    if snapshot.wip_total >= snapshot.global_max:
+        reasons.append("wip_total reached or exceeded global_max")
+    if snapshot.at_cap_projects:
+        reasons.append("one or more projects are already at cap")
+    if snapshot.wip_total >= ratio_threshold and snapshot.wip_total < snapshot.global_max:
+        reasons.append("wip_total crossed the friction threshold")
+    if snapshot.backlog_delta > 0:
+        reasons.append("backlog is growing")
+    if level == 0 and not reasons:
+        reasons.append("wip and backlog are within policy thresholds")
+    if level == 3:
+        reasons.append("global cap pressure persisted long enough to block writes")
+    return reasons
+
+
+def evaluate_snapshot(
+    snapshot: TalosSnapshot,
+    prior_state: Optional[TalosState] = None,
+    policy: Optional[TalosPolicy] = None,
+) -> TalosEvaluation:
+    active_state = prior_state or TalosState()
+    active_policy = policy or TalosPolicy()
+    persisted_pressure_cycles = (
+        active_state.count if active_state.level >= 2 and snapshot.wip_total >= snapshot.global_max else 0
+    )
+    level = compute_talos_level(
+        wip_total=snapshot.wip_total,
+        global_max=snapshot.global_max,
+        at_cap_projects=snapshot.at_cap_projects,
+        backlog_delta=snapshot.backlog_delta,
+        cycles_at_current_level=persisted_pressure_cycles,
+        policy=active_policy,
+    )
+    pressure_streak_continues = active_state.level >= 2 and level >= 2 and snapshot.wip_total >= snapshot.global_max
+    cycles_at_level = active_state.count + 1 if level == active_state.level or pressure_streak_continues else 1
+    return TalosEvaluation(
+        level=level,
+        previous_level=active_state.level,
+        cycles_at_level=cycles_at_level,
+        wip_total=snapshot.wip_total,
+        global_max=snapshot.global_max,
+        backlog_total=snapshot.backlog_total,
+        backlog_delta=snapshot.backlog_delta,
+        at_cap_projects=list(snapshot.at_cap_projects),
+        reasons=explain_level(snapshot=snapshot, level=level, policy=active_policy),
+        source=snapshot.source,
+        timestamp=snapshot.timestamp,
+    )
+
+
+def next_state_from_evaluation(evaluation: TalosEvaluation) -> TalosState:
+    return TalosState(level=evaluation.level, count=evaluation.cycles_at_level, last_backlog=evaluation.backlog_total)
+
+
+def load_cycles() -> dict:
+    """Backward-compatible wrapper for loading persisted Talos cycle state."""
+    return load_state().to_dict()
+
+
+def save_cycles(level: int, count: int, last_backlog: int) -> None:
+    """Backward-compatible wrapper for saving persisted Talos cycle state."""
+    save_state(level=level, count=count, last_backlog=last_backlog)

talos/scorer.py

@@ -1,64 +1,30 @@
-"""Pure arithmetic scoring and persistence helpers for Talos."""
+"""Pure arithmetic scoring helpers for Talos."""
 
 from __future__ import annotations
 
-import json
-import logging
-import os
-from pathlib import Path
+from typing import List, Optional
 
-logger = logging.getLogger(__name__)
+from .models import TalosPolicy
 
-
-def _default_cycles_file() -> Path:
-    configured = os.environ.get("TALOS_CYCLES_FILE")
-    if configured:
-        return Path(configured).expanduser()
-    return Path.home() / ".config" / "talos" / "cycles.json"
-
-
-TALOS_CYCLES_FILE = _default_cycles_file()
+DEFAULT_POLICY = TalosPolicy()
 
 
 def compute_talos_level(
     wip_total: int,
     global_max: int,
-    at_cap_projects: list,
+    at_cap_projects: List[str],
     backlog_delta: int,
     cycles_at_current_level: int,
+    policy: Optional[TalosPolicy] = None,
 ) -> int:
-    """Return Talos enforcement level from deterministic pipeline inputs."""
+    """Return Talos enforcement level from deterministic inputs."""
+    active_policy = policy or DEFAULT_POLICY
     if global_max <= 0:
         return 0
-    if wip_total >= global_max and cycles_at_current_level >= 2:
+    if wip_total >= global_max and cycles_at_current_level >= active_policy.write_block_cycles:
         return 3
     if wip_total >= global_max or len(at_cap_projects) > 0:
         return 2
-    if wip_total >= global_max * 0.8 or backlog_delta > 0:
+    if wip_total >= global_max * active_policy.friction_ratio or backlog_delta > 0:
         return 1
     return 0
-
-
-def load_cycles() -> dict:
-    """Load persisted cycle counter, returning defaults if missing or corrupt."""
-    try:
-        data = json.loads(TALOS_CYCLES_FILE.read_text())
-        return {
-            "level": int(data.get("level", 0)),
-            "count": int(data.get("count", 0)),
-            "last_backlog": int(data.get("last_backlog", 0)),
-        }
-    except Exception:
-        return {"level": 0, "count": 0, "last_backlog": 0}
-
-
-def save_cycles(level: int, count: int, last_backlog: int) -> None:
-    """Persist cycle counter across restarts and log failures."""
-    try:
-        TALOS_CYCLES_FILE.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
-        payload = json.dumps({"level": level, "count": count, "last_backlog": last_backlog})
-        tmp_file = TALOS_CYCLES_FILE.with_suffix(f"{TALOS_CYCLES_FILE.suffix}.tmp")
-        tmp_file.write_text(payload, encoding="utf-8")
-        tmp_file.replace(TALOS_CYCLES_FILE)
-    except Exception:
-        logger.warning("Failed to persist cycles to %s", TALOS_CYCLES_FILE, exc_info=True)