jpmschweitzer/settled-reach
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
4628458ab03e48a7109f985f409e81ae7dd63c8a
settled-reach/tooling/db/ticket
T
jpmschweitzerandClaude Opus 4.6 dd75203d11 refactor(process): replace persistent team worktrees with ephemeral sprint branches 
Persistent worktrees (server, client, copy, audio, visual, ci, planning,
maintenance) caused agents crossing boundaries, stuck agents leaving
uncommitted work, and index.lock collisions. Replaced with ephemeral
sprint branches (sprint-{N}/{team}) and worktrees created on demand.

Changes:
- New start-sprint script replaces start-session (dynamic tabs per active team)
- Sprint teardown integrated into sprint-start skill (A1c step)
- SR_DB_PATH env var for database access from any directory
- CLAUDE.md team boundaries rewritten (scope-based, not directory-based)
- Agent Rule 0 updated to team scope dirs instead of worktree isolation
- PR review uses git show instead of cross-directory reads
- Briefing template updated for sprint-{N}/{team} branch naming
- Deleted worktree-update skill (obsolete)
- Removed WORKTREE_TEAM env var and cross-directory Read permissions
- All 8 persistent worktrees removed

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-05 00:33:02 +02:00

422 lines
17 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] [--sprint 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 sprint [--active]
ticket sprint assign <id> <sprint_id>
ticket deps <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]
All output is JSON on stdout.
"""
import json
import os
import sqlite3
import sys
from pathlib import Path
SCRIPT_DIR = Path(__file__).resolve().parent
CONFIG_PATH = SCRIPT_DIR / "config.json"
WORKTREE_ROOT = (SCRIPT_DIR / ".." / "..").resolve()
# Database path: SR_DB_PATH env var (absolute), or fallback to parent directory heuristic.
DB_PATH = Path(os.environ["SR_DB_PATH"]).resolve() if os.environ.get("SR_DB_PATH") else (WORKTREE_ROOT / ".." / "settledreach.db").resolve()
def load_config():
with open(CONFIG_PATH, "r") as f:
cfg = json.load(f)
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", "team"])
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"])
if "team" in flags:
# Match exact team name within comma-separated list
conditions.append("(',' || t.team || ',' LIKE '%,' || ? || ',%')")
params.append(flags["team"])
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.team, 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, 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]
# 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
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
# Header line
print(f"#{t['id']}: {t['title']}")
# Metadata line
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("sprint_id"):
parts.append(f"Sprint:{t['sprint_id']}")
if t.get("decision_ref"):
parts.append(f"Ref:{t['decision_ref']}")
print(f" {' | '.join(parts)}")
# Description
desc = t.get("description") or ""
if desc:
# Truncate long descriptions
if len(desc) > 200:
desc = desc[:197] + "..."
print(f" {desc}")
# Dependencies
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
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, 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})
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
HELP = """ticket — project ticket CLI
Usage:
ticket list [--status S] [--priority P] [--epic N] [--sprint 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/branch
ticket unassign <id> Remove assignment
ticket team <id> <teams> Set team(s) (comma-separated, e.g. server,client)
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] [--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"""
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 == "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()
Reference in New Issue View Git Blame Copy Permalink