jpmschweitzer/settled-reach
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(db): add decisions sync script and schema tables

Browse Source 
Add decisions_sync.py that parses decisions/*.md markdown headings,
extracts metadata (type, domain, status, round, date), and upserts
into SQLite. Two-pass approach: decisions first, then cross-references
to avoid FK violations on forward references.

Schema adds decisions table (52 rows) and decision_refs table (29 rows)
with cascading deletes. Makefile gains decisions-sync, decisions-coverage,
decisions-active, and decisions-orphan targets. Sync runs as part of
make setup.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jpmschweitzer
2026-02-11 09:29:48 +01:00
co-authored by Claude Opus 4.6
parent 4f5ee0274b
commit 485b02db67
4 changed files with 472 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/connectors/decisions-sync
Executable
+4
View File
@@ -0,0 +1,4 @@
#!/usr/bin/env bash
# Sync decisions/*.md domain files into the SQLite database. Whitelistable command.
# Usage: decisions-sync
exec python3 "$(dirname "$0")/decisions_sync.py" sync "$@"
db/connectors/decisions_sync.py
+414
View File
@@ -0,0 +1,414 @@
#!/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"
SCHEMA_PATH = SCRIPT_DIR.parent / "schema.sql"
REPO_ROOT = SCRIPT_DIR.parent.parent
DECISIONS_DIR = REPO_ROOT / "decisions"
# ---------------------------------------------------------------------------
# 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)
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
# ---------------------------------------------------------------------------
# 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(REPO_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()
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
HELP_TEXT = """\
Commonwealth Decisions Sync
Usage:
decisions_sync.py sync Parse decisions/*.md and upsert into SQLite
decisions_sync.py --help Show this help message
Parses all markdown files in decisions/ (excluding README.md), extracts
decision blocks (D-NNN, Q-NNN, R-NNN), and syncs them into the decisions
and decision_refs tables.
Idempotent: safe to run repeatedly. References are rebuilt on every sync.
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)
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()
db/schema.sql
+32
View File
@@ -59,3 +59,35 @@ CREATE INDEX IF NOT EXISTS idx_tickets_sprint ON tickets(sprint_id);
CREATE INDEX IF NOT EXISTS idx_tickets_assigned ON tickets(assigned_to);
CREATE INDEX IF NOT EXISTS idx_tickets_decision ON tickets(decision_ref);
CREATE INDEX IF NOT EXISTS idx_history_ticket ON ticket_history(ticket_id);
-- ---------------------------------------------------------------------------
-- Decision Sync Tables
-- Populated by: python3 db/connectors/decisions_sync.py
-- Source: decisions/*.md domain files
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS decisions (
id TEXT PRIMARY KEY,
type TEXT NOT NULL CHECK(type IN ('confirmed', 'question', 'rejected')),
domain TEXT NOT NULL,
title TEXT NOT NULL,
status TEXT DEFAULT 'active' CHECK(status IN ('active', 'superseded', 'resolved', 'open')),
round INTEGER,
date TEXT,
file_path TEXT NOT NULL,
synced_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS decision_refs (
source_id TEXT NOT NULL REFERENCES decisions(id) ON DELETE CASCADE,
target_id TEXT NOT NULL REFERENCES decisions(id) ON DELETE CASCADE,
ref_type TEXT NOT NULL CHECK(ref_type IN ('supersedes', 'references', 'resolves', 'depends_on')),
note TEXT,
PRIMARY KEY (source_id, target_id, ref_type)
);
CREATE INDEX IF NOT EXISTS idx_decisions_type ON decisions(type);
CREATE INDEX IF NOT EXISTS idx_decisions_domain ON decisions(domain);
CREATE INDEX IF NOT EXISTS idx_decisions_status ON decisions(status);
CREATE INDEX IF NOT EXISTS idx_decision_refs_source ON decision_refs(source_id);
CREATE INDEX IF NOT EXISTS idx_decision_refs_target ON decision_refs(target_id);