Claudia
734f96cfcf
All checks were successful
Tests / test (push) Successful in 2s
All ~/clawd/ references replaced with configurable paths: - CORTEX_HOME (default: ~/.cortex) - CORTEX_MEMORY_DIR, CORTEX_CONFIG, CORTEX_GROWTH_LOG, CORTEX_ROADMAP - permanent_files configurable via config.json - Tests pass both with and without env vars set - 169/169 tests green
339 lines
10 KiB
Python
339 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Roadmap Manager — track tasks, deadlines, dependencies, generate reports.
|
|
|
|
Usage:
|
|
python3 roadmap.py list [--status open] [--priority P0,P1] [--tag security]
|
|
python3 roadmap.py add "title" --priority P1 --deadline 2026-02-19 --tag infra
|
|
python3 roadmap.py update <id> --status done
|
|
python3 roadmap.py overdue
|
|
python3 roadmap.py upcoming [--days 7]
|
|
python3 roadmap.py report
|
|
python3 roadmap.py check-deps
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import uuid
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from cortex.config import roadmap_file
|
|
ROADMAP_FILE = roadmap_file()
|
|
VALID_STATUSES = {"open", "in_progress", "blocked", "done"}
|
|
VALID_PRIORITIES = {"P0", "P1", "P2", "P3"}
|
|
|
|
|
|
def load_roadmap(path: Path | None = None) -> dict:
|
|
p = path or ROADMAP_FILE
|
|
if p.exists():
|
|
return json.loads(p.read_text())
|
|
return {"items": []}
|
|
|
|
|
|
def save_roadmap(data: dict, path: Path | None = None):
|
|
p = path or ROADMAP_FILE
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
p.write_text(json.dumps(data, indent=2, ensure_ascii=False))
|
|
|
|
|
|
def find_item(data: dict, item_id: str) -> dict | None:
|
|
for item in data["items"]:
|
|
if item["id"] == item_id or item["id"].startswith(item_id):
|
|
return item
|
|
return None
|
|
|
|
|
|
def now_iso() -> str:
|
|
return datetime.now().isoformat(timespec="seconds")
|
|
|
|
|
|
def parse_date(s: str | None) -> datetime | None:
|
|
if not s:
|
|
return None
|
|
for fmt in ["%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"]:
|
|
try:
|
|
return datetime.strptime(s, fmt)
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
|
|
# --- Commands ---
|
|
|
|
def cmd_list(args, data: dict):
|
|
items = data["items"]
|
|
if args.status:
|
|
statuses = set(args.status.split(","))
|
|
items = [i for i in items if i["status"] in statuses]
|
|
if args.priority:
|
|
priorities = set(args.priority.split(","))
|
|
items = [i for i in items if i["priority"] in priorities]
|
|
if args.tag:
|
|
tags = set(args.tag.split(","))
|
|
items = [i for i in items if tags & set(i.get("tags", []))]
|
|
|
|
if not items:
|
|
print("No items found.")
|
|
return
|
|
|
|
# Sort by priority then deadline
|
|
prio_order = {"P0": 0, "P1": 1, "P2": 2, "P3": 3}
|
|
items.sort(key=lambda x: (prio_order.get(x["priority"], 9),
|
|
x.get("deadline") or "9999"))
|
|
|
|
for i in items:
|
|
deadline = i.get("deadline", "")
|
|
dl_str = f" ⏰ {deadline}" if deadline else ""
|
|
status_icon = {"open": "⬚", "in_progress": "🔄", "blocked": "🚫", "done": "✅"
|
|
}.get(i["status"], "?")
|
|
tags = " ".join(f"#{t}" for t in i.get("tags", []))
|
|
print(f" {status_icon} [{i['priority']}] {i['title']}{dl_str} {tags}")
|
|
print(f" id: {i['id'][:8]}")
|
|
|
|
|
|
def cmd_add(args, data: dict):
|
|
item = {
|
|
"id": str(uuid.uuid4()),
|
|
"title": args.title,
|
|
"status": "open",
|
|
"priority": args.priority or "P2",
|
|
"deadline": args.deadline,
|
|
"depends_on": [],
|
|
"tags": args.tag.split(",") if args.tag else [],
|
|
"created": now_iso(),
|
|
"updated": now_iso(),
|
|
"notes": args.notes or "",
|
|
}
|
|
if item["priority"] not in VALID_PRIORITIES:
|
|
print(f"Invalid priority: {item['priority']}")
|
|
sys.exit(1)
|
|
data["items"].append(item)
|
|
save_roadmap(data)
|
|
print(f"Added: {item['title']} (id: {item['id'][:8]})")
|
|
|
|
|
|
def cmd_update(args, data: dict):
|
|
item = find_item(data, args.id)
|
|
if not item:
|
|
print(f"Item not found: {args.id}")
|
|
sys.exit(1)
|
|
if args.status:
|
|
if args.status not in VALID_STATUSES:
|
|
print(f"Invalid status: {args.status}")
|
|
sys.exit(1)
|
|
item["status"] = args.status
|
|
if args.priority:
|
|
item["priority"] = args.priority
|
|
if args.deadline:
|
|
item["deadline"] = args.deadline
|
|
if args.notes:
|
|
item["notes"] = args.notes
|
|
if args.tag:
|
|
item["tags"] = args.tag.split(",")
|
|
item["updated"] = now_iso()
|
|
save_roadmap(data)
|
|
print(f"Updated: {item['title']}")
|
|
|
|
|
|
def cmd_overdue(args, data: dict):
|
|
now = datetime.now()
|
|
overdue = []
|
|
for item in data["items"]:
|
|
if item["status"] == "done":
|
|
continue
|
|
dl = parse_date(item.get("deadline"))
|
|
if dl and dl < now:
|
|
overdue.append(item)
|
|
|
|
if not overdue:
|
|
print("No overdue items. ✅")
|
|
return
|
|
|
|
prio_order = {"P0": 0, "P1": 1, "P2": 2, "P3": 3}
|
|
overdue.sort(key=lambda x: prio_order.get(x["priority"], 9))
|
|
|
|
has_critical = any(i["priority"] in ("P0", "P1") for i in overdue)
|
|
print(f"🚨 {len(overdue)} overdue item(s):")
|
|
for i in overdue:
|
|
print(f" [{i['priority']}] {i['title']} — due {i['deadline']}")
|
|
|
|
if has_critical:
|
|
sys.exit(1)
|
|
|
|
|
|
def cmd_upcoming(args, data: dict):
|
|
now = datetime.now()
|
|
days = args.days or 7
|
|
cutoff = now + timedelta(days=days)
|
|
upcoming = []
|
|
for item in data["items"]:
|
|
if item["status"] == "done":
|
|
continue
|
|
dl = parse_date(item.get("deadline"))
|
|
if dl and now <= dl <= cutoff:
|
|
upcoming.append(item)
|
|
|
|
if not upcoming:
|
|
print(f"No items due in the next {days} days.")
|
|
return
|
|
|
|
upcoming.sort(key=lambda x: x.get("deadline", ""))
|
|
print(f"📅 {len(upcoming)} item(s) due in the next {days} days:")
|
|
for i in upcoming:
|
|
print(f" [{i['priority']}] {i['title']} — due {i['deadline']}")
|
|
|
|
|
|
def cmd_report(args, data: dict):
|
|
now = datetime.now()
|
|
lines = [
|
|
f"# Roadmap Status Report",
|
|
f"*Generated: {now.strftime('%Y-%m-%d %H:%M')}*",
|
|
"",
|
|
]
|
|
|
|
# Summary
|
|
by_status = {}
|
|
for item in data["items"]:
|
|
by_status.setdefault(item["status"], []).append(item)
|
|
|
|
lines.append("## Summary")
|
|
for status in ["open", "in_progress", "blocked", "done"]:
|
|
count = len(by_status.get(status, []))
|
|
icon = {"open": "⬚", "in_progress": "🔄", "blocked": "🚫", "done": "✅"}[status]
|
|
lines.append(f"- {icon} {status}: {count}")
|
|
lines.append("")
|
|
|
|
# Overdue
|
|
overdue = [i for i in data["items"] if i["status"] != "done"
|
|
and parse_date(i.get("deadline")) and parse_date(i["deadline"]) < now]
|
|
if overdue:
|
|
lines.append("## 🚨 Overdue")
|
|
for i in overdue:
|
|
lines.append(f"- **[{i['priority']}]** {i['title']} — due {i['deadline']}")
|
|
lines.append("")
|
|
|
|
# Upcoming (7 days)
|
|
cutoff = now + timedelta(days=7)
|
|
upcoming = [i for i in data["items"] if i["status"] != "done"
|
|
and parse_date(i.get("deadline"))
|
|
and now <= parse_date(i["deadline"]) <= cutoff]
|
|
if upcoming:
|
|
lines.append("## 📅 Upcoming (7 days)")
|
|
for i in sorted(upcoming, key=lambda x: x["deadline"]):
|
|
lines.append(f"- **[{i['priority']}]** {i['title']} — due {i['deadline']}")
|
|
lines.append("")
|
|
|
|
# Stale items (in_progress > 7 days without update)
|
|
stale = []
|
|
for item in data["items"]:
|
|
if item["status"] == "in_progress":
|
|
updated = parse_date(item.get("updated"))
|
|
if updated and (now - updated).days > 7:
|
|
stale.append(item)
|
|
if stale:
|
|
lines.append("## ⏳ Stale (in_progress >7 days)")
|
|
for i in stale:
|
|
lines.append(f"- **[{i['priority']}]** {i['title']} — last updated {i['updated']}")
|
|
lines.append("")
|
|
|
|
# Active work
|
|
active = by_status.get("in_progress", [])
|
|
if active:
|
|
lines.append("## 🔄 In Progress")
|
|
for i in active:
|
|
dl = f" (due {i['deadline']})" if i.get("deadline") else ""
|
|
lines.append(f"- **[{i['priority']}]** {i['title']}{dl}")
|
|
lines.append("")
|
|
|
|
print("\n".join(lines))
|
|
|
|
|
|
def cmd_check_deps(args, data: dict):
|
|
id_status = {i["id"]: i for i in data["items"]}
|
|
issues = []
|
|
for item in data["items"]:
|
|
if item["status"] == "done":
|
|
continue
|
|
for dep_id in item.get("depends_on", []):
|
|
dep = id_status.get(dep_id)
|
|
if not dep:
|
|
issues.append((item, dep_id, "missing"))
|
|
elif dep["status"] != "done":
|
|
issues.append((item, dep_id, dep["status"]))
|
|
|
|
if not issues:
|
|
print("No dependency issues. ✅")
|
|
return
|
|
|
|
print(f"⚠️ {len(issues)} dependency issue(s):")
|
|
for item, dep_id, status in issues:
|
|
if status == "missing":
|
|
print(f" [{item['priority']}] {item['title']} → depends on unknown {dep_id[:8]}")
|
|
else:
|
|
dep = id_status[dep_id]
|
|
print(f" [{item['priority']}] {item['title']} → blocked by '{dep['title']}' ({status})")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Roadmap Manager")
|
|
sub = parser.add_subparsers(dest="command")
|
|
|
|
# list
|
|
p_list = sub.add_parser("list")
|
|
p_list.add_argument("--status", type=str)
|
|
p_list.add_argument("--priority", type=str)
|
|
p_list.add_argument("--tag", type=str)
|
|
|
|
# add
|
|
p_add = sub.add_parser("add")
|
|
p_add.add_argument("title", type=str)
|
|
p_add.add_argument("--priority", type=str, default="P2")
|
|
p_add.add_argument("--deadline", type=str)
|
|
p_add.add_argument("--tag", type=str)
|
|
p_add.add_argument("--notes", type=str)
|
|
|
|
# update
|
|
p_upd = sub.add_parser("update")
|
|
p_upd.add_argument("id", type=str)
|
|
p_upd.add_argument("--status", type=str)
|
|
p_upd.add_argument("--priority", type=str)
|
|
p_upd.add_argument("--deadline", type=str)
|
|
p_upd.add_argument("--tag", type=str)
|
|
p_upd.add_argument("--notes", type=str)
|
|
|
|
# overdue
|
|
sub.add_parser("overdue")
|
|
|
|
# upcoming
|
|
p_up = sub.add_parser("upcoming")
|
|
p_up.add_argument("--days", type=int, default=7)
|
|
|
|
# report
|
|
sub.add_parser("report")
|
|
|
|
# check-deps
|
|
sub.add_parser("check-deps")
|
|
|
|
args = parser.parse_args()
|
|
if not args.command:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
data = load_roadmap()
|
|
|
|
cmd_map = {
|
|
"list": cmd_list,
|
|
"add": cmd_add,
|
|
"update": cmd_update,
|
|
"overdue": cmd_overdue,
|
|
"upcoming": cmd_upcoming,
|
|
"report": cmd_report,
|
|
"check-deps": cmd_check_deps,
|
|
}
|
|
cmd_map[args.command](args, data)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|