From 5d1d01b06de92e097cbfbe5601f2788b99b0d721 Mon Sep 17 00:00:00 2001 From: John Malone Date: Mon, 25 May 2026 10:04:19 -0500 Subject: [PATCH 1/2] Add Talos runtime evaluation contract --- talos/__init__.py | 18 ++++- talos/models.py | 94 +++++++++++++++++++++++++ talos/runtime.py | 157 ++++++++++++++++++++++++++++++++++++++++++ talos/scorer.py | 54 +++------------ tests/test_cycles.py | 43 +++++++----- tests/test_models.py | 42 +++++++++++ tests/test_runtime.py | 75 ++++++++++++++++++++ tests/test_scorer.py | 5 ++ 8 files changed, 424 insertions(+), 64 deletions(-) create mode 100644 talos/models.py create mode 100644 talos/runtime.py create mode 100644 tests/test_models.py create mode 100644 tests/test_runtime.py diff --git a/talos/__init__.py b/talos/__init__.py index e03deef..68dc16d 100644 --- a/talos/__init__.py +++ b/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", +] diff --git a/talos/models.py b/talos/models.py new file mode 100644 index 0000000..576915a --- /dev/null +++ b/talos/models.py @@ -0,0 +1,94 @@ +"""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": + return cls( + wip_total=int(payload["wip_total"]), + global_max=int(payload["global_max"]), + at_cap_projects=list(payload.get("at_cap_projects", [])), + backlog_total=int(payload.get("backlog_total", 0)), + backlog_delta=int(payload.get("backlog_delta", 0)), + projects=list(payload.get("projects", [])), + 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) diff --git a/talos/runtime.py b/talos/runtime.py new file mode 100644 index 0000000..3adbdc5 --- /dev/null +++ b/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) diff --git a/talos/scorer.py b/talos/scorer.py index 33d5c98..4afed0e 100644 --- a/talos/scorer.py +++ b/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) diff --git a/tests/test_cycles.py b/tests/test_cycles.py index 7e073f2..838e987 100644 --- a/tests/test_cycles.py +++ b/tests/test_cycles.py @@ -1,55 +1,62 @@ from pathlib import Path -import talos.scorer as scorer +import talos.runtime as runtime def test_default_cycles_file_uses_secure_home_config_location(monkeypatch) -> None: monkeypatch.delenv("TALOS_CYCLES_FILE", raising=False) - monkeypatch.setattr(scorer.Path, "home", lambda: Path("/tmp/fake-home")) - assert scorer._default_cycles_file() == Path("/tmp/fake-home/.config/talos/cycles.json") + monkeypatch.setattr(runtime.Path, "home", lambda: Path("/tmp/fake-home")) + assert runtime._path_from_env("TALOS_CYCLES_FILE", "cycles.json") == Path("/tmp/fake-home/.config/talos/cycles.json") def test_default_cycles_file_honors_environment_override(monkeypatch) -> None: monkeypatch.setenv("TALOS_CYCLES_FILE", "/tmp/custom/talos-cycles.json") - assert scorer._default_cycles_file() == Path("/tmp/custom/talos-cycles.json") + assert runtime._path_from_env("TALOS_CYCLES_FILE", "cycles.json") == Path("/tmp/custom/talos-cycles.json") def test_load_cycles_returns_defaults_when_file_is_missing(tmp_path: Path, monkeypatch) -> None: - monkeypatch.setattr(scorer, "TALOS_CYCLES_FILE", tmp_path / "missing.json") - assert scorer.load_cycles() == {"level": 0, "count": 0, "last_backlog": 0} + monkeypatch.setattr(runtime, "TALOS_CYCLES_FILE", tmp_path / "missing.json") + assert runtime.load_cycles() == {"level": 0, "count": 0, "last_backlog": 0} def test_load_cycles_returns_defaults_when_file_is_corrupt(tmp_path: Path, monkeypatch) -> None: target = tmp_path / "cycles.json" target.write_text("not-json") - monkeypatch.setattr(scorer, "TALOS_CYCLES_FILE", target) - assert scorer.load_cycles() == {"level": 0, "count": 0, "last_backlog": 0} + monkeypatch.setattr(runtime, "TALOS_CYCLES_FILE", target) + assert runtime.load_cycles() == {"level": 0, "count": 0, "last_backlog": 0} def test_save_cycles_and_load_cycles_round_trip(tmp_path: Path, monkeypatch) -> None: target = tmp_path / "cycles.json" - monkeypatch.setattr(scorer, "TALOS_CYCLES_FILE", target) - scorer.save_cycles(level=2, count=4, last_backlog=11) - assert scorer.load_cycles() == {"level": 2, "count": 4, "last_backlog": 11} + monkeypatch.setattr(runtime, "TALOS_CYCLES_FILE", target) + runtime.save_cycles(level=2, count=4, last_backlog=11) + assert runtime.load_cycles() == {"level": 2, "count": 4, "last_backlog": 11} def test_save_cycles_logs_failures(monkeypatch, caplog) -> None: class BrokenPath: parent = Path("/tmp") + suffix = ".json" - def write_text(self, _payload: str) -> None: + def write_text(self, _payload: str, encoding: str = "utf-8") -> None: raise OSError("boom") - monkeypatch.setattr(scorer, "TALOS_CYCLES_FILE", BrokenPath()) + def with_suffix(self, suffix: str): + return self + + def replace(self, target): + return target + + monkeypatch.setattr(runtime, "TALOS_CYCLES_FILE", BrokenPath()) with caplog.at_level("WARNING"): - scorer.save_cycles(level=1, count=2, last_backlog=3) - assert "Failed to persist cycles" in caplog.text + runtime.save_cycles(level=1, count=2, last_backlog=3) + assert "Failed to persist Talos state" in caplog.text def test_save_cycles_writes_atomically(tmp_path: Path, monkeypatch) -> None: target = tmp_path / "cycles.json" - monkeypatch.setattr(scorer, "TALOS_CYCLES_FILE", target) - scorer.save_cycles(level=7, count=8, last_backlog=9) + monkeypatch.setattr(runtime, "TALOS_CYCLES_FILE", target) + runtime.save_cycles(level=7, count=8, last_backlog=9) assert target.exists() assert not target.with_suffix(".json.tmp").exists() - assert scorer.load_cycles() == {"level": 7, "count": 8, "last_backlog": 9} + assert runtime.load_cycles() == {"level": 7, "count": 8, "last_backlog": 9} diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..94be591 --- /dev/null +++ b/tests/test_models.py @@ -0,0 +1,42 @@ +from talos.models import TalosEvaluation, TalosPolicy, TalosSnapshot, TalosState + + +def test_policy_round_trip() -> None: + policy = TalosPolicy.from_dict({"friction_ratio": 0.9, "write_block_cycles": 3}) + assert policy.to_dict() == {"friction_ratio": 0.9, "write_block_cycles": 3} + + +def test_snapshot_round_trip() -> None: + snapshot = TalosSnapshot.from_dict( + { + "wip_total": 8, + "global_max": 10, + "at_cap_projects": ["alpha"], + "backlog_total": 5, + "backlog_delta": 1, + "projects": [{"name": "alpha", "wip": 4}], + "source": "rowan", + "timestamp": "2026-05-25T12:00:00Z", + } + ) + assert snapshot.to_dict()["source"] == "rowan" + + +def test_state_round_trip() -> None: + state = TalosState.from_dict({"level": 2, "count": 3, "last_backlog": 7}) + assert state.to_dict() == {"level": 2, "count": 3, "last_backlog": 7} + + +def test_evaluation_to_dict() -> None: + evaluation = TalosEvaluation( + level=1, + previous_level=0, + cycles_at_level=1, + wip_total=8, + global_max=10, + backlog_total=4, + backlog_delta=1, + at_cap_projects=[], + reasons=["backlog is growing"], + ) + assert evaluation.to_dict()["reasons"] == ["backlog is growing"] diff --git a/tests/test_runtime.py b/tests/test_runtime.py new file mode 100644 index 0000000..ea7a11a --- /dev/null +++ b/tests/test_runtime.py @@ -0,0 +1,75 @@ +import json +from pathlib import Path + +from talos.models import TalosPolicy, TalosSnapshot, TalosState +from talos.runtime import append_history, evaluate_snapshot, load_state, next_state_from_evaluation, save_state + + +def test_evaluate_snapshot_builds_structured_output() -> None: + snapshot = TalosSnapshot( + wip_total=10, + global_max=10, + at_cap_projects=["alpha"], + backlog_total=12, + backlog_delta=2, + source="test", + timestamp="2026-05-25T12:00:00Z", + ) + evaluation = evaluate_snapshot(snapshot, prior_state=TalosState(level=2, count=2), policy=TalosPolicy()) + assert evaluation.level == 3 + assert evaluation.cycles_at_level == 3 + assert evaluation.source == "test" + assert "global cap pressure persisted long enough to block writes" in evaluation.reasons + + +def test_state_round_trip(tmp_path: Path) -> None: + target = tmp_path / "cycles.json" + save_state(level=2, count=4, last_backlog=11, path=target) + assert load_state(path=target).to_dict() == {"level": 2, "count": 4, "last_backlog": 11} + + +def test_history_appends_json_lines(tmp_path: Path) -> None: + target = tmp_path / "history.jsonl" + snapshot = TalosSnapshot(wip_total=4, global_max=10, backlog_total=2, backlog_delta=0) + evaluation = evaluate_snapshot(snapshot, prior_state=TalosState()) + append_history(evaluation, path=target) + lines = target.read_text().strip().splitlines() + assert len(lines) == 1 + assert json.loads(lines[0])["level"] == 0 + + +def test_next_state_matches_evaluation() -> None: + snapshot = TalosSnapshot(wip_total=8, global_max=10, backlog_total=5, backlog_delta=1) + evaluation = evaluate_snapshot(snapshot, prior_state=TalosState(level=1, count=2)) + state = next_state_from_evaluation(evaluation) + assert state.level == evaluation.level + assert state.count == evaluation.cycles_at_level + + +def test_load_policy_reads_json(tmp_path: Path) -> None: + target = tmp_path / "policy.json" + target.write_text('{"friction_ratio": 0.9, "write_block_cycles": 4}') + policy = __import__("talos.runtime", fromlist=["load_policy"]).load_policy(path=target) + assert policy.to_dict() == {"friction_ratio": 0.9, "write_block_cycles": 4} + + +def test_save_status_writes_structured_json(tmp_path: Path) -> None: + runtime = __import__("talos.runtime", fromlist=["save_status", "TalosSnapshot", "TalosState", "evaluate_snapshot"]) + target = tmp_path / "status.json" + evaluation = evaluate_snapshot(TalosSnapshot(wip_total=4, global_max=10), prior_state=TalosState()) + runtime.save_status(evaluation, path=target) + assert json.loads(target.read_text())["level"] == 0 + + +def test_read_snapshot_loads_json_file(tmp_path: Path) -> None: + runtime = __import__("talos.runtime", fromlist=["read_snapshot"]) + target = tmp_path / "snapshot.json" + target.write_text('{"wip_total": 5, "global_max": 10, "backlog_delta": 1}') + snapshot = runtime.read_snapshot(target) + assert snapshot.wip_total == 5 + + +def test_explain_level_handles_zero_cap() -> None: + runtime = __import__("talos.runtime", fromlist=["explain_level"]) + reasons = runtime.explain_level(TalosSnapshot(wip_total=1, global_max=0), level=0) + assert reasons == ["global_max<=0 so Talos stayed at clean level 0"] diff --git a/tests/test_scorer.py b/tests/test_scorer.py index 9b953b6..a68bde3 100644 --- a/tests/test_scorer.py +++ b/tests/test_scorer.py @@ -1,4 +1,5 @@ from talos import compute_talos_level +from talos.models import TalosPolicy def test_returns_clean_when_global_cap_is_invalid() -> None: @@ -9,6 +10,10 @@ def test_returns_clean_when_below_thresholds() -> None: assert compute_talos_level(3, 10, [], 0, 0) == 0 +def test_policy_can_raise_friction_threshold() -> None: + assert compute_talos_level(8, 10, [], 0, 0, policy=TalosPolicy(friction_ratio=0.9, write_block_cycles=2)) == 0 + + def test_returns_friction_when_backlog_grows() -> None: assert compute_talos_level(3, 10, [], 1, 0) == 1 From ff260012353fe6577b18846bf137af96b37859a9 Mon Sep 17 00:00:00 2001 From: John Malone Date: Mon, 25 May 2026 12:31:35 -0500 Subject: [PATCH 2/2] Tighten Talos model and runtime coverage --- talos/models.py | 10 ++++++-- tests/test_models.py | 55 ++++++++++++++++++++++++++++++++----------- tests/test_runtime.py | 49 +++++++++++++++++++++++++++++++++++++- tests/test_scorer.py | 4 ++++ 4 files changed, 101 insertions(+), 17 deletions(-) diff --git a/talos/models.py b/talos/models.py index 576915a..e243c71 100644 --- a/talos/models.py +++ b/talos/models.py @@ -39,13 +39,19 @@ class TalosSnapshot: @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=list(payload.get("at_cap_projects", [])), + at_cap_projects=at_cap_projects.copy(), backlog_total=int(payload.get("backlog_total", 0)), backlog_delta=int(payload.get("backlog_delta", 0)), - projects=list(payload.get("projects", [])), + projects=projects.copy(), source=str(payload.get("source", "manual")), timestamp=str(payload.get("timestamp", "")), ) diff --git a/tests/test_models.py b/tests/test_models.py index 94be591..4aad74e 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -7,19 +7,34 @@ def test_policy_round_trip() -> None: def test_snapshot_round_trip() -> None: - snapshot = TalosSnapshot.from_dict( - { - "wip_total": 8, - "global_max": 10, - "at_cap_projects": ["alpha"], - "backlog_total": 5, - "backlog_delta": 1, - "projects": [{"name": "alpha", "wip": 4}], - "source": "rowan", - "timestamp": "2026-05-25T12:00:00Z", - } - ) - assert snapshot.to_dict()["source"] == "rowan" + payload = { + "wip_total": 8, + "global_max": 10, + "at_cap_projects": ["alpha"], + "backlog_total": 5, + "backlog_delta": 1, + "projects": [{"name": "alpha", "wip": 4}], + "source": "rowan", + "timestamp": "2026-05-25T12:00:00Z", + } + snapshot = TalosSnapshot.from_dict(payload) + assert snapshot.to_dict() == payload + + +def test_snapshot_rejects_non_list_fields() -> None: + try: + TalosSnapshot.from_dict({"wip_total": 8, "global_max": 10, "at_cap_projects": "alpha"}) + except TypeError as exc: + assert "at_cap_projects must be a list" in str(exc) + else: # pragma: no cover + raise AssertionError("expected TypeError for invalid at_cap_projects") + + try: + TalosSnapshot.from_dict({"wip_total": 8, "global_max": 10, "projects": "alpha"}) + except TypeError as exc: + assert "projects must be a list" in str(exc) + else: # pragma: no cover + raise AssertionError("expected TypeError for invalid projects") def test_state_round_trip() -> None: @@ -39,4 +54,16 @@ def test_evaluation_to_dict() -> None: at_cap_projects=[], reasons=["backlog is growing"], ) - assert evaluation.to_dict()["reasons"] == ["backlog is growing"] + assert evaluation.to_dict() == { + "level": 1, + "previous_level": 0, + "cycles_at_level": 1, + "wip_total": 8, + "global_max": 10, + "backlog_total": 4, + "backlog_delta": 1, + "at_cap_projects": [], + "reasons": ["backlog is growing"], + "source": "manual", + "timestamp": "", + } diff --git a/tests/test_runtime.py b/tests/test_runtime.py index ea7a11a..5b71660 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -1,8 +1,18 @@ import json from pathlib import Path +import pytest + from talos.models import TalosPolicy, TalosSnapshot, TalosState -from talos.runtime import append_history, evaluate_snapshot, load_state, next_state_from_evaluation, save_state +from talos.runtime import ( + append_history, + evaluate_snapshot, + load_policy, + load_state, + next_state_from_evaluation, + save_state, + save_status, +) def test_evaluate_snapshot_builds_structured_output() -> None: @@ -53,6 +63,12 @@ def test_load_policy_reads_json(tmp_path: Path) -> None: assert policy.to_dict() == {"friction_ratio": 0.9, "write_block_cycles": 4} +def test_load_policy_returns_default_when_file_is_invalid(tmp_path: Path) -> None: + target = tmp_path / "policy.json" + target.write_text("not-json") + assert load_policy(path=target).to_dict() == {"friction_ratio": 0.8, "write_block_cycles": 2} + + def test_save_status_writes_structured_json(tmp_path: Path) -> None: runtime = __import__("talos.runtime", fromlist=["save_status", "TalosSnapshot", "TalosState", "evaluate_snapshot"]) target = tmp_path / "status.json" @@ -61,6 +77,18 @@ def test_save_status_writes_structured_json(tmp_path: Path) -> None: assert json.loads(target.read_text())["level"] == 0 +def test_save_status_logs_failures(monkeypatch, caplog, tmp_path: Path) -> None: + evaluation = evaluate_snapshot(TalosSnapshot(wip_total=4, global_max=10), prior_state=TalosState()) + + def boom(_target: Path, _payload: str) -> None: + raise OSError("boom") + + monkeypatch.setattr("talos.runtime._atomic_write_text", boom) + with caplog.at_level("WARNING"): + save_status(evaluation, path=tmp_path / "status.json") + assert "Failed to persist Talos status" in caplog.text + + def test_read_snapshot_loads_json_file(tmp_path: Path) -> None: runtime = __import__("talos.runtime", fromlist=["read_snapshot"]) target = tmp_path / "snapshot.json" @@ -73,3 +101,22 @@ def test_explain_level_handles_zero_cap() -> None: runtime = __import__("talos.runtime", fromlist=["explain_level"]) reasons = runtime.explain_level(TalosSnapshot(wip_total=1, global_max=0), level=0) assert reasons == ["global_max<=0 so Talos stayed at clean level 0"] + + +def test_append_history_logs_failures(caplog, tmp_path: Path) -> None: + target = tmp_path / "history.jsonl" + target.write_text("") + evaluation = evaluate_snapshot(TalosSnapshot(wip_total=4, global_max=10), prior_state=TalosState()) + + original_open = Path.open + + def broken_open(self: Path, *args, **kwargs): + if self == target: + raise OSError("boom") + return original_open(self, *args, **kwargs) + + with pytest.MonkeyPatch.context() as monkeypatch: + monkeypatch.setattr(Path, "open", broken_open) + with caplog.at_level("WARNING"): + append_history(evaluation, path=target) + assert "Failed to append Talos history" in caplog.text diff --git a/tests/test_scorer.py b/tests/test_scorer.py index a68bde3..7d43ba1 100644 --- a/tests/test_scorer.py +++ b/tests/test_scorer.py @@ -14,6 +14,10 @@ def test_policy_can_raise_friction_threshold() -> None: assert compute_talos_level(8, 10, [], 0, 0, policy=TalosPolicy(friction_ratio=0.9, write_block_cycles=2)) == 0 +def test_policy_can_raise_write_block_threshold() -> None: + assert compute_talos_level(10, 10, [], 0, 2, policy=TalosPolicy(write_block_cycles=3)) == 2 + + def test_returns_friction_when_backlog_grows() -> None: assert compute_talos_level(3, 10, [], 1, 0) == 1