refactor(db): split import_economics.py; single generator-source registry (T-1067)

import_economics.py 2,620 → 309 lines — a thin orchestrator keeping the
exact CLI, single-transaction/rollback contract, and exit codes. The 16
import steps, MIGRATION_SQL, brands shell-out, validators, and stamp
write now live in tooling/economy-db/economy_import/ (db, migration,
economy, corporations, brands, bodies, atlas, specialization, traits,
validators, stamp, paths, errors). Full type hints throughout.

tooling/generator_sources.py replaces the triplicated source registry
(importer / stamp checker / pr-process watch list — the skill now derives
its list via --list). The registry stamps itself, and economy_import/
modules are globbed fail-closed, so a future module is stamped the moment
it exists — closing the silently-weakened-stamp failure mode.

Rider: connector config helpers centralized in tooling/db/common.py.

Byte-identical behavior proven: full-import table dump diff EMPTY over
107,843 lines / 37 tables (volatile timestamp fields excluded); dry-run
output parity; generated_brands.toml sha unchanged. make test-tooling
PASS; ruff clean.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

.claude/rules/asset-pipeline.md

@@ -35,10 +35,11 @@ invalidate the `import_economics` meta stamp even though the Python file
 itself didn't change.
 
 The full set of source files contributing to the meta stamp SHA (the Python
-importer, the Rust brand binary sources, `tooling/schema_version.py`, and the
+importer entrypoint + its `economy_import` module package, the Rust brand
-authored economics TOMLs) is registered in the `GENERATOR_SOURCES` dict at the
+binary sources, `tooling/schema_version.py`, the authored economics TOMLs, and
-top of `tooling/check-systems-db-stamp` — that dict is the single source of
+the registry itself) is defined once in `tooling/generator_sources.py` (T-1067)
-truth, mirrored by `IMPORT_ECONOMICS_SOURCES` in `import_economics.py`.
+— imported by both the importer's stamp writer and `check-systems-db-stamp`,
+and listed via `python3 tooling/generator_sources.py --list`.
 
 The surviving planet-gen importers (`import_heightmaps`,
 `import_province_boundaries`) are one-time build imports baked into the
@@ -122,9 +123,11 @@ automatically before pushing.
 
 The check script is `tooling/check-systems-db-stamp`.  Run it interactively with
 `make check-systems-db` or `python3 tooling/check-systems-db-stamp --verbose`.  The
-`GENERATOR_SOURCES` dict at the top of that script is the single registry — when
+`GENERATOR_SOURCES` dict in `tooling/generator_sources.py` is the single registry
-you add a new generator or source file, update it there and mirror the change in
+(T-1067) — the check script and the importer's stamp writer both import it, and
-the `/pr-push` skill's source-file watch list.
+the `/pr-push` skill derives its source-file watch list from
+`python3 tooling/generator_sources.py --list`.  When you add a new generator or
+source file, register it there and nowhere else.
 
 ---
 

.claude/skills/pr-process/SKILL.md

@@ -219,24 +219,23 @@ If clean, continue.
 Check whether any file in the **source-file watch list** was modified on this branch
 versus `origin/main`. This list covers generator code AND the data files that feed them.
 
