jpmschweitzerandClaude Opus 4.6
bc226d8baf
Consolidates all connector scripts under tooling/ per project structure conventions. Symlink at db/connectors → tooling/db/ preserves backwards compatibility (remove after Sprint 22). Updated references in CLAUDE.md, Makefile, DEVOPS.md, all skill files, agent files, rules, schema comments, and Sprint 21 briefings. Python scripts updated with correct SCHEMA_PATH (now relative to WORKTREE_ROOT/db/schema.sql). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
743 lines
24 KiB
Python
Executable File
743 lines
24 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Sprint CLI — orchestrates sprint lifecycle and context for agents.
|
|
|
|
Calls the ticket CLI for data queries (no SQL duplication).
|
|
Direct DB access only for sprint lifecycle mutations.
|
|
|
|
Usage:
|
|
sprint status [--sprint N] [--team T] Sprint progress and ticket overview
|
|
sprint sweep [--sprint N] Health check: grouped tickets, issues, team summary (JSON)
|
|
sprint start [--sprint N] Activate a planned sprint
|
|
sprint stop [--sprint N] Complete an active sprint
|
|
sprint start-work [--sprint N] [--team T] Full context dump for starting work
|
|
sprint prepare [--sprint N] [--team T] Prepare next sprint (candidates + gaps)
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
TICKET_CLI = str(SCRIPT_DIR / "ticket")
|
|
WORKTREE_ROOT = (SCRIPT_DIR / ".." / "..").resolve()
|
|
DB_PATH = (WORKTREE_ROOT / ".." / "settledreach.db").resolve()
|
|
PROJECT_ROOT = WORKTREE_ROOT
|
|
|
|
REMINDER = """---
|
|
Reminder: Keep ticket status up to date after finishing work.
|
|
tooling/db/ticket status <id> in_progress (when starting)
|
|
tooling/db/ticket status <id> done (when finished)"""
|
|
|
|
|
|
def run_ticket(*args):
|
|
"""Call the ticket CLI and return parsed JSON."""
|
|
result = subprocess.run(
|
|
[sys.executable, TICKET_CLI] + list(args),
|
|
capture_output=True, text=True
|
|
)
|
|
if result.returncode != 0:
|
|
return {"ok": False, "error": result.stderr.strip()}
|
|
try:
|
|
return json.loads(result.stdout)
|
|
except json.JSONDecodeError:
|
|
return {"ok": False, "error": f"Bad ticket output: {result.stdout[:200]}"}
|
|
|
|
|
|
def get_connection():
|
|
"""Direct DB connection for lifecycle mutations only."""
|
|
conn = sqlite3.connect(str(DB_PATH))
|
|
conn.execute("PRAGMA journal_mode=WAL;")
|
|
conn.execute("PRAGMA foreign_keys=ON;")
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
def parse_flags(args, known_flags):
|
|
"""Parse --flag value pairs from args, return (flags_dict, positional_args)."""
|
|
flags = {}
|
|
positional = []
|
|
i = 0
|
|
while i < len(args):
|
|
if args[i].startswith("--") and args[i][2:] in known_flags:
|
|
key = args[i][2:]
|
|
if i + 1 < len(args):
|
|
flags[key] = args[i + 1]
|
|
i += 2
|
|
else:
|
|
positional.append(args[i])
|
|
i += 1
|
|
else:
|
|
positional.append(args[i])
|
|
i += 1
|
|
return flags, positional
|
|
|
|
|
|
def detect_team(flags):
|
|
"""Detect team from flags or git branch."""
|
|
if "team" in flags:
|
|
return flags["team"]
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "branch", "--show-current"],
|
|
capture_output=True, text=True, cwd=str(PROJECT_ROOT)
|
|
)
|
|
branch = result.stdout.strip()
|
|
if branch and branch != "main":
|
|
return branch
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def get_all_sprints():
|
|
"""Get all sprints via ticket CLI."""
|
|
data = run_ticket("sprint")
|
|
if not data.get("ok"):
|
|
return []
|
|
return data.get("sprints", [])
|
|
|
|
|
|
def detect_sprint(flags, prefer_status=None):
|
|
"""Detect sprint from flags or by status preference.
|
|
|
|
prefer_status: which status to prefer when auto-detecting.
|
|
'active' for status/start-work/stop
|
|
'planning' for start
|
|
None for prepare (targets next sprint)
|
|
"""
|
|
if "sprint" in flags:
|
|
sprint_id = int(flags["sprint"])
|
|
sprints = get_all_sprints()
|
|
for s in sprints:
|
|
if s["id"] == sprint_id:
|
|
return s
|
|
print(f"Error: Sprint {sprint_id} not found.")
|
|
sys.exit(1)
|
|
|
|
sprints = get_all_sprints()
|
|
if not sprints:
|
|
print("Error: No sprints found in database.")
|
|
sys.exit(1)
|
|
|
|
if prefer_status:
|
|
matching = [s for s in sprints if s["status"] == prefer_status]
|
|
if len(matching) == 1:
|
|
return matching[0]
|
|
if len(matching) > 1:
|
|
ids = ", ".join(str(s["id"]) for s in matching)
|
|
print(f"Error: Multiple {prefer_status} sprints: {ids}. Use --sprint N to specify.")
|
|
sys.exit(1)
|
|
# Fall through: no match for preferred status
|
|
if prefer_status == "active":
|
|
# No active sprint
|
|
print("Error: No active sprint. Use --sprint N to specify.")
|
|
sys.exit(1)
|
|
if prefer_status == "planning":
|
|
print("Error: No sprint in planning status. Use sprint prepare first.")
|
|
sys.exit(1)
|
|
|
|
return None
|
|
|
|
|
|
def detect_sprint_for_prepare(flags):
|
|
"""For prepare: target the next sprint after the most recent one."""
|
|
if "sprint" in flags:
|
|
sprint_id = int(flags["sprint"])
|
|
sprints = get_all_sprints()
|
|
for s in sprints:
|
|
if s["id"] == sprint_id:
|
|
return s
|
|
# Sprint doesn't exist yet — return a stub
|
|
return {"id": sprint_id, "status": "new", "name": None}
|
|
|
|
sprints = get_all_sprints()
|
|
# If there's a planning sprint, use it
|
|
planning = [s for s in sprints if s["status"] == "planning"]
|
|
if len(planning) == 1:
|
|
return planning[0]
|
|
if len(planning) > 1:
|
|
ids = ", ".join(str(s["id"]) for s in planning)
|
|
print(f"Error: Multiple planning sprints: {ids}. Use --sprint N to specify.")
|
|
sys.exit(1)
|
|
|
|
# Otherwise target max_id + 1
|
|
if sprints:
|
|
next_id = max(s["id"] for s in sprints) + 1
|
|
return {"id": next_id, "status": "new", "name": None}
|
|
|
|
return {"id": 1, "status": "new", "name": None}
|
|
|
|
|
|
def get_tickets_for_sprint(sprint_id, team=None):
|
|
"""Get tickets for a sprint, optionally filtered by team."""
|
|
args = ["list", "--sprint", str(sprint_id)]
|
|
if team:
|
|
args += ["--team", team]
|
|
data = run_ticket(*args)
|
|
if not data.get("ok"):
|
|
return []
|
|
return data.get("rows", [])
|
|
|
|
|
|
def get_ticket_deps(ticket_id):
|
|
"""Get dependencies for a ticket."""
|
|
data = run_ticket("deps", str(ticket_id))
|
|
if not data.get("ok"):
|
|
return {"blocked_by": [], "blocks": []}
|
|
return data
|
|
|
|
|
|
def get_ticket_detail(ticket_id):
|
|
"""Get full ticket detail."""
|
|
data = run_ticket("show", str(ticket_id))
|
|
if not data.get("ok"):
|
|
return None
|
|
return data.get("ticket")
|
|
|
|
|
|
def briefing_path(sprint_id, team):
|
|
"""Find the briefing file for a sprint/team if it exists."""
|
|
p = PROJECT_ROOT / "docs" / "sprints" / f"sprint-{sprint_id}" / f"{team}.md"
|
|
if p.exists():
|
|
return str(p.relative_to(PROJECT_ROOT))
|
|
return None
|
|
|
|
|
|
def format_ticket_table(tickets):
|
|
"""Format tickets as an aligned table."""
|
|
if not tickets:
|
|
print(" (none)")
|
|
return
|
|
# Header
|
|
print(f" {'#':<6} {'Title':<50} {'Status':<12} {'Assigned':<10} {'Priority'}")
|
|
print(f" {'---':<6} {'---':<50} {'---':<12} {'---':<10} {'---'}")
|
|
for t in tickets:
|
|
title = t.get("title", "")
|
|
if len(title) > 48:
|
|
title = title[:45] + "..."
|
|
assigned = t.get("assigned_to") or ""
|
|
print(f" {t['id']:<6} {title:<50} {t['status']:<12} {assigned:<10} {t['priority']}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Commands
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_status(args):
|
|
flags, _ = parse_flags(args, ["sprint", "team"])
|
|
sprint = detect_sprint(flags, prefer_status="active")
|
|
team = detect_team(flags)
|
|
|
|
tickets = get_tickets_for_sprint(sprint["id"], team)
|
|
|
|
# Header
|
|
name = sprint.get("name", f"Sprint {sprint['id']}")
|
|
print(f"=== {name} ({sprint['status']}) ===")
|
|
if sprint.get("goal"):
|
|
print(f"Goal: {sprint['goal']}")
|
|
parts = []
|
|
if sprint.get("start_date"):
|
|
parts.append(f"Started: {sprint['start_date']}")
|
|
if sprint.get("end_date"):
|
|
parts.append(f"Ended: {sprint['end_date']}")
|
|
if team:
|
|
parts.append(f"Team: {team}")
|
|
if parts:
|
|
print(" | ".join(parts))
|
|
print()
|
|
|
|
# Progress
|
|
total = len(tickets)
|
|
done = sum(1 for t in tickets if t["status"] == "done")
|
|
pct = int(done / total * 100) if total > 0 else 0
|
|
print(f"Progress: {done}/{total} done ({pct}%)")
|
|
|
|
# Status breakdown
|
|
statuses = {}
|
|
for t in tickets:
|
|
statuses[t["status"]] = statuses.get(t["status"], 0) + 1
|
|
status_parts = []
|
|
for s in ["backlog", "ready", "in_progress", "review", "done", "cancelled"]:
|
|
if s in statuses:
|
|
status_parts.append(f"{s}: {statuses[s]}")
|
|
if status_parts:
|
|
print(f" {' | '.join(status_parts)}")
|
|
print()
|
|
|
|
# Ticket table
|
|
print("Tickets:")
|
|
format_ticket_table(tickets)
|
|
print()
|
|
|
|
# Blocked tickets
|
|
blocked_lines = []
|
|
for t in tickets:
|
|
if t["status"] == "done":
|
|
continue
|
|
deps = get_ticket_deps(t["id"])
|
|
for b in deps.get("blocked_by", []):
|
|
if b["status"] != "done":
|
|
blocked_lines.append(f" #{t['id']} blocked by #{b['id']} ({b['status']})")
|
|
if blocked_lines:
|
|
print("Blocked:")
|
|
for line in blocked_lines:
|
|
print(line)
|
|
print()
|
|
|
|
# Briefing
|
|
if team:
|
|
bp = briefing_path(sprint["id"], team)
|
|
if bp:
|
|
print(f"Briefing: {bp}")
|
|
else:
|
|
# Show all available briefings
|
|
briefings = []
|
|
for t_name in ["server", "client", "copy", "audio", "visual", "ci", "joint"]:
|
|
bp = briefing_path(sprint["id"], t_name)
|
|
if bp:
|
|
briefings.append(bp)
|
|
if briefings:
|
|
print("Briefings:")
|
|
for bp in briefings:
|
|
print(f" {bp}")
|
|
|
|
print()
|
|
print(REMINDER)
|
|
|
|
|
|
def cmd_start(args):
|
|
flags, _ = parse_flags(args, ["sprint"])
|
|
sprint = detect_sprint(flags, prefer_status="planning")
|
|
|
|
if sprint["status"] != "planning":
|
|
print(f"Error: Sprint {sprint['id']} is '{sprint['status']}', expected 'planning'.")
|
|
sys.exit(1)
|
|
|
|
# Check ticket count
|
|
tickets = get_tickets_for_sprint(sprint["id"])
|
|
if not tickets:
|
|
print(f"Error: Sprint {sprint['id']} has no tickets. Run sprint prepare first.")
|
|
sys.exit(1)
|
|
|
|
# Activate
|
|
conn = get_connection()
|
|
conn.execute(
|
|
"UPDATE sprints SET status='active', start_date=date('now') WHERE id=?",
|
|
(sprint["id"],)
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
# Summary
|
|
teams = {}
|
|
for t in tickets:
|
|
team = t.get("team") or "unassigned"
|
|
teams[team] = teams.get(team, 0) + 1
|
|
|
|
name = sprint.get("name", f"Sprint {sprint['id']}")
|
|
print(f"Started: {name}")
|
|
print(f"Tickets: {len(tickets)}")
|
|
for team, count in sorted(teams.items()):
|
|
print(f" {team}: {count}")
|
|
print()
|
|
print(REMINDER)
|
|
|
|
|
|
def cmd_stop(args):
|
|
flags, _ = parse_flags(args, ["sprint"])
|
|
sprint = detect_sprint(flags, prefer_status="active")
|
|
|
|
if sprint["status"] != "active":
|
|
print(f"Error: Sprint {sprint['id']} is '{sprint['status']}', expected 'active'.")
|
|
sys.exit(1)
|
|
|
|
tickets = get_tickets_for_sprint(sprint["id"])
|
|
done = [t for t in tickets if t["status"] == "done"]
|
|
cancelled = [t for t in tickets if t["status"] == "cancelled"]
|
|
incomplete = [t for t in tickets if t["status"] not in ("done", "cancelled")]
|
|
|
|
# Auto-close: mark picked-up tickets (in_progress, review) as done.
|
|
# Work merged to main before sprint close means the ticket is done —
|
|
# agents just forget to update status. Backlog/ready tickets were
|
|
# never started, so they stay as carry-over candidates.
|
|
picked_up_statuses = ("in_progress", "review")
|
|
picked_up = [t for t in incomplete if t["status"] in picked_up_statuses]
|
|
auto_closed = []
|
|
if picked_up:
|
|
conn = get_connection()
|
|
for t in picked_up:
|
|
conn.execute("UPDATE tickets SET status='done' WHERE id=?", (t["id"],))
|
|
auto_closed.append(t)
|
|
conn.commit()
|
|
conn.close()
|
|
# Move auto-closed into done count, remove from incomplete
|
|
done = done + auto_closed
|
|
incomplete = [t for t in incomplete if t["status"] not in picked_up_statuses]
|
|
|
|
# Complete the sprint
|
|
conn = get_connection()
|
|
conn.execute(
|
|
"UPDATE sprints SET status='completed', end_date=date('now') WHERE id=?",
|
|
(sprint["id"],)
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
name = sprint.get("name", f"Sprint {sprint['id']}")
|
|
print(f"Completed: {name}")
|
|
print(f"Done: {len(done)}/{len(tickets)}")
|
|
if cancelled:
|
|
print(f"Cancelled: {len(cancelled)}")
|
|
if auto_closed:
|
|
print(f"Auto-closed: {len(auto_closed)} tickets marked done on sprint close:")
|
|
for t in auto_closed:
|
|
print(f" #{t['id']}: {t['title']} ({t['status']} → done)")
|
|
print()
|
|
|
|
if incomplete:
|
|
print("Carry-over candidates (never started):")
|
|
format_ticket_table(incomplete)
|
|
print()
|
|
|
|
print(REMINDER)
|
|
|
|
|
|
def cmd_start_work(args):
|
|
flags, _ = parse_flags(args, ["sprint", "team"])
|
|
sprint = detect_sprint(flags, prefer_status="active")
|
|
team = detect_team(flags)
|
|
|
|
if sprint["status"] != "active":
|
|
print(f"Error: Sprint {sprint['id']} is '{sprint['status']}', expected 'active'.")
|
|
sys.exit(1)
|
|
|
|
tickets = get_tickets_for_sprint(sprint["id"], team)
|
|
|
|
# Header
|
|
name = sprint.get("name", f"Sprint {sprint['id']}")
|
|
team_label = f" \u2014 {team.title()}" if team else ""
|
|
print(f"=== {name}{team_label} ===")
|
|
if sprint.get("goal"):
|
|
print(f"Goal: {sprint['goal']}")
|
|
parts = [f"Status: {sprint['status']}"]
|
|
if sprint.get("start_date"):
|
|
parts.append(f"Started: {sprint['start_date']}")
|
|
print(" | ".join(parts))
|
|
print()
|
|
|
|
# Briefing
|
|
if team:
|
|
bp = briefing_path(sprint["id"], team)
|
|
if bp:
|
|
print(f"Briefing: {bp}")
|
|
# Also check joint briefing
|
|
jbp = briefing_path(sprint["id"], "joint")
|
|
if jbp:
|
|
print(f"Joint briefing: {jbp}")
|
|
|
|
# Collect decision refs
|
|
decision_refs = set()
|
|
for t in tickets:
|
|
detail = get_ticket_detail(t["id"])
|
|
if detail and detail.get("decision_ref"):
|
|
decision_refs.add(detail["decision_ref"])
|
|
if decision_refs:
|
|
print(f"Decisions: {', '.join(sorted(decision_refs))}")
|
|
print()
|
|
|
|
# Build dependency map
|
|
blocked_by_map = {} # ticket_id -> [blocker tickets]
|
|
blocks_map = {} # ticket_id -> [blocked ticket ids]
|
|
for t in tickets:
|
|
deps = get_ticket_deps(t["id"])
|
|
open_blockers = [b for b in deps.get("blocked_by", []) if b["status"] != "done"]
|
|
if open_blockers:
|
|
blocked_by_map[t["id"]] = open_blockers
|
|
blocking = deps.get("blocks", [])
|
|
if blocking:
|
|
blocks_map[t["id"]] = blocking
|
|
|
|
# Categorize
|
|
done_tickets = [t for t in tickets if t["status"] == "done"]
|
|
blocked_tickets = [t for t in tickets if t["status"] != "done" and t["id"] in blocked_by_map]
|
|
actionable_tickets = [t for t in tickets if t["status"] != "done" and t["id"] not in blocked_by_map]
|
|
|
|
# Actionable
|
|
if actionable_tickets:
|
|
print("Actionable (not blocked, not done):")
|
|
for t in actionable_tickets:
|
|
print(f" #{t['id']}: {t['title']}")
|
|
# Metadata line
|
|
meta = [t.get("type", ""), f"P:{t['priority']}", f"S:{t['status']}"]
|
|
if t.get("assigned_to"):
|
|
meta.append(f"@{t['assigned_to']}")
|
|
if t.get("team"):
|
|
meta.append(f"Team:{t['team']}")
|
|
detail = get_ticket_detail(t["id"])
|
|
if detail and detail.get("decision_ref"):
|
|
meta.append(f"Ref:{detail['decision_ref']}")
|
|
print(f" {' | '.join(meta)}")
|
|
if t["id"] in blocks_map:
|
|
block_ids = ", ".join(f"#{b['id']}" for b in blocks_map[t["id"]])
|
|
print(f" Blocks: {block_ids}")
|
|
print()
|
|
|
|
# Blocked
|
|
if blocked_tickets:
|
|
print("Blocked:")
|
|
for t in blocked_tickets:
|
|
blockers = blocked_by_map[t["id"]]
|
|
blocker_str = ", ".join(f"#{b['id']} ({b['status']})" for b in blockers)
|
|
print(f" #{t['id']}: {t['title']} \u2190 blocked by {blocker_str}")
|
|
print()
|
|
|
|
# Done
|
|
if done_tickets:
|
|
print("Done:")
|
|
for t in done_tickets:
|
|
print(f" #{t['id']}: {t['title']} \u2713")
|
|
print()
|
|
|
|
print(REMINDER)
|
|
|
|
|
|
def cmd_sweep(args):
|
|
"""Health check: grouped tickets, bookkeeping issues, team summary (JSON)."""
|
|
flags, _ = parse_flags(args, ["sprint"])
|
|
sprint = detect_sprint(flags, prefer_status="active")
|
|
|
|
tickets = get_tickets_for_sprint(sprint["id"])
|
|
|
|
# Build dependency map for blocked detection
|
|
blocked_by_map = {} # ticket_id -> [blocker_id, ...]
|
|
for t in tickets:
|
|
if t["status"] == "done":
|
|
continue
|
|
deps = get_ticket_deps(t["id"])
|
|
open_blockers = [b["id"] for b in deps.get("blocked_by", []) if b["status"] != "done"]
|
|
if open_blockers:
|
|
blocked_by_map[t["id"]] = open_blockers
|
|
|
|
# Group by status
|
|
by_status = {"done": [], "review": [], "in_progress": [], "blocked": [], "backlog": []}
|
|
for t in tickets:
|
|
entry = {
|
|
"id": t["id"],
|
|
"title": t["title"],
|
|
"team": t.get("team") or "unassigned",
|
|
"assigned_to": t.get("assigned_to"),
|
|
}
|
|
if t["status"] == "done":
|
|
by_status["done"].append(entry)
|
|
elif t["id"] in blocked_by_map:
|
|
entry["blocked_by"] = blocked_by_map[t["id"]]
|
|
by_status["blocked"].append(entry)
|
|
elif t["status"] == "review":
|
|
by_status["review"].append(entry)
|
|
elif t["status"] == "in_progress":
|
|
by_status["in_progress"].append(entry)
|
|
else:
|
|
by_status["backlog"].append(entry)
|
|
|
|
# Per-team summary
|
|
by_team = {}
|
|
for status, items in by_status.items():
|
|
for item in items:
|
|
team = item["team"]
|
|
if team not in by_team:
|
|
by_team[team] = {"backlog": 0, "in_progress": 0, "review": 0, "blocked": 0, "done": 0, "total": 0}
|
|
by_team[team][status] = by_team[team].get(status, 0) + 1
|
|
by_team[team]["total"] += 1
|
|
|
|
# Bookkeeping issues
|
|
issues = []
|
|
for t in tickets:
|
|
if t["status"] in ("in_progress", "review") and not t.get("assigned_to"):
|
|
issues.append({
|
|
"type": "unassigned_in_progress",
|
|
"detail": f"#{t['id']} unassigned {t['status']}",
|
|
"fix": f"tooling/db/ticket assign {t['id']} <agent>",
|
|
})
|
|
if t["status"] == "backlog" and sprint["status"] == "active" and t["id"] not in blocked_by_map:
|
|
issues.append({
|
|
"type": "stale_backlog",
|
|
"detail": f"#{t['id']} stale backlog",
|
|
"fix": f"tooling/db/ticket status {t['id']} in_progress",
|
|
})
|
|
if t["status"] == "done" and t.get("assigned_to"):
|
|
issues.append({
|
|
"type": "assigned_but_done",
|
|
"detail": f"#{t['id']} done, still assigned",
|
|
"fix": f"tooling/db/ticket unassign {t['id']}",
|
|
})
|
|
|
|
# Progress
|
|
total = len(tickets)
|
|
done = len(by_status["done"])
|
|
pct = int(done / total * 100) if total > 0 else 0
|
|
|
|
result = {
|
|
"sprint": {
|
|
"id": sprint["id"],
|
|
"name": sprint.get("name", f"Sprint {sprint['id']}"),
|
|
"goal": sprint.get("goal", ""),
|
|
},
|
|
"progress": {"total": total, "done": done, "pct": pct},
|
|
"by_status": by_status,
|
|
"by_team": by_team,
|
|
"issues": issues,
|
|
}
|
|
print(json.dumps(result, indent=2))
|
|
|
|
|
|
def cmd_prepare(args):
|
|
flags, _ = parse_flags(args, ["sprint", "team"])
|
|
sprint = detect_sprint_for_prepare(flags)
|
|
team = detect_team(flags)
|
|
|
|
# Create sprint record if it doesn't exist
|
|
if sprint.get("status") == "new":
|
|
conn = get_connection()
|
|
conn.execute(
|
|
"INSERT INTO sprints (id, name, status) VALUES (?, ?, 'planning')",
|
|
(sprint["id"], f"Sprint {sprint['id']}")
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
print(f"Created Sprint {sprint['id']} (planning)")
|
|
sprint["status"] = "planning"
|
|
sprint["name"] = f"Sprint {sprint['id']}"
|
|
elif sprint["status"] not in ("planning", "new"):
|
|
print(f"Warning: Sprint {sprint['id']} is '{sprint['status']}', not 'planning'.")
|
|
|
|
print(f"=== Preparing Sprint {sprint['id']} ===")
|
|
print()
|
|
|
|
# Previous sprint info
|
|
all_sprints = get_all_sprints()
|
|
prev_sprints = [s for s in all_sprints if s["id"] < sprint["id"]]
|
|
if prev_sprints:
|
|
prev = max(prev_sprints, key=lambda s: s["id"])
|
|
prev_tickets = get_tickets_for_sprint(prev["id"])
|
|
prev_done = sum(1 for t in prev_tickets if t["status"] == "done")
|
|
prev_name = prev.get("name", f"Sprint {prev['id']}")
|
|
print(f"Previous: {prev_name} ({prev['status']}, {prev_done}/{len(prev_tickets)} done)")
|
|
print()
|
|
|
|
# Carry-over candidates
|
|
incomplete = [t for t in prev_tickets if t["status"] not in ("done", "cancelled")]
|
|
if team:
|
|
incomplete = [t for t in incomplete if team in (t.get("team") or "")]
|
|
if incomplete:
|
|
print("Carry-over candidates (incomplete from previous sprint):")
|
|
format_ticket_table(incomplete)
|
|
print()
|
|
|
|
# Backlog candidates
|
|
backlog_args = ["list", "--status", "backlog"]
|
|
if team:
|
|
backlog_args += ["--team", team]
|
|
backlog_data = run_ticket(*backlog_args)
|
|
backlog = backlog_data.get("rows", []) if backlog_data.get("ok") else []
|
|
# Filter out tickets already assigned to a sprint
|
|
backlog = [t for t in backlog if not t.get("sprint_id")]
|
|
|
|
if backlog:
|
|
if team:
|
|
print(f"Backlog candidates ({team}):")
|
|
format_ticket_table(backlog)
|
|
else:
|
|
# Group by team
|
|
by_team = {}
|
|
for t in backlog:
|
|
t_team = t.get("team") or "unassigned"
|
|
by_team.setdefault(t_team, []).append(t)
|
|
print("Backlog candidates (unassigned to any sprint):")
|
|
for t_name in sorted(by_team.keys()):
|
|
print(f"\n {t_name.title()}:")
|
|
format_ticket_table(by_team[t_name])
|
|
print()
|
|
|
|
# Decision coverage gaps
|
|
conn = get_connection()
|
|
cursor = conn.execute("""
|
|
SELECT id, title FROM decisions
|
|
WHERE type='confirmed' AND status='active'
|
|
AND id NOT IN (SELECT DISTINCT decision_ref FROM tickets WHERE decision_ref IS NOT NULL)
|
|
ORDER BY id
|
|
""")
|
|
orphans = cursor.fetchall()
|
|
conn.close()
|
|
|
|
if orphans:
|
|
print("Decision coverage gaps (active decisions without tickets):")
|
|
for row in orphans:
|
|
print(f" {row[0]}: {row[1]}")
|
|
print()
|
|
|
|
# Already assigned to this sprint
|
|
assigned = get_tickets_for_sprint(sprint["id"], team)
|
|
if assigned:
|
|
print(f"Already assigned to Sprint {sprint['id']}:")
|
|
format_ticket_table(assigned)
|
|
print()
|
|
|
|
print(REMINDER)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
HELP = """sprint \u2014 sprint lifecycle and context for agents
|
|
|
|
Usage:
|
|
sprint status [--sprint N] [--team T] Sprint progress and ticket overview
|
|
sprint sweep [--sprint N] Health check: grouped tickets, issues, team summary (JSON)
|
|
sprint start [--sprint N] Activate a planned sprint
|
|
sprint stop [--sprint N] Complete an active sprint
|
|
sprint start-work [--sprint N] [--team T] Full context dump for starting work
|
|
sprint prepare [--sprint N] [--team T] Prepare next sprint (candidates + gaps)
|
|
|
|
Sprint auto-detection:
|
|
status/start-work/sweep prefer the active sprint
|
|
start prefer the planning sprint
|
|
stop prefer the active sprint
|
|
prepare target next sprint (max id + 1)
|
|
|
|
Team auto-detection:
|
|
If --team is omitted, uses the current git branch name (unless on main).
|
|
On main with no --team, shows all teams."""
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 2 or sys.argv[1] in ("--help", "-h", "help"):
|
|
print(HELP)
|
|
sys.exit(0)
|
|
|
|
cmd = sys.argv[1]
|
|
args = sys.argv[2:]
|
|
|
|
commands = {
|
|
"status": cmd_status,
|
|
"sweep": cmd_sweep,
|
|
"start": cmd_start,
|
|
"stop": cmd_stop,
|
|
"start-work": cmd_start_work,
|
|
"prepare": cmd_prepare,
|
|
}
|
|
|
|
if cmd not in commands:
|
|
print(f"Error: Unknown command '{cmd}'. Use --help for usage.")
|
|
sys.exit(1)
|
|
|
|
commands[cmd](args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|