#!/usr/bin/env python3 """ gemma_naming.py — Batch-name every empty name field in the reach's markers.json files using the Gemma 2 voice pipeline (#833, D-191 §4). Pipeline per body: 1. Load markers.json; identify feature records whose `name` is empty or null. Hand-authored names are never overwritten. 2. Build a short corridor-aware prompt per feature. 3. Stream the prompts into `sr-voice serve --stdio` (long-lived subprocess, restarted every --refresh requests to prevent KV-cache context bleed). 4. Post-process each response: strip quotes, trim whitespace, reject blocklisted Earth majors, retry with a bumped seed on collision or on blocklist hit (up to 3 attempts), fall back to a deterministic palette-driven name on persistent failure. 5. Dedup within (cultural_corridor, feature_type) so two bodies in the same corridor never ship the same river name; cross-corridor collisions are allowed (two "Aldren"s on opposite arcs is fine). 6. Write markers.json back (only if any field changed). 7. Sync every touched body's atlas_* rows in systems.db so `atlas_cities.name`, `atlas_rivers.name`, etc. pick up the new strings without needing a follow-up generate_atlas.py pass. Usage: tooling/planet-gen/gemma_naming.py # full batch, real model tooling/planet-gen/gemma_naming.py --body GJ380c # single body tooling/planet-gen/gemma_naming.py --limit 5 --verbose # smoke test tooling/planet-gen/gemma_naming.py --mock # mock-stdio.sh (no model) tooling/planet-gen/gemma_naming.py \\ --sr-voice /var/mnt/data/projects/settled-reach/main/server/sr-voice/target/release/sr-voice \\ --model /var/mnt/data/projects/settled-reach/main/server/models/gemma2.gguf Exit codes: 0 pipeline completed (possibly with skipped bodies) 1 fatal error (subprocess crash, missing binary, missing schema) """ import argparse import datetime import hashlib import json import re import subprocess import sys import time from pathlib import Path TOOLING_DIR = Path(__file__).resolve().parent REPO_ROOT = (TOOLING_DIR / ".." / "..").resolve() # Reuse the atlas DB sync logic from generate_atlas.py so there is one # authoritative path for atlas_* row updates. sys.path.insert(0, str(TOOLING_DIR)) from generate_atlas import ( # noqa: E402 GRID_H, GRID_W, ensure_atlas_schema, sync_markers_to_db, ) import sqlite3 # noqa: E402 DB_PATH = REPO_ROOT / "server" / "data" / "systems.db" WIKI_SYSTEMS = REPO_ROOT / "wiki" / "star-systems" BLOCKLIST_PATH = TOOLING_DIR / "earth_blocklist.txt" # When --dump-prompts is set, name_feature short-circuits: it builds the # attempt-0 prompt for each feature, writes a JSONL line to this file, # and returns a unique deterministic placeholder so the pipeline runs to # completion without touching a real LLM. Lets us capture the exact prompt # set any backend (Gemma CPU, Gemma GPU, Haiku) would receive on first # attempt — so an offline backend can replay them and the resulting names # can be A/B compared. _CAPTURE_FILE = None # set in main() when --dump-prompts is used # Default binary + model paths. The sr-voice binary is platform-specific # (GPU backend baked in per-build) and lives OUTSIDE any git worktree so # it survives sprint-worktree cleanup: # # ~/Projects/settled-reach/binaries/sr-voice-rocm (AMD / ROCm) # ~/Projects/settled-reach/binaries/sr-voice-cuda (NVIDIA, future) # ~/Projects/settled-reach/binaries/sr-voice-vulkan (cross-vendor, future) # ~/Projects/settled-reach/binaries/sr-voice-cpu (fallback) # # See #850 for the multi-backend release-binary ticket. The container # used to build these is created via the distrobox recipe documented in # the sr-voice README. # # The Gemma 2 model weights live under main/server/models and are shared # across worktrees (too large to duplicate). HOME_PROJECTS = Path.home() / "Projects" / "settled-reach" BINARIES_DIR = HOME_PROJECTS / "binaries" MAIN_WORKDIR = Path("/var/mnt/data/projects/settled-reach/main") def _find_sr_voice() -> Path: """Resolve the default sr-voice binary path. Preference order: 1. $HOME/Projects/settled-reach/binaries/sr-voice-rocm — persistent across worktree lifetimes, the canonical dev location. 2. main workdir's target/release/sr-voice — legacy, for backward-compatibility with older layouts. """ rocm_bin = BINARIES_DIR / "sr-voice-rocm" if rocm_bin.exists(): return rocm_bin return MAIN_WORKDIR / "server" / "sr-voice" / "target" / "release" / "sr-voice" DEFAULT_SR_VOICE = _find_sr_voice() DEFAULT_MODEL = MAIN_WORKDIR / "server" / "models" / "gemma2.gguf" MOCK_STDIO = REPO_ROOT / "server" / "sr-voice" / "mock-stdio.sh" # --------------------------------------------------------------------------- # Tee logger — stdout + log file in one call # --------------------------------------------------------------------------- class Logger: """Write lines to stdout AND an optional log file. Every message gets a prefix of the form `[HH:MM:SS +00h03m]`: - HH:MM:SS is wall-clock local time, - +NNhMMm is the elapsed time since the Logger was constructed. The elapsed offset tells the user at a glance how long the run has been going without scrolling back to the banner line. Flushes after every line so a kill -9 loses at most one entry. """ def __init__(self, log_path: Path | None): self.log_path = log_path self.started_at = time.monotonic() self.fh = None if log_path is not None: log_path.parent.mkdir(parents=True, exist_ok=True) # Truncate on open so each run starts fresh — the user can # rename an old log before kicking off the next run. self.fh = log_path.open("w", buffering=1) # line buffered def _elapsed(self) -> str: secs = int(time.monotonic() - self.started_at) return f"+{secs // 3600:02d}h{(secs % 3600) // 60:02d}m" def _prefix(self) -> str: clock = datetime.datetime.now().strftime("%H:%M:%S") return f"[{clock} {self._elapsed()}]" def __call__(self, msg: str = "") -> None: line = f"{self._prefix()} {msg}" if msg else "" print(line, flush=True) if self.fh is not None: self.fh.write(line + "\n") self.fh.flush() def raw(self, msg: str = "") -> None: """Print without the timestamp prefix (for banner lines).""" print(msg, flush=True) if self.fh is not None: self.fh.write(msg + "\n") self.fh.flush() def close(self) -> None: if self.fh is not None: self.fh.close() self.fh = None # --------------------------------------------------------------------------- # Corridor palettes (D-191 §4, glossary.md §Corridors, decisions/economics.md D-175) # --------------------------------------------------------------------------- # Each palette is the cultural inflection the prompt asks Gemma to # produce names in. Palette keys match the values of # `star_systems.geographic_sector` directly — that column is the real # source of corridor identity in systems.db (cultural_corridor is a # legacy field that was never populated beyond sol-gateway-axis). CORRIDOR_SUBSTYLES: dict[str, list[dict[str, str]]] = { # Each corridor has a list of sub-style inflections. The pipeline # picks one per body via hash(body_id) so neighbouring bodies on the # same planet get different registers, and Gemma's narrow per-register # vocabulary (~15 stems) stays fresh across hundreds of bodies. # # IMPORTANT: the inflection is a DOMINANT bias, not a hard lock. # A British surveyor on an east_reach moon still names a river after # their aunt in Dorset. Each sub-style explicitly names its register # AND invites diaspora variety. "core": [ {"inflection": "English countryside, rural, agricultural settlers", "examples": "Thornbury, Bramblewood, Millbrook, Ashford, Weston"}, {"inflection": "British colonial settlement era", "examples": "New Bristol, Port Augusta, Kingstown, Admiralty, Georgetown"}, {"inflection": "American frontier, practical, geographic", "examples": "Dusty Creek, Twin Oaks, Cedar Flat, Hawk's Hollow, Red Bluff"}, {"inflection": "American municipal, administrative, cosmopolitan", "examples": "Prospect Heights, Liberty, Union, Meridian, Commonwealth"}, {"inflection": "Classical references, institutional, civic", "examples": "Concordia, Aurelius, Prefecture, Senate Landing, Forum"}, {"inflection": "Australian and New Zealand settler", "examples": "Redfern, Glenelg, Wollongong, Kaikoura, Hawke's Bay"}, ], "north_reach": [ {"inflection": "English rural, village and parish names", "examples": "Wolcott, Mildern, Ashbourne, Briarfell, Tarndale"}, {"inflection": "Scottish Highland and Lowland place-names", "examples": "Glenmoray, Dunfermline, Kinross, Brae, Dalwhinnie"}, {"inflection": "Australian outback, station and property names", "examples": "Redfern, Birdsville, Tennant, Woomera, Coober"}, {"inflection": "Irish rural and coastal settlement", "examples": "Ballymore, Kilrush, Dunmore, Tralee, Skellig"}, {"inflection": "South African English settler", "examples": "Grahamstown, Oudtshoorn, Stellenbosch, Graaff, Beaufort"}, ], "south_reach": [ {"inflection": "Portuguese colonial era, Iberian", "examples": "Monteforte, Serra, Tavira, Oliveira, Porto Novo"}, {"inflection": "Brazilian interior, frontier settlement", "examples": "Ribeirão, Campo Largo, Várzea, Ilhabela, Pinheiro"}, {"inflection": "East African Swahili coastal", "examples": "Inhambane, Kilimi, Ngola, Manhica, Quelimane"}, {"inflection": "Cape Verdean and West African", "examples": "Cabo, Moçambo, Ribeira, Mindelo, Tarrafal"}, {"inflection": "Angolan and Mozambican settlement", "examples": "Huambo, Lobito, Nampula, Lichinga, Benguela"}, ], "east_reach": [ {"inflection": "Korean place-name tradition", "examples": "Hanyang, Seorak, Baektu, Saeyeon, Taegong"}, {"inflection": "Japanese rural and coastal settlement", "examples": "Takamine, Ginoza, Tsukuri, Aomori, Fukagawa"}, {"inflection": "Taiwanese and Hakka settler", "examples": "Jiufen, Beigang, Hsinchu, Meinong, Tainan"}, {"inflection": "Filipino settler community", "examples": "Batangas, Legazpi, Tuguegarao, Zambales, Tarlac"}, {"inflection": "Mixed East Asian diaspora, cosmopolitan", "examples": "Naruhan, Morimine, Kōzan, Midori, Kawasaki"}, ], "west_reach": [ {"inflection": "German settlement, orderly and compound names", "examples": "Altdorf, Drachenberg, Feldberg, Krakenberg, Lüneborg"}, {"inflection": "Dutch colonial, low-country", "examples": "Kloosterdam, Hoogland, Oudewater, Nieuwpoort, Voorhout"}, {"inflection": "Nordic and Scandinavian", "examples": "Sørholm, Torsfell, Bergfjord, Lindeborg, Nordhölm"}, {"inflection": "Polish and Czech settler", "examples": "Krakowice, Bystrica, Wieliczka, Tarnów, Ostrava"}, {"inflection": "Baltic and Finnish settler", "examples": "Järvenpää, Tallinna, Pärnu, Turku, Rakvere"}, ], "deep_frontier": [ {"inflection": "frontier founder-name era, surname-first, any Earth culture", "examples": "Okafor Reach, Stenner Cross, Weller Hold, Pruitt Basin"}, {"inflection": "frontier descriptive, geographic features named by surveyors", "examples": "Red Mesa, Dry Fork, Iron Flat, Long Ridge, Dust Basin"}, {"inflection": "frontier outpost, functional and military", "examples": "Forward Post, Relay Station, Survey Camp, Waypoint, Anchor"}, ], } # Legacy aliases CORRIDOR_SUBSTYLES["sol-gateway-axis"] = CORRIDOR_SUBSTYLES["core"] CORRIDOR_SUBSTYLES["inner_corridor"] = CORRIDOR_SUBSTYLES["core"] CORRIDOR_SUBSTYLES["inner_orbit"] = CORRIDOR_SUBSTYLES["core"] CORRIDOR_SUBSTYLES["frontier"] = CORRIDOR_SUBSTYLES["deep_frontier"] DEFAULT_SUBSTYLES = CORRIDOR_SUBSTYLES["core"] # Processing order for the main run — core first so those bodies win # the dedup race and the outer sectors fall into the palette fallback # path when names collide. SECTOR_PRIORITY: dict[str, int] = { "core": 0, "north_reach": 1, "south_reach": 2, "east_reach": 3, "west_reach": 4, "deep_frontier": 5, } def palette_for(corridor: str | None, system_id: str = "") -> dict[str, str]: """Pick a sub-style for this system within its corridor. All bodies in the same system get the same sub-style (consistent cultural register per star system). Different systems rotate through the sub-style list via hash(system_id). """ substyles = CORRIDOR_SUBSTYLES.get(corridor or "core", DEFAULT_SUBSTYLES) idx = int(hashlib.sha256(system_id.encode()).hexdigest()[:8], 16) % len(substyles) return substyles[idx] # --------------------------------------------------------------------------- # Feature prompt templates — few-shot format with rotating example pools # --------------------------------------------------------------------------- # Gemma 2 2B is small and noisy on free-form instruction prompts — it # loves to echo the prompt back as "[River Name]" / "NAME: ..." / # "**River: X**" etc. The fix is few-shot: show concrete # `Style → Answer` examples so the model completes a pattern instead # of generating to an open-ended instruction. # # Two lessons from earlier iterations: # 1. The examples are the ONLY thing the model actually learns from. # If they're all epic/classical (Wolcott Beck, Nakamura Stream), # the model completes in epic/classical register for every body. # Grounded outputs (Cooper's Creek, West Ridge, Mill Run) require # grounded examples. # 2. A single static example set produces uniform output: same prompt # + similar seeds → similar completions. Rotating through a pool # of example sets per call injects variation and nudges the # sampler into different regions of the output distribution. # # Each feature type has a POOL of example sets. `_build_prompt()` picks # one set deterministically per (body_id, local_id) so the same feature # always gets the same prompt but neighbouring features get different # prompts. The pools emphasise grounded / first-person / prosaic names # with the occasional classical one — matching how real settlers on # frontier worlds actually named places. # # Preamble wording matters too: "Settlers name …" reminds the model that # these are human-chosen names, not fantasy coinages. The negative # constraint "Most names are mundane" reinforces the grounded bias. # Each pool entry is a list of `(style_label, example_name)` pairs. The # style labels are cross-corridor — they teach Gemma the pattern, not # a specific corridor's vocabulary. The target corridor's inflection # gets substituted at the END of the prompt. _RIVER_POOLS: list[list[tuple[str, str]]] = [ # Pool 0 — possessive, surnames dominant with a first-name mixed in [ ("British/Australian", "Cooper's Creek"), ("Irish", "Maura's Run"), # first name ("Dutch", "Van Dael's Beek"), ("Italian", "Fiume Bruno"), ("Japanese", "Tanaka Stream"), ("Polish", "Kowalski Potok"), ], # Pool 1 — compass / descriptive, mixed cultures [ ("Australian", "West Brook"), ("Nordic", "Nordälven"), ("French", "Ruisseau du Nord"), ("Japanese", "Kita-gawa"), ("Swahili", "Mto wa Kaskazini"), ("Hungarian", "Északi Patak"), ], # Pool 2 — colour / feature observation [ ("Irish", "Blackwater"), ("German", "Braunbach"), ("Spanish", "Río Verde"), ("Russian", "Chornaya Rechka"), ("Korean", "Ha-gang"), ("Portuguese", "Ribeira Negra"), ], # Pool 3 — short single-word / old-world prosaic [ ("British", "Mill Run"), ("Dutch", "Oude Wetering"), ("Nordic", "Stenbäck"), ("Japanese", "Sakura-gawa"), ("Portuguese", "Ribeiro Seco"), ("Czech", "Starý Potok"), ], # Pool 4 — founder surname + feature [ ("British/Australian", "Garner Creek"), ("Dutch", "Meijer Beek"), ("Nordic", "Sveinsström"), ("Korean", "Choi Stream"), ("Italian", "Fiume Marconi"), ("Greek", "Petrakis Rema"), ], # Pool 6 — founder FIRST name possessive (Clifford's Bay shape) # Added so first-name-possessive naming joins the rotation alongside # the surname pools without replacing any of them. [ ("British", "Clifford's Bay"), ("Irish", "Maura's Run"), ("Japanese", "Yuki's Pool"), ("Italian", "Rio di Marco"), ("Portuguese", "Rio de Ana"), ("French", "Rivière d'Elena"), ], # Pool 5 — classical / institutional / Latinate (occasional ~17%) [ ("British/Australian", "Aqueduct Run"), ("Italian", "Acqua Vetusta"), ("Spanish", "Río Antiguo"), ("French", "Vieille Rivière"), ("German", "Altwasser"), ], ] _MOUNTAIN_POOLS: list[list[tuple[str, str]]] = [ # Pool 0 — compass / direct observation (the "Western Ridge" shape) [ ("British/Australian", "Western Ridge"), ("Dutch", "Noordrug"), ("Nordic", "Sørkammen"), ("Japanese", "Minami-yama"), ("Portuguese", "Serra do Sul"), ("Hungarian", "Északi Hát"), ], # Pool 1 — surname + feature, cosmopolitan [ ("British/Australian", "Drayton Hills"), ("Italian", "Monti Rovere"), ("Dutch", "Van Dijk Heuvels"), ("Korean", "Park Sanmaek"), ("Polish", "Góry Brzeskie"), ("French", "Crête Valmont"), ], # Pool 2 — colour / shape descriptor [ ("British", "The Long Spine"), ("Japanese", "Shiro-yama"), ("Russian", "Bely Khrebet"), ("Portuguese", "Serra Branca"), ("German", "Blauberg"), ("Spanish", "Sierra Roja"), ], # Pool 3 — short single-word [ ("British", "Fell Back"), ("Japanese", "Takamine"), ("Dutch", "Klipfjord"), ("Nordic", "Torsfell"), ("Portuguese", "Cabeço"), ("German", "Eichfels"), ], # Pool 4 — something-the-settlers-said (The-word / definite-article) [ ("British", "The Backbone"), ("Spanish", "El Espinazo"), ("Italian", "La Schiena"), ("Russian", "Khrebet"), ("Portuguese", "O Dorso"), ("French", "L'Épine"), ], # Pool 5 — classical / institutional / Latinate (occasional ~17%) [ ("British/Australian", "Cassian Range"), ("Italian", "Monti Augusti"), ("Portuguese", "Monte Augusto"), ("Latin", "Mons Cassianus"), ("French", "Massif Aurélien"), ], # Pool 6 — founder FIRST name + feature [ ("British", "Clifford's Ridge"), ("Irish", "Maeve's Back"), ("Japanese", "Keiko's Peak"), ("Spanish", "Sierra de Elena"), ("French", "Crête de Pierre"), ("Russian", "Anushka Khrebet"), ], ] _LAKE_POOLS: list[list[tuple[str, str]]] = [ [ ("British", "Cold Tarn"), ("Dutch", "Winterplas"), ("Nordic", "Kalltjärn"), ("Japanese", "Shizuko"), ("Portuguese", "Lagoa Funda"), ("Finnish", "Kylmäjärvi"), ], [ ("British/Australian", "Mildern Mere"), ("Italian", "Lago d'Argento"), ("Japanese", "Aoike"), ("Polish", "Jezioro Srebrne"), ("German", "Bergsee"), ("French", "Lac Clair"), ], [ ("British", "Three Oaks Pool"), ("Dutch", "Driehoekplas"), ("Japanese", "Midori-ike"), ("Hungarian", "Három Tölgy Tava"), ("Portuguese", "Poça Grande"), ("Spanish", "Laguna Grande"), ], ] _OCEAN_POOLS: list[list[tuple[str, str]]] = [ [ ("British", "Tarnsea"), ("Japanese", "Aomi"), ("Nordic", "Nordhav"), ("Portuguese", "Mar do Sul"), ("Dutch", "Zuidzee"), ("Italian", "Mare Meridio"), ], [ ("British", "The Long Main"), ("Japanese", "Kuro-umi"), ("Nordic", "Stormsø"), ("Portuguese", "Mar Profundo"), ("Russian", "Bolshoye More"), ("French", "Grand Large"), ], ] _SEA_POOLS: list[list[tuple[str, str]]] = [ [ ("British", "Harven Sea"), ("Japanese", "Minami-kai"), ("Nordic", "Sønderhav"), ("Portuguese", "Mar de Quelim"), ("Italian", "Mare Toscano"), ("Dutch", "Zeebocht"), ], [ ("British", "Cold Gulf"), ("Japanese", "Nagi-kai"), ("Nordic", "Iskullfjord"), ("Portuguese", "Golfo das Ilhas"), ("Polish", "Zatoka Zimna"), ("German", "Tiefbucht"), ], ] _CITY_CAPITAL_POOLS: list[list[tuple[str, str]]] = [ # Pool 0 — founder / homestead / surname-town [ ("British/Australian", "Holmwood"), ("Dutch", "Van Damhoeve"), ("Japanese", "Yuna"), ("Italian", "Borgo Marconi"), ("Polish", "Kowalowo"), ("Portuguese", "Vila Moreira"), ], # Pool 1 — compass + old-country place name [ ("British", "Westfield"), ("Nordic", "Sørholm"), ("Japanese", "Kita-sato"), ("French", "Saint-Nord"), ("Hungarian", "Kelethegy"), ("German", "Südkamp"), ], # Pool 2 — short rooted stem (farm / town / kiln etc) [ ("British", "Kiln"), ("Dutch", "Stenen"), ("Japanese", "Sora"), ("Portuguese", "Paço"), ("Italian", "Forno"), ("Czech", "Starovice"), ], # Pool 3 — explicitly mundane / functional [ ("British", "Landing"), ("Nordic", "Brygga"), ("Japanese", "Habu"), ("Portuguese", "Cabo"), ("Dutch", "Haven"), ("French", "Débarquement"), ], # Pool 4 — classical / institutional / Latinate (occasional ~20%) [ ("British/Australian", "Meridian"), ("Italian", "Augusta"), ("Portuguese", "Porto Imperial"), ("Latin", "Solarium"), ("French", "Saint-Aurélien"), ], # Pool 5 — founder FIRST name settlement [ ("British", "Clifford's Landing"), ("Irish", "Maura's Cross"), ("Japanese", "Yuki-mura"), ("Italian", "Villa di Marco"), ("Portuguese", "Vila Helena"), ("French", "Chez Pierre"), ], ] _CITY_SECONDARY_POOLS: list[list[tuple[str, str]]] = [ [ ("British/Australian", "Carberry"), ("Korean/Japanese", "Yurigawa"), ("Dutch", "Kleindorp"), ("Italian", "Piccola Villa"), ("Polish", "Nowawieś"), ("Portuguese", "Ribeirão"), ], [ ("British/Australian", "Garner's Cross"), ("French", "Sainte-Marie"), ("Japanese", "Tanaka-no-mura"), ("Nordic", "Sveinsby"), ("Hungarian", "Kiskút"), ("Portuguese", "Vila Nova"), ], [ ("British", "Kelstern"), ("Japanese", "Shirakawa"), ("Dutch", "Hoogland"), ("Czech", "Starovice"), ("Spanish", "Alta Vista"), ("German", "Talhöhe"), ], [ ("British", "Mill End"), ("Japanese", "Shimo-machi"), ("Nordic", "Nedreby"), ("French", "Les Moulins"), ("Portuguese", "Marginal"), ("Italian", "Fondobasso"), ], # Classical / Latinate (occasional) [ ("British", "Prospect"), ("Japanese", "Seishin"), ("Italian", "Porta Aurea"), ("Portuguese", "Pórtico"), ("French", "Consulat"), ], # Founder FIRST-name settlements [ ("British", "Clifford's Ferry"), ("Irish", "Maeve's Quay"), ("Japanese", "Yuki-no-mura"), ("Italian", "Casa Elena"), ("Portuguese", "Vila de Ana"), ("French", "Saint-Martin"), ], ] _POI_TRANSIT_POOLS: list[list[tuple[str, str]]] = [ [ ("British", "Holmwood Gate Terminal"), ("Japanese", "Yurigawa Transit"), ("Dutch", "Noordpoort Gate Terminal"), ("Portuguese", "Porto Exchange"), ("French", "Gare du Nord Concourse"), ], [ ("British", "West Gate Terminal"), ("Nordic", "Brygga Transit"), ("Italian", "Porta Vecchia"), ("Japanese", "Kita-sato Transit"), ("German", "Steinhof Gate Terminal"), ], ] _POI_INSTITUTIONAL_POOLS: list[list[tuple[str, str]]] = [ [ ("British", "Holmwood Assembly Hall"), ("Japanese", "Seungmun Archive"), ("Italian", "Palazzo Civico"), ("Portuguese", "Câmara Municipal"), ("German", "Altes Rathaus"), ], [ ("British", "Founders' Registry"), ("French", "Registre Général"), ("Japanese", "Kō Records Office"), ("Polish", "Archiwum Miejskie"), ("Dutch", "Burgerhuis"), ], ] _POI_CULTURAL_POOLS: list[list[tuple[str, str]]] = [ [ ("British", "The Commons"), ("Japanese", "Yurigawa Grounds"), ("Italian", "Piazza Nuova"), ("Portuguese", "Praça do Vento"), ("German", "Marktplatz"), ], [ ("British", "The Meeting House"), ("French", "Place des Fondateurs"), ("Japanese", "Sakura Grounds"), ("Polish", "Rynek Stary"), ("Nordic", "Gamle Torget"), ], ] # Map feature type → pools + preamble + length hint + extra context hook. _PROMPT_CONFIG: dict[str, dict] = { "river": { "pools": _RIVER_POOLS, "subject": "rivers", "length_hint": "1-3 words", }, "ocean": { "pools": _OCEAN_POOLS, "subject": "oceans", "length_hint": "1-3 words", }, "sea": { "pools": _SEA_POOLS, "subject": "seas", "length_hint": "1-3 words", }, "lake": { "pools": _LAKE_POOLS, "subject": "lakes", "length_hint": "1-3 words", }, "mountain_range": { "pools": _MOUNTAIN_POOLS, "subject": "mountain ranges", "length_hint": "1-3 words", }, "city_capital": { "pools": _CITY_CAPITAL_POOLS, "subject": "their capital town", "length_hint": "1-2 words", "include_planet": True, }, "city_secondary": { "pools": _CITY_SECONDARY_POOLS, "subject": "their secondary towns", "length_hint": "1-2 words", "include_planet": True, }, "poi_transit": { "pools": _POI_TRANSIT_POOLS, "subject": "gate terminals / transit hubs", "length_hint": "2-3 words ending in 'Gate Terminal', 'Transit', " "'Exchange', or 'Concourse'", }, "poi_institutional": { "pools": _POI_INSTITUTIONAL_POOLS, "subject": "institutional landmarks", "length_hint": "2-4 words", }, "poi_cultural": { "pools": _POI_CULTURAL_POOLS, "subject": "cultural landmarks", "length_hint": "2-4 words", }, } def _build_prompt( feature_type: str, inflection: str, planet_class: str, body_id: str, local_id: str, attempt: int, system_hook: str | None = None, system_name: str | None = None, body_name: str | None = None, ) -> str: """Assemble a few-shot prompt for the given feature type. The example pool rotates per call via a deterministic hash of (body_id, local_id, attempt) — this puts a finger on the sampling scales so neighbouring features on the same body don't all draw from an identical prompt and collapse to identical outputs. `system_hook` is the pre-extracted gttr.md characterisation for this system (e.g. "RAN is, by all accounts, a very nice place to live, provided you are the sort of person whose grandparents owned the land"). When present it gets woven into the preamble so Gemma has per-system cultural context — this is the single biggest lever for producing names that feel like they belong to THIS world rather than any world in this corridor. When absent the prompt degrades to the corridor inflection alone. """ cfg = _PROMPT_CONFIG.get(feature_type) if cfg is None: return "" pools: list[list[tuple[str, str]]] = cfg["pools"] # Deterministic pool pick: same feature always hits the same pool on # attempt 0; retries rotate forward so a rejected name gets a # different example set, not just a different seed. salt = int( hashlib.sha256(f"{body_id}|{local_id}|{attempt}".encode()).hexdigest()[:8], 16, ) pool = pools[salt % len(pools)] subject = cfg["subject"] # Use "named X" for collective/plural subjects, "called X" for # singular possessive ones ("their capital town"). Heuristic: if the # subject starts with "their", use "called"; otherwise "named". verb = "called" if subject.startswith("their ") else "named" # Context block. When system / body proper names are available, # label them explicitly so Gemma has the canonical identifiers in # addition to the gttr hook. Redundancy helps on a small model — # the hook already opens with the system name but putting it in a # labeled "System:" line makes it harder to gloss over. When the # gttr hook exists too, it goes on its own line as the cultural # one-liner. context_lines: list[str] = [] ident_parts = [] if system_name: ident_parts.append(f"System: {system_name}") if body_name: ident_parts.append(f"Planet: {body_name}") if ident_parts: context_lines.append(". ".join(ident_parts) + ".") if system_hook: context_lines.append(f"About the system: {system_hook}") preamble = ( f"Settlers {verb} {subject} after themselves, after what they saw, " f"or after places back home. Most names are mundane, short, and " f"direct — a surname, a compass direction, a feature, a practical " f"description. Classical or epic names are rare. " f"Reply with ONLY the name, {cfg['length_hint']}, no brackets, " f"no quotes, no markdown, no label." ) lines = [preamble, ""] if context_lines: lines.extend(context_lines) lines.append("") for style, example in pool: lines.append(f"Style: {style}. Answer: {example}") lines.append("") tail = f"Style: {inflection}." if cfg.get("include_planet"): tail += f" Planet: {planet_class}." tail += " Answer:" lines.append(tail) return "\n".join(lines) # --------------------------------------------------------------------------- # Blocklist # --------------------------------------------------------------------------- def load_blocklist(path: Path = BLOCKLIST_PATH) -> set[str]: """Load earth_blocklist.txt into a lowercase set for exact-match checks.""" if not path.exists(): return set() blocked: set[str] = set() for line in path.read_text().splitlines(): line = line.strip() if not line or line.startswith("#"): continue blocked.add(line.lower()) return blocked _THE_PREFIX = re.compile(r"^the\s+", re.IGNORECASE) def is_blocked(name: str, blocklist: set[str]) -> bool: """True iff the case-insensitive name appears in the blocklist. Also strips a leading "The " before comparing, so "The Great Divide" and "Great Divide" both match a `great divide` entry. Prefixed variants like 'Nouveau Paris', 'Neu Berlin', 'New Tokyo' still do NOT match because those prefixes are substantive (a new place), while "The" is just the definite article. """ lc = name.strip().lower() if lc in blocklist: return True stripped = _THE_PREFIX.sub("", lc).strip() return stripped in blocklist def is_placeholder(name: str) -> bool: """True iff the cleaned name looks like Gemma echoed the prompt or drifted into verbose poetry rather than producing a usable name. Catches: - Empty, <3 chars, or pure punctuation after cleaning. - Any whole-word token from _PLACEHOLDER_TOKENS appearing in the name ('Name', 'Example', 'Placeholder', 'TBD', …). - Strings that start with 'River ', 'City ', 'Lake ', etc. with a single-letter suffix (clear prompt fragments). - 5+ word outputs — the prompt requests 1-3 words; when Gemma drifts into 'The Grand Lake of the Astral Sea' territory it's generating description, not a name. 4 words is the practical upper bound for clean Latin/Korean/Germanic 3-token names with a 'The' prefix. """ s = name.strip() if len(s) < 3: return True if not re.search(r"[A-Za-z]", s): return True lowered = s.lower() tokens = re.findall(r"[a-z]+", lowered) if any(tok in _PLACEHOLDER_TOKENS for tok in tokens): return True if len(tokens) >= 5: return True if re.fullmatch( r"(river|lake|ocean|sea|city|capital|town|mountain|range|peak|poi|" r"gate|terminal)\s+[a-z]", lowered ): return True return False # --------------------------------------------------------------------------- # Post-processing # --------------------------------------------------------------------------- _TRAILING_PUNCT = re.compile(r"[\s\.,;:!?\"'`\-]+$") _LEADING_PUNCT = re.compile(r"^[\s\.,;:!?\"'`\-]+") # Strip surrounding square brackets the model likes to emit # (`[River Name]`, `[Lake Foo]`, `[Optional: Bar]`). _BRACKET_WRAP = re.compile(r"^\[\s*(.*?)\s*\]$") # Tokens that indicate the model failed the instruction and echoed the # prompt back. If any of these appear as whole words (case-insensitive) # in the cleaned output, reject the whole name and retry. _PLACEHOLDER_TOKENS = { "name", "names", "placeholder", "example", "example1", "exampleone", "optional", "tbd", "todo", } # Labels Gemma likes to prepend to the answer. Matched case-insensitively # at the start of the line, with optional whitespace and a `:` or `-`. _LABEL_PREFIX = re.compile( r"^(name|answer|output|result|response|city|town|capital|village|river|" r"stream|ocean|sea|lake|bay|gulf|mountain|range|peak|ridge|spine|poi|" r"landmark|terminal|gate|optional|alternative|alt|suggestion|" r"example|note)\s*[:\-]\s*", re.IGNORECASE, ) # Strip markdown bold/italic wrappers (`**X**`, `*X*`, `__X__`, `_X_`) # that Gemma sometimes emits even when told "no quotes or explanation". _MD_BOLD = re.compile(r"^\*+\s*(.*?)\s*\*+$") _MD_UNDER = re.compile(r"^_+\s*(.*?)\s*_+$") def post_process(raw: str) -> str: """Extract a clean single-line name from Gemma's raw output. Handles, in order: - First non-empty line only (Gemma often continues with an explanation). - Strip surrounding markdown bold/italic (`**X**`, `*X*`, `__X__`, `_X_`). - Strip label prefixes the model likes to prepend (`Name:`, `City:`, `River:`, `Ocean:`, `Mountain:`, etc.) — case-insensitive, with or without a `:` or `-` separator. - Strip surrounding quotes / punctuation. - Collapse internal whitespace. - Truncate at 40 chars as a hard safety limit. The result is the canonical form used for both writing to disk AND dedup comparison, so "River: Aureus" and a later raw "Aureus" normalize to the same string and collide as intended. """ text = (raw or "").strip() if not text: return "" # First non-empty line only. for line in text.splitlines(): line = line.strip() if line: text = line break else: return "" # Strip surrounding markdown bold/italic, square brackets, label # prefixes, and stray asterisks/underscores in alternation until the # text stops shrinking. This handles every combination: `**Foo**`, # `**City: Foo**`, `City: **Foo**`, `**City:** Foo`, `_Name: Bar_`, # `[River Foo]`, `[City: Bar]`, etc. while True: before = text m = _MD_BOLD.match(text) or _MD_UNDER.match(text) or _BRACKET_WRAP.match(text) if m: text = m.group(1).strip() text = text.strip("*_ \t-[]") new = _LABEL_PREFIX.sub("", text) if new != text: text = new.strip() if text == before: break # Strip surrounding quotes / punctuation. text = _LEADING_PUNCT.sub("", text) text = _TRAILING_PUNCT.sub("", text) # Collapse internal whitespace. text = re.sub(r"\s+", " ", text).strip() # Hard length safety. if len(text) > 40: text = text[:40].rstrip() return text # --------------------------------------------------------------------------- # Fallback palette generator (deterministic, no LLM) # --------------------------------------------------------------------------- # Fallback stems — 40+ per corridor drawn from grounded, cosmopolitan # place-name roots. Each corridor now spans multiple cultural registers # so fallback names inherit the same diversity the few-shot pools teach. # The old 10-stem Latin-institutional lists produced very same-y # "Concord Ridge / Axis Spine / Senatus Heights" clusters when dense # core bodies exhausted the LLM's unique output budget. # # Stems are deliberately a mix of real surnames, real place-name # fragments, and administrative English — because settlers named # places that way. _FALLBACK_STEMS: dict[str, list[str]] = { "north_reach": [ # British / Australian / Irish place-name stems "Ashford", "Bellfield", "Briarcroft", "Clifford", "Drayton", "Elmhurst", "Fairfield", "Garner", "Holmwood", "Irvington", "Kelsworth", "Larkfield", "Mildern", "Netherbrook", "Oakdale", "Pendleton", "Redfern", "Standen", "Tarndale", "Upwold", "Waltham", "Wolcott", "Ashbourne", "Briarfell", "Harford", "Kelston", "Pennhowe", "Rowanmoor", "Threlwood", "Carberry", "Glendale", "Hollowmere", "Ingledene", "Kirkstone", "Maybourne", "Newbridge", "Pinegrove", "Ridgemont", "Silverpine", "Tallgrass", ], "south_reach": [ # Portuguese / Brazilian / Swahili / Cape Verdean stems "Alves", "Brandão", "Cabral", "Dantas", "Évora", "Figueira", "Gomes", "Hlanganani", "Inhambane", "Jabali", "Kilimi", "Lourenço", "Manhica", "Natal", "Oliveira", "Pinheiro", "Quelimane", "Ribeiro", "Serra", "Tavira", "Vargas", "Monteforte", "Serravale", "Cabo", "Ilhabela", "Moçambo", "Ngola", "Porto", "Ribeira", "Costa", "Marinha", "Várzea", "Campo", "Outeiro", "Pedregal", "Telhado", ], "east_reach": [ # Korean / Japanese / Taiwanese stems "Aomori", "Baektu", "Chōshi", "Daisen", "Etajima", "Fukagawa", "Gogen", "Hanyang", "Izumi", "Junan", "Kita", "Mori", "Nagare", "Okami", "Saeyeon", "Takamine", "Unari", "Yurigawa", "Ginoza", "Naruhan", "Morimine", "Taegong", "Tsukuri", "Seorak", "Shiroyama", "Kōzan", "Hirano", "Nakayama", "Midori", "Kawasaki", "Tsukushi", "Higashi", "Nishimura", "Miyazaki", "Tachibana", ], "west_reach": [ # German / Dutch / Nordic / Polish / Czech stems "Altdorf", "Bergfjord", "Drachenberg", "Eikhof", "Feldberg", "Hellekade", "Hoogland", "Kloosterdam", "Lindeborg", "Nordhölm", "Oudewater", "Sørholm", "Straend", "Torsfell", "Vanebach", "Viskrans", "Aalburg", "Blankhof", "Dalhem", "Grunendal", "Halvorsen", "Järvenpää", "Krakenberg", "Lüneborg", "Meinhof", "Nieuwpoort", "Osterdal", "Reinwoud", "Svarteberg", "Tjornholm", "Urlinden", "Voorhout", "Weserhof", "Östby", "Brabanthof", ], "core": [ # Administrative English / Gateway-era, with a side of classical # for the occasional institutional feel "Ashfield", "Bellview", "Cedarbrook", "Concord", "Crestline", "Easthill", "Fairmont", "Foxgrove", "Gateway", "Glenbrook", "Haversham", "Holloway", "Ironwood", "Kirkwood", "Lakeside", "Libertyhill", "Linden", "Marchwood", "Meridian", "Midvale", "Newbridge", "Northfield", "Oakdale", "Pinehurst", "Riverside", "Rosemont", "Southgate", "Standen", "Stonewall", "Sunnyvale", "Tallwood", "Union", "Uplands", "Valewood", "Westbrook", "Willowgate", "Windmere", "Claremont", "Fernwood", "Hartley", "Cardinal", "Prefecture", "Senate", "Assembly", "Federal", ], # Aliases for legacy geographic_sector names "inner_corridor": [ "Ashfield", "Bellview", "Cedarbrook", "Concord", "Crestline", "Easthill", "Fairmont", "Foxgrove", "Gateway", "Glenbrook", "Meridian", "Midvale", "Northfield", "Oakdale", "Pinehurst", "Riverside", "Rosemont", "Southgate", "Uplands", "Westbrook", ], "inner_orbit": [ "Ashfield", "Bellview", "Cedarbrook", "Concord", "Crestline", "Easthill", "Fairmont", "Foxgrove", "Gateway", "Glenbrook", "Meridian", "Midvale", "Northfield", "Oakdale", "Pinehurst", "Riverside", "Rosemont", "Southgate", "Uplands", "Westbrook", ], "sol-gateway-axis": [ "Ashfield", "Bellview", "Cedarbrook", "Concord", "Crestline", "Easthill", "Fairmont", "Foxgrove", "Gateway", "Glenbrook", "Meridian", "Midvale", "Northfield", "Oakdale", "Pinehurst", "Riverside", "Rosemont", "Southgate", "Uplands", "Westbrook", ], "deep_frontier": [ "Okafor", "Stenner", "Weller", "Pruitt", "Hale", "Kettle", "Bowman", "Alder", "Risher", "Vickery", "Morgan", "Zhou", "Novak", "Lindström", "Barbosa", "Farrow", "Tillman", "Kellogg", "Grayson", "Whitby", "Pickering", "Cortland", "Deacon", "Huckle", "Stonebrook", "Clayridge", "Ironvale", "Redrock", "Blackwater", "Dustgate", ], "frontier": [ "Okafor", "Stenner", "Weller", "Pruitt", "Hale", "Kettle", "Bowman", "Alder", "Risher", "Vickery", ], } _FALLBACK_SUFFIXES: dict[str, list[str]] = { "river": ["Run", "Water", "Beck", "Rill", "Course", "Brook", "Stream", "Flow", "Creek"], "ocean": ["Sea", "Expanse", "Deep", "Reach", "Main", "Vast"], "sea": ["Sea", "Gulf", "Basin", "Bay", "Firth"], "lake": ["Lake", "Mere", "Tarn", "Pool", "Loch", "Pond"], "mountain_range": ["Range", "Ridge", "Spine", "Heights", "Scarp", "Hills", "Peaks", "Back", "Crest", "Fell"], "city_capital": ["Hold", "Prime", "Seat", "Court", "Landing", "Cross", "Hub"], "city_secondary": ["Cross", "Reach", "Hollow", "Fields", "Stand", "Ferry", "Mill", "End", "Gate"], "poi_transit": ["Gate Terminal", "Transit", "Concourse", "Exchange", "Junction"], "poi_institutional": ["Archive", "Hall", "Assembly", "Registry", "Council", "Office"], "poi_cultural": ["Commons", "Grounds", "Circle", "Square", "Park", "Plaza"], } def fallback_name(corridor: str, feature_type: str, salt: int) -> str: """Deterministic palette-driven fallback when the LLM can't produce a usable name after retries. Picks a stem + suffix using `salt` so the same (body, feature) always gets the same fallback.""" stems = _FALLBACK_STEMS.get(corridor) or _FALLBACK_STEMS["inner_corridor"] suffixes = _FALLBACK_SUFFIXES.get(feature_type, ["Place"]) stem = stems[salt % len(stems)] suffix = suffixes[(salt // max(1, len(stems))) % len(suffixes)] return f"{stem} {suffix}" # --------------------------------------------------------------------------- # Subprocess manager # --------------------------------------------------------------------------- class VoiceSubprocess: """Thin wrapper around sr-voice stdio mode. The subprocess holds a KV cache across requests — after N_REFRESH requests we tear it down and start a fresh one so earlier prompts don't pollute later ones (context bleed). Real model load takes a few seconds; mock mode is instantaneous. """ def __init__( self, sr_voice_bin: Path, model_path: Path | None, mock: bool, refresh_every: int, verbose: bool, distrobox: str | None = None, ): self.sr_voice_bin = sr_voice_bin self.model_path = model_path self.mock = mock self.refresh_every = max(1, refresh_every) self.verbose = verbose self.distrobox = distrobox self.proc: subprocess.Popen | None = None self.request_count = 0 self._start() def _build_cmd(self) -> list[str]: if self.mock: return [str(MOCK_STDIO), "serve", "--stdio"] inner_cmd = [str(self.sr_voice_bin), "serve", "--stdio"] if self.model_path is not None: inner_cmd += ["--model", str(self.model_path)] # If a distrobox container was requested, invoke the binary # inside the container. Needed when the built binary depends on # libs (e.g. libhipblas.so.2) that only exist in the container, # not on the host. `distrobox enter -- ` # forwards stdin/stdout through the container's podman exec # pipe, which is exactly what VoiceSubprocess needs — the # stdio JSONL protocol flows through unchanged. if self.distrobox: return ["distrobox", "enter", self.distrobox, "--"] + inner_cmd return inner_cmd def _start(self) -> None: cmd = self._build_cmd() if self.verbose: print(f" [subprocess] starting: {' '.join(cmd)}", flush=True) self.proc = subprocess.Popen( cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL if not self.verbose else None, text=True, bufsize=1, # line-buffered ) self.request_count = 0 # Mock prints a stderr banner synchronously; real sr-voice prints # stderr while loading the model. Either way we drive it request- # reply so no readiness probe is needed — the first request just # blocks until the model is ready. def request(self, prompt: str, seed: int) -> str: """Send one JSONL request, read one JSONL response, return raw text. On subprocess death, restart once and retry. On restart threshold, tear down and restart cleanly before the request. """ if self.request_count >= self.refresh_every: if self.verbose: print( f" [subprocess] refresh after {self.request_count} requests", flush=True, ) self._stop() self._start() req = json.dumps({"prompt": prompt, "seed": seed}) assert self.proc is not None and self.proc.stdin is not None and self.proc.stdout is not None try: self.proc.stdin.write(req + "\n") self.proc.stdin.flush() line = self.proc.stdout.readline() except (BrokenPipeError, OSError) as e: print(f" [subprocess] pipe broken ({e}) — restarting", flush=True) self._stop() self._start() self.proc.stdin.write(req + "\n") # type: ignore[union-attr] self.proc.stdin.flush() # type: ignore[union-attr] line = self.proc.stdout.readline() # type: ignore[union-attr] self.request_count += 1 if not line: raise RuntimeError("sr-voice returned empty response (subprocess died?)") try: resp = json.loads(line.strip()) except json.JSONDecodeError as e: raise RuntimeError(f"sr-voice returned non-JSON: {line!r} ({e})") from e if "error" in resp: raise RuntimeError(f"sr-voice error: {resp['error']}") return resp.get("text", "") or "" def _stop(self) -> None: if self.proc is None: return try: if self.proc.stdin: self.proc.stdin.close() except Exception: pass try: self.proc.terminate() self.proc.wait(timeout=5) except subprocess.TimeoutExpired: self.proc.kill() self.proc.wait() except Exception: pass self.proc = None def close(self) -> None: self._stop() def __enter__(self) -> "VoiceSubprocess": return self def __exit__(self, *exc) -> None: self.close() # --------------------------------------------------------------------------- # Naming (per-feature request + retry) # --------------------------------------------------------------------------- def _seed_for(world_seed: int, body_id: str, local_id: str, attempt: int) -> int: """Deterministic per-feature seed. Same (world, body, feature) always starts from the same seed; retries bump `attempt` to get a different sample path without losing determinism.""" h = hashlib.sha256( f"{world_seed}|{body_id}|{local_id}|{attempt}".encode() ).hexdigest() return int(h[:16], 16) def body_population_band(population: int) -> str: if population <= 0: return "uninhabited" if population < 100_000: return "outpost (<100k)" if population < 10_000_000: return "small (<10M)" if population < 100_000_000: return "medium (<100M)" if population < 1_000_000_000: return "large (<1B)" return "megaworld (1B+)" def name_feature( voice: VoiceSubprocess, feature_type: str, ctx: dict, blocklist: set[str], corpus: dict[tuple, set[str]], body_used: set[str], system_hook: str | None, system_id: str, world_seed: int, body_id: str, local_id: str, hop: int, log: "Logger", verbose: bool, max_attempts: int = 5, ) -> str: """Request a name from Gemma, enforce blocklist + per-system dedup + per-body cross-type dedup, skip on persistent failure. Dedup scopes: - `corpus[(system_id, feature_type)]` — no two features of the same type in the same system share a name. - `body_used` — per-body set across ALL feature types. Prevents the same name from appearing as a river AND an ocean AND a mountain range on the same world. """ corridor = ctx.get("cultural_corridor") or "core" if feature_type not in _PROMPT_CONFIG: return fallback_name(corridor, feature_type, _seed_for(world_seed, body_id, local_id, 0)) palette = palette_for(corridor, system_id) planet_class = ctx.get("planet_class") or "habitable" dedup_key = (system_id, feature_type) used = corpus.setdefault(dedup_key, set()) # Capture mode: build the attempt-0 prompt, log it, return a unique # placeholder. No LLM call, no validation gauntlet — we want the raw # prompt set as Gemma would first see it, before retries kick in. if _CAPTURE_FILE is not None: prompt = _build_prompt( feature_type, inflection=palette["inflection"], planet_class=planet_class, body_id=body_id, local_id=local_id, attempt=0, system_hook=system_hook, system_name=ctx.get("system_proper_name"), body_name=ctx.get("body_proper_name"), ) seed = _seed_for(world_seed, body_id, local_id, 0) _CAPTURE_FILE.write(json.dumps({ "body_id": body_id, "local_id": local_id, "feature_type": feature_type, "corridor": corridor, "seed": seed, "prompt": prompt, }) + "\n") _CAPTURE_FILE.flush() # Unique deterministic placeholder. Embeds the seed so every # call returns a distinct string — passes dedup. Two letters + # hex keep it short, no _PLACEHOLDER_TOKENS, no length issues. placeholder = f"Zq{seed:08x}" used.add(placeholder) body_used.add(placeholder) return placeholder def _is_duplicate(candidate: str) -> bool: lc = candidate.lower() return ( lc in (n.lower() for n in used) or lc in (n.lower() for n in body_used) ) def _commit_name(final: str) -> None: used.add(final) body_used.add(final) rejections: dict[str, int] = {"empty": 0, "placeholder": 0, "blocklist": 0, "dedup": 0, "error": 0} for attempt in range(max_attempts): seed = _seed_for(world_seed, body_id, local_id, attempt) prompt = _build_prompt( feature_type, inflection=palette["inflection"], planet_class=planet_class, body_id=body_id, local_id=local_id, attempt=attempt, system_hook=system_hook, system_name=ctx.get("system_proper_name"), body_name=ctx.get("body_proper_name"), ) try: raw = voice.request(prompt, seed) except RuntimeError as e: rejections["error"] += 1 log(f" error {body_id}/{local_id} attempt {attempt}: {e}") continue cleaned = post_process(raw) if not cleaned: rejections["empty"] += 1 log(f" empty {body_id}/{local_id} attempt {attempt}") continue if is_placeholder(cleaned): rejections["placeholder"] += 1 log(f" placeholder '{cleaned}' {body_id}/{local_id}") continue if is_blocked(cleaned, blocklist): rejections["blocklist"] += 1 log(f" blocklist '{cleaned}' {body_id}/{local_id}") continue if _is_duplicate(cleaned): rejections["dedup"] += 1 log(f" dedup '{cleaned}' {body_id}/{local_id}") continue _commit_name(cleaned) return cleaned # All attempts exhausted. Leave the name as None so the preserved # path skips it on a fill round — a second pass with a fresh corpus # will pick it up and try again with different dedup pressure. reasons = " ".join(f"{k}={v}" for k, v in rejections.items() if v > 0) log(f" skipped: {body_id}/{local_id} [{reasons}]") return None # --------------------------------------------------------------------------- # Body walker # --------------------------------------------------------------------------- def load_body_context(body_id: str, system_id: str, conn: sqlite3.Connection) -> dict: row = conn.execute( """ SELECT b.planet_class, b.settlement_pattern, COALESCE(b.cultural_corridor, s.cultural_corridor, s.geographic_sector), b.population, b.economic_role, b.proper_name, s.proper_name FROM bodies b JOIN star_systems s ON b.system_id = s.system_id WHERE b.body_id = ? """, (body_id,), ).fetchone() if row is None: return { "planet_class": None, "settlement_pattern": None, "cultural_corridor": None, "population": 0, "economic_role": None, "pop_band": "unknown", "body_proper_name": None, "system_proper_name": None, } (planet_class, settlement_pattern, corridor, population, economic_role, body_proper_name, system_proper_name) = row return { "planet_class": planet_class, "settlement_pattern": settlement_pattern, "cultural_corridor": corridor, "population": population or 0, "economic_role": economic_role, "pop_band": body_population_band(population or 0), "body_proper_name": body_proper_name, "system_proper_name": system_proper_name, } def load_system_gttr_hooks(conn: sqlite3.Connection) -> dict[str, str]: """Return `{system_id: gttr_hook}` for every system with a populated hook. The gttr_hook column is filled by `tooling/db/populate_gttr_hook.py` from wiki/star-systems//gttr.md and captures the system's canonical characterisation in 30-45 words ("where the rules live", "forty years old and still in the draft", …). The naming pipeline injects this into the prompt preamble so Gemma has per-system cultural context without having to parse thousands of lines of wiki markdown at runtime. Systems whose gttr_hook is NULL return nothing from this map — the caller falls back to the corridor palette alone. """ rows = conn.execute( "SELECT system_id, gttr_hook FROM star_systems " "WHERE gttr_hook IS NOT NULL AND gttr_hook != ''" ).fetchall() return {row[0]: row[1] for row in rows} def load_body_hop_order(conn: sqlite3.Connection) -> dict[str, tuple[int, str]]: """Return a `{body_id: (hop_distance_from_gateway, body_id)}` map used as a stable sort key so the pipeline walks the reach from core outward: Gateway (hop 0) first, then hop 1, hop 2, ... all the way to the deep frontier. Ordering core-first gives those bodies first crack at every unique Gemma output and lets outer sectors fall into the palette fallback when they lose the dedup race. """ rows = conn.execute( """ SELECT b.body_id, COALESCE(sg.hop_distance_from_gateway, 99) AS hop FROM bodies b LEFT JOIN system_gates sg ON b.system_id = sg.system_id """ ).fetchall() return {body_id: (hop, body_id) for body_id, hop in rows} def _is_blank(value) -> bool: return value is None or (isinstance(value, str) and value.strip() == "") def _feature_type_for_city(city: dict) -> str: return "city_capital" if city.get("kind") == "capital" else "city_secondary" def _feature_type_for_poi(poi: dict) -> str: kind = (poi.get("kind") or "").lower() if kind in ("transit", "gate_terminal"): return "poi_transit" if kind in ("institutional", "corporate", "government"): return "poi_institutional" if kind in ("cultural", "commercial", "heritage"): return "poi_cultural" return "poi_institutional" def _feature_type_for_ocean(water: dict) -> str: kind = (water.get("kind") or "ocean").lower() if kind == "lake": return "lake" if kind == "sea": return "sea" return "ocean" def process_body( body_id: str, system_id: str, markers_path: Path, voice: VoiceSubprocess, conn: sqlite3.Connection, blocklist: set[str], corpus: dict[tuple, set[str]], system_hook: str | None, world_seed: int, hop: int, log: "Logger", verbose: bool, ) -> dict: """Fill every empty name field in this body's markers.json. Returns a summary counts dict plus a `generated` dict mapping section name to the list of new names produced (for the main loop to log).""" try: markers = json.loads(markers_path.read_text()) except json.JSONDecodeError as e: return {"error": f"invalid JSON: {e}"} grid = markers.get("grid") or {} if grid.get("w") != GRID_W or grid.get("h") != GRID_H: return { "error": ( f"grid mismatch {grid} != {{'w': {GRID_W}, 'h': {GRID_H}}}" ) } ctx = load_body_context(body_id, system_id, conn) corridor = ctx.get("cultural_corridor") or "core" counts = { "cities": 0, "rivers": 0, "oceans": 0, "mountain_ranges": 0, "pois": 0, "preserved": 0, } generated: dict[str, list[str]] = { "cities": [], "rivers": [], "oceans": [], "mountain_ranges": [], "pois": [], } changed = False # Per-body dedup set — no name may appear twice on the same body, # even across feature types. Seeded with every hand-authored name # already present so templates (Lendel, Edict, Estrade, …) keep their body_used: set[str] = set() for key in ("cities", "rivers", "oceans", "mountain_ranges", "pois"): for feat in markers.get(key) or []: name = feat.get("name") if name and isinstance(name, str) and name.strip(): body_used.add(name.strip()) # Cities for city in markers.get("cities") or []: if not _is_blank(city.get("name")): corpus.setdefault((system_id, _feature_type_for_city(city)), set()).add( city["name"] ) counts["preserved"] += 1 continue feature_type = _feature_type_for_city(city) name = name_feature( voice, feature_type, ctx, blocklist, corpus, body_used, system_hook, system_id, world_seed, body_id, city.get("id") or "city_?", hop, log, verbose, ) if name is not None: city["name"] = name counts["cities"] += 1 generated["cities"].append(name) changed = True # Rivers for river in markers.get("rivers") or []: if not _is_blank(river.get("name")): corpus.setdefault((system_id, "river"), set()).add(river["name"]) counts["preserved"] += 1 continue name = name_feature( voice, "river", ctx, blocklist, corpus, body_used, system_hook, system_id, world_seed, body_id, river.get("id") or "river_?", hop, log, verbose, ) if name is not None: river["name"] = name counts["rivers"] += 1 generated["rivers"].append(name) changed = True # Oceans / seas / lakes for water in markers.get("oceans") or []: if not _is_blank(water.get("name")): corpus.setdefault((system_id, _feature_type_for_ocean(water)), set()).add( water["name"] ) counts["preserved"] += 1 continue feature_type = _feature_type_for_ocean(water) name = name_feature( voice, feature_type, ctx, blocklist, corpus, body_used, system_hook, system_id, world_seed, body_id, water.get("id") or "water_?", hop, log, verbose, ) if name is not None: water["name"] = name counts["oceans"] += 1 generated["oceans"].append(name) changed = True # Mountain ranges for rng_feat in markers.get("mountain_ranges") or []: if not _is_blank(rng_feat.get("name")): corpus.setdefault((system_id, "mountain_range"), set()).add( rng_feat["name"] ) counts["preserved"] += 1 continue name = name_feature( voice, "mountain_range", ctx, blocklist, corpus, body_used, system_hook, system_id, world_seed, body_id, rng_feat.get("id") or "range_?", hop, log, verbose, ) if name is not None: rng_feat["name"] = name counts["mountain_ranges"] += 1 generated["mountain_ranges"].append(name) changed = True # POIs for poi in markers.get("pois") or []: if not _is_blank(poi.get("name")): corpus.setdefault((system_id, _feature_type_for_poi(poi)), set()).add( poi["name"] ) counts["preserved"] += 1 continue feature_type = _feature_type_for_poi(poi) name = name_feature( voice, feature_type, ctx, blocklist, corpus, body_used, system_hook, system_id, world_seed, body_id, poi.get("id") or "poi_?", hop, log, verbose, ) if name is not None: poi["name"] = name counts["pois"] += 1 generated["pois"].append(name) changed = True if changed: markers_path.write_text(json.dumps(markers, indent=2) + "\n") # Refresh atlas_* rows for this body so DB queries pick up the # new names without a separate generate_atlas.py pass. Commit # immediately so a mid-run crash / kill loses at most one body # of DB state — the markers.json files are already persisted # above, atomically, via Path.write_text. sync_markers_to_db(conn, body_id, markers) conn.commit() counts["generated"] = generated return counts # --------------------------------------------------------------------------- # Body discovery # --------------------------------------------------------------------------- def _body_id_from_path(markers_path: Path) -> tuple[str, str]: """From wiki/star-systems/GJ-244A/bodies/GJ244Ad/markers.json return ('GJ244Ad', 'GJ 244A').""" body_id = markers_path.parent.name system_slug = markers_path.parent.parent.parent.name system_id = system_slug.replace("-", " ", 1) if system_slug.startswith("GJ-") else system_slug return body_id, system_id def discover_bodies( body_filter: str | None, limit: int | None, hop_order: dict[str, tuple[int, str]], ) -> list[Path]: """Discover every markers.json under wiki/star-systems/ and return the list sorted by hop distance from Gateway ascending (core first, deep frontier last). Files whose body_id is not in the hop_order map — e.g. bodies deleted from the DB but still carrying a markers.json — sort to the end with a sentinel hop of 99 so they don't pollute early dedup decisions. """ all_markers = list(WIKI_SYSTEMS.glob("*/bodies/*/markers.json")) if body_filter: all_markers = [ p for p in all_markers if _body_id_from_path(p)[0] == body_filter ] def sort_key(path: Path) -> tuple[int, str]: body_id = _body_id_from_path(path)[0] return hop_order.get(body_id, (99, body_id)) all_markers.sort(key=sort_key) if limit is not None: all_markers = all_markers[:limit] return all_markers # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser( description="Batch-name atlas features via Gemma 2 voice pipeline (#833)" ) parser.add_argument("--db", default=str(DB_PATH), help="Path to systems.db") parser.add_argument("--body", help="Process only this body_id") parser.add_argument( "--limit", type=int, help="Process at most N bodies (in sorted order). Smoke testing.", ) parser.add_argument( "--mock", action="store_true", help="Use mock-stdio.sh instead of the real sr-voice binary", ) parser.add_argument( "--sr-voice", default=str(DEFAULT_SR_VOICE), help="Path to the sr-voice release binary", ) parser.add_argument( "--model", default=str(DEFAULT_MODEL), help="Path to the Gemma 2 GGUF model (ignored in --mock mode)", ) parser.add_argument( "--distrobox", default=None, help="Name of a distrobox container to invoke sr-voice inside. " "Use this when the built binary depends on libraries (e.g. " "libhipblas.so.2) that are only installed in the container, " "not on the host. Example: --distrobox reach-build. " "Default: run sr-voice directly on the host.", ) parser.add_argument( "--refresh", type=int, default=1000, help="Restart the voice subprocess every N requests (default: 200) " "to prevent KV-cache context bleed", ) parser.add_argument( "--seed", type=int, default=42, help="World seed for deterministic naming (default: 42)", ) parser.add_argument( "--log", default=str(REPO_ROOT / ".tmp" / "gemma_naming.log"), help="Path to a log file. Every status line is written to both " "stdout and the log. Default: .tmp/gemma_naming.log. " "Pass '-' to disable file logging.", ) parser.add_argument( "--verbose", action="store_true", help="Print per-feature retry detail (noisy) and subprocess " "lifecycle events", ) parser.add_argument( "--dump-prompts", default=None, help="Capture mode: write the attempt-0 prompt for every feature " "to this JSONL path and return unique placeholders, without " "calling any LLM. Use to feed the same prompt set to a " "different backend (e.g. a Claude agent) for A/B comparison.", ) args = parser.parse_args() global _CAPTURE_FILE if args.dump_prompts: Path(args.dump_prompts).parent.mkdir(parents=True, exist_ok=True) _CAPTURE_FILE = open(args.dump_prompts, "w") # Capture mode short-circuits inside name_feature and never # actually calls the voice subprocess. Force --mock so we boot # the cheap mock-stdio shell instead of loading a real model. args.mock = True log_path = None if args.log == "-" else Path(args.log) db_path = Path(args.db) if not db_path.exists(): print(f"error: {db_path} not found", file=sys.stderr) sys.exit(1) sr_voice_bin = Path(args.sr_voice) model_path = Path(args.model) if args.model else None if not args.mock: if not sr_voice_bin.exists(): print( f"error: sr-voice binary not found at {sr_voice_bin}\n" "Either build it (`make build-sr-voice` in the main workdir) " "or run with --mock for a dry-fire test.", file=sys.stderr, ) sys.exit(1) if model_path and not model_path.exists(): print( f"error: model not found at {model_path}\n" "Either download the Gemma 2 GGUF or run with --mock.", file=sys.stderr, ) sys.exit(1) else: if not MOCK_STDIO.exists(): print(f"error: mock stdio script not found at {MOCK_STDIO}", file=sys.stderr) sys.exit(1) log = Logger(log_path) blocklist = load_blocklist() conn = sqlite3.connect(str(db_path), timeout=30.0) # WAL mode + busy_timeout so two concurrent shards serialize writes # without locking errors. WAL is a pragma-level switch, safe to # re-apply on every connect. conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA busy_timeout=15000") conn.execute("PRAGMA foreign_keys=ON") ensure_atlas_schema(conn) hop_order = load_body_hop_order(conn) system_gttr_hooks = load_system_gttr_hooks(conn) markers_paths = discover_bodies(args.body, args.limit, hop_order) if not markers_paths: log(f"error: no markers.json found (body={args.body})") conn.close() sys.exit(1) # Count distinct systems so the progress lines can report # `systems X/Y done` alongside `bodies X/Y done`. total_systems = len({_body_id_from_path(p)[1] for p in markers_paths}) seen_systems: set[str] = set() first_hop = hop_order.get(_body_id_from_path(markers_paths[0])[0], (99, ""))[0] last_hop = hop_order.get(_body_id_from_path(markers_paths[-1])[0], (99, ""))[0] log.raw("") log.raw(f" Gemma 2 Batch Naming Pipeline (#833)") log.raw(f" DB: {db_path}") log.raw(f" Mode: {'MOCK' if args.mock else 'REAL'}") log.raw(f" sr-voice: {MOCK_STDIO if args.mock else sr_voice_bin}") if not args.mock: log.raw(f" model: {model_path}") log.raw(f" seed: {args.seed} refresh: every {args.refresh} requests") log.raw(f" {len(markers_paths)} markers.json files to process") log.raw(f" hop {first_hop} → hop {last_hop}, core-first ordering") log.raw(f" {total_systems} distinct systems") log.raw(f" blocklist: {len(blocklist)} Earth-major entries") if log.log_path is not None: log.raw(f" log: {log.log_path}") log.raw("") log.raw( " Resume: re-run this command any time. Bodies whose markers.json " "already has non-empty name fields will be skipped (preserved path)." ) log.raw("") corpus: dict[tuple, set[str]] = {} # Cache of system_id → proper_name so we can emit a header line the # first time we hit each system without re-querying per body. system_name_cache: dict[str, str] = { row[0]: row[1] or "" for row in conn.execute( "SELECT system_id, proper_name FROM star_systems" ).fetchall() } body_name_cache: dict[str, str] = { row[0]: row[1] or "" for row in conn.execute( "SELECT body_id, proper_name FROM bodies" ).fetchall() } last_system_id: str | None = None totals = { "cities": 0, "rivers": 0, "oceans": 0, "mountain_ranges": 0, "pois": 0, "preserved": 0, "errors": 0, } bodies_touched = 0 t_total = time.time() try: with VoiceSubprocess( sr_voice_bin=sr_voice_bin, model_path=None if args.mock else model_path, mock=args.mock, refresh_every=args.refresh, verbose=args.verbose, distrobox=args.distrobox, ) as voice: for i, markers_path in enumerate(markers_paths): body_id, system_id = _body_id_from_path(markers_path) # System header: print when we enter a new system, so the # user can see which part of the reach we're in. Includes # the proper_name if the system has one (e.g. "Tau Ceti", # "p Eridani", "Groombridge"). if system_id != last_system_id: last_system_id = system_id sys_proper = system_name_cache.get(system_id, "") sys_hop = hop_order.get(body_id, (99, ""))[0] label = f"{system_id}" if sys_proper: label = f"{system_id} — {sys_proper}" log(f" ── SYSTEM {len(seen_systems)+1}/{total_systems} " f"{label} (hop {sys_hop})") seen_systems.add(system_id) t0 = time.time() hop = hop_order.get(body_id, (99, ""))[0] counts = process_body( body_id=body_id, system_id=system_id, markers_path=markers_path, voice=voice, conn=conn, blocklist=blocklist, corpus=corpus, system_hook=system_gttr_hooks.get(system_id), world_seed=args.seed, hop=hop, log=log, verbose=args.verbose, ) elapsed = time.time() - t0 body_progress = f"body {i+1}/{len(markers_paths)}" sys_progress = f"sys {len(seen_systems)}/{total_systems}" hop = hop_order.get(body_id, (99, ""))[0] progress = f"{body_progress} {sys_progress} hop={hop}" # Append the body's proper name if it has one ("Threshold", # "Arden", "Earth") so the log reads like a tour through # the reach rather than a wall of body_id slugs. body_proper = body_name_cache.get(body_id, "") body_label = body_id if not body_proper else f"{body_id:14s} ({body_proper})" body_label = body_label if body_proper else f"{body_id:14s}" if "error" in counts: totals["errors"] += 1 log(f" [{progress}] {body_label} ERROR: {counts['error']}") continue generated: dict[str, list[str]] = counts.pop("generated", {}) or { "cities": [], "rivers": [], "oceans": [], "mountain_ranges": [], "pois": [], } touched = ( counts["cities"] + counts["rivers"] + counts["oceans"] + counts["mountain_ranges"] + counts["pois"] ) if touched: bodies_touched += 1 for k in ("cities", "rivers", "oceans", "mountain_ranges", "pois", "preserved"): totals[k] += counts[k] log( f" [{progress}] {body_label} +{touched} names " f"({elapsed:.1f}s) — " f"cities={counts['cities']} rivers={counts['rivers']} " f"oceans={counts['oceans']} mtns={counts['mountain_ranges']} " f"pois={counts['pois']}" ) # Print the new names so the user can eyeball quality # as the run progresses. for section_label, key in ( ("cities", "cities"), ("rivers", "rivers"), ("waters", "oceans"), ("mtns", "mountain_ranges"), ("pois", "pois"), ): names = generated.get(key) or [] if names: preview = ", ".join(names[:10]) if len(names) > 10: preview += f", … (+{len(names) - 10} more)" log(f" {section_label:7s} {preview}") else: totals["preserved"] += counts["preserved"] log( f" [{progress}] {body_label} " f"(skip — {counts['preserved']} names already set)" ) # Periodic cumulative snapshot so the log has regular # checkpoint lines the user can scroll to. if (i + 1) % 25 == 0 or (i + 1) == len(markers_paths): cum = ( totals["cities"] + totals["rivers"] + totals["oceans"] + totals["mountain_ranges"] + totals["pois"] ) rate = cum / max(time.time() - t_total, 1e-6) remaining = len(markers_paths) - (i + 1) if remaining > 0 and (i + 1) > 0: per_body = (time.time() - t_total) / (i + 1) eta_s = int(per_body * remaining) eta = f"{eta_s // 3600}h{(eta_s % 3600) // 60:02d}m" else: eta = "--" log( f" >> CHECKPOINT bodies {i+1}/{len(markers_paths)} " f"systems {len(seen_systems)}/{total_systems} " f"names {cum} {rate:.1f}/s eta {eta}" ) # Final explicit commit for anything we accumulated since # the last per-body commit (should be no-op since we commit # per body, but defensive). conn.commit() finally: conn.close() elapsed_total = time.time() - t_total log.raw("") log.raw(f" Done: {elapsed_total:.0f}s ({elapsed_total/60:.1f} min)") log.raw(f" bodies processed: {len(markers_paths)}") log.raw(f" bodies touched: {bodies_touched}") log.raw(f" systems seen: {len(seen_systems)}/{total_systems}") log.raw(f" cities named: {totals['cities']}") log.raw(f" rivers named: {totals['rivers']}") log.raw(f" oceans named: {totals['oceans']}") log.raw(f" mountains named: {totals['mountain_ranges']}") log.raw(f" pois named: {totals['pois']}") log.raw(f" preserved: {totals['preserved']}") log.raw(f" errors: {totals['errors']}") log.raw("") log.close() if totals["errors"] > 0: sys.exit(1) if __name__ == "__main__": main()