jpmschweitzerandClaude Opus 4.6
bc226d8baf
Consolidates all connector scripts under tooling/ per project structure conventions. Symlink at db/connectors → tooling/db/ preserves backwards compatibility (remove after Sprint 22). Updated references in CLAUDE.md, Makefile, DEVOPS.md, all skill files, agent files, rules, schema comments, and Sprint 21 briefings. Python scripts updated with correct SCHEMA_PATH (now relative to WORKTREE_ROOT/db/schema.sql). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
539 lines
18 KiB
Python
539 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Commonwealth Decisions Sync — parse decisions/*.md domain files into SQLite.
|
|
|
|
Reads all markdown files from the decisions/ directory, parses decision blocks
|
|
(D-NNN, Q-NNN, R-NNN), extracts metadata, and upserts into the decisions and
|
|
decision_refs tables.
|
|
|
|
Usage:
|
|
python3 decisions_sync.py sync Parse and upsert all decisions
|
|
python3 decisions_sync.py --help Show this help message
|
|
"""
|
|
|
|
import json
|
|
import re
|
|
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"
|
|
DECISIONS_DIR = WORKTREE_ROOT / "decisions"
|
|
# Shared database lives in the parent of all worktrees (three levels up from tooling/db/).
|
|
DB_PATH = (WORKTREE_ROOT / ".." / "settledreach.db").resolve()
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config / DB (same pattern as sqlite_connector.py)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Parsing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Matches headings like: ### D-008: Action pillar design principles
|
|
HEADING_RE = re.compile(r"^###\s+((?:D|Q|R)-\d{3}):\s+(.+)$")
|
|
|
|
# Matches metadata lines like: - **Date:** 2026-02-08
|
|
DATE_RE = re.compile(r"^\s*-\s+\*\*(?:Date|Rejected):\*\*\s+(\d{4}-\d{2}-\d{2})")
|
|
STATUS_RE = re.compile(r"^\s*-\s+\*\*Status:\*\*\s+(.+)")
|
|
ROUND_RE = re.compile(r"Round\s+(\d+)", re.IGNORECASE)
|
|
|
|
# Cross-reference patterns in body text
|
|
REF_RE = re.compile(r"(?:D|Q|R)-\d{3}")
|
|
|
|
# Contextual reference patterns (on specific metadata lines)
|
|
SUPERSEDES_RE = re.compile(r"^\s*-\s+\*\*Supersedes:\*\*", re.IGNORECASE)
|
|
SUPERSEDED_BY_RE = re.compile(r"^\s*-\s+\*\*Superseded\s+by:\*\*", re.IGNORECASE)
|
|
RESOLVES_RE = re.compile(r"^\s*-\s+\*\*Resolves:\*\*", re.IGNORECASE)
|
|
CROSS_REF_RE = re.compile(r"^\s*-\s+\*\*Cross-reference:\*\*", re.IGNORECASE)
|
|
DEPENDS_RE = re.compile(r"^\s*-\s+\*\*Depends\s+on:\*\*", re.IGNORECASE)
|
|
|
|
# Title may include [SUPERSEDED] suffix
|
|
SUPERSEDED_TITLE_RE = re.compile(r"\s*\[SUPERSEDED\]\s*$", re.IGNORECASE)
|
|
|
|
|
|
def classify_id(decision_id):
|
|
"""Return the type string for a decision ID prefix."""
|
|
prefix = decision_id[0]
|
|
return {"D": "confirmed", "Q": "question", "R": "rejected"}[prefix]
|
|
|
|
|
|
def infer_status(decision_id, title, body_lines):
|
|
"""Infer the status of a decision from its content."""
|
|
id_type = classify_id(decision_id)
|
|
|
|
# Rejected alternatives are always 'rejected' (maps to our status concept)
|
|
if id_type == "rejected":
|
|
return "active"
|
|
|
|
# Check for [SUPERSEDED] in title
|
|
if SUPERSEDED_TITLE_RE.search(title):
|
|
return "superseded"
|
|
|
|
# Check body for "Superseded by:" line
|
|
for line in body_lines:
|
|
if SUPERSEDED_BY_RE.match(line):
|
|
return "superseded"
|
|
|
|
# Questions: check if resolved
|
|
if id_type == "question":
|
|
for line in body_lines:
|
|
m = STATUS_RE.match(line)
|
|
if m:
|
|
status_text = m.group(1).strip()
|
|
lower = status_text.lower()
|
|
# "Partially resolved/scoped" or qualified "X resolved...Remaining" = still open
|
|
if "partial" in lower or "remaining" in lower:
|
|
return "open"
|
|
# Clean "Resolved ->" pattern = fully resolved
|
|
if lower.startswith("resolved"):
|
|
return "resolved"
|
|
# Everything else (not yet discussed, etc.) = open
|
|
return "open"
|
|
return "open"
|
|
|
|
return "active"
|
|
|
|
|
|
def extract_round(body_lines):
|
|
"""Try to find a Round number from the decision body."""
|
|
for line in body_lines:
|
|
m = ROUND_RE.search(line)
|
|
if m:
|
|
return int(m.group(1))
|
|
return None
|
|
|
|
|
|
def extract_date(body_lines):
|
|
"""Extract date from metadata lines."""
|
|
for line in body_lines:
|
|
m = DATE_RE.match(line)
|
|
if m:
|
|
return m.group(1)
|
|
return None
|
|
|
|
|
|
def extract_refs(decision_id, body_lines):
|
|
"""
|
|
Extract typed references from the body of a decision block.
|
|
|
|
Returns a list of (target_id, ref_type, note) tuples.
|
|
"""
|
|
refs = []
|
|
seen = set()
|
|
|
|
for line in body_lines:
|
|
# Determine the ref_type based on the line context
|
|
if SUPERSEDES_RE.match(line):
|
|
ref_type = "supersedes"
|
|
elif SUPERSEDED_BY_RE.match(line):
|
|
# The *other* decision supersedes *this* one.
|
|
# We record it as the other decision superseding us,
|
|
# but from our perspective we store it as a reference.
|
|
# The canonical direction: source supersedes target.
|
|
# Here source=other, target=us. We'll record source=us,
|
|
# target=other with ref_type='references' (since we're
|
|
# the superseded party; the superseder's block carries
|
|
# the 'supersedes' ref).
|
|
ref_type = "references"
|
|
elif RESOLVES_RE.match(line):
|
|
ref_type = "resolves"
|
|
elif DEPENDS_RE.match(line):
|
|
ref_type = "depends_on"
|
|
elif CROSS_REF_RE.match(line):
|
|
ref_type = "references"
|
|
else:
|
|
ref_type = "references"
|
|
|
|
# Find all decision IDs on this line
|
|
for target in REF_RE.findall(line):
|
|
if target == decision_id:
|
|
continue # skip self-references
|
|
key = (target, ref_type)
|
|
if key not in seen:
|
|
seen.add(key)
|
|
note = line.strip().lstrip("- ").rstrip()
|
|
# Truncate note to something reasonable
|
|
if len(note) > 200:
|
|
note = note[:197] + "..."
|
|
refs.append((target, ref_type, note))
|
|
|
|
return refs
|
|
|
|
|
|
def parse_file(filepath):
|
|
"""
|
|
Parse a single decisions/*.md file into a list of decision dicts.
|
|
|
|
Each dict has: id, type, domain, title, status, round, date, file_path,
|
|
and a refs list of (target_id, ref_type, note).
|
|
"""
|
|
domain = filepath.stem # e.g. "architecture" from "architecture.md"
|
|
rel_path = str(filepath.relative_to(WORKTREE_ROOT))
|
|
text = filepath.read_text(encoding="utf-8")
|
|
lines = text.split("\n")
|
|
|
|
decisions = []
|
|
current_id = None
|
|
current_title = None
|
|
current_body = []
|
|
|
|
def flush():
|
|
if current_id is None:
|
|
return
|
|
clean_title = SUPERSEDED_TITLE_RE.sub("", current_title).strip()
|
|
decisions.append({
|
|
"id": current_id,
|
|
"type": classify_id(current_id),
|
|
"domain": domain,
|
|
"title": clean_title,
|
|
"status": infer_status(current_id, current_title, current_body),
|
|
"round": extract_round(current_body),
|
|
"date": extract_date(current_body),
|
|
"file_path": rel_path,
|
|
"refs": extract_refs(current_id, current_body),
|
|
})
|
|
|
|
for line in lines:
|
|
m = HEADING_RE.match(line)
|
|
if m:
|
|
flush()
|
|
current_id = m.group(1)
|
|
current_title = m.group(2)
|
|
current_body = []
|
|
elif current_id is not None:
|
|
# Stop collecting body at the next --- separator or new ### heading
|
|
if line.strip() == "---":
|
|
flush()
|
|
current_id = None
|
|
current_title = None
|
|
current_body = []
|
|
else:
|
|
current_body.append(line)
|
|
|
|
# Flush final block (file may not end with ---)
|
|
flush()
|
|
|
|
return decisions
|
|
|
|
|
|
def parse_all():
|
|
"""Parse all decisions/*.md files. Returns (decisions_list, warnings)."""
|
|
if not DECISIONS_DIR.is_dir():
|
|
return [], [f"Decisions directory not found: {DECISIONS_DIR}"]
|
|
|
|
all_decisions = []
|
|
warnings = []
|
|
|
|
md_files = sorted(DECISIONS_DIR.glob("*.md"))
|
|
# Skip README.md
|
|
md_files = [f for f in md_files if f.name.lower() != "readme.md"]
|
|
|
|
if not md_files:
|
|
warnings.append(f"No .md files found in {DECISIONS_DIR}")
|
|
return all_decisions, warnings
|
|
|
|
for filepath in md_files:
|
|
try:
|
|
decisions = parse_file(filepath)
|
|
all_decisions.extend(decisions)
|
|
except Exception as exc:
|
|
warnings.append(f"Error parsing {filepath.name}: {exc}")
|
|
|
|
return all_decisions, warnings
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Database sync
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def sync(cfg):
|
|
"""Parse all decision files and upsert into the database."""
|
|
decisions, warnings = parse_all()
|
|
|
|
if not decisions and warnings:
|
|
return {
|
|
"ok": False,
|
|
"error": "No decisions parsed",
|
|
"warnings": warnings,
|
|
}
|
|
|
|
# Collect all known IDs for reference validation
|
|
known_ids = {d["id"] for d in decisions}
|
|
|
|
conn = get_connection(cfg)
|
|
try:
|
|
# Ensure tables exist (idempotent)
|
|
schema_sql = SCHEMA_PATH.read_text()
|
|
conn.executescript(schema_sql)
|
|
|
|
upserted = 0
|
|
refs_created = 0
|
|
broken_refs = []
|
|
|
|
# Clear existing refs (we rebuild every sync)
|
|
conn.execute("DELETE FROM decision_refs")
|
|
|
|
# Pass 1: Upsert all decisions (so foreign keys resolve in pass 2)
|
|
for d in decisions:
|
|
conn.execute(
|
|
"""INSERT INTO decisions (id, type, domain, title, status, round, date, file_path, synced_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))
|
|
ON CONFLICT(id) DO UPDATE SET
|
|
type = excluded.type,
|
|
domain = excluded.domain,
|
|
title = excluded.title,
|
|
status = excluded.status,
|
|
round = excluded.round,
|
|
date = excluded.date,
|
|
file_path = excluded.file_path,
|
|
synced_at = datetime('now')""",
|
|
(d["id"], d["type"], d["domain"], d["title"],
|
|
d["status"], d["round"], d["date"], d["file_path"]),
|
|
)
|
|
upserted += 1
|
|
|
|
# Pass 2: Insert all references (all targets now exist)
|
|
for d in decisions:
|
|
for target_id, ref_type, note in d["refs"]:
|
|
if target_id not in known_ids:
|
|
broken_refs.append(
|
|
f"{d['id']} -> {target_id} ({ref_type}): target not found"
|
|
)
|
|
warnings.append(
|
|
f"Broken reference: {d['id']} -> {target_id} "
|
|
f"({ref_type}) in {d['file_path']}"
|
|
)
|
|
continue
|
|
|
|
try:
|
|
conn.execute(
|
|
"""INSERT OR IGNORE INTO decision_refs
|
|
(source_id, target_id, ref_type, note)
|
|
VALUES (?, ?, ?, ?)""",
|
|
(d["id"], target_id, ref_type, note),
|
|
)
|
|
refs_created += 1
|
|
except sqlite3.IntegrityError:
|
|
pass # duplicate ref, skip
|
|
|
|
conn.commit()
|
|
|
|
return {
|
|
"ok": True,
|
|
"decisions_synced": upserted,
|
|
"refs_created": refs_created,
|
|
"broken_refs": len(broken_refs),
|
|
"warnings": warnings,
|
|
"summary": (
|
|
f"Synced {upserted} decisions, "
|
|
f"{refs_created} refs created, "
|
|
f"{len(broken_refs)} broken refs, "
|
|
f"{len(warnings)} warnings"
|
|
),
|
|
}
|
|
|
|
except sqlite3.Error as exc:
|
|
conn.rollback()
|
|
return {"ok": False, "error": str(exc), "warnings": warnings}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ID claiming — database is authority for ID allocation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def next_id(cfg, prefix=None):
|
|
"""Return the next available ID for a given prefix (D, Q, R) or all."""
|
|
conn = get_connection(cfg)
|
|
try:
|
|
result = {}
|
|
prefixes = [prefix.upper()] if prefix else ["D", "Q", "R"]
|
|
for p in prefixes:
|
|
# Check both DB and markdown files for the highest ID
|
|
row = conn.execute(
|
|
"SELECT MAX(CAST(SUBSTR(id, 3) AS INTEGER)) as max_num "
|
|
"FROM decisions WHERE id LIKE ?",
|
|
(f"{p}-%",),
|
|
).fetchone()
|
|
db_max = row["max_num"] if row and row["max_num"] else 0
|
|
|
|
# Also scan markdown files in case they're ahead of the DB
|
|
md_max = 0
|
|
for filepath in sorted(DECISIONS_DIR.glob("*.md")):
|
|
if filepath.name.lower() == "readme.md":
|
|
continue
|
|
text = filepath.read_text(encoding="utf-8")
|
|
for m in re.finditer(rf"^###\s+{p}-(\d{{3}}):", text, re.MULTILINE):
|
|
num = int(m.group(1))
|
|
if num > md_max:
|
|
md_max = num
|
|
|
|
highest = max(db_max, md_max)
|
|
next_num = highest + 1
|
|
result[p] = f"{p}-{next_num:03d}"
|
|
|
|
return {"ok": True, **result}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def claim_id(cfg, prefix, domain, title):
|
|
"""Claim the next available ID and insert a placeholder into the DB."""
|
|
if prefix not in ("D", "Q", "R"):
|
|
return {"ok": False, "error": f"Invalid prefix: {prefix}. Must be D, Q, or R."}
|
|
|
|
type_map = {"D": "confirmed", "Q": "question", "R": "rejected"}
|
|
status_map = {"D": "active", "Q": "open", "R": "active"}
|
|
|
|
nxt = next_id(cfg, prefix)
|
|
if not nxt.get("ok"):
|
|
return nxt
|
|
|
|
new_id = nxt[prefix]
|
|
conn = get_connection(cfg)
|
|
try:
|
|
conn.execute(
|
|
"""INSERT INTO decisions (id, type, domain, title, status, file_path, synced_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, datetime('now'))""",
|
|
(new_id, type_map[prefix], domain, title, status_map[prefix],
|
|
f"decisions/{domain}.md"),
|
|
)
|
|
conn.commit()
|
|
return {"ok": True, "id": new_id, "domain": domain, "title": title}
|
|
except sqlite3.IntegrityError as exc:
|
|
conn.rollback()
|
|
return {"ok": False, "error": f"ID conflict: {exc}"}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def check_dupes(cfg):
|
|
"""Check for duplicate decision IDs across all markdown files."""
|
|
# Pre-existing collisions too deeply embedded to renumber (139+ references).
|
|
# New collisions are prevented by the claim workflow.
|
|
KNOWN_EXCEPTIONS = {"D-035"}
|
|
|
|
id_locations = {} # id -> [(file, line_number)]
|
|
warnings = []
|
|
|
|
for filepath in sorted(DECISIONS_DIR.glob("*.md")):
|
|
if filepath.name.lower() == "readme.md":
|
|
continue
|
|
text = filepath.read_text(encoding="utf-8")
|
|
for i, line in enumerate(text.split("\n"), 1):
|
|
m = HEADING_RE.match(line)
|
|
if m:
|
|
did = m.group(1)
|
|
if did not in id_locations:
|
|
id_locations[did] = []
|
|
id_locations[did].append((filepath.name, i))
|
|
|
|
dupes = {did: locs for did, locs in id_locations.items()
|
|
if len(locs) > 1 and did not in KNOWN_EXCEPTIONS}
|
|
|
|
if dupes:
|
|
for did, locs in sorted(dupes.items()):
|
|
loc_str = ", ".join(f"{f}:{ln}" for f, ln in locs)
|
|
warnings.append(f"DUPLICATE {did}: {loc_str}")
|
|
|
|
return {
|
|
"ok": len(dupes) == 0,
|
|
"total_ids": len(id_locations),
|
|
"duplicates": len(dupes),
|
|
"known_exceptions": list(KNOWN_EXCEPTIONS),
|
|
"details": warnings,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
HELP_TEXT = """\
|
|
Commonwealth Decisions Sync & ID Management
|
|
|
|
Usage:
|
|
decisions_sync.py sync Parse decisions/*.md and upsert into SQLite
|
|
decisions_sync.py next [D|Q|R] Show next available ID (all prefixes or one)
|
|
decisions_sync.py claim <D|Q|R> <domain> [title] Claim next ID and insert placeholder
|
|
decisions_sync.py check-dupes Check for duplicate IDs across markdown files
|
|
decisions_sync.py --help Show this help message
|
|
|
|
ID claiming workflow:
|
|
1. Agent calls 'claim D architecture "Per-game save dirs"'
|
|
2. Gets back D-085 (or whatever is next)
|
|
3. Agent writes D-085 in the appropriate domain file
|
|
4. Pre-commit hook runs check-dupes to catch collisions
|
|
|
|
Config: {config}
|
|
Schema: {schema}
|
|
Source: {decisions}
|
|
""".format(config=CONFIG_PATH, schema=SCHEMA_PATH, decisions=DECISIONS_DIR)
|
|
|
|
|
|
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 == "sync":
|
|
result = sync(cfg)
|
|
elif cmd == "next":
|
|
result = next_id(cfg, sys.argv[2] if len(sys.argv) > 2 else None)
|
|
elif cmd == "claim":
|
|
if len(sys.argv) < 4:
|
|
result = {"ok": False, "error": "Usage: claim <D|Q|R> <domain> [title]"}
|
|
else:
|
|
prefix = sys.argv[2].upper()
|
|
domain = sys.argv[3]
|
|
title = " ".join(sys.argv[4:]) if len(sys.argv) > 4 else "(unclaimed)"
|
|
result = claim_id(cfg, prefix, domain, title)
|
|
elif cmd == "check-dupes":
|
|
result = check_dupes(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()
|