From 2f994c1795d0401636736f379ce9098401742786 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Mon, 1 Jun 2026 12:44:03 +0200 Subject: [PATCH] feat(economics): D-237 specialization CI guardrails + coverage report (#1015) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- tooling/economy-db/import_economics.py | 269 +++++++++++++++++++++++-- 1 file changed, 253 insertions(+), 16 deletions(-) diff --git a/tooling/economy-db/import_economics.py b/tooling/economy-db/import_economics.py index 3f4c3c7ee..add4b6626 100755 --- a/tooling/economy-db/import_economics.py +++ b/tooling/economy-db/import_economics.py @@ -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']}"