feat(economics): D-237 specialization CI guardrails + coverage report (#1015)

Adds the V-SES / V-FAC guardrail suite to import_system_specialization():

Always-on hard errors (abort the import transaction): V-SES-01 economic
value in vocabulary; V-SES-03 production_ubiquity_override >= commodity
catalog concentration (HUB values with empty override exempt — they
intentionally project lower); V-SES-04 cultural value in the combined
activity+heritage vocabulary; V-SES-05 faction in 8-value vocab; V-SES-06
vocab commodity FK; plus unknown system_id.

Completeness gates — V-SES-02 (every inhabited system resolves a non-null
economic value) and V-FAC-01 (every inhabited named system has authored
faction) — are HARD only under --strict-specialization (default off). Their
preconditions are the #1014 heuristic fallback (blocked by #982) and the
#1016 content pass; until those land they print as warnings so regen-db
stays green.

Soft warnings (always print): W-SES-01 MonopolySource D-177 review list,
W-SES-02 single-value concentration, W-SES-08 estate_farming + large pop,
W-FAC-01 compact + TRACTUS currency, W-FAC-02 compact + MonopolySource
(terroir/marble exempt), W-FAC-03 syndic_dominant w/o corp HQ, W-FAC-04
compact_sympathetic + MARK currency. Coverage report always prints
economic/cultural/faction counts + BulkClass/ProductionUbiquity dists.

Cultural vocabulary is _CULTURAL_ACTIVITY | _CULTURAL_HERITAGE module
constants — #1016 adds heritage values there in one place.

Validated on a copy of the live DB: default exits 0 (gates as warnings, real
W-SES-08 hit on Arbour); strict exits 1 on V-SES-02 (Sol, Struve) and rolls
back cleanly; dry-run exits 0.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-01 12:44:03 +02:00
co-authored by Claude Opus 4.8
parent 883c1ab0b1
commit 2f994c1795
+253 -16
View File
@@ -1608,9 +1608,33 @@ _FACTION_VOCAB = {
"veil_institute", "independent", "disputed", "mixed",
}
# Combined cultural_specialization vocabulary (D-237; miri-round3 §2). Two value
# kinds in one column: activity/character and founding-heritage. EXTENSIBLE — the
# #1016 content pass adds heritage values here as more GTTR systems are reviewed;
# add the new value to this set and CI accepts it. V-SES-04 validates against it.
_CULTURAL_ACTIVITY = {
"scholarly", "artistic", "institutional", "commercial", "agrarian",
"industrial_heritage", "medical_elite", "ecological", "military",
"financial_technocratic", "cosmopolitan", "compact_cooperative",
}
_CULTURAL_HERITAGE = {
"scottish", "french_provencal", "vietnamese", "afrikaans_cape", "zulu",
"chinese", "italian_northern", "tagalog", "norse_compact", "irish",
"portuguese_iberian", "korean", "japanese", "swahili", "hausa",
"bengali", "punjabi", "yoruba",
}
_CULTURAL_VOCAB = _CULTURAL_ACTIVITY | _CULTURAL_HERITAGE
def import_system_specialization(conn: sqlite3.Connection, dry_run: bool) -> dict:
"""Import the D-237 authored specialization layer.
# Catalog production_ubiquity concentration ranking for V-SES-03 (override may
# only be >= the commodity's global default). regional ≈ common tier.
_UBIQUITY_RANK = {
"ubiquitous": 0, "common": 1, "regional": 1, "concentrated": 2, "monopolistic": 3,
}
def import_system_specialization(conn: sqlite3.Connection, dry_run: bool,
strict: bool = False) -> dict:
"""Import + validate the D-237 authored specialization layer (#1013, #1015).
Reads two TOMLs:
- specialization_vocabulary.toml -> specialization_vocabulary table
@@ -1626,17 +1650,33 @@ def import_system_specialization(conn: sqlite3.Connection, dry_run: bool) -> dic
is shared with other derivation paths, so it is ONLY overwritten for systems
present in the TOML (per #1013) — never globally cleared.
CI guardrails (#1015):
Always-on hard errors (abort): V-SES-01 (econ value in vocab), V-SES-03
(override >= catalog concentration), V-SES-04 (cultural value in vocab),
V-SES-05 (faction in vocab), V-SES-06 (vocab commodity FK), plus unknown
system_id.
Completeness gates V-SES-02 (every inhabited system resolves a non-null
economic value) and V-FAC-01 (every inhabited named system has authored
dominant_faction) are HARD only under `strict` — their preconditions are
the #1014 fallback (blocked by #982) and the #1016 content pass. Until
those land, they emit warnings; flip `--strict-specialization` on once
both are complete so regen-db enforces them.
Soft warnings (W-SES-*, W-FAC-*) and the coverage report always print.
Returns a coverage dict for the caller's report. Raises _ImportAborted on a
hard validation failure (FK, enum, or unknown system_id).
hard validation failure.
"""
with open(SPECIALIZATION_VOCAB_TOML, "rb") as f:
vocab = tomllib.load(f)
with open(SYSTEM_SPECIALIZATION_TOML, "rb") as f:
systems = tomllib.load(f)
commodity_ids = {
r[0] for r in conn.execute("SELECT commodity_id FROM commodities").fetchall()
commodity_pu = {
r[0]: r[1] for r in conn.execute(
"SELECT commodity_id, production_ubiquity FROM commodities"
).fetchall()
}
commodity_ids = set(commodity_pu)
system_ids = {
r[0] for r in conn.execute("SELECT system_id FROM star_systems").fetchall()
}
@@ -1649,13 +1689,13 @@ def import_system_specialization(conn: sqlite3.Connection, dry_run: bool) -> dic
errors: list[str] = []
# --- Vocabulary: FK + enum validation -------------------------------
# --- Vocabulary: FK + enum validation (V-SES-06, V-SES-03) ----------
vocab_rows = []
for spec_id, v in vocab.items():
cid = v.get("commodity_id")
if cid not in commodity_ids:
if cid not in commodity_ids: # V-SES-06
errors.append(
f"specialization_vocabulary '{spec_id}': commodity_id "
f"V-SES-06: specialization_vocabulary '{spec_id}': commodity_id "
f"'{cid}' not in commodities catalog"
)
bc = v.get("bulk_class_projected")
@@ -1672,26 +1712,50 @@ def import_system_specialization(conn: sqlite3.Connection, dry_run: bool) -> dic
f"(expected one of {sorted(_PRODUCTION_UBIQUITY)})"
)
override = v.get("production_ubiquity_override") or None # "" -> NULL
vocab_rows.append((spec_id, cid, override, bc, pu, v.get("description", "")))
# V-SES-03: a non-empty override may only raise (or equal) the
# commodity's global concentration — never claim a globally scarce good
# is locally more common. Empty override = HUB value (intentionally
# projects below catalog; exempt — see vocab TOML header).
if override is not None and cid in commodity_pu:
cat_rank = _UBIQUITY_RANK.get(commodity_pu[cid], -1)
ovr_rank = _UBIQUITY_RANK.get(override, -1)
if ovr_rank < 0:
errors.append(
f"V-SES-03: specialization_vocabulary '{spec_id}': "
f"production_ubiquity_override '{override}' not a catalog term"
)
elif ovr_rank < cat_rank:
errors.append(
f"V-SES-03: specialization_vocabulary '{spec_id}': override "
f"'{override}' is less concentrated than commodity "
f"'{cid}' catalog default '{commodity_pu[cid]}' — incoherent"
)
valid_spec_ids = set(vocab.keys())
# --- System stanzas: id + value validation --------------------------
# --- System stanzas: id + value validation (V-SES-01/04/05) ---------
for sid, s in systems.items():
if sid not in system_ids:
errors.append(
f"system_specialization '{sid}': not a known star_systems.system_id"
)
es = s.get("economic_specialization")
if es is not None and es not in valid_spec_ids:
if es is not None and es not in valid_spec_ids: # V-SES-01
errors.append(
f"system_specialization '{sid}': economic_specialization "
f"V-SES-01: system_specialization '{sid}': economic_specialization "
f"'{es}' not in specialization_vocabulary"
)
df = s.get("dominant_faction")
if df is not None and df not in _FACTION_VOCAB:
cs = s.get("cultural_specialization")
if cs is not None and cs not in _CULTURAL_VOCAB: # V-SES-04
errors.append(
f"system_specialization '{sid}': dominant_faction "
f"V-SES-04: system_specialization '{sid}': cultural_specialization "
f"'{cs}' not in the activity+heritage vocabulary "
f"(add new heritage values to _CULTURAL_HERITAGE)"
)
df = s.get("dominant_faction")
if df is not None and df not in _FACTION_VOCAB: # V-SES-05
errors.append(
f"V-SES-05: system_specialization '{sid}': dominant_faction "
f"'{df}' invalid (expected one of {sorted(_FACTION_VOCAB)})"
)
@@ -1715,6 +1779,7 @@ def import_system_specialization(conn: sqlite3.Connection, dry_run: bool) -> dic
coverage["economic"] += 1 if s.get("economic_specialization") else 0
coverage["cultural"] += 1 if s.get("cultural_specialization") else 0
coverage["faction"] += 1 if s.get("dominant_faction") else 0
_specialization_checks(conn, vocab, systems, strict)
return coverage
# --- Repopulate vocabulary table ------------------------------------
@@ -1760,9 +1825,174 @@ def import_system_specialization(conn: sqlite3.Connection, dry_run: bool) -> dic
else:
coverage["missing_faction_row"].append(sid)
# --- Completeness gates + soft warnings + coverage report -----------
# Run AFTER the UPSERTs so they see the freshly-written DB state.
_specialization_checks(conn, vocab, systems, strict)
return coverage
def _specialization_checks(conn, vocab, systems, strict):
"""V-SES-02 / V-FAC-01 completeness gates, soft warnings, coverage report.
Reads the post-UPSERT DB state. Gates are warnings unless `strict` (their
preconditions — the #1014 fallback and the #1016 content pass — are not yet
in place). Raises _ImportAborted only when strict and a gate fails.
"""
gate_failures: list[str] = []
warnings: list[str] = []
# Population: integer where present. Inhabited = population > 0.
inhabited = [
(r[0], r[1]) for r in conn.execute(
"SELECT system_id, population FROM system_economy "
"WHERE population IS NOT NULL AND population > 0"
).fetchall()
]
econ = {
r[0]: r[1] for r in conn.execute(
"SELECT system_id, economic_specialization FROM system_economy"
).fetchall()
}
cult = {
r[0]: r[1] for r in conn.execute(
"SELECT system_id, cultural_specialization FROM system_economy"
).fetchall()
}
faction = {
r[0]: r[1] for r in conn.execute(
"SELECT system_id, dominant_faction FROM system_factions"
).fetchall()
}
currency = {
r[0]: r[1] for r in conn.execute(
"SELECT system_id, currency_zone FROM star_systems"
).fetchall()
}
# "Named" = has authored GTTR identity (proper_name or gttr_hook).
named = {
r[0] for r in conn.execute(
"SELECT system_id FROM star_systems "
"WHERE (proper_name IS NOT NULL AND proper_name != '') "
" OR (gttr_hook IS NOT NULL AND gttr_hook != '')"
).fetchall()
}
# Tier-1 monopolist corp HQ presence per system (best-effort; tables may be
# sparse pre-#1016). primary_operation/headquarters live on corp_presence.
hq_systems = set()
try:
hq_systems = {
r[0] for r in conn.execute(
"SELECT DISTINCT location_id FROM corp_presence "
"WHERE primary_operation IS NOT NULL"
).fetchall()
}
except sqlite3.OperationalError:
pass
vocab_pu = {k: (v.get("production_ubiquity_projected")) for k, v in vocab.items()}
vocab_commodity = {k: v.get("commodity_id") for k, v in vocab.items()}
# V-SES-02: every inhabited system must resolve a non-null economic value
# (authored here, or via the #1014 fallback once it exists).
for sid, _pop in inhabited:
if not econ.get(sid):
gate_failures.append(
f"V-SES-02: inhabited system '{sid}' has no economic_specialization "
f"(authored or fallback)"
)
# V-FAC-01: every inhabited NAMED system must have an authored faction.
for sid, _pop in inhabited:
if sid in named and not faction.get(sid):
gate_failures.append(
f"V-FAC-01: inhabited named system '{sid}' has no dominant_faction"
)
# --- Soft warnings --------------------------------------------------
# W-SES-01: MonopolySource systems for D-177 human review.
monopoly = [
sid for sid, e in econ.items()
if e and vocab_pu.get(e) == "MonopolySource"
]
if monopoly:
warnings.append(f"W-SES-01: MonopolySource systems [D-177 review]: {sorted(monopoly)}")
# W-SES-02: >25% of authored-economic systems share one value.
from collections import Counter
econ_counts = Counter(e for e in econ.values() if e)
n_authored_econ = sum(econ_counts.values())
if n_authored_econ:
for val, cnt in econ_counts.items():
if cnt > 0.25 * n_authored_econ and cnt > 2:
warnings.append(
f"W-SES-02: '{val}' covers {cnt}/{n_authored_econ} "
f"({100*cnt//n_authored_econ}%) of authored-economic systems"
)
# W-SES-08: estate_farming + large population (probably breadbasket).
for sid, pop in inhabited:
if econ.get(sid) == "estate_farming" and pop and pop > 5_000_000:
warnings.append(
f"W-SES-08: '{sid}' is estate_farming with population {pop} "
f"(probably breadbasket)"
)
# W-FAC-01: compact + tractus_primary currency (D-172 violation).
# W-FAC-04: compact_sympathetic + mark_primary (may be full member).
for sid, f in faction.items():
if not f:
continue
cz = (currency.get(sid) or "").upper()
if f == "compact" and cz == "TRACTUS_PRIMARY":
warnings.append(f"W-FAC-01: '{sid}' compact + TRACTUS_PRIMARY currency (D-172)")
if f == "compact_sympathetic" and cz == "MARK_PRIMARY":
warnings.append(f"W-FAC-04: '{sid}' compact_sympathetic + MARK_PRIMARY (may be full member)")
# W-FAC-02: compact + MonopolySource extraction (Compact self-sufficiency),
# excluding terroir_* / marble_monopoly (lore-sanctioned monopolies).
_exempt = {"marble_monopoly", "terroir_agriculture", "terroir_spirits", "terroir_organics"}
for sid, f in faction.items():
e = econ.get(sid)
if f == "compact" and e and vocab_pu.get(e) == "MonopolySource" and e not in _exempt:
warnings.append(f"W-FAC-02: '{sid}' compact + MonopolySource '{e}' (self-sufficiency doctrine)")
# W-FAC-03: syndic_dominant + no corp HQ presence (ungrounded pin).
for sid, f in faction.items():
if f == "syndic_dominant" and sid not in hq_systems:
warnings.append(f"W-FAC-03: '{sid}' syndic_dominant but no corp HQ in corp_presence (ungrounded)")
# --- Coverage report (always) ---------------------------------------
n_inhabited = len(inhabited)
n_named = len(named)
n_econ = sum(1 for e in econ.values() if e)
n_cult = sum(1 for c in cult.values() if c)
n_fac = sum(1 for f in faction.values() if f)
bulk_dist = Counter()
pu_dist = Counter()
for e in econ.values():
if e and e in vocab:
bulk_dist[vocab[e].get("bulk_class_projected")] += 1
pu_dist[vocab[e].get("production_ubiquity_projected")] += 1
print(" Specialization coverage (D-237):")
print(f" economic: {n_econ} authored ({n_inhabited} inhabited; rest on fallback once #1014 lands)")
print(f" cultural: {n_cult} authored ({n_named} named; rest on corridor default)")
print(f" faction: {n_fac} authored ({n_named} named; rest on derivation)")
print(f" BulkClass: " + " | ".join(f"{k}={v}" for k, v in sorted(bulk_dist.items())))
print(f" ProductionUbiquity: " + " | ".join(f"{k}={v}" for k, v in sorted(pu_dist.items())))
if warnings:
print(f" Specialization warnings ({len(warnings)}):")
for w in warnings:
print(f" - {w}")
if gate_failures:
if strict:
print(f" SPECIALIZATION COMPLETENESS FAILURES ({len(gate_failures)}) [strict]:")
for g in gate_failures:
print(f" - {g}")
raise _ImportAborted()
else:
print(
f" Specialization completeness: {len(gate_failures)} gate item(s) "
f"pending (#1014 fallback / #1016 content pass) — warnings only until "
f"--strict-specialization"
)
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
@@ -1771,6 +2001,11 @@ def main():
parser = argparse.ArgumentParser(description="Import economics data into systems.db")
parser.add_argument("--db", default=str(DB_PATH), help="Path to systems.db")
parser.add_argument("--dry-run", action="store_true", help="Validate without writing")
parser.add_argument(
"--strict-specialization", action="store_true",
help="Treat D-237 completeness gates (V-SES-02, V-FAC-01) as hard errors. "
"Off by default until the #1014 fallback and #1016 content pass land.",
)
args = parser.parse_args()
db_path = Path(args.db)
@@ -1859,7 +2094,9 @@ def main():
# (FK) and before currency zones. UPSERTs onto pre-existing
# system_economy / system_factions rows; unauthored systems stay NULL.
print(" [4b/10] Importing system specialization (D-237)...")
spec = import_system_specialization(conn, args.dry_run)
spec = import_system_specialization(
conn, args.dry_run, strict=args.strict_specialization
)
print(
f" vocab {spec['vocab']} | economic {spec['economic']} | "
f"cultural {spec['cultural']} | faction {spec['faction']}"