235 lines
8.3 KiB
Python
235 lines
8.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Fleet progression evaluator for the Paperclips-inspired infrastructure epic.
|
|
|
|
Refs: timmy-home #547
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from urllib import request
|
|
from typing import Any
|
|
|
|
DEFAULT_BASE_URL = "https://forge.alexanderwhitestone.com/api/v1"
|
|
DEFAULT_OWNER = "Timmy_Foundation"
|
|
DEFAULT_REPO = "timmy-home"
|
|
DEFAULT_TOKEN_FILE = Path.home() / ".config" / "gitea" / "token"
|
|
DEFAULT_SPEC_FILE = Path(__file__).resolve().parent.parent / "configs" / "fleet_progression.json"
|
|
|
|
DEFAULT_RESOURCES = {
|
|
"uptime_percent_30d": 0.0,
|
|
"capacity_utilization": 0.0,
|
|
"innovation": 0.0,
|
|
"all_models_local": False,
|
|
"sovereign_stable_days": 0,
|
|
"human_free_days": 0,
|
|
}
|
|
|
|
|
|
class GiteaClient:
|
|
def __init__(self, token: str, owner: str = DEFAULT_OWNER, repo: str = DEFAULT_REPO, base_url: str = DEFAULT_BASE_URL):
|
|
self.token = token
|
|
self.owner = owner
|
|
self.repo = repo
|
|
self.base_url = base_url.rstrip("/")
|
|
|
|
def get_issue(self, issue_number: int):
|
|
headers = {"Authorization": f"token {self.token}"}
|
|
req = request.Request(
|
|
f"{self.base_url}/repos/{self.owner}/{self.repo}/issues/{issue_number}",
|
|
headers=headers,
|
|
)
|
|
with request.urlopen(req, timeout=30) as resp:
|
|
return json.loads(resp.read().decode())
|
|
|
|
|
|
def load_spec(path: Path | None = None):
|
|
target = path or DEFAULT_SPEC_FILE
|
|
return json.loads(target.read_text())
|
|
|
|
|
|
def load_issue_states(spec: dict[str, Any], token_file: Path = DEFAULT_TOKEN_FILE):
|
|
if not token_file.exists():
|
|
raise FileNotFoundError(f"Token file not found: {token_file}")
|
|
|
|
token = token_file.read_text().strip()
|
|
client = GiteaClient(token=token)
|
|
issue_states = {}
|
|
for phase in spec["phases"]:
|
|
issue = client.get_issue(phase["issue_number"])
|
|
issue_states[phase["issue_number"]] = issue["state"]
|
|
return issue_states
|
|
|
|
|
|
def _evaluate_rule(rule: dict[str, Any], issue_states: dict[int, str], resources: dict[str, Any]):
|
|
rule_type = rule["type"]
|
|
rule_id = rule["id"]
|
|
|
|
if rule_type == "always":
|
|
return {"rule": rule_id, "passed": True, "actual": True, "expected": True}
|
|
|
|
if rule_type == "issue_closed":
|
|
issue_number = int(rule["issue"])
|
|
actual = str(issue_states.get(issue_number, "open"))
|
|
return {
|
|
"rule": rule_id,
|
|
"passed": actual == "closed",
|
|
"actual": actual,
|
|
"expected": "closed",
|
|
}
|
|
|
|
if rule_type == "resource_gte":
|
|
resource = rule["resource"]
|
|
actual = resources.get(resource, 0)
|
|
expected = rule["value"]
|
|
return {
|
|
"rule": rule_id,
|
|
"passed": actual >= expected,
|
|
"actual": actual,
|
|
"expected": f">={expected}",
|
|
}
|
|
|
|
if rule_type == "resource_gt":
|
|
resource = rule["resource"]
|
|
actual = resources.get(resource, 0)
|
|
expected = rule["value"]
|
|
return {
|
|
"rule": rule_id,
|
|
"passed": actual > expected,
|
|
"actual": actual,
|
|
"expected": f">{expected}",
|
|
}
|
|
|
|
if rule_type == "resource_true":
|
|
resource = rule["resource"]
|
|
actual = bool(resources.get(resource, False))
|
|
return {
|
|
"rule": rule_id,
|
|
"passed": actual is True,
|
|
"actual": actual,
|
|
"expected": True,
|
|
}
|
|
|
|
raise ValueError(f"Unsupported rule type: {rule_type}")
|
|
|
|
|
|
def evaluate_progression(spec: dict[str, Any], issue_states: dict[int, str], resources: dict[str, Any] | None = None):
|
|
merged_resources = {**DEFAULT_RESOURCES, **(resources or {})}
|
|
phase_results = []
|
|
|
|
for phase in spec["phases"]:
|
|
issue_number = phase["issue_number"]
|
|
completed = str(issue_states.get(issue_number, "open")) == "closed"
|
|
rule_results = [
|
|
_evaluate_rule(rule, issue_states, merged_resources)
|
|
for rule in phase.get("unlock_rules", [])
|
|
]
|
|
blocking = [item for item in rule_results if not item["passed"]]
|
|
unlocked = not blocking
|
|
phase_results.append(
|
|
{
|
|
"number": phase["number"],
|
|
"issue_number": issue_number,
|
|
"key": phase["key"],
|
|
"name": phase["name"],
|
|
"summary": phase["summary"],
|
|
"completed": completed,
|
|
"unlocked": unlocked,
|
|
"available_to_work": unlocked and not completed,
|
|
"passed_requirements": [item for item in rule_results if item["passed"]],
|
|
"blocking_requirements": blocking,
|
|
}
|
|
)
|
|
|
|
unlocked_phases = [phase for phase in phase_results if phase["unlocked"]]
|
|
current_phase = unlocked_phases[-1] if unlocked_phases else phase_results[0]
|
|
next_locked_phase = next((phase for phase in phase_results if not phase["unlocked"]), None)
|
|
epic_complete = all(phase["completed"] for phase in phase_results) and phase_results[-1]["unlocked"]
|
|
|
|
return {
|
|
"epic_issue": spec["epic_issue"],
|
|
"epic_title": spec["epic_title"],
|
|
"resources": merged_resources,
|
|
"issue_states": {str(k): v for k, v in issue_states.items()},
|
|
"phases": phase_results,
|
|
"current_phase": current_phase,
|
|
"next_locked_phase": next_locked_phase,
|
|
"epic_complete": epic_complete,
|
|
}
|
|
|
|
|
|
def parse_args():
|
|
parser = argparse.ArgumentParser(description="Evaluate current fleet progression against the Paperclips-inspired epic.")
|
|
parser.add_argument("--spec-file", type=Path, default=DEFAULT_SPEC_FILE)
|
|
parser.add_argument("--token-file", type=Path, default=DEFAULT_TOKEN_FILE)
|
|
parser.add_argument("--issue-state-file", type=Path, help="Optional JSON file of issue_number -> state overrides")
|
|
parser.add_argument("--resource-file", type=Path, help="Optional JSON file with resource values")
|
|
parser.add_argument("--uptime-percent-30d", type=float)
|
|
parser.add_argument("--capacity-utilization", type=float)
|
|
parser.add_argument("--innovation", type=float)
|
|
parser.add_argument("--all-models-local", action="store_true")
|
|
parser.add_argument("--sovereign-stable-days", type=int)
|
|
parser.add_argument("--human-free-days", type=int)
|
|
parser.add_argument("--json", action="store_true")
|
|
return parser.parse_args()
|
|
|
|
|
|
def _load_resources(args):
|
|
resources = dict(DEFAULT_RESOURCES)
|
|
if args.resource_file:
|
|
resources.update(json.loads(args.resource_file.read_text()))
|
|
|
|
overrides = {
|
|
"uptime_percent_30d": args.uptime_percent_30d,
|
|
"capacity_utilization": args.capacity_utilization,
|
|
"innovation": args.innovation,
|
|
"sovereign_stable_days": args.sovereign_stable_days,
|
|
"human_free_days": args.human_free_days,
|
|
}
|
|
for key, value in overrides.items():
|
|
if value is not None:
|
|
resources[key] = value
|
|
if args.all_models_local:
|
|
resources["all_models_local"] = True
|
|
return resources
|
|
|
|
|
|
def _load_issue_states(args, spec):
|
|
if args.issue_state_file:
|
|
raw = json.loads(args.issue_state_file.read_text())
|
|
return {int(k): v for k, v in raw.items()}
|
|
return load_issue_states(spec, token_file=args.token_file)
|
|
|
|
|
|
def main():
|
|
args = parse_args()
|
|
spec = load_spec(args.spec_file)
|
|
issue_states = _load_issue_states(args, spec)
|
|
resources = _load_resources(args)
|
|
result = evaluate_progression(spec, issue_states, resources)
|
|
|
|
if args.json:
|
|
print(json.dumps(result, indent=2))
|
|
return
|
|
|
|
print("--- Fleet Progression Evaluator ---")
|
|
print(f"Epic #{result['epic_issue']}: {result['epic_title']}")
|
|
print(f"Current phase: {result['current_phase']['number']} — {result['current_phase']['name']}")
|
|
if result["next_locked_phase"]:
|
|
print(f"Next locked phase: {result['next_locked_phase']['number']} — {result['next_locked_phase']['name']}")
|
|
print(f"Epic complete: {result['epic_complete']}")
|
|
print()
|
|
for phase in result["phases"]:
|
|
state = "COMPLETE" if phase["completed"] else "ACTIVE" if phase["available_to_work"] else "LOCKED"
|
|
print(f"Phase {phase['number']} [{state}] {phase['name']}")
|
|
if phase["blocking_requirements"]:
|
|
for blocker in phase["blocking_requirements"]:
|
|
print(f" - blocked by {blocker['rule']}: actual={blocker['actual']} expected={blocker['expected']}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|