-The generator-source paths below **must stay in sync** with `GENERATOR_SOURCES` in
+The stamped generator sources come from the shared registry
-`tooling/check-systems-db-stamp` (PR #136 review T7) — if you add a new source file
+`tooling/generator_sources.py` (T-1067) — the same module the stamp writer and
-to the stamp, add it here too, and vice versa. Drift between the two lists reintroduces
+`tooling/check-systems-db-stamp` use, so the lists can no longer drift (PR #136
-exactly the silent-stale-DB class of bug this skill exists to prevent.
+review T7). The CLI call below expands to one repo-relative path per line; the
+extra hardcoded entries are non-stamped watch items (retired/one-time planet-gen
+importers, the schema DDL whose SHA is stamped separately, and the data
+directories that feed the generators).
 
 ```bash
 git diff --name-only origin/main...HEAD -- \
-  tooling/economy-db/import_economics.py \
+  $(python3 tooling/generator_sources.py --list) \
   tooling/planet-gen/generate_atlas.py \
   tooling/planet-gen/gemma_naming.py \
   tooling/planet-gen/naming_core.py \
   tooling/planet-gen/import_city_names.py \
   tooling/planet-gen/import_heightmaps.py \
   tooling/planet-gen/import_province_boundaries.py \
-  server/src/bin/generate_brands/main.rs \
-  server/src/bin/generate_brands/names.rs \
-  server/src/bin/shared/surname_corpus.rs \
-  tooling/generate-brands \
   server/data/systems-schema.sql \
   wiki/star-systems/ \
   wiki/economics/ \

.pql/changelog/ticket_history/2026-06.sql

@@ -1851,3 +1851,4 @@ simulation/input.rs: 2,739 lines, imports 9 domains (bookmark, bridge::debug, kn
 Extract per-domain input handlers into their owning modules behind a thin dispatch table; split dialogue into talk-selection / confrontation / response-assembly. Mechanical, no behavior change; gate with the existing determinism.rs byte-identical goldens.
 
 Closed 2026-06-12, commit 257d979ae. input.rs 2,739→858 (thin dispatch; 9 type_complexity allows dissolved via PlayerInputQuery alias); dialogue.rs → dialogue/{mod,selection,response,confrontation} with public paths preserved. Documented seam call: process_walk_away sits with confrontation (shared D-064/D-063 world-response shape). Zero behavior change proven: determinism + golden_suite byte-identical, 1,504 lib tests unchanged, 23+53 tests moved with subjects, scheduling registrations untouched. Full gate green on push.', NULL, '2026-06-12 15:08:41', '2026-06-12 15:08:41', '2026-06-12 15:08:41', NULL, '20710d422729e5868eff93917077f01f', 2) ON CONFLICT(hash) DO NOTHING;
+INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBPTXRX067NBNG76MV84MDQ0', 'status', 'backlog', 'in_progress', NULL, '2026-06-12 17:47:35', '2026-06-12 17:47:35', '2026-06-12 17:47:35', NULL, 'dc7b623e9675f67fb0f3db230525db59', 2) ON CONFLICT(hash) DO NOTHING;

.pql/changelog/tickets/2026-06.sql

@@ -2110,3 +2110,8 @@ simulation/input.rs: 2,739 lines, imports 9 domains (bookmark, bridge::debug, kn
 Extract per-domain input handlers into their owning modules behind a thin dispatch table; split dialogue into talk-selection / confrontation / response-assembly. Mechanical, no behavior change; gate with the existing determinism.rs byte-identical goldens.
 
 Closed 2026-06-12, commit 257d979ae. input.rs 2,739→858 (thin dispatch; 9 type_complexity allows dissolved via PlayerInputQuery alias); dialogue.rs → dialogue/{mod,selection,response,confrontation} with public paths preserved. Documented seam call: process_walk_away sits with confrontation (shared D-064/D-063 world-response shape). Zero behavior change proven: determinism + golden_suite byte-identical, 1,504 lib tests unchanged, 23+53 tests moved with subjects, scheduling registrations untouched. Full gate green on push.', 'done', 'medium', NULL, 'server', NULL, '2026-06-12 10:40:59', '2026-06-12 15:08:41', NULL, '5a2a8338136b821745012637f22608ec', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > tickets.updated_at OR (excluded.updated_at = tickets.updated_at AND excluded.hash > tickets.hash);
+INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBPTXRX067NBNG76MV84MDQ0', 'task', '06FBPPMZNNEV052DBYYY3A897C', 'Split the 2,617-line import_economics.py monolith; single shared generator-source registry', '(description follows in first append)
+
+Filed 2026-06-12 from the fable-ous.md audit (S-19; citations adversarially verified).
+tooling/economy-db/import_economics.py (2,617 lines, 31 top-level defs) owns schema migration, brand regeneration (shells to Rust), 16 import steps, validation, and the meta stamp. Its source-file list is hand-maintained in THREE places guarded only by comments: IMPORT_ECONOMICS_SOURCES (:81-99), GENERATOR_SOURCES (tooling/check-systems-db-stamp:48-63), and the /pr-push watch list (.claude/skills/pr-process/SKILL.md:222-243). Error handling itself is good (single transaction, rollback, distinct exit codes at :2350-2613).
+Fix: extract one shared tooling/generator_sources.py imported by the script + stamp checker and read by /pr-push; then split importers/validators into modules under tooling/economy-db/; finish type hints during the split. Also fold in S-25''s connector dedup: move CONFIG_PATH/load_config/get_base_url into tooling/db/common.py (currently copy-pasted across audio/trellis/image connectors).', 'in_progress', 'medium', NULL, NULL, NULL, '2026-06-12 10:40:59', '2026-06-12 17:47:35', NULL, '46bb09b660cefbc43f3bcfb0dfcb894e', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > tickets.updated_at OR (excluded.updated_at = tickets.updated_at AND excluded.hash > tickets.hash);

tooling/check-systems-db-stamp

@@ -19,7 +19,6 @@ Usage (interactive):
 Decision refs: #855 (generator versioning), #857 (pre-push hook)
 """
 
-import hashlib
 import re
 import sqlite3
 import sys
@@ -32,52 +31,12 @@ REPO_ROOT = Path(__file__).resolve().parent.parent
 
 DB_PATH = REPO_ROOT / "server" / "data" / "systems.db"
 
-# Maps generator_name (as stored in meta.generator_name) to the source
+# The generator-source registry and the SHA helper live in the shared module
-# file(s) whose SHA is stamped.  The SHA is computed as SHA-1 of the
+# tooling/generator_sources.py (T-1067) — the single source of truth, also
-# concatenated bytes of all files in sorted order.
+# imported by the importer's stamp writer and consumed by /pr-process via
-#
+# `python3 tooling/generator_sources.py --list`.
-# import_economics' source set includes the Rust generate_brands binary it now
+sys.path.insert(0, str(Path(__file__).resolve().parent))
-# invokes as a subroutine (#136 review T2/H3).  Keep this list in sync with
+from generator_sources import GENERATOR_SOURCES, file_sha1  # noqa: E402
-# IMPORT_ECONOMICS_SOURCES in tooling/economy-db/import_economics.py.
-# The atlas geometry generator (generate_atlas) was retired in #951 (D-223).
-# import_economics is now the sole regen-db generator that writes systems.db; it
-# owns the atlas index (names-only pool + empty geometry tables).  The surviving
-# planet-gen importers (import_heightmaps, import_province_boundaries) are
-# one-time build imports baked into the committed DB, not part of regen-db, so
-# they are intentionally not stamped here.
-GENERATOR_SOURCES: dict[str, list[Path]] = {
-    "import_economics": [
-        REPO_ROOT / "tooling" / "economy-db" / "import_economics.py",
-        REPO_ROOT / "server" / "src" / "bin" / "generate_brands" / "main.rs",
-        REPO_ROOT / "server" / "src" / "bin" / "generate_brands" / "names.rs",
-        # Shared surname corpus extracted from the two names.rs copies (T-1064).
-        REPO_ROOT / "server" / "src" / "bin" / "shared" / "surname_corpus.rs",
-        REPO_ROOT / "tooling" / "generate-brands",
-        REPO_ROOT / "tooling" / "schema_version.py",
-        # D-237 authored specialization layer data TOMLs (#1013). Keep in sync
-        # with IMPORT_ECONOMICS_SOURCES in tooling/economy-db/import_economics.py.
-        REPO_ROOT / "wiki" / "economics" / "specialization_vocabulary.toml",
-        REPO_ROOT / "wiki" / "economics" / "system_specialization.toml",
-        # D-232 architecture-flavor trait catalog + hero bias (#993).
-        REPO_ROOT / "wiki" / "economics" / "architecture_trait_catalog.toml",
-        REPO_ROOT / "wiki" / "economics" / "architecture_trait_bias.toml",
-    ],
-}
-
-
-def file_sha1(*paths: Path) -> str:
-    """SHA-1 of concatenated file contents (sorted paths).
-
-    Missing files raise FileNotFoundError rather than silently contributing
-    an empty-string hash (H2): a ghost SHA could mask real breakage when
-    stored and current SHAs converge on the empty-bytes digest.
-    """
-    h = hashlib.sha1()
-    for p in sorted(paths):
-        if not p.exists():
-            raise FileNotFoundError(f"generator source not found: {p}")
-        h.update(p.read_bytes())
-    return h.hexdigest()
 
 
 def check(verbose: bool = False) -> int:
@@ -172,7 +131,7 @@ def check(verbose: bool = False) -> int:
             file=sys.stderr,
         )
         print(
-            "  Update GENERATOR_SOURCES in tooling/check-systems-db-stamp to "
+            "  Update GENERATOR_SOURCES in tooling/generator_sources.py to "
             "register them before pushing.",
             file=sys.stderr,
         )

tooling/db/audio_connector.py

@@ -15,7 +15,7 @@ from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
 
-from common import ensure_venv  # noqa: E402
+from common import ensure_venv, get_base_url as _get_base_url  # noqa: E402
 
 ensure_venv()
 
@@ -27,15 +27,9 @@ import time
 import urllib.error
 import urllib.request
 
-CONFIG_PATH = os.path.join(os.path.dirname(__file__), "config.json")
 
-def load_config():
+def get_base_url() -> str:
-    with open(CONFIG_PATH) as f:
+    return _get_base_url("stable_audio_url", "http://tower-of-joy:11500")
-        return json.load(f)
-
-def get_base_url():
-    config = load_config()
-    return config.get("stable_audio_url", "http://tower-of-joy:11500")
 
 def health():
     """Check if the Stable Audio API is reachable."""

tooling/db/common.py

@@ -3,11 +3,14 @@
 
 Provides `ensure_venv`, used by the asset connectors (audio_connector,
 audio_batch, image_connector, trellis_connector) to re-exec into the project
-.venv before their third-party imports. The former settledreach.db connection
+.venv before their third-party imports, plus the connector config helpers
-helpers were removed when the ticket/decision tooling was retired (pql migration
+(`load_config`, `get_base_url`, `get_api_key`) that were formerly copy-pasted
-Phase 6); planning now lives in pql (`.pql/`).
+across the connectors (T-1067 rider, S-25). The former settledreach.db
+connection helpers were removed when the ticket/decision tooling was retired
+(pql migration Phase 6); planning now lives in pql (`.pql/`).
 """
 
+import json
 import os
 import sys
 from pathlib import Path
@@ -18,6 +21,7 @@ from pathlib import Path
 
 SCRIPT_DIR = Path(__file__).resolve().parent
 WORKTREE_ROOT = (SCRIPT_DIR / ".." / "..").resolve()
+CONFIG_PATH = SCRIPT_DIR / "config.json"
 
 
 # ---------------------------------------------------------------------------
@@ -53,3 +57,42 @@ def ensure_venv() -> None:
 
     # Re-exec into the venv Python, preserving all arguments.
     os.execv(str(venv_python), [str(venv_python)] + sys.argv)
+
+
+def load_config() -> dict:
+    """Load the shared endpoint configuration from tooling/db/config.json."""
+    with open(CONFIG_PATH) as f:
+        return json.load(f)
+
+
+def get_base_url(key: str, default: str) -> str:
+    """Resolve a service base URL from config.json, with a fallback default.
+
+    Usage::
+
+        base = get_base_url("trellis_url", "http://tower-of-joy:11510")
+    """
+    return load_config().get(key, default)
+
+
+def get_api_key(env_var: str, config_key: str) -> str:
+    """Get an API key from the environment or config.json.
+
+    Checks the ``env_var`` environment variable first, then ``config_key`` in
+    config.json. Prints a JSON error and exits 1 if neither is set — connector
+    scripts emit machine-readable JSON on all paths.
+    """
+    key = os.environ.get(env_var)
+    if key:
+        return key
+    try:
+        with open(CONFIG_PATH) as f:
+            config = json.load(f)
+            return config.get(config_key, "")
+    except Exception:
+        pass
+    print(json.dumps({
+        "ok": False,
+        "error": f"No {env_var} found in environment or config.json"
+    }, indent=2))
+    sys.exit(1)

tooling/db/image_connector.py

@@ -15,7 +15,7 @@ from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
 
-from common import ensure_venv  # noqa: E402
+from common import ensure_venv, get_api_key as _get_api_key  # noqa: E402
 
 ensure_venv()
 
@@ -25,26 +25,12 @@ import os
 import urllib.error
 import urllib.request
 
-CONFIG_PATH = os.path.join(os.path.dirname(__file__), "config.json")
 DEFAULT_OUTPUT_DIR = os.path.expanduser("~/Pictures/mcp-images")
 
 
-def get_api_key():
+def get_api_key() -> str:
     """Get Gemini API key from env or config."""
-    key = os.environ.get("GEMINI_API_KEY")
+    return _get_api_key("GEMINI_API_KEY", "gemini_api_key")
-    if key:
-        return key
-    try:
-        with open(CONFIG_PATH) as f:
-            config = json.load(f)
-            return config.get("gemini_api_key", "")
-    except Exception:
-        pass
-    print(json.dumps({
-        "ok": False,
-        "error": "No GEMINI_API_KEY found in environment or config.json"
-    }, indent=2))
-    sys.exit(1)
 
 
 def health():

tooling/db/trellis_connector.py

@@ -41,7 +41,7 @@ from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
 
-from common import ensure_venv  # noqa: E402
+from common import ensure_venv, get_base_url as _get_base_url  # noqa: E402
 
 ensure_venv()
 
@@ -53,17 +53,9 @@ import urllib.error
 import urllib.parse
 import urllib.request
 
-CONFIG_PATH = os.path.join(os.path.dirname(__file__), "config.json")
 
-
+def get_base_url() -> str:
-def load_config():
+    return _get_base_url("trellis_url", "http://tower-of-joy:11510")
-    with open(CONFIG_PATH) as f:
-        return json.load(f)
-
-
-def get_base_url():
-    config = load_config()
-    return config.get("trellis_url", "http://tower-of-joy:11510")
 
 
 def health():

tooling/economy-db/economy_import/__init__.py

@@ -0,0 +1,38 @@
+"""
+economy_import — module package behind tooling/economy-db/import_economics.py.
+
+Split out of the former single-file importer (T-1067).  The entrypoint
+``import_economics.py`` remains the CLI (Makefile regen-db, make test-tooling)
+and orchestrates the single-transaction import; the steps live here:
+
+  paths.py           source-file path constants (re-exports stamped paths
+                     from tooling/generator_sources.py)
+  errors.py          ImportAborted control-flow exception
+  db.py              connection + FK-safe table-clear helpers
+  migration.py       MIGRATION_SQL / COLUMN_MIGRATIONS schema migration
+  economy.py         gate links, commodities, chains, currency zones,
+                     gate energy, system_fiscal
+  corporations.py    wiki corp parsing, D-182 sync, corp_presence
+  brands.py          generate_brands shell-out + D-189 brand layer
+  bodies.py          body_radius_km / axial_tilt_deg backfills
+  atlas.py           atlas index schema + city-name pool (D-223, D-207)
+  specialization.py  D-237 authored specialization layer
+  traits.py          D-232 architecture-flavor trait templates + hero bias
+  validators.py      structural + D-175 coverage validation
+  stamp.py           meta-table generator stamp (#855, #856)
+
+Every module here is part of the import_economics meta-stamp source set —
+tooling/generator_sources.py globs this directory, so adding a module
+automatically extends staleness detection.
+"""
+
+import sys
+from pathlib import Path
+
+# Bootstrap: make tooling/ importable (generator_sources, schema_version)
+# before any submodule needs them.  tooling/economy-db is not a package
+# (hyphenated dir), so the entrypoint puts it on sys.path and this package
+# adds tooling/ itself.
+_TOOLING_DIR = Path(__file__).resolve().parents[2]
+if str(_TOOLING_DIR) not in sys.path:
+    sys.path.insert(0, str(_TOOLING_DIR))

tooling/economy-db/economy_import/atlas.py

@@ -0,0 +1,250 @@
+"""Atlas index ownership (D-223, #951): canonical atlas_* schema, the
+names-only city pool, and the corp-HQ city cross-reference (D-207)."""
+
+import glob
+import json
+import re
+import sqlite3
+
+from .paths import CORPORATIONS_DIR, SCHEMA_SQL, WIKI_STAR_SYSTEMS
+
+# Atlas geometry index tables (D-191). These hold computed positions — city
+# centres, road/river/rail polylines, ocean/mountain extents. Under D-223 the
+# Python atlas geometry generator was retired (#951); the deterministic
+# server-side cascade (Phase 4) is the sole producer of this geometry. We keep
+# the tables (the Atlas viewer #960 and the cascade read them) but empty them on
+# every regen so the committed DB carries no stale prototype geometry — the
+# empty tables are the gap the server cascade fills.
+_ATLAS_GEOMETRY_TABLES: tuple[str, ...] = (
+    "atlas_cities",
+    "atlas_roads",
+    "atlas_railroads",
+    "atlas_pois",
+    "atlas_rivers",
+    "atlas_oceans",
+    "atlas_mountain_ranges",
+    "atlas_body_grids",
+)
+
+_ATLAS_INDEX_BEGIN_MARKER = "-- BEGIN ATLAS INDEX"
+_ATLAS_INDEX_END_MARKER = "-- END ATLAS INDEX"
+
+
+def ensure_atlas_index_schema(conn: sqlite3.Connection, dry_run: bool) -> None:
+    """Apply the canonical atlas_* DDL and empty the geometry tables (D-223, #951).
+
+    systems-schema.sql is the single source of truth for the atlas index tables
+    (the BEGIN/END ATLAS INDEX block). The retired generate_atlas.py used to
+    apply this block; import_economics now owns it, since it is the only
+    regen-db generator that touches systems.db's atlas tables. The block is all
+    CREATE ... IF NOT EXISTS, so applying it on the committed DB is a no-op and
+    on a fresh DB it creates the geometry tables.
+
+    After ensuring the schema, the geometry tables are cleared: their geometry
+    now comes from the server cascade, not from authored markers (D-223).
+    """
+    text = SCHEMA_SQL.read_text()
+    try:
+        start = text.index(_ATLAS_INDEX_BEGIN_MARKER)
+        end = text.index(_ATLAS_INDEX_END_MARKER, start)
+    except ValueError as e:
+        raise RuntimeError(
+            f"systems-schema.sql is missing the {_ATLAS_INDEX_BEGIN_MARKER}/"
+            f"{_ATLAS_INDEX_END_MARKER} block — has the schema been restructured?"
+        ) from e
+    conn.executescript(text[start:end])
+    if not dry_run:
+        for table in _ATLAS_GEOMETRY_TABLES:
+            conn.execute(f"DELETE FROM {table}")
+
+
+def populate_atlas_city_names(conn: sqlite3.Connection, dry_run: bool) -> int:
+    """Populate atlas_city_names from the names-only markers.json pool (D-223, #951).
+
+    Scans wiki/star-systems/*/bodies/*/markers.json for the flavoured city
+    name pool at `names.cities` and inserts one atlas_city_names row per name:
+      - body_id       : directory name (e.g. GJ0e)
+      - name          : pooled city name
+      - kind          : 'city' — capital is chosen at placement (#955)
+      - economic_role : inherited from bodies.economic_role; fallback 'mixed'
+      - population    : 0 — assigned by the server cascade at placement (#955)
+      - corp_id       : NULL — populated by populate_atlas_city_names_corps (#909)
+      - reserved      : 0
+
+    markers.json is a names-only flavoured pool (D-223): it carries no geometry
+    or population. The deterministic server cascade attaches these names to
+    computed settlements and assigns population/kind/position at placement time;
+    this importer just loads the pool.
+
+    Deterministic rebuild: clears atlas_city_names first (the FK cascade clears
+    atlas_city_positions), so re-runs are idempotent — there is no UNIQUE on
+    (body_id, name), so without the clear a re-run would accumulate duplicates.
+    Skips body directories not found in the bodies table (missing FK).
+    """
+    # Build body_id -> economic_role map
+    body_roles: dict[str, str] = {}
+    for body_id, role in conn.execute(
+        "SELECT body_id, economic_role FROM bodies"
+    ).fetchall():
+        body_roles[body_id] = role or "mixed"
+
+    valid_body_ids: set[str] = set(body_roles.keys())
+
+    # Sol (system 'GJ 0') is permanently exempt from the normal generators
+    # (D-223, #951): its bodies use real Earth/Mars/Luna geography via
+    # sol_import.py and keep geometry-bearing markers.json as preserved config.
+    # Sol names come from its own (future) scripted integration, not the names
+    # pool — skip Sol bodies here regardless of their markers format.
+    sol_body_ids: set[str] = {
+        r[0] for r in conn.execute(
+            "SELECT body_id FROM bodies WHERE system_id = 'GJ 0'"
+        ).fetchall()
+    }
+
+    rows: list[tuple] = []
+    skipped_bodies: list[str] = []
+
+    pattern = str(WIKI_STAR_SYSTEMS / "*" / "bodies" / "*" / "markers.json")
+    for markers_path in sorted(glob.glob(pattern)):
+        body_id = markers_path.split("/bodies/")[1].split("/")[0]
+        if body_id not in valid_body_ids or body_id in sol_body_ids:
+            if body_id not in valid_body_ids:
+                skipped_bodies.append(body_id)
+            continue
+
+        with open(markers_path) as fh:
+            data = json.load(fh)
+
+        names_pool = (data.get("names") or {}).get("cities") or []
+        economic_role = body_roles[body_id]
+        for raw_name in names_pool:
+            name = (raw_name or "").strip()
+            if not name:
+                continue
+            # kind defaults to 'city'; population 0 until placement (#955).
+            rows.append((body_id, name, "city", economic_role, 0))
+
+    if skipped_bodies:
+        unique = sorted(set(skipped_bodies))
+        print(f"        warning: {len(unique)} body dirs not in DB — skipped: {unique[:5]}")
+
+    if not dry_run:
+        conn.execute("DELETE FROM atlas_city_names")
+        if rows:
+            conn.executemany(
+                """INSERT INTO atlas_city_names
+                   (body_id, name, kind, economic_role, population)
+                   VALUES (?, ?, ?, ?, ?)""",
+                rows,
+            )
+
+    return len(rows)
+
+
+def populate_atlas_city_names_corps(conn: sqlite3.Connection, dry_run: bool) -> tuple[int, int]:
+    """Cross-reference corp HQ city names into atlas_city_names (D-207, #909).
+
+    For each corporation with a parseable headquarters field ("City (SYSTEM_ID)"):
+      - If atlas_city_names already has a row with matching name on a body in that
+        system: UPDATE the row to set corp_id.
+      - Otherwise: INSERT a reserved row (reserved=1) so the name is protected.
+        Attaches to the most-populated body in the system (fallback: any body).
+
+    Returns (n_updated, n_inserted).
+    """
+    # Build system_id -> sorted bodies (by population desc, then body_id)
+    sys_bodies: dict[str, list[tuple[int, str, str]]] = {}
+    for body_id, sys_id, pop, role in conn.execute(
+        "SELECT body_id, system_id, COALESCE(population, 0), COALESCE(economic_role, 'mixed') FROM bodies"
+    ).fetchall():
+        sys_bodies.setdefault(sys_id, []).append((pop, body_id, role))
+    for v in sys_bodies.values():
+        v.sort(key=lambda x: (-x[0], x[1]))
+
+    # Build (body_id, name_lower) -> id index for existing atlas_city_names rows
+    existing: dict[tuple[str, str], int] = {}
+    body_to_sys: dict[str, str] = {
+        r[0]: r[1]
+        for r in conn.execute("SELECT body_id, system_id FROM bodies").fetchall()
+    }
+    for row_id, body_id, name in conn.execute(
+        "SELECT id, body_id, name FROM atlas_city_names"
+    ).fetchall():
+        existing[(body_id, name.lower())] = row_id
+
+    # Build system_id -> set of body_ids for quick lookup
+    sys_body_ids: dict[str, set[str]] = {}
+    for body_id, sys_id in body_to_sys.items():
+        sys_body_ids.setdefault(sys_id, set()).add(body_id)
+
+    updated: list[tuple[str, int]] = []   # (corp_id, atlas_row_id)
+    inserted: list[tuple] = []            # insert rows
+
+    for corp_id, headquarters_system in conn.execute(
+        "SELECT corp_id, headquarters_system FROM corporations WHERE headquarters_system IS NOT NULL"
+    ).fetchall():
+        # Retrieve original headquarters string from wiki to get city name
+        md_file = CORPORATIONS_DIR / f"{corp_id}.md"
+        if not md_file.exists():
+            continue
+        hq_raw = ""
+        with open(md_file) as f:
+            in_fm = False
+            for line in f:
+                if line.strip() == "---":
+                    if not in_fm:
+                        in_fm = True
+                        continue
+                    else:
+                        break
+                if in_fm and line.startswith("headquarters:"):
+                    hq_raw = line.split(":", 1)[1].strip().strip('"')
+                    break
+        if not hq_raw:
+            continue
+        m = re.search(r"\(([^)]+)\)", hq_raw)
+        city_name = hq_raw[: m.start()].strip() if m else hq_raw.strip()
+        if not city_name:
+            continue
+
+        # Try to find a matching atlas_city_names row in the same system
+        body_ids_in_sys = sys_body_ids.get(headquarters_system, set())
+        match_id: int | None = None
+        # sorted() for determinism: on a name collision across bodies in the
+        # same system, set iteration order is not stable (D-010 #4).
+        for body_id in sorted(body_ids_in_sys):
+            key = (body_id, city_name.lower())
+            if key in existing:
+                match_id = existing[key]
+                break
+
+        if match_id is not None:
+            updated.append((corp_id, match_id))
+        else:
+            # Sol (system 'GJ 0') is exempt from the normal generators (D-223,
+            # #951) — do not synthesize a reserved corp-HQ row on a Sol body;
+            # Sol's atlas data comes from its own scripted integration.
+            if headquarters_system == "GJ 0":
+                continue
+            # Insert a reserved row on the most-populated body in the system
+            candidates = sys_bodies.get(headquarters_system, [])
+            if not candidates:
+                continue
+            _, target_body_id, body_role = candidates[0]
+            inserted.append((target_body_id, city_name, "city", body_role, 0, corp_id, 1))
+
+    if not dry_run:
+        for corp_id, row_id in updated:
+            conn.execute(
+                "UPDATE atlas_city_names SET corp_id = ? WHERE id = ?",
+                (corp_id, row_id),
+            )
+        if inserted:
+            conn.executemany(
+                """INSERT OR IGNORE INTO atlas_city_names
+                   (body_id, name, kind, economic_role, population, corp_id, reserved)
+                   VALUES (?, ?, ?, ?, ?, ?, ?)""",
+                inserted,
+            )
+
+    return len(updated), len(inserted)

tooling/economy-db/economy_import/bodies.py

@@ -0,0 +1,204 @@
+"""Body-attribute backfills: body_radius_km (D-204) and axial_tilt_deg (T-1024)."""
+
+import hashlib
+import sqlite3
+from pathlib import Path
+
+from .paths import WIKI_STAR_SYSTEMS
+
+
+def populate_body_radius_km(conn: sqlite3.Connection, dry_run: bool) -> int:
+    """Populate body_radius_km column from planet_class fallback (D-204, #910).
+
+    Applies the fallback lookup table to rows where body_radius_km IS NULL.
+    Does not overwrite rows where body_radius_km is already set (authoritative data).
+
+    Fallback values (km):
+        super_earth   -> 8000
+        earth_like    -> 6371
+        earth         -> 6371  (alternate spelling)
+        sub_earth     -> 4500
+        ocean_world   -> 6500
+        arid          -> 5800
+        frozen        -> 4500
+        ice_world     -> 3000
+        barren        -> 4500
+        volcanic      -> 5500
+        gas_giant     -> 0 (no settlements, skip)
+        moon          -> 1737
+        other/unknown -> 6371 (Earth default)
+    """
+    PLANET_CLASS_RADIUS = {
+        "super_earth": 8000.0,
+        "earth_like":  6371.0,
+        "earth":       6371.0,
+        "sub_earth":   3500.0,
+        "ocean_world": 6500.0,
+        "arid":        5800.0,
+        "frozen":      3500.0,
+        "ice_world":   3000.0,
+        "barren":      3500.0,
+        "volcanic":    5500.0,
+        "temperate":   6371.0,
+        "moon":        1737.0,
+    }
+    DEFAULT_RADIUS = 6371.0
+    SKIP_RADIUS_TYPES = {"oort_cloud", "asteroid_belt"}
+
+    GAS_GIANT_RADIUS = {
+        "gas_giant": 50000.0,
+        "ice_giant": 25000.0,
+    }
+    GAS_GIANT_DEFAULT = 45000.0
+    GAS_GIANT_SCATTER = 0.20  # ±20%
+
+    rows = conn.execute(
+        "SELECT body_id, planet_class, body_type, mass_class FROM bodies WHERE body_radius_km IS NULL"
+    ).fetchall()
+
+    SCATTER_FRACTION = 0.15  # ±15% for rocky bodies
+
+    updates: list[tuple[float, str]] = []
+    for body_id, planet_class, body_type, mass_class in rows:
+        if body_type in SKIP_RADIUS_TYPES:
+            continue
+
+        h = int(hashlib.sha256(body_id.encode()).hexdigest()[:8], 16)
+        scatter_val = (h / 0xFFFFFFFF) * 2.0 - 1.0  # [-1.0, 1.0]
+
+        if body_type == "gas_giant":
+            mc = (mass_class or "").lower()
+            base_radius = GAS_GIANT_RADIUS.get(mc, GAS_GIANT_DEFAULT)
+            radius = round(base_radius * (1.0 + scatter_val * GAS_GIANT_SCATTER), 1)
+        elif body_type == "moon":
+            base_radius = 1400.0
+            scatter_frac = 0.86     # ±86% → ~196–2604 km
+            radius = round(base_radius * (1.0 + scatter_val * scatter_frac), 1)
+        else:
+            base_radius = PLANET_CLASS_RADIUS.get(
+                (planet_class or "").lower(), DEFAULT_RADIUS
+            )
+            radius = round(base_radius * (1.0 + scatter_val * SCATTER_FRACTION), 1)
+        updates.append((radius, body_id))
+
+    if not dry_run and updates:
+        conn.executemany(
+            "UPDATE bodies SET body_radius_km = ? WHERE body_id = ?", updates
+        )
+
+    return len(updates)
+
+
+def populate_axial_tilt_deg(conn: sqlite3.Connection, dry_run: bool) -> int:
+    """Populate bodies.axial_tilt_deg from planet-gen body-def frontmatter (T-1024).
+
+    Reads wiki/star-systems/*/bodies/*/index.md YAML frontmatter and extracts
+    ``orbit.axial_tilt_deg``. Only updates rows where axial_tilt_deg IS NULL
+    (preserves any future authoritative column writes).
+
+    Source: body_definition_parser.py writes ``orbit.axial_tilt_deg`` into each
+    body's index.md during the planet-gen batch run. This function mirrors
+    ``populate_body_radius_km`` in structure.
+    """
+    import yaml  # stdlib-compatible subset via PyYAML if available, else manual parse
+
+    def _parse_frontmatter_yaml(text: str) -> dict:
+        """Extract YAML frontmatter block from a markdown file."""
+        lines = text.split("\n")
+        if not lines or lines[0].strip() != "---":
+            return {}
+        end_idx = None
+        for i, line in enumerate(lines[1:], 1):
+            if line.strip() == "---":
+                end_idx = i
+                break
+        if end_idx is None:
+            return {}
+        fm_text = "\n".join(lines[1:end_idx])
+        try:
+            result = yaml.safe_load(fm_text)
+            return result if isinstance(result, dict) else {}
+        except Exception:
+            return {}
+
+    # Check if yaml is available; if not, use manual extraction.
+    try:
+        import yaml as _yaml_check  # noqa: F401
+        has_yaml = True
+    except ImportError:
+        has_yaml = False
+
+    if not has_yaml:
+        # Fallback: manual extraction of axial_tilt_deg from YAML frontmatter.
+        # Scans for "  axial_tilt_deg: <value>" under an "orbit:" block.
+        def _parse_frontmatter_manual(text: str) -> dict:
+            lines = text.split("\n")
+            if not lines or lines[0].strip() != "---":
+                return {}
+            in_orbit = False
+            result: dict = {}
+            for line in lines[1:]:
+                if line.strip() == "---":
+                    break
+                stripped = line.strip()
+                if stripped == "orbit:":
+                    in_orbit = True
+                    continue
+                if in_orbit:
+                    # Detect leaving the orbit block (non-indented key).
+                    if line and not line.startswith(" ") and not line.startswith("\t"):
+                        in_orbit = False
+                    elif stripped.startswith("axial_tilt_deg:"):
+                        _, _, val = stripped.partition(":")
+                        try:
+                            result["axial_tilt_deg"] = float(val.strip())
+                        except ValueError:
+                            pass
+            return result
+
+        def _parse_frontmatter_yaml(text: str) -> dict:  # type: ignore[misc]
+            return _parse_frontmatter_manual(text)
+
+    # Find all body index.md files.
+    body_dir_pattern = WIKI_STAR_SYSTEMS / "*" / "bodies" / "*" / "index.md"
+    import glob as _glob
+    body_files = sorted(_glob.glob(str(body_dir_pattern)))
+
+    # Build body_id → axial_tilt_deg mapping from frontmatter.
+    tilt_map: dict[str, float] = {}
+    for fpath in body_files:
+        try:
+            text = Path(fpath).read_text(encoding="utf-8")
+        except OSError:
+            continue
+        fm = _parse_frontmatter_yaml(text)
+        if not isinstance(fm, dict):
+            continue
+        body_id = fm.get("id")
+        orbit = fm.get("orbit", {})
+        if isinstance(orbit, dict):
+            tilt = orbit.get("axial_tilt_deg")
+        else:
+            tilt = None
+        if body_id and tilt is not None:
+            try:
+                tilt_map[str(body_id)] = float(tilt)
+            except (TypeError, ValueError):
+                pass
+
+    # Get all bodies where axial_tilt_deg IS NULL and body_id is in tilt_map.
+    rows = conn.execute(
+        "SELECT body_id FROM bodies WHERE axial_tilt_deg IS NULL"
+    ).fetchall()
+
+    updates: list[tuple[float, str]] = []
+    for (body_id,) in rows:
+        if body_id in tilt_map:
+            updates.append((tilt_map[body_id], body_id))
+
+    if not dry_run and updates:
+        conn.executemany(
+            "UPDATE bodies SET axial_tilt_deg = ? WHERE body_id = ?", updates
+        )
+
+    return len(updates)

tooling/economy-db/economy_import/brands.py

@@ -0,0 +1,249 @@
+"""Brand layer (D-189, #827): generate_brands shell-out, TOML import, validation."""
+
+import sqlite3
+import subprocess
+import sys
+import tomllib
+from pathlib import Path
+
+from .errors import ImportAborted
+from .paths import BRANDS_TOML, GENERATE_BRANDS_WRAPPER, GENERATED_BRANDS_TOML, REPO_ROOT
+
+VALID_BRAND_CATEGORIES: set[str] = {
+    "terroir", "heritage_craft", "tech_premium", "cultural",
+    "service_premium", "commodity_branded", "design_heritage", "platform_catalogue",
+}
+VALID_VALUE_TRAJECTORIES: set[str] = {"appreciating", "depreciating", "timeless"}
+VALID_SCARCITY_CLASSES: set[str] = {"capped", "constrained", "scalable", "unlimited"}
+VALID_BRAND_TIERS: set[str] = {"halo", "volume"}
+VALID_CURRENCY_DENOMINATIONS: set[str] = {"tractus", "mark", "mixed", "sol_adjacent"}
+VALID_PRICE_TIERS: set[str] = {"mass", "premium", "luxury", "flagship", "institutional"}
+
+
+def regenerate_brands() -> None:
+    """Run the Rust generate_brands binary to refresh generated_brands.toml.
+
+    Invoked as the first step of import_economics' main flow so the TOML on disk
+    always matches the current Rust source before the Python import reads it.
+    This replaces the former split (tooling/generate-brands run separately by
+    make regen-db) with a single, coherent brand pipeline owned by one stamp.
+
+    The wrapper script builds the binary on demand and runs it with the default
+    canonical seed=1; callers that need non-canonical seeds must still invoke
+    the wrapper directly (experimentation only — committed output must be seed=1).
+    """
+    if not GENERATE_BRANDS_WRAPPER.exists():
+        raise FileNotFoundError(
+            f"generate_brands wrapper not found at {GENERATE_BRANDS_WRAPPER}"
+        )
+    print("  [pre/10] Running generate_brands (Rust) to refresh generated_brands.toml...")
+    result = subprocess.run(
+        [str(GENERATE_BRANDS_WRAPPER)],
+        cwd=str(REPO_ROOT),
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode != 0:
+        print(result.stdout, file=sys.stderr)
+        print(result.stderr, file=sys.stderr)
+        raise ImportAborted()
+    # Print the Rust binary's own summary lines (brands generated, coverage).
+    # Indent so they fold under the pre-step heading.
+    for line in result.stdout.splitlines():
+        if line.strip():
+            print(f"        {line}")
+
+
+def _load_brand_file(path: Path) -> tuple[list, list]:
+    """Load brand_products and brand_inputs from a TOML file. Returns empty lists if missing."""
+    if not path.exists():
+        return [], []
+    with open(path, "rb") as f:
+        data = tomllib.load(f)
+    return data.get("brand_products", []), data.get("brand_inputs", [])
+
+
+def import_brands(
+    conn: sqlite3.Connection, dry_run: bool
+) -> tuple[int, int]:
+    """Import brand_products and brand_inputs from brands.toml and generated_brands.toml.
+
+    Hand-authored brands (brands.toml) are imported first; generated brands
+    (generated_brands.toml, produced by `tooling/generate-brands`) are merged in.
+    Returns (n_products, n_inputs).
+    """
+    if not BRANDS_TOML.exists():
+        print("        warning: brands.toml not found — brand layer skipped")
+        return 0, 0
+
+    products_authored, inputs_authored = _load_brand_file(BRANDS_TOML)
+    products_generated, inputs_generated = _load_brand_file(GENERATED_BRANDS_TOML)
+
+    if products_generated:
+        print(f"        merging {len(products_generated)} generated brand_products from generated_brands.toml")
+
+    products = products_authored + products_generated
+    inputs = inputs_authored + inputs_generated
+
+    product_rows: list[tuple] = []
+    for p in products:
+        product_rows.append((
+            p["brand_product_id"],
+            p["corp_id"],
+            p["product_name"],
+            p["brand_category"],
+            p["value_trajectory"],
+            p["scarcity_class"],
+            p.get("product_subcategory"),
+            p.get("base_premium_multiplier", 1.0),
+            p.get("premium_floor", 0.0),
+            p.get("origin_system"),
+            int(p.get("terroir_locked", False)),
+            p.get("currency_denomination", "tractus"),
+            int(p.get("shadow_viable", False)),
+            p["brand_tier"],
+            p.get("halo_brand_id"),
+            p.get("price_tier"),
+        ))
+
+    input_rows: list[tuple] = []
+    for inp in inputs:
+        input_rows.append((
+            inp["brand_product_id"],
+            inp["commodity_id"],
+            inp["quantity"],
+        ))
+
+    if not dry_run:
+        conn.executemany(
+            """INSERT OR REPLACE INTO brand_products (
+                brand_product_id, corp_id, product_name, brand_category,
+                value_trajectory, scarcity_class, product_subcategory,
+                base_premium_multiplier, premium_floor, origin_system,
+                terroir_locked, currency_denomination, shadow_viable,
+                brand_tier, halo_brand_id, price_tier
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+            product_rows,
+        )
+        conn.executemany(
+            """INSERT OR REPLACE INTO brand_inputs
+               (brand_product_id, commodity_id, quantity) VALUES (?, ?, ?)""",
+            input_rows,
+        )
+
+    return len(product_rows), len(input_rows)
+
+
+def validate_brands(conn: sqlite3.Connection) -> list[str]:
+    """Brand layer structural validation rules V-B01 through V-B06.
+
+    V-B01: Every brand_products row has a valid corp_id (FK to corporations).
+    V-B02: Every brand_inputs row has valid brand_product_id and commodity_id FKs.
+    V-B03: Every halo brand has at least one brand_inputs entry (demand stub must consume).
+    V-B04: Every volume tier must reference an existing halo brand_product_id.
+    V-B05: No brand_product_id is used as halo_brand_id by a non-volume-tier product.
+    V-B06: Every enum column (brand_category, value_trajectory, scarcity_class,
+           brand_tier, currency_denomination) is a member of its VALID_* set.
+    """
+    errors: list[str] = []
+
+    # V-B01: brand_products → corporations FK
+    orphan_corps = conn.execute("""
+        SELECT bp.brand_product_id, bp.corp_id
+        FROM brand_products bp
+        LEFT JOIN corporations c ON bp.corp_id = c.corp_id
+        WHERE c.corp_id IS NULL
+    """).fetchall()
+    for pid, corp_id in orphan_corps:
+        errors.append(
+            f"V-B01: brand_product '{pid}' references unknown corp_id '{corp_id}'"
+        )
+
+    # V-B02: brand_inputs → brand_products and brand_inputs → commodities FKs
+    orphan_inputs_bp = conn.execute("""
+        SELECT bi.brand_product_id, bi.commodity_id
+        FROM brand_inputs bi
+        LEFT JOIN brand_products bp ON bi.brand_product_id = bp.brand_product_id
+        WHERE bp.brand_product_id IS NULL
+    """).fetchall()
+    for pid, cid in orphan_inputs_bp:
+        errors.append(
+            f"V-B02: brand_inputs row ({pid}, {cid}) references unknown brand_product_id"
+        )
+
+    orphan_inputs_comm = conn.execute("""
+        SELECT bi.brand_product_id, bi.commodity_id
+        FROM brand_inputs bi
+        LEFT JOIN commodities c ON bi.commodity_id = c.commodity_id
+        WHERE c.commodity_id IS NULL
+    """).fetchall()
+    for pid, cid in orphan_inputs_comm:
+        errors.append(
+            f"V-B02: brand_inputs row ({pid}, {cid}) references unknown commodity_id '{cid}'"
+        )
+
+    # V-B03: every halo brand has at least one brand_inputs entry
+    halo_no_inputs = conn.execute("""
+        SELECT bp.brand_product_id
+        FROM brand_products bp
+        WHERE bp.brand_tier = 'halo'
+          AND bp.brand_product_id NOT IN (SELECT brand_product_id FROM brand_inputs)
+    """).fetchall()
+    for (pid,) in halo_no_inputs:
+        errors.append(
+            f"V-B03: halo brand '{pid}' has no brand_inputs entries "
+            f"(must consume at least one commodity as a demand node)"
+        )
+
+    # V-B04: volume tiers reference valid halo_brand_id
+    volume_bad_halo = conn.execute("""
+        SELECT bp.brand_product_id, bp.halo_brand_id
+        FROM brand_products bp
+        WHERE bp.brand_tier = 'volume'
+          AND (bp.halo_brand_id IS NULL
+               OR bp.halo_brand_id NOT IN (SELECT brand_product_id FROM brand_products))
+    """).fetchall()
+    for pid, halo_id in volume_bad_halo:
+        errors.append(
+            f"V-B04: volume brand '{pid}' has invalid halo_brand_id '{halo_id}'"
+        )
+
+    # V-B05: halo_brand_id must only point to halo-tier products
+    halo_points_to_non_halo = conn.execute("""
+        SELECT child.brand_product_id, child.halo_brand_id, parent.brand_tier
+        FROM brand_products child
+        JOIN brand_products parent ON child.halo_brand_id = parent.brand_product_id
+        WHERE child.brand_tier = 'volume'
+          AND parent.brand_tier != 'halo'
+    """).fetchall()
+    for child_id, halo_id, parent_tier in halo_points_to_non_halo:
+        errors.append(
+            f"V-B05: volume brand '{child_id}' points to '{halo_id}' "
+            f"which has brand_tier='{parent_tier}', not 'halo'"
+        )
+
+    # V-B06: every enum column is in its VALID_* set. The SQL columns are
+    # plain TEXT without CHECK constraints, so a typo like `terrior` would
+    # otherwise silently import.
+    enum_checks: list[tuple[str, set[str]]] = [
+        ("brand_category", VALID_BRAND_CATEGORIES),
+        ("value_trajectory", VALID_VALUE_TRAJECTORIES),
+        ("scarcity_class", VALID_SCARCITY_CLASSES),
+        ("brand_tier", VALID_BRAND_TIERS),
+        ("currency_denomination", VALID_CURRENCY_DENOMINATIONS),
+        ("price_tier", VALID_PRICE_TIERS),
+    ]
+    for column, valid_set in enum_checks:
+        bad = conn.execute(
+            f"SELECT brand_product_id, {column} FROM brand_products"
+        ).fetchall()
+        for pid, value in bad:
+            if value is None:
+                continue  # nullable columns (e.g. price_tier) may be unset
+            if value not in valid_set:
+                errors.append(
+                    f"V-B06: brand_product '{pid}' has {column}='{value}' — "
+                    f"must be one of {sorted(valid_set)}"
+                )
+
+    return errors

tooling/economy-db/economy_import/corporations.py

@@ -0,0 +1,238 @@
+"""Corporation wiki parsing, D-182 sync, and corp_presence population."""
+
+import re
+import sqlite3
+from pathlib import Path
+
+from .paths import CORPORATIONS_DIR
+
+# ---------------------------------------------------------------------------
+# Corporation wiki parsing
+# ---------------------------------------------------------------------------
+
+
+def _parse_corp_frontmatter(path: Path) -> dict[str, str | list[str]] | None:
+    """Parse YAML frontmatter from a wiki corporation markdown file."""
+    text = path.read_text()
+    lines = text.split("\n")
+    if not lines or lines[0].strip() != "---":
+        return None
+    end_idx = None
+    for i, line in enumerate(lines[1:], 1):
+        if line.strip() == "---":
+            end_idx = i
+            break
+    if end_idx is None:
+        return None
+    fm: dict[str, str | list[str]] = {}
+    for line in lines[1:end_idx]:
+        if ":" not in line:
+            continue
+        key, _, val = line.partition(":")
+        key = key.strip()
+        val = val.strip()
+        if val.startswith("[") and val.endswith("]"):
+            items = [x.strip().strip('"').strip("'") for x in val[1:-1].split(",")]
+            fm[key] = [item for item in items if item]
+        else:
+            fm[key] = val.strip('"').strip("'")
+    return fm
+
+
+def load_wiki_corps() -> list[dict]:
+    """Load all wiki corporation files. Returns list of parsed corp records."""
+    corps: list[dict] = []
+    for md_file in sorted(CORPORATIONS_DIR.glob("*.md")):
+        if md_file.name == "index.md":
+            continue
+        fm = _parse_corp_frontmatter(md_file)
+        if not fm or not fm.get("slug") or not fm.get("title"):
+            continue
+        hq = fm.get("headquarters", "")
+        m = re.search(r"\(([^)]+)\)", hq)
+        system_id = m.group(1) if m else None
+        corps.append({
+            "corp_id": fm["slug"],
+            "proper_name": fm["title"],
+            "system_id": system_id,
+            "tags": fm.get("tags", []),
+            "scope": fm.get("scope", ""),
+        })
+    return corps
+
+
+# ---------------------------------------------------------------------------
+# Corporation sync (D-182: wiki is source of truth)
+# ---------------------------------------------------------------------------
+
+
+def sync_corporations(
+    conn: sqlite3.Connection, wiki_corps: list[dict], dry_run: bool
+) -> list[str]:
+    """Sync wiki corps to DB. Hard error on proper_name divergence (D-182).
+
+    Returns list of error strings. Inserts corps that exist in wiki but not DB.
+    Corps that exist only in DB (legacy records) are left untouched.
+    headquarters_system is only written if the system_id exists in star_systems
+    (to avoid FK violations when atlas hasn't yet registered the system).
+    """
+    errors: list[str] = []
+    existing = {
+        r[0]: r[1]
+        for r in conn.execute("SELECT corp_id, proper_name FROM corporations").fetchall()
+    }
+    valid_systems = {
+        r[0] for r in conn.execute("SELECT system_id FROM star_systems").fetchall()
+    }
+
+    to_insert: list[tuple] = []
+    for corp in wiki_corps:
+        corp_id = corp["corp_id"]
+        proper_name = corp["proper_name"]
+        if corp_id in existing:
+            if existing[corp_id] != proper_name:
+                errors.append(
+                    f"name divergence: corp_id='{corp_id}' "
+                    f"wiki='{proper_name}' db='{existing[corp_id]}'"
+                )
+        else:
+            system_id = corp.get("system_id")
+            hq_system = system_id if system_id and system_id in valid_systems else None
+            if system_id and system_id not in valid_systems:
+                print(f"        warning: {corp_id} HQ system '{system_id}' not in DB, "
+                      f"headquarters_system set to NULL")
+            to_insert.append((
+                corp_id,
+                proper_name,
+                "corporation",
+                corp.get("scope") or None,
+                hq_system,
+            ))
+
+    if not dry_run and not errors:
+        conn.executemany(
+            """INSERT OR IGNORE INTO corporations
+               (corp_id, proper_name, corp_type, scope, headquarters_system)
+               VALUES (?, ?, ?, ?, ?)""",
+            to_insert,
+        )
+
+    return errors
+
+
+# ---------------------------------------------------------------------------
+# Corp presence population
+# ---------------------------------------------------------------------------
+
+
+def _resolve_hq_location(
+    conn: sqlite3.Connection,
+    system_id: str,
+    headquarters_body: str | None,
+) -> tuple[str, str] | None:
+    """Resolve a corp's HQ to a (location_id, location_type) pair.
+
+    Resolution order:
+    1. Use headquarters_body from corporations table if set (body or station).
+    2. Most-populated body in the system.
+    3. Any body in the system.
+    4. Any station in the system.
+    Returns None if no body or station found.
+    """
+    if headquarters_body:
+        # Determine whether it's a body or station
+        body = conn.execute(
+            "SELECT body_id FROM bodies WHERE body_id = ?", (headquarters_body,)
+        ).fetchone()
+        if body:
+            return (headquarters_body, "body")
+        station = conn.execute(
+            "SELECT station_id FROM stations WHERE station_id = ?",
+            (headquarters_body,),
+        ).fetchone()
+        if station:
+            return (headquarters_body, "station")
+
+    # Most-populated body
+    body = conn.execute(
+        """SELECT body_id FROM bodies WHERE system_id = ?
+           ORDER BY population DESC LIMIT 1""",
+        (system_id,),
+    ).fetchone()
+    if body:
+        return (body[0], "body")
+
+    # Any station
+    station = conn.execute(
+        "SELECT station_id FROM stations WHERE system_id = ? LIMIT 1",
+        (system_id,),
+    ).fetchone()
+    if station:
+        return (station[0], "station")
+
+    return None
+
+
+def import_corp_presence(
+    conn: sqlite3.Connection,
+    wiki_corps: list[dict],
+    commodity_ids: set[str],
+    dry_run: bool,
+) -> int:
+    """Populate corp_presence from wiki headquarters data.
+
+    Each corporation gets one presence row at its headquarters body or station.
+    location_type is 'body' or 'station' per schema (D-182).
+    primary_operation is set to the first commodity tag matching a known commodity ID.
+    """
+    valid_systems = {
+        r[0] for r in conn.execute("SELECT system_id FROM star_systems").fetchall()
+    }
+
+    # Load headquarters_body from corporations table (set during import)
+    hq_body_map: dict[str, str | None] = {
+        r[0]: r[1]
+        for r in conn.execute(
+            "SELECT corp_id, headquarters_body FROM corporations"
+        ).fetchall()
+    }
+
+    rows: list[tuple] = []
+    skipped: list[str] = []
+    for corp in wiki_corps:
+        system_id = corp.get("system_id")
+        if not system_id:
+            skipped.append(f"{corp['corp_id']} (no headquarters system parsed)")
+            continue
+        if system_id not in valid_systems:
+            skipped.append(f"{corp['corp_id']} (system '{system_id}' not in DB)")
+            continue
+
+        hq_body = hq_body_map.get(corp["corp_id"])
+        location = _resolve_hq_location(conn, system_id, hq_body)
+        if not location:
+            skipped.append(
+                f"{corp['corp_id']} (no body/station found in system '{system_id}')"
+            )
+            continue
+
+        location_id, location_type = location
+        primary_op = next(
+            (tag for tag in corp.get("tags", []) if tag in commodity_ids), None
+        )
+        rows.append((corp["corp_id"], location_id, location_type, primary_op))
+
+    if skipped:
+        for s in skipped:
+            print(f"        warning: skipped corp_presence for {s}")
+
+    if not dry_run:
+        conn.execute("DELETE FROM corp_presence")
+        conn.executemany(
+            """INSERT OR IGNORE INTO corp_presence
+               (corp_id, location_id, location_type, primary_operation)
+               VALUES (?, ?, ?, ?)""",
+            rows,
+        )
+
+    return len(rows)

tooling/economy-db/economy_import/db.py

@@ -0,0 +1,33 @@
+"""Connection and table-maintenance helpers for the economics import."""
+
+import sqlite3
+from pathlib import Path
+
+
+def connect(db_path: Path) -> sqlite3.Connection:
+    """Open systems.db with foreign-key enforcement on."""
+    conn = sqlite3.connect(str(db_path))
+    conn.execute("PRAGMA foreign_keys=ON")
+    return conn
+
+
+def clear_economics_tables(conn: sqlite3.Connection) -> None:
+    """Clear economics tables in FK-safe order (children before parents).
+
+    corp_presence is cleared here; the corporations table is append-only
+    (never cleared).  Runs inside the caller's transaction, so a failure
+    later in the import rolls these DELETEs back too.
+    """
+    conn.execute("DELETE FROM corp_presence")
+    conn.execute("DELETE FROM brand_inputs")
+    conn.execute("DELETE FROM brand_products")
+    conn.execute("DELETE FROM system_fiscal")
+    conn.execute("DELETE FROM corp_financial_state")
+    conn.execute("DELETE FROM corp_lifecycle_events")
+    conn.execute("DELETE FROM chain_inputs")
+    conn.execute("DELETE FROM production_chains")
+    # specialization_vocabulary FK-references commodities — clear it
+    # before commodities so the FK-on delete does not fail (D-237).
+    conn.execute("DELETE FROM specialization_vocabulary")
+    conn.execute("DELETE FROM commodities")
+    conn.execute("DELETE FROM gate_links")

tooling/economy-db/economy_import/economy.py

@@ -0,0 +1,254 @@
+"""Core economy imports: gate links, commodities, production chains,
+currency zones, gate energy connectivity, and system fiscal parameters."""
+
+import json
+import sqlite3
+import tomllib
+
+from .paths import CHAINS_TOML, COMMODITIES_TOML, CURRENCY_ZONES_TOML, STAR_MAP
+
+# ---------------------------------------------------------------------------
+# Gate links
+# ---------------------------------------------------------------------------
+
+
+def import_gate_links(conn: sqlite3.Connection, dry_run: bool) -> int:
+    with open(STAR_MAP) as f:
+        data = json.load(f)
+
+    edges = data["edges"]
+    system_ids = {r[0] for r in conn.execute("SELECT system_id FROM star_systems").fetchall()}
+
+    rows: list[tuple[str, str]] = []
+    skipped: list[str] = []
+    for a, b in edges:
+        if a not in system_ids:
+            skipped.append(a)
+            continue
+        if b not in system_ids:
+            skipped.append(b)
+            continue
+        rows.append((a, b))
+        rows.append((b, a))
+
+    if skipped:
+        unique_skipped = sorted(set(skipped))
+        print(f"  warning: {len(unique_skipped)} system(s) in star-map.json not in DB: {unique_skipped[:5]}...")
+
+    if not dry_run:
+        conn.executemany(
+            "INSERT OR IGNORE INTO gate_links (from_system_id, to_system_id) VALUES (?, ?)",
+            rows,
+        )
+
+    return len(rows)
+
+
+# ---------------------------------------------------------------------------
+# Commodities
+# ---------------------------------------------------------------------------
+
+
+def import_commodities(conn: sqlite3.Connection, dry_run: bool) -> int:
+    with open(COMMODITIES_TOML, "rb") as f:
+        data = tomllib.load(f)
+
+    rows: list[tuple] = []
+    for cid, c in data.items():
+        rows.append((
+            cid,
+            c["name"],
+            c["tier"],
+            c["elasticity"],
+            c["base_price"],
+            c.get("bulk_class"),
+            c.get("unit"),
+            c.get("production_ubiquity"),
+            c.get("demand_model"),
+            int(c.get("commission_certifiable", False)),
+            int(c.get("compact_contested", False)),
+            int(c.get("shadow_viable", False)),
+            c.get("panic_threshold_weeks", 0),
+            c.get("description"),
+        ))
+
+    if not dry_run:
+        conn.executemany(
+            """INSERT INTO commodities (
+                commodity_id, name, tier, elasticity, base_price,
+                bulk_class, unit, production_ubiquity, demand_model,
+                commission_certifiable, compact_contested, shadow_viable,
+                panic_threshold_weeks, description
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+            rows,
+        )
+
+    return len(rows)
+
+
+# ---------------------------------------------------------------------------
+# Production chains + inputs
+# ---------------------------------------------------------------------------
+
+
+def import_chains(conn: sqlite3.Connection, dry_run: bool) -> tuple[int, int]:
+    with open(CHAINS_TOML, "rb") as f:
+        data = tomllib.load(f)
+
+    chain_rows: list[tuple] = []
+    input_rows: list[tuple] = []
+    for chain_id, c in data.items():
+        chain_rows.append((
+            chain_id,
+            c["output"],
+            c.get("output_quantity", 1.0),
+            int(c.get("location_bound", False)),
+            c.get("description"),
+        ))
+        for inp in c.get("inputs", []):
+            input_rows.append((
+                chain_id,
+                inp["commodity"],
+                inp["quantity"],
+            ))
+
+    if not dry_run:
+        conn.executemany(
+            """INSERT INTO production_chains (
+                chain_id, output_commodity_id, output_quantity,
+                location_bound, description
+            ) VALUES (?, ?, ?, ?, ?)""",
+            chain_rows,
+        )
+        conn.executemany(
+            """INSERT INTO chain_inputs (
+                chain_id, input_commodity_id, quantity
+            ) VALUES (?, ?, ?)""",
+            input_rows,
+        )
+
+    return len(chain_rows), len(input_rows)
+
+
+# ---------------------------------------------------------------------------
+# Currency zones
+# ---------------------------------------------------------------------------
+
+
+def set_currency_zones(conn: sqlite3.Connection, dry_run: bool) -> dict[str, int | str]:
+    """Set currency_zone on star_systems from wiki/economics/currency_zones.toml.
+
+    Default: TRACTUS_PRIMARY. Sol (GJ 0): MIXED (set before file is read).
+    MARK_PRIMARY and MIXED assignments come from the TOML file (D-172).
+    """
+    if dry_run:
+        return {"TRACTUS_PRIMARY": "all", "MIXED": "GJ 0 + toml"}
+
+    # Default everything to TRACTUS_PRIMARY
+    conn.execute("UPDATE star_systems SET currency_zone = 'TRACTUS_PRIMARY'")
+
+    # Sol system is MIXED (Earth legacy currency presence — set before TOML load)
+    conn.execute("UPDATE star_systems SET currency_zone = 'MIXED' WHERE system_id = 'GJ 0'")
+
+    # Load MARK_PRIMARY and MIXED assignments from authored TOML (D-172)
+    if CURRENCY_ZONES_TOML.exists():
+        with open(CURRENCY_ZONES_TOML, "rb") as f:
+            zones = tomllib.load(f)
+
+        mark_ids = [entry["system_id"] for entry in zones.get("mark_primary", [])]
+        mixed_ids = [entry["system_id"] for entry in zones.get("mixed", [])]
+
+        for sid in mark_ids:
+            conn.execute(
+                "UPDATE star_systems SET currency_zone = 'MARK_PRIMARY' WHERE system_id = ?",
+                (sid,),
+            )
+        for sid in mixed_ids:
+            conn.execute(
+                "UPDATE star_systems SET currency_zone = 'MIXED' WHERE system_id = ?",
+                (sid,),
+            )
+    else:
+        print("        warning: wiki/economics/currency_zones.toml not found — "
+              "all systems default to TRACTUS_PRIMARY / Sol to MIXED")
+
+    counts: dict[str, int | str] = {}
+    for row in conn.execute("SELECT currency_zone, COUNT(*) FROM star_systems GROUP BY currency_zone"):
+        counts[row[0]] = row[1]
+
+    return counts
+
+
+# ---------------------------------------------------------------------------
+# Gate energy connectivity (D-186)
+# ---------------------------------------------------------------------------
+
+
+def set_gate_energy(conn: sqlite3.Connection, dry_run: bool) -> dict[str, int | str]:
+    """Set gate_energy_connected on star_systems based on currency_zone.
+
+    MARK_PRIMARY zones default to false (Compact refused Gate Corp dependency).
+    All other zones default to true.
+    """
+    if dry_run:
+        return {"on_grid": "non-MARK_PRIMARY", "off_grid": "MARK_PRIMARY"}
+
+    # Default: all systems on-grid
+    conn.execute("UPDATE star_systems SET gate_energy_connected = 1 WHERE gate_energy_connected IS NULL")
+
+    # MARK_PRIMARY zones are off-grid (Compact energy sovereignty)
+    conn.execute("UPDATE star_systems SET gate_energy_connected = 0 WHERE currency_zone = 'MARK_PRIMARY'")
+
+    counts: dict[str, int | str] = {}
+    for row in conn.execute(
+        "SELECT gate_energy_connected, COUNT(*) FROM star_systems GROUP BY gate_energy_connected"
+    ):
+        label = "on_grid" if row[0] == 1 else "off_grid"
+        counts[label] = row[1]
+
+    return counts
+
+
+# ---------------------------------------------------------------------------
+# System fiscal parameters (D-189 section 6)
+# ---------------------------------------------------------------------------
+
+
+def import_system_fiscal(conn: sqlite3.Connection, dry_run: bool) -> int:
+    """Populate system_fiscal with hardcoded Phase 2 values.
+
+    Phase 2 values (NOT derived from D-189 §6 yet):
+      - corp_tax_rate = 0.22 (flat default)
+      - collection_efficiency = 0.85 (mid-reach average placeholder)
+
+    The D-189 §6 formula `collection_efficiency = 1.0 - shadow_economy_intensity × 0.6`
+    is deliberately NOT implemented here — `shadow_economy_intensity` is not
+    yet per-system in the DB (pending the shadow_economy.toml pipeline). When
+    that pipeline lands, replace the hardcoded 0.85 with the derivation and
+    wire `shadow_economy_intensity` through the SELECT. Tracked as a Phase 3
+    follow-up.
+    """
+    inhabited = conn.execute("""
+        SELECT ss.system_id, COALESCE(se.population, 0)
+        FROM star_systems ss
+        LEFT JOIN system_economy se ON ss.system_id = se.system_id
+        WHERE ss.inhabited_planet_count > 0 OR se.population > 0
+        ORDER BY ss.system_id
+    """).fetchall()
+
+    PHASE2_CORP_TAX_RATE = 0.22
+    PHASE2_COLLECTION_EFFICIENCY = 0.85
+
+    rows = [
+        (system_id, PHASE2_CORP_TAX_RATE, PHASE2_COLLECTION_EFFICIENCY)
+        for system_id, _pop in inhabited
+    ]
+
+    if not dry_run:
+        conn.executemany(
+            """INSERT OR IGNORE INTO system_fiscal
+               (system_id, corp_tax_rate, collection_efficiency) VALUES (?, ?, ?)""",
+            rows,
+        )
+
+    return len(rows)

tooling/economy-db/economy_import/errors.py

@@ -0,0 +1,7 @@
+"""Control-flow exceptions shared by the economy_import modules."""
+
+
+class ImportAborted(Exception):
+    """Raised internally by an import/validation step when it wants a clean
+    rollback + exit 1. Caught only by the entrypoint's main(); error messages
+    are printed before raising so the user sees them."""

tooling/economy-db/economy_import/migration.py

@@ -0,0 +1,349 @@
+"""Schema migration for systems.db — new tables and columns on existing DBs.
+
+MIGRATION_SQL is the sanctioned migration path for systems.db (see
+.claude/rules/asset-pipeline.md): idempotent CREATE TABLE/INDEX IF NOT EXISTS
+statements plus data normalizations, executed at the top of every import run
+inside the same transaction that clears + reimports data.  ALTER TABLE column
+additions go through COLUMN_MIGRATIONS (SQLite has no IF NOT EXISTS for ALTER).
+"""
+
+import sqlite3
+
+MIGRATION_SQL: str = """
+-- Economics tables (idempotent — safe to re-run)
+
+-- Brand layer tables (D-189, #827)
+CREATE TABLE IF NOT EXISTS brand_products (
+    brand_product_id TEXT PRIMARY KEY,
+    corp_id TEXT NOT NULL REFERENCES corporations(corp_id),
+    product_name TEXT NOT NULL,
+    brand_category TEXT NOT NULL,
+    value_trajectory TEXT NOT NULL,
+    scarcity_class TEXT NOT NULL,
+    product_subcategory TEXT,
+    base_premium_multiplier REAL NOT NULL DEFAULT 1.0,
+    premium_floor REAL NOT NULL DEFAULT 0.0,
+    origin_system TEXT REFERENCES star_systems(system_id),
+    terroir_locked INTEGER NOT NULL DEFAULT 0,
+    currency_denomination TEXT NOT NULL DEFAULT 'tractus',
+    shadow_viable INTEGER NOT NULL DEFAULT 0,
+    brand_tier TEXT NOT NULL,
+    halo_brand_id TEXT REFERENCES brand_products(brand_product_id),
+    price_tier TEXT,
+    updated_at TEXT DEFAULT (datetime('now'))
+);
+
+CREATE TABLE IF NOT EXISTS brand_inputs (
+    brand_product_id TEXT NOT NULL REFERENCES brand_products(brand_product_id),
+    commodity_id TEXT NOT NULL REFERENCES commodities(commodity_id),
+    quantity REAL NOT NULL,
+    PRIMARY KEY (brand_product_id, commodity_id)
+);
+
+CREATE TABLE IF NOT EXISTS system_fiscal (
+    system_id TEXT PRIMARY KEY REFERENCES star_systems(system_id),
+    corp_tax_rate REAL NOT NULL DEFAULT 0.22,
+    collection_efficiency REAL NOT NULL DEFAULT 1.0,
+    updated_at TEXT DEFAULT (datetime('now'))
+);
+
+CREATE TABLE IF NOT EXISTS corp_financial_state (
+    corp_id TEXT PRIMARY KEY REFERENCES corporations(corp_id),
+    health_metric REAL NOT NULL DEFAULT 1.0,
+    updated_at TEXT DEFAULT (datetime('now'))
+);
+
+CREATE TABLE IF NOT EXISTS corp_lifecycle_events (
+    event_id INTEGER PRIMARY KEY AUTOINCREMENT,
+    corp_id TEXT NOT NULL REFERENCES corporations(corp_id),
+    event_type TEXT NOT NULL,
+    event_tick INTEGER NOT NULL DEFAULT 0,
+    event_data TEXT,
+    created_at TEXT DEFAULT (datetime('now'))
+);
+
+CREATE INDEX IF NOT EXISTS idx_brand_products_corp_category ON brand_products(corp_id, brand_category);
+CREATE INDEX IF NOT EXISTS idx_brand_products_origin ON brand_products(origin_system);
+CREATE INDEX IF NOT EXISTS idx_brand_products_tier ON brand_products(brand_tier);
+CREATE INDEX IF NOT EXISTS idx_brand_inputs_commodity ON brand_inputs(commodity_id);
+CREATE INDEX IF NOT EXISTS idx_corp_lifecycle_events_corp ON corp_lifecycle_events(corp_id);
+
+CREATE TABLE IF NOT EXISTS gate_links (
+    from_system_id TEXT NOT NULL REFERENCES star_systems(system_id),
+    to_system_id TEXT NOT NULL REFERENCES star_systems(system_id),
+    PRIMARY KEY (from_system_id, to_system_id)
+);
+
+CREATE TABLE IF NOT EXISTS commodities (
+    commodity_id TEXT PRIMARY KEY,
+    name TEXT NOT NULL,
+    tier TEXT NOT NULL,
+    elasticity TEXT NOT NULL,
+    base_price REAL NOT NULL,
+    bulk_class TEXT,
+    unit TEXT,
+    production_ubiquity TEXT,
+    demand_model TEXT,
+    commission_certifiable INTEGER DEFAULT 0,
+    compact_contested INTEGER DEFAULT 0,
+    shadow_viable INTEGER DEFAULT 0,
+    panic_threshold_weeks INTEGER DEFAULT 0,
+    description TEXT,
+    updated_at TEXT DEFAULT (datetime('now'))
+);
+
+-- D-237 authored specialization layer vocabulary (must follow commodities for FK).
+-- Mirrors the canonical DDL in systems-schema.sql; here so the migration path
+-- (existing DBs) gets the table, not just fresh systems-schema.sql builds.
+CREATE TABLE IF NOT EXISTS specialization_vocabulary (
+    specialization_id            TEXT PRIMARY KEY,
+    commodity_id                 TEXT NOT NULL REFERENCES commodities(commodity_id),
+    production_ubiquity_override  TEXT,
+    bulk_class_projected         TEXT NOT NULL,
+    production_ubiquity_projected TEXT NOT NULL,
+    description                  TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_spec_vocab_commodity ON specialization_vocabulary(commodity_id);
+
+CREATE TABLE IF NOT EXISTS production_chains (
+    chain_id TEXT PRIMARY KEY,
+    output_commodity_id TEXT NOT NULL REFERENCES commodities(commodity_id),
+    output_quantity REAL NOT NULL DEFAULT 1.0,
+    location_bound INTEGER DEFAULT 0,
+    description TEXT,
+    updated_at TEXT DEFAULT (datetime('now'))
+);
+
+CREATE TABLE IF NOT EXISTS chain_inputs (
+    chain_id TEXT NOT NULL REFERENCES production_chains(chain_id),
+    input_commodity_id TEXT NOT NULL REFERENCES commodities(commodity_id),
+    quantity REAL NOT NULL,
+    PRIMARY KEY (chain_id, input_commodity_id)
+);
+
+CREATE TABLE IF NOT EXISTS corp_presence (
+    corp_id TEXT NOT NULL REFERENCES corporations(corp_id),
+    location_id TEXT NOT NULL,
+    location_type TEXT NOT NULL,
+    primary_operation TEXT,
+    updated_at TEXT DEFAULT (datetime('now')),
+    PRIMARY KEY (corp_id, location_id)
+);
+
+-- New indexes
+CREATE INDEX IF NOT EXISTS idx_star_systems_currency ON star_systems(currency_zone);
+CREATE INDEX IF NOT EXISTS idx_gate_links_from ON gate_links(from_system_id);
+CREATE INDEX IF NOT EXISTS idx_gate_links_to ON gate_links(to_system_id);
+CREATE INDEX IF NOT EXISTS idx_commodities_tier ON commodities(tier);
+CREATE INDEX IF NOT EXISTS idx_production_chains_output ON production_chains(output_commodity_id);
+CREATE INDEX IF NOT EXISTS idx_chain_inputs_commodity ON chain_inputs(input_commodity_id);
+CREATE INDEX IF NOT EXISTS idx_corp_presence_corp ON corp_presence(corp_id);
+CREATE INDEX IF NOT EXISTS idx_corp_presence_location ON corp_presence(location_id);
+
+-- Generator metadata stamp (#855, #856)
+CREATE TABLE IF NOT EXISTS meta (
+    generator_name TEXT PRIMARY KEY,
+    schema_version TEXT NOT NULL,
+    generator_sha TEXT NOT NULL,
+    generated_at TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
+-- Drop the pre-merge 'generate_brands' stamp row if it exists (PR #136 review T2/H3).
+-- The Rust brand binary is now a subroutine of import_economics — its source
+-- SHA contributes to the 'import_economics' stamp — so it no longer merits its
+-- own meta row.  This DELETE makes the check-systems-db-stamp "unknown generator"
+-- path (fail-closed per T6) compatible with older DBs that still have the row.
+DELETE FROM meta WHERE generator_name = 'generate_brands';
+
+-- Drop the retired 'generate_atlas' stamp row if it exists (#951, D-223).
+-- The atlas geometry generator was retired; import_economics now owns the
+-- atlas index, so generate_atlas no longer merits its own meta row.  Without
+-- this DELETE, check-systems-db-stamp's fail-closed "unknown generator" path
+-- (T6) would reject any committed DB that still carries the old row.
+DELETE FROM meta WHERE generator_name = 'generate_atlas';
+
+-- Drop the retired heightmap BLOB table (D-202 amended, #963): canonical
+-- elevation is now a per-body 16-bit grayscale heightmap.png file, not a DB
+-- BLOB. The Rust loader reads the PNG; nothing reads this table anymore.
+DROP TABLE IF EXISTS atlas_body_heightmaps;
+
+-- City name reservations (D-207, #902)
+CREATE TABLE IF NOT EXISTS atlas_city_names (
+    id               INTEGER PRIMARY KEY AUTOINCREMENT,
+    body_id          TEXT NOT NULL REFERENCES bodies(body_id) ON DELETE CASCADE,
+    name             TEXT NOT NULL,
+    kind             TEXT NOT NULL DEFAULT 'city',
+    economic_role    TEXT NOT NULL,
+    population       INTEGER NOT NULL,
+    settlement_class TEXT,
+    corp_id          TEXT REFERENCES corporations(corp_id),
+    reserved         INTEGER NOT NULL DEFAULT 0,
+    updated_at       TEXT NOT NULL DEFAULT (datetime('now'))
+);
+CREATE INDEX IF NOT EXISTS idx_atlas_city_names_body ON atlas_city_names(body_id);
+CREATE INDEX IF NOT EXISTS idx_atlas_city_names_kind ON atlas_city_names(kind);
+CREATE INDEX IF NOT EXISTS idx_atlas_city_names_corp ON atlas_city_names(corp_id);
+
+-- Geographic feature name reservations (#903)
+CREATE TABLE IF NOT EXISTS atlas_feature_names (
+    id           INTEGER PRIMARY KEY AUTOINCREMENT,
+    body_id      TEXT NOT NULL REFERENCES bodies(body_id) ON DELETE CASCADE,
+    name         TEXT NOT NULL,
+    feature_type TEXT NOT NULL,
+    priority     INTEGER NOT NULL DEFAULT 0,
+    updated_at   TEXT NOT NULL DEFAULT (datetime('now'))
+);
+CREATE INDEX IF NOT EXISTS idx_atlas_feature_names_body ON atlas_feature_names(body_id);
+
+-- Province boundaries (D-205, #904)
+CREATE TABLE IF NOT EXISTS atlas_province_boundaries (
+    body_id   TEXT NOT NULL REFERENCES bodies(body_id) ON DELETE CASCADE,
+    basin_id  INTEGER NOT NULL,
+    path      TEXT NOT NULL,
+    area_pct  REAL NOT NULL,
+    PRIMARY KEY (body_id, basin_id)
+);
+CREATE INDEX IF NOT EXISTS idx_atlas_province_boundaries_body ON atlas_province_boundaries(body_id);
+
+-- City positions — attractor-matched placement output (D-211, #34)
+CREATE TABLE IF NOT EXISTS atlas_city_positions (
+    city_names_id  INTEGER PRIMARY KEY REFERENCES atlas_city_names(id) ON DELETE CASCADE,
+    body_id        TEXT NOT NULL REFERENCES bodies(body_id) ON DELETE CASCADE,
+    row            INTEGER NOT NULL,
+    col            INTEGER NOT NULL,
+    attractor_type TEXT NOT NULL,
+    score          REAL NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_atlas_city_positions_body ON atlas_city_positions(body_id);
+
+-- Architecture-flavor trait templates (D-232, #993). trait_templates = the
+-- shared catalog (baked from architecture_trait_catalog.toml); atlas_body_trait_bias
+-- = sparse per-body hero pins (#1017). List/map fields are JSON; numerics are
+-- integer basis-points (D-010). Retires the round-2 atlas_body_culture tables.
+DROP TABLE IF EXISTS atlas_body_culture_era;
+DROP TABLE IF EXISTS atlas_body_culture;
+CREATE TABLE IF NOT EXISTS trait_templates (
+    tag                      TEXT PRIMARY KEY,
+    label                    TEXT NOT NULL,
+    cultural_description     TEXT,
+    corridor_pool            TEXT NOT NULL DEFAULT 'baseline',
+    geographic_sector        TEXT,
+    bulk_class_gate          TEXT,
+    production_ubiquity_gate TEXT,
+    min_prosperity_bps       INTEGER NOT NULL DEFAULT 0,
+    base_weight              INTEGER NOT NULL DEFAULT 10000,
+    weight_mods              TEXT,
+    zone_affinity            TEXT,
+    allow_tags               TEXT,
+    block_tags               TEXT,
+    era_scope                TEXT,
+    visual_bundle            TEXT,
+    updated_at               TEXT NOT NULL DEFAULT (datetime('now'))
+);
+CREATE INDEX IF NOT EXISTS idx_trait_templates_pool ON trait_templates(corridor_pool);
+CREATE INDEX IF NOT EXISTS idx_trait_templates_sector ON trait_templates(geographic_sector);
+CREATE TABLE IF NOT EXISTS atlas_body_trait_bias (
+    id                    INTEGER PRIMARY KEY AUTOINCREMENT,
+    body_id               TEXT NOT NULL REFERENCES bodies(body_id) ON DELETE CASCADE,
+    template_tag          TEXT NOT NULL REFERENCES trait_templates(tag) ON DELETE CASCADE,
+    bias_kind             TEXT NOT NULL,
+    weight_multiplier_bps INTEGER,
+    note                  TEXT,
+    updated_at            TEXT NOT NULL DEFAULT (datetime('now')),
+    UNIQUE (body_id, template_tag)
+);
+CREATE INDEX IF NOT EXISTS idx_atlas_body_trait_bias_body ON atlas_body_trait_bias(body_id);
+CREATE INDEX IF NOT EXISTS idx_atlas_body_trait_bias_tag ON atlas_body_trait_bias(template_tag);
+
+-- Normalize bodies.economic_role to the D-194 canonical 10-value set (#911).
+-- Idempotent: each UPDATE is a no-op if the old value is already gone.
+UPDATE bodies SET economic_role = 'agricultural'  WHERE economic_role IN ('agriculture', 'mixed-agriculture');
+UPDATE bodies SET economic_role = 'extraction'    WHERE economic_role IN ('mining', 'resource_extraction', 'energy');
+UPDATE bodies SET economic_role = 'transit_hub'   WHERE economic_role = 'transit';
+UPDATE bodies SET economic_role = 'service_mixed' WHERE economic_role IN ('commercial', 'coordination');
+UPDATE bodies SET economic_role = 'residential'   WHERE economic_role = 'frontier';
+
+-- Backfill bodies.founding_age_years for all inhabited bodies in player scope (D-216 amendment,
+-- ticket #1000).  Idempotent: WHERE clause limits to NULL rows, so a re-run is a no-op.
+--
+-- Strategy: COALESCE(events-first, wave-fallback)
+--   events-first:  the system's colonial_charter event age_years (authored; 9 systems in live DB).
+--                  system_history is keyed PRIMARY KEY on system_id, so the wave subquery returns
+--                  at most one row; LIMIT 1 is used on historical_events for safety (max one
+--                  colonial_charter per system in the data).
+--   wave-fallback: canonical founding-edge of each settlement_wave range per D-216:
+--                  wave_1=600, wave_2=500, wave_3=300, wave_4=100, wave_5=40.
+--
+-- Exclusions (founding_age_years stays NULL):
+--   origin    — Sol; out of player scope per D-236.
+--   unsettled — 24 systems with inhabited=0; the EXISTS guard also excludes them.
+UPDATE bodies
+SET founding_age_years = COALESCE(
+    (
+        SELECT he.age_years
+        FROM historical_events he
+        WHERE he.system_id = bodies.system_id
+          AND he.event_type = 'colonial_charter'
+        ORDER BY he.sort_order ASC
+        LIMIT 1
+    ),
+    (
+        SELECT CASE sh.settlement_wave
+            WHEN 'wave_1' THEN 600
+            WHEN 'wave_2' THEN 500
+            WHEN 'wave_3' THEN 300
+            WHEN 'wave_4' THEN 100
+            WHEN 'wave_5' THEN 40
+            ELSE NULL -- unexpected settlement_wave: add a WHEN above
+        END
+        FROM system_history sh
+        WHERE sh.system_id = bodies.system_id
+    )
+)
+WHERE founding_age_years IS NULL
+  AND inhabited = 1
+  AND EXISTS (
+      SELECT 1
+      FROM system_history sh2
+      WHERE sh2.system_id = bodies.system_id
+        AND sh2.settlement_wave NOT IN ('origin', 'unsettled')
+  );
+"""
+
+# Columns to add to existing tables (ALTER TABLE is idempotent via try/except)
+COLUMN_MIGRATIONS: list[tuple[str, str, str]] = [
+    ("star_systems", "currency_zone", "TEXT DEFAULT 'TRACTUS_PRIMARY'"),
+    ("star_systems", "gate_energy_connected", "INTEGER DEFAULT 1"),
+    ("corporations", "behavioral_archetype", "TEXT"),
+    ("corporations", "supply_chain_role", "TEXT"),
+    ("corporations", "shadow_economy_access", "INTEGER DEFAULT 0"),
+    ("brand_products", "price_tier", "TEXT"),
+    ("bodies", "body_radius_km", "REAL"),  # D-204 — physical radius in km, nullable
+    ("bodies", "axial_tilt_deg", "REAL"),  # T-1024, D-239 §2 — axial tilt from body-def frontmatter
+    ("meta", "schema_sha", "TEXT"),
+    ("atlas_city_names", "settlement_class", "TEXT"),  # D-196 — NULL until placement (#37)
+    ("system_economy", "economic_specialization", "TEXT"),  # D-237 — authored specialization layer
+    ("system_economy", "cultural_specialization", "TEXT"),  # D-237 — authored specialization layer
+]
+
+
+def _add_column(conn: sqlite3.Connection, table: str, col: str, col_type: str) -> None:
+    """Add a column if it doesn't exist. SQLite has no IF NOT EXISTS for ALTER."""
+    try:
+        conn.execute(f"ALTER TABLE {table} ADD COLUMN {col} {col_type}")
+    except sqlite3.OperationalError as e:
+        if "duplicate column" in str(e).lower():
+            pass  # already exists
+        else:
+            raise
+
+
+def apply_schema_migrations(conn: sqlite3.Connection) -> None:
+    """Apply COLUMN_MIGRATIONS then MIGRATION_SQL (idempotent, in-transaction).
+
+    Note: ``executescript`` implicitly commits any pending transaction before
+    running the script — same semantics as the pre-split importer.
+    """
+    for table, col, col_type in COLUMN_MIGRATIONS:
+        _add_column(conn, table, col, col_type)
+    conn.executescript(MIGRATION_SQL)

tooling/economy-db/economy_import/paths.py

@@ -0,0 +1,50 @@
+"""Source-file path constants for the economy_import modules.
+
+Single façade: modules import their input paths from here.  Paths that are
+also part of the import_economics meta-stamp source set are defined once in
+tooling/generator_sources.py and re-exported, so the stamp registry and the
+importer can never disagree about where a stamped source lives.
+"""
+
+from pathlib import Path
+
+from generator_sources import (
+    ARCHITECTURE_TRAIT_BIAS_TOML,
+    ARCHITECTURE_TRAIT_CATALOG_TOML,
+    GENERATE_BRANDS_WRAPPER,
+    REPO_ROOT,
+    SPECIALIZATION_VOCAB_TOML,
+    SYSTEM_SPECIALIZATION_TOML,
+)
+
+__all__ = [
+    "ARCHITECTURE_TRAIT_BIAS_TOML",
+    "ARCHITECTURE_TRAIT_CATALOG_TOML",
+    "BRANDS_TOML",
+    "CHAINS_TOML",
+    "COMMODITIES_TOML",
+    "CORPORATIONS_DIR",
+    "CURRENCY_ZONES_TOML",
+    "DB_PATH",
+    "GENERATED_BRANDS_TOML",
+    "GENERATE_BRANDS_WRAPPER",
+    "REPO_ROOT",
+    "SCHEMA_SQL",
+    "SPECIALIZATION_VOCAB_TOML",
+    "STAR_MAP",
+    "SYSTEM_SPECIALIZATION_TOML",
+    "WIKI_STAR_SYSTEMS",
+]
+
+DB_PATH: Path = REPO_ROOT / "server" / "data" / "systems.db"
+STAR_MAP: Path = REPO_ROOT / "docs" / "design" / "star-map.json"
+COMMODITIES_TOML: Path = REPO_ROOT / "wiki" / "economics" / "commodities.toml"
+CHAINS_TOML: Path = REPO_ROOT / "wiki" / "economics" / "production_chains.toml"
+CURRENCY_ZONES_TOML: Path = REPO_ROOT / "wiki" / "economics" / "currency_zones.toml"
+SCHEMA_SQL: Path = REPO_ROOT / "server" / "data" / "systems-schema.sql"
+CORPORATIONS_DIR: Path = REPO_ROOT / "wiki" / "corporations"
+WIKI_STAR_SYSTEMS: Path = REPO_ROOT / "wiki" / "star-systems"
+BRANDS_TOML: Path = REPO_ROOT / "wiki" / "economics" / "corporations" / "brands.toml"
+GENERATED_BRANDS_TOML: Path = (
+    REPO_ROOT / "wiki" / "economics" / "corporations" / "generated_brands.toml"
+)

tooling/economy-db/economy_import/specialization.py

@@ -0,0 +1,428 @@
+"""System specialization — D-237 authored layer (#1013, #1015)."""
+
+import sqlite3
+import tomllib
+from collections import Counter
+
+from .errors import ImportAborted
+from .paths import SPECIALIZATION_VOCAB_TOML, SYSTEM_SPECIALIZATION_TOML
+
+# Authoritative D-233 projected enums. The full CI guardrail suite (V-SES-*)
+# lands in #1015; this module performs the FK + enum sanity the import itself
+# needs to stay sound. NOTE for #1015: the D-237 "equal-or-higher" override rule
+# must NOT hard-fail HUB specializations (shipbuilding, transit_hub) whose local
+# production_ubiquity_projected is intentionally below their commodity's global
+# default — see specialization_vocabulary.toml header.
+_BULK_CLASSES: set[str] = {"BulkSolid", "BulkLiquid", "PrecisionDense", "Perishable", "NonPhysical"}
+_PRODUCTION_UBIQUITY: set[str] = {"Ubiquitous", "Common", "Specialist", "MonopolySource"}
+_FACTION_VOCAB: set[str] = {
+    "concord_assembly", "compact", "compact_sympathetic", "syndic_dominant",
+    "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: set[str] = {
+    "scholarly", "artistic", "institutional", "commercial", "agrarian",
+    "industrial_heritage", "medical_elite", "ecological", "military",
+    "financial_technocratic", "cosmopolitan", "compact_cooperative",
+}
+# Canonical 47-value heritage taxonomy (#1016, D-237). Source of truth:
+# docs/workshops/system-economic-specialization/heritage-taxonomy-draft.md.
+# Real-world people/nationality granularity, lowercase_snake. Heritage wins over
+# activity values when both apply. Pin only when a system's founding heritage
+# DIVERGES from its corridor baseline (D-167/D-232); corridor-typical systems
+# stay NULL and take the corridor default.
+_CULTURAL_HERITAGE: set[str] = {
+    # British Isles & Anglo-diaspora
+    "anglo", "scottish", "irish", "welsh",
+    # Iberian, Latin & Lusophone
+    "portuguese", "brazilian", "afro_brazilian", "cape_verdean", "angolan",
+    "sao_tomean", "spanish", "canarian", "italian", "french",
+    # Northern / Central / Eastern European
+    "german", "dutch", "nordic", "finnish", "polish", "czech", "russian",
+    "luxembourgish", "hungarian",
+    # Sub-Saharan African
+    "afrikaans", "cape_malay", "zulu", "xhosa", "herero", "shona", "swahili",
+    "igbo", "yoruba", "hausa", "akan",
+    # South Asian
+    "indian", "bengali", "punjabi", "konkan",
+    # East & Southeast Asian
+    "chinese", "korean", "japanese", "vietnamese", "tagalog",
+    # Pacific
+    "maori",
+    # Middle East / North Africa / Central Asia
+    "arab", "persian", "turkic",
+}
+_CULTURAL_VOCAB: set[str] = _CULTURAL_ACTIVITY | _CULTURAL_HERITAGE
+
+# Catalog production_ubiquity concentration ranking for V-SES-03 (override may
+# only be >= the commodity's global default). regional ≈ common tier.
+_UBIQUITY_RANK: dict[str, int] = {
+    "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
+        (FK-validated against commodities; MUST run after import_commodities).
+      - system_specialization.toml -> UPSERTs economic_specialization +
+        cultural_specialization onto system_economy, and dominant_faction onto
+        system_factions, for authored (hero) systems only.
+
+    Authored lore wins; unauthored systems are left NULL for the generator's
+    heuristic fallback. economic_specialization / cultural_specialization are
+    owned exclusively by this importer, so they are cleared to NULL first for
+    idempotency (a removed stanza must not leave a stale value). dominant_faction
+    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.
+    """
+    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_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()
+    }
+    economy_system_ids = {
+        r[0] for r in conn.execute("SELECT system_id FROM system_economy").fetchall()
+    }
+    faction_system_ids = {
+        r[0] for r in conn.execute("SELECT system_id FROM system_factions").fetchall()
+    }
+
+    errors: list[str] = []
+
+    # --- Vocabulary: FK + enum validation (V-SES-06, V-SES-03) ----------
+    vocab_rows: list[tuple] = []
+    for spec_id, v in vocab.items():
+        cid = v.get("commodity_id")
+        if cid not in commodity_ids:  # V-SES-06
+            errors.append(
+                f"V-SES-06: specialization_vocabulary '{spec_id}': commodity_id "
+                f"'{cid}' not in commodities catalog"
+            )
+        bc = v.get("bulk_class_projected")
+        if bc not in _BULK_CLASSES:
+            errors.append(
+                f"specialization_vocabulary '{spec_id}': bulk_class_projected "
+                f"'{bc}' invalid (expected one of {sorted(_BULK_CLASSES)})"
+            )
+        pu = v.get("production_ubiquity_projected")
+        if pu not in _PRODUCTION_UBIQUITY:
+            errors.append(
+                f"specialization_vocabulary '{spec_id}': "
+                f"production_ubiquity_projected '{pu}' invalid "
+                f"(expected one of {sorted(_PRODUCTION_UBIQUITY)})"
+            )
+        override = v.get("production_ubiquity_override") or None  # "" -> NULL
+        # 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"
+                )
+        vocab_rows.append((spec_id, cid, override, bc, pu, v.get("description", "")))
+
+    valid_spec_ids = set(vocab.keys())
+
+    # --- 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:  # V-SES-01
+            errors.append(
+                f"V-SES-01: system_specialization '{sid}': economic_specialization "
+                f"'{es}' not in specialization_vocabulary"
+            )
+        cs = s.get("cultural_specialization")
+        if cs is not None and cs not in _CULTURAL_VOCAB:  # V-SES-04
+            errors.append(
+                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)})"
+            )
+
+    if errors:
+        print(f"  SPECIALIZATION ERRORS ({len(errors)}):")
+        for e in errors:
+            print(f"    - {e}")
+        raise ImportAborted()
+
+    coverage: dict = {
+        "vocab": len(vocab_rows),
+        "economic": 0,
+        "cultural": 0,
+        "faction": 0,
+        "missing_economy_row": [],
+        "missing_faction_row": [],
+    }
+
+    if dry_run:
+        for sid, s in systems.items():
+            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 ------------------------------------
+    conn.execute("DELETE FROM specialization_vocabulary")
+    conn.executemany(
+        """INSERT INTO specialization_vocabulary (
+            specialization_id, commodity_id, production_ubiquity_override,
+            bulk_class_projected, production_ubiquity_projected, description
+        ) VALUES (?, ?, ?, ?, ?, ?)""",
+        vocab_rows,
+    )
+
+    # --- Clear importer-owned columns (idempotency) ---------------------
+    conn.execute(
+        "UPDATE system_economy SET economic_specialization = NULL, "
+        "cultural_specialization = NULL"
+    )
+
+    # --- UPSERT per-system authored fields ------------------------------
+    for sid, s in systems.items():
+        es = s.get("economic_specialization")
+        cs = s.get("cultural_specialization")
+        if sid in economy_system_ids:
+            conn.execute(
+                "UPDATE system_economy SET economic_specialization = ?, "
+                "cultural_specialization = ? WHERE system_id = ?",
+                (es, cs, sid),
+            )
+            coverage["economic"] += 1 if es else 0
+            coverage["cultural"] += 1 if cs else 0
+        else:
+            coverage["missing_economy_row"].append(sid)
+
+        df = s.get("dominant_faction")
+        if df is not None:
+            if sid in faction_system_ids:
+                conn.execute(
+                    "UPDATE system_factions SET dominant_faction = ? "
+                    "WHERE system_id = ?",
+                    (df, sid),
+                )
+                coverage["faction"] += 1
+            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: sqlite3.Connection, vocab: dict, systems: dict, strict: bool
+) -> None:
+    """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[str] = 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.
+    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 = Counter()
+    pu_dist: Counter = 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("    BulkClass: " + " | ".join(f"{k}={v}" for k, v in sorted(bulk_dist.items())))
+    print("    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"
+            )

tooling/economy-db/economy_import/stamp.py

@@ -0,0 +1,39 @@
+"""Meta-table generator stamp (#855, #856).
+
+Records the source-SHA of the generator that produced the DB so the pre-push
+hook (tooling/check-systems-db-stamp) can detect stale snapshots.  The source
+set comes from tooling/generator_sources.py — the single shared registry.
+"""
+
+import sqlite3
+from pathlib import Path
+
+from generator_sources import file_sha1
+from schema_version import SCHEMA_VERSION
+
+from .paths import SCHEMA_SQL
+
+
+def write_stamp(conn: sqlite3.Connection, generator_name: str, *source_files: Path) -> None:
+    """Upsert a row in the meta table recording this generator's current source SHA.
+
+    Called after every successful non-dry-run commit.  Idempotent: running
+    twice on the same sources writes the same sha with an updated timestamp.
+
+    Only one stamp is written by this package: ``import_economics``, whose source
+    set includes the Rust binary it invokes (see IMPORT_ECONOMICS_SOURCES in
+    tooling/generator_sources.py).  generate_brands does NOT write a stamp
+    of its own — it's a subroutine of import_economics, not an independent DB
+    writer (PR #136 review T2/H3).
+
+    The meta table is created by the MIGRATION_SQL block in migration.py; this
+    function assumes it exists (caller must run migrations first).
+    """
+    schema_sha = file_sha1(SCHEMA_SQL)
+    generator_sha = file_sha1(*source_files)
+    conn.execute(
+        """INSERT OR REPLACE INTO meta
+               (generator_name, schema_version, schema_sha, generator_sha, generated_at)
+           VALUES (?, ?, ?, ?, datetime('now'))""",
+        (generator_name, SCHEMA_VERSION, schema_sha, generator_sha),
+    )

tooling/economy-db/economy_import/traits.py

@@ -0,0 +1,170 @@
+"""Architecture-flavor trait templates (D-232, #993): catalog baker + hero bias."""
+
+import json
+import sqlite3
+import tomllib
+
+from .errors import ImportAborted
+from .paths import ARCHITECTURE_TRAIT_BIAS_TOML, ARCHITECTURE_TRAIT_CATALOG_TOML
+
+_TRAIT_CORRIDOR_POOLS: set[str] = {"baseline", "heritage", "cross_corridor"}
+_TRAIT_BIAS_KINDS: set[str] = {"pin", "boost", "suppress"}
+# JSON-encoded list/map columns on trait_templates (TOML inline arrays/tables ->
+# JSON text the generator parses).
+_TRAIT_JSON_LIST: tuple[str, ...] = ("bulk_class_gate", "production_ubiquity_gate", "allow_tags", "block_tags")
+_TRAIT_JSON_MAP: tuple[str, ...] = ("weight_mods", "zone_affinity", "visual_bundle")
+
+
+def populate_trait_templates(conn: sqlite3.Connection, dry_run: bool) -> int:
+    """Bake the D-232 architecture-flavor catalog into trait_templates (#993).
+
+    Reads ARCHITECTURE_TRAIT_CATALOG_TOML (`[templates.<tag>]` stanzas) and
+    rebuilds the table. List/map fields are stored as JSON text; numeric
+    eligibility is integer basis-points (D-010). The catalog *content* is
+    authored in #1005 — this baker is the mechanism. Absent source -> 0 rows
+    (the table still exists for the downstream pipeline). Deterministic rebuild:
+    clears trait_templates (cascading atlas_body_trait_bias) first.
+    """
+    if not ARCHITECTURE_TRAIT_CATALOG_TOML.exists():
+        if not dry_run:
+            conn.execute("DELETE FROM atlas_body_trait_bias")
+            conn.execute("DELETE FROM trait_templates")
+        return 0
+    with open(ARCHITECTURE_TRAIT_CATALOG_TOML, "rb") as f:
+        data = tomllib.load(f)
+    templates = data.get("templates", {})
+    errors: list[str] = []
+    rows: list[tuple] = []
+    for tag, t in templates.items():
+        pool = t.get("corridor_pool", "baseline")
+        if pool not in _TRAIT_CORRIDOR_POOLS:
+            errors.append(f"trait_templates '{tag}': corridor_pool '{pool}' invalid")
+        if "label" not in t:
+            errors.append(f"trait_templates '{tag}': missing required 'label'")
+        for k in (*_TRAIT_JSON_LIST, *_TRAIT_JSON_MAP):
+            # any provided list/map field must JSON-encode cleanly
+            if k in t:
+                try:
+                    json.dumps(t[k])
+                except (TypeError, ValueError):
+                    errors.append(f"trait_templates '{tag}': field '{k}' not JSON-serialisable")
+        rows.append((
+            tag, t.get("label", ""), t.get("cultural_description"),
+            pool, t.get("geographic_sector"),
+            json.dumps(t["bulk_class_gate"]) if t.get("bulk_class_gate") else None,
+            json.dumps(t["production_ubiquity_gate"]) if t.get("production_ubiquity_gate") else None,
+            int(t.get("min_prosperity_bps", 0)),
+            int(t.get("base_weight", 10000)),
+            json.dumps(t["weight_mods"]) if t.get("weight_mods") else None,
+            json.dumps(t["zone_affinity"]) if t.get("zone_affinity") else None,
+            json.dumps(t["allow_tags"]) if t.get("allow_tags") else None,
+            json.dumps(t["block_tags"]) if t.get("block_tags") else None,
+            t.get("era_scope"),
+            json.dumps(t["visual_bundle"]) if t.get("visual_bundle") else None,
+        ))
+    if errors:
+        print(f"  TRAIT TEMPLATE ERRORS ({len(errors)}):")
+        for e in errors:
+            print(f"    - {e}")
+        raise ImportAborted()
+    # CI guardrails (D-232, Nigel): >=5 templates eligible per BulkClass (a
+    # template with no bulk_class_gate is eligible for all); no single template
+    # may exceed 60% of its eligible pool's base weight. Skipped when the catalog
+    # is empty (the bootstrap/absent-source case). Integer math (no float).
+    if templates:
+        gerrors: list[str] = []
+        for bc in ("BulkSolid", "BulkLiquid", "PrecisionDense", "Perishable", "NonPhysical"):
+            elig = [t for t in templates.values()
+                    if not t.get("bulk_class_gate") or bc in t["bulk_class_gate"]]
+            if len(elig) < 5:
+                gerrors.append(
+                    f"V-TT-01: only {len(elig)} template(s) eligible for BulkClass "
+                    f"{bc} (need >=5)"
+                )
+            total = sum(int(t.get("base_weight", 10000)) for t in elig)
+            for t in elig:
+                w = int(t.get("base_weight", 10000))
+                if total and w * 5 > total * 3:  # w/total > 0.60
+                    gerrors.append(
+                        f"V-TT-02: template '{t.get('label')}' is "
+                        f"{w * 100 // total}% of the {bc} pool weight (>60%)"
+                    )
+        if gerrors:
+            print(f"  TRAIT TEMPLATE GUARDRAIL FAILURES ({len(gerrors)}):")
+            for e in gerrors:
+                print(f"    - {e}")
+            raise ImportAborted()
+    # Validation/guardrails run on dry-run too; only mutate when committing.
+    if not dry_run:
+        conn.execute("DELETE FROM atlas_body_trait_bias")
+        conn.execute("DELETE FROM trait_templates")
+        conn.executemany(
+            """INSERT INTO trait_templates
+               (tag, label, cultural_description, corridor_pool, geographic_sector,
+                bulk_class_gate, production_ubiquity_gate, min_prosperity_bps,
+                base_weight, weight_mods, zone_affinity, allow_tags, block_tags,
+                era_scope, visual_bundle)
+               VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
+            rows,
+        )
+    return len(rows)
+
+
+def populate_atlas_body_trait_bias(conn: sqlite3.Connection, dry_run: bool) -> int:
+    """Bake the sparse per-body hero pins into atlas_body_trait_bias (#993).
+
+    Reads ARCHITECTURE_TRAIT_BIAS_TOML (`[[bias]]` array). FK-validates body_id
+    against bodies and template_tag against trait_templates (which must be baked
+    first), and validates bias_kind + the basis-point multiplier ranges
+    (boost 10001..30000 = <=3x; suppress 3300..9999 = >=0.33x never 0; pin: no
+    multiplier). Hero-pin *content* is authored in #1017. Absent source -> 0
+    rows. Must run AFTER populate_trait_templates.
+    """
+    if not ARCHITECTURE_TRAIT_BIAS_TOML.exists():
+        if not dry_run:
+            conn.execute("DELETE FROM atlas_body_trait_bias")
+        return 0
+    with open(ARCHITECTURE_TRAIT_BIAS_TOML, "rb") as f:
+        data = tomllib.load(f)
+    entries = data.get("bias", [])
+    body_ids = {r[0] for r in conn.execute("SELECT body_id FROM bodies")}
+    tags = {r[0] for r in conn.execute("SELECT tag FROM trait_templates")}
+    errors: list[str] = []
+    rows: list[tuple] = []
+    seen: set[tuple] = set()
+    for i, b in enumerate(entries):
+        bid = b.get("body_id")
+        tag = b.get("template_tag")
+        kind = b.get("bias_kind")
+        mult = b.get("weight_multiplier_bps")
+        loc = f"bias[{i}] ({bid}/{tag})"
+        if bid not in body_ids:
+            errors.append(f"{loc}: body_id not in bodies")
+        if tag not in tags:
+            errors.append(f"{loc}: template_tag not in trait_templates")
+        if kind not in _TRAIT_BIAS_KINDS:
+            errors.append(f"{loc}: bias_kind '{kind}' invalid (pin|boost|suppress)")
+        if (bid, tag) in seen:
+            errors.append(f"{loc}: duplicate (body_id, template_tag)")
+        seen.add((bid, tag))
+        if kind == "boost" and not (mult and 10001 <= mult <= 30000):
+            errors.append(f"{loc}: boost weight_multiplier_bps must be 10001..30000 (<=3x), got {mult}")
+        if kind == "suppress" and not (mult and 3300 <= mult <= 9999):
+            errors.append(f"{loc}: suppress weight_multiplier_bps must be 3300..9999 (>=0.33x, never 0), got {mult}")
+        if kind == "pin" and mult is not None:
+            errors.append(f"{loc}: pin is mandatory and must not carry weight_multiplier_bps (got {mult})")
+        rows.append((bid, tag, kind, mult, b.get("note")))
+    if errors:
+        print(f"  TRAIT BIAS ERRORS ({len(errors)}):")
+        for e in errors:
+            print(f"    - {e}")
+        raise ImportAborted()
+    if not dry_run:
+        conn.execute("DELETE FROM atlas_body_trait_bias")
+        conn.executemany(
+            """INSERT INTO atlas_body_trait_bias
+               (body_id, template_tag, bias_kind, weight_multiplier_bps, note)
+               VALUES (?,?,?,?,?)""",
+            rows,
+        )
+    return len(rows)

tooling/economy-db/economy_import/validators.py

@@ -0,0 +1,137 @@
+"""Structural-integrity and D-175 coverage validation.
+
+``validate`` is a pre-commit hard blocker; the coverage validators run after
+commit (Phase 2 gate, exit 2) — see the entrypoint's main() for the contract.
+"""
+
+import sqlite3
+
+
+def validate(conn: sqlite3.Connection) -> list[str]:
+    """Validate structural integrity of imported data.
+
+    Checks FK integrity, chain commodity references, and chain completeness.
+    These are hard blockers — broken data must not be committed.
+
+    Coverage validation (commodity/system thresholds) is separate and runs
+    after commit via validate_commodity_coverage() and validate_system_coverage().
+    """
+    errors: list[str] = []
+
+    # FK integrity
+    fk_issues = conn.execute("PRAGMA foreign_key_check").fetchall()
+    if fk_issues:
+        for issue in fk_issues[:10]:
+            errors.append(f"FK violation: table={issue[0]} rowid={issue[1]} "
+                          f"parent={issue[2]} fkid={issue[3]}")
+
+    # Chain inputs reference valid commodities
+    orphan_inputs = conn.execute("""
+        SELECT ci.chain_id, ci.input_commodity_id
+        FROM chain_inputs ci
+        LEFT JOIN commodities c ON ci.input_commodity_id = c.commodity_id
+        WHERE c.commodity_id IS NULL
+    """).fetchall()
+    for chain_id, cid in orphan_inputs:
+        errors.append(f"chain_inputs: chain '{chain_id}' references unknown commodity '{cid}'")
+
+    # Chain outputs reference valid commodities
+    orphan_outputs = conn.execute("""
+        SELECT pc.chain_id, pc.output_commodity_id
+        FROM production_chains pc
+        LEFT JOIN commodities c ON pc.output_commodity_id = c.commodity_id
+        WHERE c.commodity_id IS NULL
+    """).fetchall()
+    for chain_id, cid in orphan_outputs:
+        errors.append(f"production_chains: chain '{chain_id}' outputs unknown commodity '{cid}'")
+
+    # economic_role must be one of the D-194 canonical 10 values
+    valid_roles = {
+        'manufacturing', 'financial', 'agricultural', 'extraction',
+        'service_mixed', 'institutional', 'transit_hub', 'research',
+        'military', 'residential',
+    }
+    bad_roles = conn.execute("""
+        SELECT DISTINCT economic_role, COUNT(*) as cnt
+        FROM bodies
+        WHERE economic_role IS NOT NULL
+          AND economic_role NOT IN (
+              'manufacturing', 'financial', 'agricultural', 'extraction',
+              'service_mixed', 'institutional', 'transit_hub', 'research',
+              'military', 'residential'
+          )
+        GROUP BY economic_role
+    """).fetchall()
+    for role, cnt in bad_roles:
+        errors.append(
+            f"bodies.economic_role: non-canonical value '{role}' on {cnt} row(s) — "
+            f"valid values: {sorted(valid_roles)}"
+        )
+
+    # Chain completeness: every intermediate commodity must have at least one producer
+    missing_chains = conn.execute("""
+        SELECT c.commodity_id, c.name
+        FROM commodities c
+        WHERE c.tier = 'intermediate'
+          AND c.commodity_id NOT IN (SELECT output_commodity_id FROM production_chains)
+        ORDER BY c.commodity_id
+    """).fetchall()
+    for cid, name in missing_chains:
+        errors.append(f"chain completeness: no production chain produces intermediate '{cid}' ({name})")
+
+    return errors
+
+
+def validate_commodity_coverage(
+    conn: sqlite3.Connection, wiki_corps: list[dict], commodity_ids: set[str]
+) -> list[str]:
+    """3+ corporations per major commodity type (raw + intermediate). D-175."""
+    errors: list[str] = []
+    major = [
+        r[0]
+        for r in conn.execute(
+            "SELECT commodity_id FROM commodities "
+            "WHERE tier IN ('raw', 'intermediate') ORDER BY commodity_id"
+        ).fetchall()
+    ]
+
+    # Build commodity → corp set from wiki tags filtered to known commodity IDs
+    coverage: dict[str, set[str]] = {cid: set() for cid in major}
+    for corp in wiki_corps:
+        for tag in corp.get("tags", []):
+            if tag in coverage:
+                coverage[tag].add(corp["corp_id"])
+
+    for cid in major:
+        n = len(coverage[cid])
+        if n < 3:
+            corp_list = sorted(coverage[cid]) if coverage[cid] else ["none"]
+            errors.append(
+                f"commodity coverage: '{cid}' has {n}/3 corp(s) — {corp_list}"
+            )
+
+    return errors
+
+
+def validate_system_coverage(
+    conn: sqlite3.Connection, wiki_corps: list[dict]
+) -> list[str]:
+    """1+ corporation per inhabited system with population > 100K. D-175.
+
+    Uses wiki_corps headquarters data (not DB corp_presence) so this check
+    is accurate in both dry-run and real-run modes.
+    """
+    covered = {c["system_id"] for c in wiki_corps if c.get("system_id")}
+    populated = conn.execute("""
+        SELECT se.system_id, ss.proper_name, se.population
+        FROM system_economy se
+        JOIN star_systems ss ON se.system_id = ss.system_id
+        WHERE se.population > 100000
+        ORDER BY se.system_id
+    """).fetchall()
+
+    return [
+        f"system coverage: no corp presence in '{sid}' ({name}, pop={pop:,})"
+        for sid, name, pop in populated
+        if sid not in covered
+    ]

tooling/generator_sources.py

@@ -0,0 +1,196 @@
+#!/usr/bin/env python3
+"""
+generator_sources — single source of truth for systems.db generator source sets.
+
+Each generator that writes a stamp row to systems.db's ``meta`` table records the
+SHA-1 of the concatenated bytes of its source files (sorted by path).  This module
+is the ONE place that defines which files belong to each generator's stamped set
+(T-1067).  It replaces the formerly triplicated lists in:
+
+  - ``IMPORT_ECONOMICS_SOURCES`` in tooling/economy-db/import_economics.py
+  - ``GENERATOR_SOURCES`` in tooling/check-systems-db-stamp
+  - the hardcoded watch list in .claude/skills/pr-process/SKILL.md
+    (now derived via ``python3 tooling/generator_sources.py --list``)
+
+Consumers:
+  - tooling/economy-db (the importer) — imports ``IMPORT_ECONOMICS_SOURCES`` and
+    ``file_sha1`` to write the stamp after a successful run.
+  - tooling/check-systems-db-stamp — imports ``GENERATOR_SOURCES`` and
+    ``file_sha1`` to verify the stamp before a push.
+  - /pr-process (skill) — shells out to the ``--list`` CLI for its watch list.
+
+This file is itself a member of the stamped source set: a registry change (adding
+or removing a source) must flip the stamp and force ``make regen-db``, otherwise
+edits to the list would be invisible to staleness detection.
+
+The economy_import package modules are collected by glob rather than listed by
+hand, so a module added during a future split is stamped automatically — a
+hand-maintained list that misses a module silently weakens staleness detection.
+
+Usage:
+    python3 tooling/generator_sources.py --list [generator_name]
+"""
+
+import argparse
+import hashlib
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+
+# ---------------------------------------------------------------------------
+# import_economics source set
+# ---------------------------------------------------------------------------
+
+# Entrypoint + split module package (T-1067).
+IMPORT_ECONOMICS_ENTRYPOINT: Path = (
+    REPO_ROOT / "tooling" / "economy-db" / "import_economics.py"
+)
+ECONOMY_IMPORT_PACKAGE_DIR: Path = (
+    REPO_ROOT / "tooling" / "economy-db" / "economy_import"
+)
+
+# Rust sources for the generate_brands subroutine.  import_economics shells out
+# to tooling/generate-brands as part of its normal flow (see economy_import/
+# brands.py), so all of these contribute to its effective source SHA: any change
+# to them must invalidate the meta stamp even though Python hasn't changed.
+GENERATE_BRANDS_RS: Path = (
+    REPO_ROOT / "server" / "src" / "bin" / "generate_brands" / "main.rs"
+)
+GENERATE_BRANDS_NAMES_RS: Path = (
+    REPO_ROOT / "server" / "src" / "bin" / "generate_brands" / "names.rs"
+)
+# Shared surname corpus extracted from the two names.rs copies (T-1064).
+GENERATE_BRANDS_SURNAMES_RS: Path = (
+    REPO_ROOT / "server" / "src" / "bin" / "shared" / "surname_corpus.rs"
+)
+GENERATE_BRANDS_WRAPPER: Path = REPO_ROOT / "tooling" / "generate-brands"
+
+# D-237 authored specialization layer: these data TOMLs feed the DB, so a change
+# to either must flip the stamp and force a regen (#1013).
+SPECIALIZATION_VOCAB_TOML: Path = (
+    REPO_ROOT / "wiki" / "economics" / "specialization_vocabulary.toml"
+)
+SYSTEM_SPECIALIZATION_TOML: Path = (
+    REPO_ROOT / "wiki" / "economics" / "system_specialization.toml"
+)
+# D-232 architecture-flavor trait-template catalog + sparse hero bias (#993).
+ARCHITECTURE_TRAIT_CATALOG_TOML: Path = (
+    REPO_ROOT / "wiki" / "economics" / "architecture_trait_catalog.toml"
+)
+ARCHITECTURE_TRAIT_BIAS_TOML: Path = (
+    REPO_ROOT / "wiki" / "economics" / "architecture_trait_bias.toml"
+)
+
+
+def _economy_import_modules() -> tuple[Path, ...]:
+    """All Python modules of the economy_import package, collected by glob.
+
+    Globbing (rather than hand-listing) guarantees a future module is stamped
+    the moment it exists.  Fail closed if the package is missing — an empty
+    set here would silently weaken staleness detection.
+    """
+    modules = tuple(sorted(ECONOMY_IMPORT_PACKAGE_DIR.glob("*.py")))
+    if not modules:
+        raise RuntimeError(
+            f"economy_import package not found at {ECONOMY_IMPORT_PACKAGE_DIR} — "
+            "the import_economics stamp source set would be incomplete"
+        )
+    return modules
+
+
+# Canonical source set for import_economics' meta stamp.  Covers the Python
+# entrypoint and its module package, the Rust binary it invokes (generate_brands
+# main.rs + names.rs + surname corpus + wrapper script), the shared schema
+# version constant, the authored data TOMLs, and this registry itself.
+IMPORT_ECONOMICS_SOURCES: tuple[Path, ...] = (
+    Path(__file__).resolve(),  # registry changes must stale the stamp
+    IMPORT_ECONOMICS_ENTRYPOINT,
+    *_economy_import_modules(),
+    GENERATE_BRANDS_RS,
+    GENERATE_BRANDS_NAMES_RS,
+    GENERATE_BRANDS_SURNAMES_RS,
+    GENERATE_BRANDS_WRAPPER,
+    REPO_ROOT / "tooling" / "schema_version.py",
+    SPECIALIZATION_VOCAB_TOML,
+    SYSTEM_SPECIALIZATION_TOML,
+    ARCHITECTURE_TRAIT_CATALOG_TOML,
+    ARCHITECTURE_TRAIT_BIAS_TOML,
+)
+
+# ---------------------------------------------------------------------------
+# Registry
+# ---------------------------------------------------------------------------
+
+# Maps generator_name (as stored in meta.generator_name) to the source file(s)
+# whose SHA is stamped.  The SHA is computed as SHA-1 of the concatenated bytes
+# of all files in sorted order.
+#
+# The atlas geometry generator (generate_atlas) was retired in #951 (D-223).
+# import_economics is now the sole regen-db generator that writes systems.db; it
+# owns the atlas index (names-only pool + empty geometry tables).  The surviving
+# planet-gen importers (import_heightmaps, import_province_boundaries) are
+# one-time build imports baked into the committed DB, not part of regen-db, so
+# they are intentionally not stamped here.
+GENERATOR_SOURCES: dict[str, tuple[Path, ...]] = {
+    "import_economics": IMPORT_ECONOMICS_SOURCES,
+}
+
+
+def file_sha1(*paths: Path) -> str:
+    """SHA-1 of concatenated file contents (sorted paths).
+
+    Missing files raise FileNotFoundError rather than silently contributing
+    an empty-string hash (H2): a ghost SHA could mask real breakage when
+    stored and current SHAs converge on the empty-bytes digest.
+    """
+    h = hashlib.sha1()
+    for p in sorted(paths):
+        if not p.exists():
+            raise FileNotFoundError(f"generator source not found: {p}")
+        h.update(p.read_bytes())
+    return h.hexdigest()
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Single source of truth for systems.db generator source sets"
+    )
+    parser.add_argument(
+        "--list",
+        action="store_true",
+        help="Print the stamped source files, one repo-relative path per line",
+    )
+    parser.add_argument(
+        "generator",
+        nargs="?",
+        default=None,
+        help="Restrict --list to one generator (default: all)",
+    )
+    args = parser.parse_args()
+
+    if not args.list:
+        parser.print_help()
+        sys.exit(2)
+
+    if args.generator is not None:
+        if args.generator not in GENERATOR_SOURCES:
+            print(
+                f"error: unknown generator '{args.generator}' — "
+                f"known: {sorted(GENERATOR_SOURCES)}",
+                file=sys.stderr,
+            )
+            sys.exit(1)
+        names = [args.generator]
+    else:
+        names = sorted(GENERATOR_SOURCES)
+
+    seen: set[Path] = set()
+    for name in names:
+        seen.update(GENERATOR_SOURCES[name])
+    for path in sorted(seen):
+        print(path.relative_to(REPO_ROOT).as_posix())
+
+
+if __name__ == "__main__":
+    main()