jpmschweitzerandClaude Opus 4.6
70a6f850c6
Uses git rev-parse --git-common-dir to find the main worktree root, then resolves the DB path there. Prevents WAL/journal pollution on feature branch worktrees that caused merge conflicts on commonwealth.db. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
378 lines
14 KiB
Python
Executable File
378 lines
14 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Ticket CLI — ergonomic interface to the project ticketing database.
|
|
|
|
Usage:
|
|
ticket list [--status S] [--priority P] [--epic N] [--sprint N] [--assigned A]
|
|
ticket show <id>
|
|
ticket done <id> [<id> ...]
|
|
ticket status <id> <new_status>
|
|
ticket assign <id> <agent>
|
|
ticket unassign <id>
|
|
ticket sprint [--active]
|
|
ticket sprint assign <id> <sprint_id>
|
|
ticket deps <id>
|
|
ticket search <keyword>
|
|
ticket create <type> <title> [--parent N] [--priority P] [--decision D]
|
|
ticket epics [--status S]
|
|
ticket children <id>
|
|
ticket count [--status S]
|
|
|
|
All output is JSON on stdout.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
CONFIG_PATH = SCRIPT_DIR / "config.json"
|
|
|
|
|
|
def _main_worktree():
|
|
"""Find the main worktree root via git's common dir.
|
|
|
|
In a worktree layout like:
|
|
settled-reach/main/ (main worktree, holds .git/)
|
|
settled-reach/server/ (linked worktree, .git is a file)
|
|
git rev-parse --git-common-dir always points to main's .git/,
|
|
so its parent is the main worktree root.
|
|
"""
|
|
try:
|
|
common = subprocess.check_output(
|
|
["git", "rev-parse", "--git-common-dir"],
|
|
stderr=subprocess.DEVNULL, text=True
|
|
).strip()
|
|
return Path(common).resolve().parent
|
|
except (subprocess.CalledProcessError, OSError):
|
|
# Fallback: assume script lives in the main worktree
|
|
return SCRIPT_DIR.parent.parent
|
|
|
|
|
|
def load_config():
|
|
with open(CONFIG_PATH, "r") as f:
|
|
cfg = json.load(f)
|
|
# Always resolve DB path in the main worktree to avoid
|
|
# WAL/journal pollution on feature branches causing merge conflicts.
|
|
main_root = _main_worktree()
|
|
db_path = (main_root / "db" / "commonwealth.db").resolve()
|
|
cfg["sqlite_db_resolved"] = str(db_path)
|
|
return cfg
|
|
|
|
|
|
def get_connection(cfg):
|
|
conn = sqlite3.connect(cfg["sqlite_db_resolved"])
|
|
conn.execute("PRAGMA journal_mode=WAL;")
|
|
conn.execute("PRAGMA foreign_keys=ON;")
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
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", "sprint", "assigned"])
|
|
conditions = []
|
|
params = []
|
|
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 "sprint" in flags:
|
|
conditions.append("t.sprint_id = ?")
|
|
params.append(int(flags["sprint"]))
|
|
if "assigned" in flags:
|
|
conditions.append("t.assigned_to = ?")
|
|
params.append(flags["assigned"])
|
|
where = " AND ".join(conditions) if conditions else "1=1"
|
|
sql = f"""SELECT t.id, t.type, t.title, t.status, t.priority, t.assigned_to,
|
|
t.parent_id, t.sprint_id
|
|
FROM tickets t 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, ticket_id):
|
|
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:
|
|
out({"ok": False, "error": f"Ticket #{ticket_id} not found"})
|
|
return
|
|
ticket = rows[0]
|
|
# Get children
|
|
children = query(conn, "SELECT id, title, status, priority FROM tickets WHERE parent_id = ? ORDER BY id", (ticket_id,))
|
|
# Get dependencies (what blocks this)
|
|
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,))
|
|
# Get dependents (what this blocks)
|
|
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,))
|
|
ticket["children"] = children
|
|
ticket["blocked_by"] = blockers
|
|
ticket["blocks"] = blocks
|
|
out({"ok": True, "ticket": ticket})
|
|
|
|
|
|
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
|
|
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_sprint(conn, args):
|
|
flags, positional = parse_flags(args, ["active"])
|
|
if positional and positional[0] == "assign" and len(positional) >= 3:
|
|
ticket_id, sprint_id = int(positional[1]), int(positional[2])
|
|
updated = execute(conn, "UPDATE tickets SET sprint_id=?, updated_at=datetime('now') WHERE id=?", (sprint_id, ticket_id))
|
|
out({"ok": True, "updated": updated, "id": ticket_id, "sprint_id": sprint_id})
|
|
return
|
|
conditions = []
|
|
params = []
|
|
if "active" in flags:
|
|
conditions.append("s.status = 'active'")
|
|
where = " AND ".join(conditions) if conditions else "1=1"
|
|
sprints = query(conn, f"""SELECT s.*, COUNT(t.id) as ticket_count,
|
|
SUM(CASE WHEN t.status='done' THEN 1 ELSE 0 END) as done_count
|
|
FROM sprints s LEFT JOIN tickets t ON t.sprint_id = s.id
|
|
WHERE {where} GROUP BY s.id ORDER BY s.id DESC""", tuple(params))
|
|
out({"ok": True, "count": len(sprints), "sprints": sprints})
|
|
|
|
|
|
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_search(conn, keyword):
|
|
rows = query(conn, """SELECT id, type, title, status, priority, assigned_to
|
|
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"])
|
|
if len(positional) < 2:
|
|
out({"ok": False, "error": "Usage: ticket create <type> <title> [--parent N] [--priority P] [--decision D]"})
|
|
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")
|
|
conn.execute(
|
|
"INSERT INTO tickets (type, title, parent_id, priority, decision_ref) VALUES (?, ?, ?, ?, ?)",
|
|
(ticket_type, title, parent_id, priority, decision_ref))
|
|
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,
|
|
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
|
|
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_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})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
HELP = """ticket — project ticket CLI
|
|
|
|
Usage:
|
|
ticket list [--status S] [--priority P] [--epic N] [--sprint N] [--assigned A]
|
|
ticket show <id> Full ticket detail with deps and children
|
|
ticket done <id> [<id> ...] Mark tickets as done
|
|
ticket status <id> <new_status> Change ticket status
|
|
ticket assign <id> <agent> Assign ticket to agent/branch
|
|
ticket unassign <id> Remove assignment
|
|
ticket sprint [--active] List sprints
|
|
ticket sprint assign <id> <sprint> Assign ticket to sprint
|
|
ticket deps <id> Show ticket dependencies
|
|
ticket search <keyword> Search tickets by title/description
|
|
ticket create <type> <title> [--parent N] [--priority P] [--decision D]
|
|
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"""
|
|
|
|
|
|
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":
|
|
if not args:
|
|
out({"ok": False, "error": "Usage: ticket show <id>"})
|
|
else:
|
|
cmd_show(conn, int(args[0]))
|
|
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 == "sprint":
|
|
cmd_sprint(conn, args)
|
|
elif cmd == "deps":
|
|
if not args:
|
|
out({"ok": False, "error": "Usage: ticket deps <id>"})
|
|
else:
|
|
cmd_deps(conn, args[0])
|
|
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)
|
|
else:
|
|
out({"ok": False, "error": f"Unknown command: {cmd}. Use --help for usage."})
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|