jpmschweitzer/settled-reach
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
3b2cd6914eeaa5e6b99bdf89993e1e71dfcc80ce
settled-reach/tooling/db/ticket
T
jpmschweitzerandClaude Opus 4.6 e6a557e8e7 feat(meta): replace sprint workflow with kanban + milestones (D-221) 
Sprint-based workflow (38 sprints) replaced by kanban + milestones.
Milestones are many-to-many with tickets and can block each other.

New: /whats-next skill (dependency-driven batch selection with Si
refinement review), /pr-process skill (renamed from pr-push, adds
review comment pickup), clerk agent + pre-push hook for D-record
consistency checks.

Deleted: sprint CLI, sprint-start/sprint-plan/sprint-status skills,
team-scoped file restrictions. Si rewritten as refinement manager.
All 19 agent briefings updated from stale PROJECT_STATE.md reference
to live ticket milestone queries.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-05-03 20:11:00 +02:00

569 lines
23 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""
Ticket CLI — ergonomic interface to the project ticketing database.
Usage:
ticket list [--status S] [--priority P] [--epic N] [--milestone N] [--assigned A] [--team T]
ticket show <id>
ticket done <id> [<id> ...]
ticket status <id> <new_status>
ticket assign <id> <agent>
ticket unassign <id>
ticket team <id> <teams>
ticket deps <id>
ticket dep add <blocker_id> <blocked_id>
ticket dep rm <blocker_id> <blocked_id>
ticket search <keyword>
ticket create <type> <title> [--parent N] [--priority P] [--decision D] [--team T] [--description TEXT]
ticket epics [--status S]
ticket children <id>
ticket count [--status S]
ticket wip
ticket milestone list [--status S]
ticket milestone create <name> [--description TEXT] [--phase N]
ticket milestone link <ticket_id> <milestone_id>
ticket milestone unlink <ticket_id> <milestone_id>
ticket milestone complete <milestone_id>
ticket milestone show <milestone_id>
ticket milestone dep <blocker_id> <blocked_id>
All output is JSON on stdout.
"""
import json
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
from common import get_connection, load_config # noqa: E402
WIP_LIMIT = 3
def query(conn, sql, params=()):
cursor = conn.execute(sql, params)
columns = [desc[0] for desc in cursor.description] if cursor.description else []
return [dict(zip(columns, row)) for row in cursor.fetchall()]
def execute(conn, sql, params=()):
cursor = conn.execute(sql, params)
conn.commit()
return cursor.rowcount
def out(data):
print(json.dumps(data, indent=2))
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
# ---------------------------------------------------------------------------
# Commands
# ---------------------------------------------------------------------------
def cmd_list(conn, args):
flags, _ = parse_flags(args, ["status", "priority", "epic", "milestone", "assigned", "team"])
conditions = []
params = []
join = ""
if "status" in flags:
conditions.append("t.status = ?")
params.append(flags["status"])
if "priority" in flags:
conditions.append("t.priority = ?")
params.append(flags["priority"])
if "epic" in flags:
conditions.append("t.parent_id = ?")
params.append(int(flags["epic"]))
if "milestone" in flags:
join = "JOIN ticket_milestones tm ON tm.ticket_id = t.id"
conditions.append("tm.milestone_id = ?")
params.append(int(flags["milestone"]))
if "assigned" in flags:
conditions.append("t.assigned_to = ?")
params.append(flags["assigned"])
if "team" in flags:
conditions.append("(',' || t.team || ',' LIKE '%,' || ? || ',%')")
params.append(flags["team"])
where = " AND ".join(conditions) if conditions else "1=1"
sql = f"""SELECT DISTINCT t.id, t.type, t.title, t.status, t.priority, t.assigned_to,
t.team, t.parent_id
FROM tickets t {join} WHERE {where}
ORDER BY
CASE t.priority WHEN 'critical' THEN 0 WHEN 'high' THEN 1
WHEN 'medium' THEN 2 ELSE 3 END, t.id"""
rows = query(conn, sql, tuple(params))
out({"ok": True, "count": len(rows), "rows": rows})
def cmd_show(conn, ids, brief=False):
tickets = []
for ticket_id in ids:
rows = query(conn, """SELECT t.*, p.title as parent_title
FROM tickets t LEFT JOIN tickets p ON t.parent_id = p.id
WHERE t.id = ?""", (ticket_id,))
if not rows:
tickets.append({"id": ticket_id, "error": f"Ticket #{ticket_id} not found"})
continue
ticket = rows[0]
children = query(conn, "SELECT id, title, status, priority FROM tickets WHERE parent_id = ? ORDER BY id", (ticket_id,))
blockers = query(conn, """SELECT t.id, t.title, t.status FROM ticket_deps d
JOIN tickets t ON d.blocker_id = t.id
WHERE d.blocked_id = ?""", (ticket_id,))
blocks = query(conn, """SELECT t.id, t.title, t.status FROM ticket_deps d
JOIN tickets t ON d.blocked_id = t.id
WHERE d.blocker_id = ?""", (ticket_id,))
milestones = query(conn, """SELECT m.id, m.name, m.status FROM ticket_milestones tm
JOIN milestones m ON tm.milestone_id = m.id
WHERE tm.ticket_id = ?""", (ticket_id,))
ticket["children"] = children
ticket["blocked_by"] = blockers
ticket["blocks"] = blocks
ticket["milestones"] = milestones
tickets.append(ticket)
if brief:
_print_brief(tickets)
elif len(tickets) == 1:
out({"ok": True, "ticket": tickets[0]})
else:
out({"ok": True, "count": len(tickets), "tickets": tickets})
def _print_brief(tickets):
for i, t in enumerate(tickets):
if "error" in t:
print(f"#{t['id']}: NOT FOUND")
continue
print(f"#{t['id']}: {t['title']}")
parts = [f"{t['type']}", f"P:{t['priority']}", f"S:{t['status']}"]
if t.get("assigned_to"):
parts.append(f"@{t['assigned_to']}")
if t.get("team"):
parts.append(f"Team:{t['team']}")
if t.get("parent_id"):
parts.append(f"Epic:#{t['parent_id']} ({t.get('parent_title', '?')})")
if t.get("milestones"):
ms = ", ".join(f"M{m['id']}" for m in t["milestones"])
parts.append(f"Milestones:{ms}")
if t.get("decision_ref"):
parts.append(f"Ref:{t['decision_ref']}")
print(f" {' | '.join(parts)}")
desc = t.get("description") or ""
if desc:
if len(desc) > 200:
desc = desc[:197] + "..."
print(f" {desc}")
if t.get("blocked_by"):
blockers = ", ".join(f"#{b['id']} ({b['status']})" for b in t["blocked_by"])
print(f" Blocked by: {blockers}")
if t.get("blocks"):
blocks = ", ".join(f"#{b['id']}" for b in t["blocks"])
print(f" Blocks: {blocks}")
if i < len(tickets) - 1:
print()
def cmd_done(conn, ids):
updated = 0
for tid in ids:
updated += execute(conn, "UPDATE tickets SET status='done', updated_at=datetime('now') WHERE id=?", (int(tid),))
out({"ok": True, "updated": updated, "ids": [int(i) for i in ids]})
def cmd_status(conn, ticket_id, new_status):
valid = ('backlog', 'ready', 'in_progress', 'review', 'done', 'cancelled')
if new_status not in valid:
out({"ok": False, "error": f"Invalid status '{new_status}'. Valid: {', '.join(valid)}"})
return
if new_status == 'in_progress':
wip = query(conn, "SELECT COUNT(*) as count FROM tickets WHERE status = 'in_progress'")
count = wip[0]["count"]
if count >= WIP_LIMIT:
print(f"WARNING: WIP limit ({WIP_LIMIT}) reached — {count} tickets already in_progress", file=sys.stderr)
updated = execute(conn, "UPDATE tickets SET status=?, updated_at=datetime('now') WHERE id=?", (new_status, int(ticket_id)))
out({"ok": True, "updated": updated, "id": int(ticket_id), "status": new_status})
def cmd_assign(conn, ticket_id, agent):
updated = execute(conn, "UPDATE tickets SET assigned_to=?, updated_at=datetime('now') WHERE id=?", (agent, int(ticket_id)))
out({"ok": True, "updated": updated, "id": int(ticket_id), "assigned_to": agent})
def cmd_unassign(conn, ticket_id):
updated = execute(conn, "UPDATE tickets SET assigned_to=NULL, updated_at=datetime('now') WHERE id=?", (int(ticket_id),))
out({"ok": True, "updated": updated, "id": int(ticket_id), "assigned_to": None})
def cmd_deps(conn, ticket_id):
blockers = query(conn, """SELECT t.id, t.title, t.status, t.priority FROM ticket_deps d
JOIN tickets t ON d.blocker_id = t.id
WHERE d.blocked_id = ? ORDER BY t.id""", (int(ticket_id),))
blocks = query(conn, """SELECT t.id, t.title, t.status, t.priority FROM ticket_deps d
JOIN tickets t ON d.blocked_id = t.id
WHERE d.blocker_id = ? ORDER BY t.id""", (int(ticket_id),))
out({"ok": True, "id": int(ticket_id), "blocked_by": blockers, "blocks": blocks})
def cmd_dep(conn, args):
if len(args) < 3:
out({"ok": False, "error": "Usage: ticket dep add|rm <blocker_id> <blocked_id>"})
return
action, blocker_id, blocked_id = args[0], int(args[1]), int(args[2])
if action == "add":
try:
execute(conn, "INSERT INTO ticket_deps (blocker_id, blocked_id) VALUES (?, ?)", (blocker_id, blocked_id))
out({"ok": True, "action": "added", "blocker_id": blocker_id, "blocked_id": blocked_id})
except Exception as e:
out({"ok": False, "error": str(e)})
elif action == "rm":
updated = execute(conn, "DELETE FROM ticket_deps WHERE blocker_id = ? AND blocked_id = ?", (blocker_id, blocked_id))
out({"ok": True, "action": "removed", "updated": updated, "blocker_id": blocker_id, "blocked_id": blocked_id})
else:
out({"ok": False, "error": f"Unknown dep action: {action}. Use 'add' or 'rm'."})
def cmd_search(conn, keyword):
rows = query(conn, """SELECT id, type, title, status, priority, assigned_to, team
FROM tickets WHERE title LIKE ? OR description LIKE ?
ORDER BY id""", (f"%{keyword}%", f"%{keyword}%"))
out({"ok": True, "count": len(rows), "rows": rows})
def cmd_create(conn, args):
flags, positional = parse_flags(args, ["parent", "priority", "decision", "team", "description"])
if len(positional) < 2:
out({"ok": False, "error": "Usage: ticket create <type> <title> [--parent N] [--priority P] [--decision D] [--team T] [--description TEXT]"})
return
ticket_type = positional[0]
title = " ".join(positional[1:])
parent_id = int(flags["parent"]) if "parent" in flags else None
priority = flags.get("priority", "medium")
decision_ref = flags.get("decision")
team = flags.get("team")
description = flags.get("description")
conn.execute(
"INSERT INTO tickets (type, title, description, parent_id, priority, decision_ref, team) VALUES (?, ?, ?, ?, ?, ?, ?)",
(ticket_type, title, description, parent_id, priority, decision_ref, team))
conn.commit()
last_id = query(conn, "SELECT last_insert_rowid() as id")[0]["id"]
out({"ok": True, "id": last_id, "title": title})
def cmd_epics(conn, args):
flags, _ = parse_flags(args, ["status"])
conditions = ["t.type = 'epic'"]
params = []
if "status" in flags:
conditions.append("t.status = ?")
params.append(flags["status"])
where = " AND ".join(conditions)
rows = query(conn, f"""SELECT t.id, t.title, t.status, t.priority, t.assigned_to, t.team,
COUNT(c.id) as child_count,
SUM(CASE WHEN c.status='done' THEN 1 ELSE 0 END) as done_count
FROM tickets t LEFT JOIN tickets c ON c.parent_id = t.id
WHERE {where} GROUP BY t.id
ORDER BY CASE t.priority WHEN 'critical' THEN 0 WHEN 'high' THEN 1
WHEN 'medium' THEN 2 ELSE 3 END, t.id""", tuple(params))
out({"ok": True, "count": len(rows), "rows": rows})
def cmd_children(conn, ticket_id):
rows = query(conn, """SELECT id, type, title, status, priority, assigned_to, team
FROM tickets WHERE parent_id = ? ORDER BY id""", (int(ticket_id),))
out({"ok": True, "count": len(rows), "parent_id": int(ticket_id), "rows": rows})
def cmd_team(conn, ticket_id, teams):
updated = execute(conn, "UPDATE tickets SET team=?, updated_at=datetime('now') WHERE id=?", (teams, int(ticket_id)))
out({"ok": True, "updated": updated, "id": int(ticket_id), "team": teams})
def cmd_count(conn, args):
flags, _ = parse_flags(args, ["status"])
if "status" in flags:
rows = query(conn, "SELECT COUNT(*) as count FROM tickets WHERE status = ?", (flags["status"],))
else:
rows = query(conn, "SELECT status, COUNT(*) as count FROM tickets GROUP BY status ORDER BY count DESC")
out({"ok": True, "rows": rows})
def cmd_wip(conn):
rows = query(conn, """SELECT id, title, assigned_to, team FROM tickets
WHERE status = 'in_progress' ORDER BY id""")
count = len(rows)
out({"ok": True, "in_progress": count, "limit": WIP_LIMIT,
"over_limit": count > WIP_LIMIT, "tickets": rows})
# ---------------------------------------------------------------------------
# Milestone commands
# ---------------------------------------------------------------------------
def cmd_milestone(conn, args):
if not args:
out({"ok": False, "error": "Usage: ticket milestone list|create|link|unlink|complete|show|dep ..."})
return
sub = args[0]
rest = args[1:]
if sub == "list":
cmd_milestone_list(conn, rest)
elif sub == "create":
cmd_milestone_create(conn, rest)
elif sub == "link":
cmd_milestone_link(conn, rest)
elif sub == "unlink":
cmd_milestone_unlink(conn, rest)
elif sub == "complete":
cmd_milestone_complete(conn, rest)
elif sub == "show":
cmd_milestone_show(conn, rest)
elif sub == "dep":
cmd_milestone_dep(conn, rest)
else:
out({"ok": False, "error": f"Unknown milestone subcommand: {sub}"})
def cmd_milestone_list(conn, args):
flags, _ = parse_flags(args, ["status"])
conditions = []
params = []
if "status" in flags:
conditions.append("m.status = ?")
params.append(flags["status"])
where = " AND ".join(conditions) if conditions else "1=1"
rows = query(conn, f"""SELECT m.*,
COUNT(DISTINCT tm.ticket_id) as ticket_count,
SUM(CASE WHEN t.status = 'done' THEN 1 ELSE 0 END) as done_count
FROM milestones m
LEFT JOIN ticket_milestones tm ON tm.milestone_id = m.id
LEFT JOIN tickets t ON t.id = tm.ticket_id
WHERE {where}
GROUP BY m.id ORDER BY m.id""", tuple(params))
out({"ok": True, "count": len(rows), "milestones": rows})
def cmd_milestone_create(conn, args):
flags, positional = parse_flags(args, ["description", "phase"])
if not positional:
out({"ok": False, "error": "Usage: ticket milestone create <name> [--description TEXT] [--phase N]"})
return
name = " ".join(positional)
description = flags.get("description")
phase = int(flags["phase"]) if "phase" in flags else None
conn.execute(
"INSERT INTO milestones (name, description, cascade_phase) VALUES (?, ?, ?)",
(name, description, phase))
conn.commit()
last_id = query(conn, "SELECT last_insert_rowid() as id")[0]["id"]
out({"ok": True, "id": last_id, "name": name})
def cmd_milestone_link(conn, args):
if len(args) < 2:
out({"ok": False, "error": "Usage: ticket milestone link <ticket_id> <milestone_id>"})
return
ticket_id, milestone_id = int(args[0]), int(args[1])
try:
execute(conn, "INSERT INTO ticket_milestones (ticket_id, milestone_id) VALUES (?, ?)", (ticket_id, milestone_id))
out({"ok": True, "action": "linked", "ticket_id": ticket_id, "milestone_id": milestone_id})
except Exception as e:
out({"ok": False, "error": str(e)})
def cmd_milestone_unlink(conn, args):
if len(args) < 2:
out({"ok": False, "error": "Usage: ticket milestone unlink <ticket_id> <milestone_id>"})
return
ticket_id, milestone_id = int(args[0]), int(args[1])
updated = execute(conn, "DELETE FROM ticket_milestones WHERE ticket_id = ? AND milestone_id = ?", (ticket_id, milestone_id))
out({"ok": True, "action": "unlinked", "updated": updated, "ticket_id": ticket_id, "milestone_id": milestone_id})
def cmd_milestone_complete(conn, args):
if not args:
out({"ok": False, "error": "Usage: ticket milestone complete <milestone_id>"})
return
milestone_id = int(args[0])
updated = execute(conn, "UPDATE milestones SET status='completed', completed_at=datetime('now') WHERE id=?", (milestone_id,))
out({"ok": True, "updated": updated, "id": milestone_id, "status": "completed"})
def cmd_milestone_show(conn, args):
if not args:
out({"ok": False, "error": "Usage: ticket milestone show <milestone_id>"})
return
milestone_id = int(args[0])
ms = query(conn, "SELECT * FROM milestones WHERE id = ?", (milestone_id,))
if not ms:
out({"ok": False, "error": f"Milestone #{milestone_id} not found"})
return
milestone = ms[0]
tickets = query(conn, """SELECT t.id, t.type, t.title, t.status, t.priority, t.assigned_to
FROM ticket_milestones tm JOIN tickets t ON t.id = tm.ticket_id
WHERE tm.milestone_id = ?
ORDER BY CASE t.priority WHEN 'critical' THEN 0 WHEN 'high' THEN 1
WHEN 'medium' THEN 2 ELSE 3 END, t.id""", (milestone_id,))
blockers = query(conn, """SELECT m.id, m.name, m.status FROM milestone_deps d
JOIN milestones m ON d.blocker_id = m.id
WHERE d.blocked_id = ?""", (milestone_id,))
blocks = query(conn, """SELECT m.id, m.name, m.status FROM milestone_deps d
JOIN milestones m ON d.blocked_id = m.id
WHERE d.blocker_id = ?""", (milestone_id,))
milestone["tickets"] = tickets
milestone["blocked_by"] = blockers
milestone["blocks"] = blocks
out({"ok": True, "milestone": milestone})
def cmd_milestone_dep(conn, args):
if len(args) < 2:
out({"ok": False, "error": "Usage: ticket milestone dep <blocker_id> <blocked_id>"})
return
blocker_id, blocked_id = int(args[0]), int(args[1])
try:
execute(conn, "INSERT INTO milestone_deps (blocker_id, blocked_id) VALUES (?, ?)", (blocker_id, blocked_id))
out({"ok": True, "action": "added", "blocker_id": blocker_id, "blocked_id": blocked_id})
except Exception as e:
out({"ok": False, "error": str(e)})
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
HELP = """ticket — project ticket CLI
Usage:
ticket list [--status S] [--priority P] [--epic N] [--milestone N] [--assigned A] [--team T]
ticket show [--brief] <id> [<id>...] Full ticket detail (--brief for summary)
ticket done <id> [<id> ...] Mark tickets as done
ticket status <id> <new_status> Change ticket status
ticket assign <id> <agent> Assign ticket to agent
ticket unassign <id> Remove assignment
ticket team <id> <teams> Set team(s) (comma-separated)
ticket deps <id> Show ticket dependencies
ticket dep add <blocker> <blocked> Add ticket dependency
ticket dep rm <blocker> <blocked> Remove ticket dependency
ticket search <keyword> Search tickets by title/description
ticket create <type> <title> [--parent N] [--priority P] [--decision D] [--team T] [--description TEXT]
ticket epics [--status S] List epics with child counts
ticket children <id> List children of a ticket
ticket count [--status S] Count tickets by status
ticket wip Show in-progress count vs WIP limit
ticket milestone list [--status S] List milestones
ticket milestone create <name> [--description TEXT] [--phase N] Create milestone
ticket milestone link <ticket> <milestone> Link ticket to milestone
ticket milestone unlink <ticket> <milestone> Unlink ticket from milestone
ticket milestone complete <milestone> Mark milestone completed
ticket milestone show <milestone> Show milestone with tickets
ticket milestone dep <blocker> <blocked> Add milestone dependency"""
def main():
if len(sys.argv) < 2 or sys.argv[1] in ("--help", "-h", "help"):
print(HELP)
sys.exit(0)
try:
cfg = load_config()
except (FileNotFoundError, json.JSONDecodeError) as exc:
out({"ok": False, "error": f"Config error: {exc}"})
sys.exit(1)
conn = get_connection(cfg)
cmd = sys.argv[1]
args = sys.argv[2:]
try:
if cmd == "list":
cmd_list(conn, args)
elif cmd == "show":
brief = "--brief" in args
id_args = [a for a in args if a != "--brief"]
if not id_args:
out({"ok": False, "error": "Usage: ticket show [--brief] <id> [<id> ...]"})
else:
cmd_show(conn, [int(a) for a in id_args], brief=brief)
elif cmd == "done":
if not args:
out({"ok": False, "error": "Usage: ticket done <id> [<id> ...]"})
else:
cmd_done(conn, args)
elif cmd == "status":
if len(args) < 2:
out({"ok": False, "error": "Usage: ticket status <id> <new_status>"})
else:
cmd_status(conn, args[0], args[1])
elif cmd == "assign":
if len(args) < 2:
out({"ok": False, "error": "Usage: ticket assign <id> <agent>"})
else:
cmd_assign(conn, args[0], args[1])
elif cmd == "unassign":
if not args:
out({"ok": False, "error": "Usage: ticket unassign <id>"})
else:
cmd_unassign(conn, args[0])
elif cmd == "team":
if len(args) < 2:
out({"ok": False, "error": "Usage: ticket team <id> <teams>"})
else:
cmd_team(conn, args[0], args[1])
elif cmd == "deps":
if not args:
out({"ok": False, "error": "Usage: ticket deps <id>"})
else:
cmd_deps(conn, args[0])
elif cmd == "dep":
cmd_dep(conn, args)
elif cmd == "search":
if not args:
out({"ok": False, "error": "Usage: ticket search <keyword>"})
else:
cmd_search(conn, " ".join(args))
elif cmd == "create":
cmd_create(conn, args)
elif cmd == "epics":
cmd_epics(conn, args)
elif cmd == "children":
if not args:
out({"ok": False, "error": "Usage: ticket children <id>"})
else:
cmd_children(conn, args[0])
elif cmd == "count":
cmd_count(conn, args)
elif cmd == "wip":
cmd_wip(conn)
elif cmd == "milestone":
cmd_milestone(conn, args)
else:
out({"ok": False, "error": f"Unknown command: {cmd}. Use --help for usage."})
finally:
conn.close()
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink