jpmschweitzerandClaude Opus 4.6
873746ff8b
1. decisions_sync.py: fix refs_created inflation (check rowcount), remove dead IntegrityError except block 2. Extract SCHEMA_VERSION to shared tooling/schema_version.py — both generators import from single source of truth 3. generate_atlas.py: narrow bare except to OperationalError + "duplicate column" check 4. check-systems-db-stamp: add cross-generator schema_version agreement assertion (defense-in-depth) 5. decision wrapper: add show + orphan-tickets to usage text 6. Add schema_version.py to all three source watch lists (GENERATOR_SOURCES, IMPORT_ECONOMICS_SOURCES, generate_atlas _write_stamp) — prevents silent staleness on version bump Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
596 lines
20 KiB
Python
596 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Settled Reach 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
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
|
|
from common import CONFIG_PATH, WORKTREE_ROOT, get_connection, load_config # noqa: E402
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Paths
|
|
# ---------------------------------------------------------------------------
|
|
|
|
SCHEMA_PATH = WORKTREE_ROOT / "db" / "schema.sql"
|
|
DECISIONS_DIR = WORKTREE_ROOT / "decisions"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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
|
|
|
|
cur = conn.execute(
|
|
"""INSERT OR IGNORE INTO decision_refs
|
|
(source_id, target_id, ref_type, note)
|
|
VALUES (?, ?, ?, ?)""",
|
|
(d["id"], target_id, ref_type, note),
|
|
)
|
|
if cur.rowcount > 0:
|
|
refs_created += 1
|
|
|
|
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 show_decision(cfg, decision_id):
|
|
"""Show a single decision with full details including linked tickets and cross-refs."""
|
|
conn = get_connection(cfg)
|
|
try:
|
|
row = conn.execute(
|
|
"SELECT * FROM decisions WHERE id = ?",
|
|
(decision_id,),
|
|
).fetchone()
|
|
|
|
if not row:
|
|
return {"ok": False, "error": f"Decision not found: {decision_id}"}
|
|
|
|
decision = dict(row)
|
|
|
|
# Implementing tickets: tickets where decision_ref = this ID
|
|
ticket_rows = conn.execute(
|
|
"SELECT id, title, status, type FROM tickets"
|
|
" WHERE decision_ref = ? ORDER BY id",
|
|
(decision_id,),
|
|
).fetchall()
|
|
decision["implementing_tickets"] = [dict(t) for t in ticket_rows]
|
|
|
|
# Cross-refs outbound: references from this decision to others
|
|
refs_out = conn.execute(
|
|
"SELECT target_id, ref_type, note FROM decision_refs"
|
|
" WHERE source_id = ? ORDER BY target_id",
|
|
(decision_id,),
|
|
).fetchall()
|
|
decision["refs_out"] = [dict(r) for r in refs_out]
|
|
|
|
# Cross-refs inbound: other decisions referencing this one
|
|
refs_in = conn.execute(
|
|
"SELECT source_id, ref_type, note FROM decision_refs"
|
|
" WHERE target_id = ? ORDER BY source_id",
|
|
(decision_id,),
|
|
).fetchall()
|
|
decision["refs_in"] = [dict(r) for r in refs_in]
|
|
|
|
return {"ok": True, "decision": decision}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def orphan_tickets(cfg):
|
|
"""List tickets whose decision_ref is set but does not match any decision in the DB."""
|
|
conn = get_connection(cfg)
|
|
try:
|
|
rows = conn.execute(
|
|
"""SELECT t.id, t.title, t.decision_ref, t.status, t.team
|
|
FROM tickets t
|
|
WHERE t.decision_ref IS NOT NULL
|
|
AND t.decision_ref != ''
|
|
AND t.decision_ref NOT IN (SELECT id FROM decisions)
|
|
ORDER BY t.decision_ref, t.id""",
|
|
).fetchall()
|
|
|
|
orphans = [dict(r) for r in rows]
|
|
|
|
return {
|
|
"ok": True,
|
|
"count": len(orphans),
|
|
"orphans": orphans,
|
|
"summary": (
|
|
f"{len(orphans)} orphan ticket(s) found"
|
|
if orphans
|
|
else "No orphan tickets — all decision_ref values are valid"
|
|
),
|
|
}
|
|
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 = """\
|
|
Settled Reach Decisions Sync & ID Management
|
|
|
|
Usage:
|
|
decisions_sync.py sync Parse decisions/*.md and upsert into SQLite
|
|
decisions_sync.py show <D-NNN> Show a decision with linked tickets + refs
|
|
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 orphan-tickets List tickets with invalid/missing decision_ref
|
|
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 == "show":
|
|
if len(sys.argv) < 3:
|
|
result = {"ok": False, "error": "Usage: show <decision_id> e.g. show D-159"}
|
|
else:
|
|
result = show_decision(cfg, sys.argv[2])
|
|
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)
|
|
elif cmd == "orphan-tickets":
|
|
result = orphan_tickets(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()
|