| |
| import argparse |
| import json |
| from datetime import datetime |
| from pathlib import Path |
|
|
| ROOT = Path(__file__).parent |
| STATE = ROOT / "tower.yaml" |
| FOCUS = ROOT / "focus.md" |
| DECISIONS = ROOT / "decisions.md" |
| RUNS = ROOT / "runs" |
|
|
| DEFAULT_STATE = { |
| "focus": None, |
| "paused": [], |
| "runs": [], |
| "next_action": "Set focus with: python tower.py focus <project>", |
| } |
|
|
|
|
| def load_state(): |
| init_files() |
| try: |
| return json.loads(STATE.read_text()) |
| except json.JSONDecodeError: |
| return DEFAULT_STATE.copy() |
|
|
|
|
| def save_state(state): |
| STATE.write_text(json.dumps(state, indent=2) + "\n") |
|
|
|
|
| def init_files(): |
| RUNS.mkdir(exist_ok=True) |
| if not STATE.exists(): |
| save_state(DEFAULT_STATE.copy()) |
| if not FOCUS.exists(): |
| FOCUS.write_text("# Focus\n\nNone\n") |
| if not DECISIONS.exists(): |
| DECISIONS.write_text("# Decisions\n\n") |
|
|
|
|
| def decision_items(): |
| lines = DECISIONS.read_text().splitlines() if DECISIONS.exists() else [] |
| return [line for line in lines if line.startswith("- [ ] ")] |
|
|
|
|
| def status(_args): |
| state = load_state() |
| active_runs = [run for run in state["runs"] if run.get("status") == "active"] |
| decisions = decision_items() |
|
|
| print("Current focus") |
| print(state["focus"] or "None") |
| print() |
|
|
| print("Active runs") |
| if active_runs: |
| for run in active_runs: |
| print(f"- {run['id']}: {run['project']} / {run['harness']} / {run['goal']}") |
| else: |
| print("None") |
| print() |
|
|
| print("Human decision queue") |
| if decisions: |
| for item in decisions: |
| print(item) |
| else: |
| print("None") |
| print() |
|
|
| print("Paused/suppressed projects") |
| if state["paused"]: |
| for project in state["paused"]: |
| print(f"- {project}") |
| else: |
| print("None") |
| print() |
|
|
| print("Next action") |
| print(state["next_action"]) |
|
|
|
|
| def set_focus(args): |
| state = load_state() |
| state["focus"] = args.project |
| state["next_action"] = f"Review active runs and decisions for {args.project}" |
| save_state(state) |
| FOCUS.write_text(f"# Focus\n\n{args.project}\n\nNext action: {state['next_action']}\n") |
| print(f"Focus: {args.project}") |
|
|
|
|
| def add_decision(args): |
| state = load_state() |
| with DECISIONS.open("a") as file: |
| file.write(f"- [ ] {args.text}\n") |
| state["next_action"] = f"Resolve decision: {args.text}" |
| save_state(state) |
| print("Decision added") |
|
|
|
|
| def add_run(args): |
| state = load_state() |
| now = datetime.now().strftime("%Y%m%d-%H%M%S") |
| safe_project = "".join(ch if ch.isalnum() or ch in "-_" else "-" for ch in args.project) |
| run_id = f"{now}-{safe_project}" |
| run = { |
| "id": run_id, |
| "project": args.project, |
| "harness": args.harness, |
| "goal": args.goal, |
| "status": "active", |
| "created_at": datetime.now().isoformat(timespec="seconds"), |
| } |
| state["runs"].append(run) |
| if state["focus"] is None: |
| state["focus"] = args.project |
| state["next_action"] = f"Check run {run_id} for {args.project}" |
| save_state(state) |
| (RUNS / f"{run_id}.md").write_text( |
| f"# Run {run_id}\n\n" |
| f"Project: {args.project}\n" |
| f"Harness: {args.harness}\n" |
| f"Goal: {args.goal}\n" |
| f"Status: active\n" |
| f"Created: {run['created_at']}\n" |
| ) |
| print(f"Run added: {run_id}") |
|
|
|
|
| def pause(args): |
| state = load_state() |
| if args.project not in state["paused"]: |
| state["paused"].append(args.project) |
| state["next_action"] = f"Choose a non-paused focus or unpause {args.project} later" |
| save_state(state) |
| print(f"Paused: {args.project}") |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(prog="tower.py") |
| sub = parser.add_subparsers(required=True) |
|
|
| p = sub.add_parser("status") |
| p.set_defaults(func=status) |
|
|
| p = sub.add_parser("focus") |
| p.add_argument("project") |
| p.set_defaults(func=set_focus) |
|
|
| p = sub.add_parser("add-decision") |
| p.add_argument("text") |
| p.set_defaults(func=add_decision) |
|
|
| p = sub.add_parser("add-run") |
| p.add_argument("--project", required=True) |
| p.add_argument("--harness", required=True) |
| p.add_argument("--goal", required=True) |
| p.set_defaults(func=add_run) |
|
|
| p = sub.add_parser("pause") |
| p.add_argument("project") |
| p.set_defaults(func=pause) |
|
|
| args = parser.parse_args() |
| args.func(args) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
