#!/usr/bin/env python3 """ Commonwealth SQLite Connector — mini MCP for ticket management. Usage: python3 sqlite_connector.py init python3 sqlite_connector.py query "SELECT * FROM tickets" python3 sqlite_connector.py execute "UPDATE tickets SET status='done' WHERE id=1" python3 sqlite_connector.py seed-decisions python3 sqlite_connector.py --help """ import json import os import sqlite3 import sys from pathlib import Path # --------------------------------------------------------------------------- # Paths # --------------------------------------------------------------------------- SCRIPT_DIR = Path(__file__).resolve().parent CONFIG_PATH = SCRIPT_DIR / "config.json" WORKTREE_ROOT = (SCRIPT_DIR / ".." / "..").resolve() SCHEMA_PATH = WORKTREE_ROOT / "db" / "schema.sql" # Shared database lives in the parent of all worktrees (three levels up from tooling/db/). DB_PATH = (WORKTREE_ROOT / ".." / "settledreach.db").resolve() def load_config(): """Load config.json and resolve the SQLite database path.""" with open(CONFIG_PATH, "r") as f: cfg = json.load(f) cfg["sqlite_db_resolved"] = str(DB_PATH) return cfg def get_connection(cfg): """Return an sqlite3 connection with WAL mode and foreign keys enabled.""" 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 # --------------------------------------------------------------------------- # Commands # --------------------------------------------------------------------------- def cmd_init(cfg): """Initialise the database from schema.sql.""" if not SCHEMA_PATH.exists(): return {"ok": False, "error": f"Schema file not found: {SCHEMA_PATH}"} schema_sql = SCHEMA_PATH.read_text() conn = get_connection(cfg) try: conn.executescript(schema_sql) conn.commit() return {"ok": True, "message": f"Database initialised at {cfg['sqlite_db_resolved']}"} except sqlite3.Error as exc: return {"ok": False, "error": str(exc)} finally: conn.close() def cmd_query(cfg, sql): """Run a SELECT query and return results as a JSON array of objects.""" conn = get_connection(cfg) try: cursor = conn.execute(sql) columns = [desc[0] for desc in cursor.description] if cursor.description else [] rows = [dict(zip(columns, row)) for row in cursor.fetchall()] return {"ok": True, "count": len(rows), "rows": rows} except sqlite3.Error as exc: return {"ok": False, "error": str(exc)} finally: conn.close() def cmd_execute(cfg, sql): """Run an INSERT/UPDATE/DELETE and return affected row count.""" conn = get_connection(cfg) try: cursor = conn.execute(sql) conn.commit() return { "ok": True, "affected_rows": cursor.rowcount, "last_id": cursor.lastrowid, } except sqlite3.Error as exc: return {"ok": False, "error": str(exc)} finally: conn.close() def cmd_seed_decisions(cfg): """Seed the database with initiatives derived from decisions and open questions.""" decisions = [ ("initiative", "Custom game, not a mod", "backlog", "medium", "D-001"), ("initiative", "Commonwealth as first campaign", "backlog", "medium", "D-003"), ("initiative", "Single character first-person story generator", "backlog", "medium", "D-005"), ("initiative", "Prototype scenario — Institute/Armstrong City/Guardians", "backlog", "medium", "D-006"), ("initiative", "Five pillars of game design", "backlog", "medium", "D-007"), ("initiative", "Action pillar design principles", "backlog", "medium", "D-008"), ("initiative", "Multiplayer — design for it, build single-player first", "backlog", "medium", "D-009"), ("initiative", "Multiplayer-ready architectural baseline", "backlog", "medium", "D-010"), ("initiative", "Fog of perception non-negotiable", "backlog", "medium", "D-011"), ("initiative", "Chunk-based map architecture", "backlog", "medium", "D-012"), ("initiative", "Diegetic insert/POI navigation", "backlog", "medium", "D-013"), ("initiative", "v0.1 map specification", "backlog", "medium", "D-014"), ("initiative", "Camera locked to character", "backlog", "medium", "D-015"), ("initiative", "Internal monologue system", "backlog", "medium", "D-016"), ("initiative", "Perception modes as character build", "backlog", "medium", "D-017"), ("initiative", "Three-range sound model", "backlog", "medium", "D-018"), ("initiative", "Top-down with 3D cutscenes", "backlog", "medium", "D-019"), ] questions = [ ("story", "Game engine selection", "ready", "critical", "Q-001"), ("story", "v0.1 prototype scope", "backlog", "medium", "Q-002"), ("story", "Art direction", "backlog", "medium", "Q-003"), ("story", "One campaign or separate eras", "backlog", "medium", "Q-004"), ("story", "Prototype scale", "backlog", "medium", "Q-005"), ("story", "Target platforms", "backlog", "medium", "Q-007"), ("story", "Licensing/distribution", "backlog", "medium", "Q-008"), ("story", "Time system", "backlog", "medium", "Q-009"), ("story", "Storyteller AI design", "backlog", "medium", "Q-010"), ("story", "Character selection roster", "backlog", "medium", "Q-011"), ] conn = get_connection(cfg) inserted = 0 skipped = 0 try: for ticket_type, title, status, priority, decision_ref in decisions + questions: # Check if a ticket with this decision_ref already exists existing = conn.execute( "SELECT id FROM tickets WHERE decision_ref = ?", (decision_ref,) ).fetchone() if existing: skipped += 1 continue conn.execute( "INSERT INTO tickets (type, title, status, priority, decision_ref) " "VALUES (?, ?, ?, ?, ?)", (ticket_type, title, status, priority, decision_ref), ) inserted += 1 conn.commit() return { "ok": True, "inserted": inserted, "skipped": skipped, "message": f"Seeded {inserted} tickets ({skipped} already existed)", } except sqlite3.Error as exc: conn.rollback() return {"ok": False, "error": str(exc)} finally: conn.close() # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- HELP_TEXT = """\ Commonwealth SQLite Connector Usage: sqlite_connector.py init Create/update database from schema.sql sqlite_connector.py query "" Run a SELECT and return JSON rows sqlite_connector.py execute "" Run INSERT/UPDATE/DELETE, return affected rows sqlite_connector.py seed-decisions Seed initiatives from decisions D-001..D-019 and Q-001..Q-011 sqlite_connector.py --help Show this help message All output is JSON on stdout. Errors also use JSON with {{"ok": false, "error": "..."}}. Config: {config} Schema: {schema} """.format(config=CONFIG_PATH, schema=SCHEMA_PATH) def main(): if len(sys.argv) < 2 or sys.argv[1] in ("--help", "-h", "help"): print(HELP_TEXT) sys.exit(0) cmd = sys.argv[1] try: cfg = load_config() except (FileNotFoundError, json.JSONDecodeError) as exc: print(json.dumps({"ok": False, "error": f"Config error: {exc}"}, indent=2)) sys.exit(1) if cmd == "init": result = cmd_init(cfg) elif cmd == "query": if len(sys.argv) < 3: result = {"ok": False, "error": "query requires a SQL string argument"} else: result = cmd_query(cfg, sys.argv[2]) elif cmd == "execute": if len(sys.argv) < 3: result = {"ok": False, "error": "execute requires a SQL string argument"} else: result = cmd_execute(cfg, sys.argv[2]) elif cmd == "seed-decisions": result = cmd_seed_decisions(cfg) else: result = {"ok": False, "error": f"Unknown command: {cmd}. Use --help for usage."} print(json.dumps(result, indent=2)) sys.exit(0 if result.get("ok") else 1) if __name__ == "__main__": main()