jpmschweitzerandClaude Opus 4.6
e7b632b02d
Set up Commonwealth game project with complete agent team (18 agents), discussion archives (12 rounds), briefing system, ticketing database, Qdrant document search, and skill definitions. Establishes the documentation and tooling foundation for engine selection and implementation phases. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
223 lines
8.3 KiB
Python
Executable File
223 lines
8.3 KiB
Python
Executable File
#!/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"
|
|
SCHEMA_PATH = SCRIPT_DIR.parent / "schema.sql"
|
|
|
|
|
|
def load_config():
|
|
"""Load config.json and resolve the SQLite database path."""
|
|
with open(CONFIG_PATH, "r") as f:
|
|
cfg = json.load(f)
|
|
# Resolve sqlite_db relative to the connectors directory
|
|
db_path = (SCRIPT_DIR / cfg["sqlite_db"]).resolve()
|
|
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 "<SQL>" Run a SELECT and return JSON rows
|
|
sqlite_connector.py execute "<SQL>" 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()
|