#!/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" # Default binary + model paths point at the main workdir (sibling worktree # where the sr-voice binary and gemma2.gguf live). Override via --sr-voice # / --model if your layout differs. MAIN_WORKDIR = Path("/var/mnt/data/projects/settled-reach/main") DEFAULT_SR_VOICE = MAIN_WORKDIR / "server" / "sr-voice" / "target" / "release" / "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 append-mode log file. Every message gets a millisecond timestamp prefix so the log is interleavable with tail -f and the user can follow progress across two parallel shards by `tail -f .tmp/gemma_naming.shard*.log`. 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.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 _ts(self) -> str: return datetime.datetime.now().strftime("%H:%M:%S") def __call__(self, msg: str = "") -> None: line = f"[{self._ts()}] {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_PALETTES: dict[str, dict[str, str]] = { "core": { "inflection": "institutional Latin / pan-Anglo / Gateway-era", "examples": "Meridian, Concord, Cardinal, Prefecture, Lumen, Foro, Axis, Senatus", }, "north_reach": { "inflection": "British / Australian / Irish", "examples": "Wolcott, Mildern, Ashbourne, Kiln, Briarfell, Tarndale", }, "south_reach": { "inflection": "Portuguese / Swahili / Cape Verdean / Brazilian", "examples": "Vargas, Inhaca, Monteforte, Serra, Ribeiro, Kilimi", }, "east_reach": { "inflection": "Korean / Japanese / Taiwanese", "examples": "Hanyang, Takamine, Seoraksan, Tsukuri, Ginoza, Baektu", }, "west_reach": { "inflection": "German / Dutch / Nordic", "examples": "Vanebach, Kloosterdam, Bergfjord, Hellekade, Straend", }, "deep_frontier": { "inflection": "founder-surname + noun (e.g. 'Okafor Reach', 'Stenner Cross')", "examples": "Okafor Reach, Stenner Cross, Weller Hold, Pruitt Basin, Vickery Hold", }, # Legacy keys retained for backward compatibility with the # cultural_corridor column on the one system that uses it. "sol-gateway-axis": { "inflection": "institutional Latin / pan-Anglo / Gateway-era", "examples": "Meridian, Concord, Cardinal, Prefecture, Lumen, Foro, Axis, Senatus", }, "inner_corridor": { "inflection": "institutional Latin / pan-Anglo", "examples": "Meridian, Concord, Prefecture, Cardinal, Lumen, Foro", }, } DEFAULT_PALETTE = CORRIDOR_PALETTES["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) -> dict[str, str]: if not corridor: return DEFAULT_PALETTE return CORRIDOR_PALETTES.get(corridor, DEFAULT_PALETTE) # --------------------------------------------------------------------------- # Feature prompt templates — few-shot format # --------------------------------------------------------------------------- # 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 3 concrete # `Input → Output` examples so the model completes a pattern instead # of generating to an open-ended instruction. # # Important rules these templates enforce: # - The examples ALWAYS show a bare name (no labels, no markdown, # no brackets, no quotes) so the completion mimics that shape. # - The examples are DIFFERENT corridors from the one being named, # to prevent Gemma from just echoing one of the examples. # - The final line ends with `Answer:` (not `NAME:`) — less likely # to collide with a real name in post-processing. # # All templates share the same few-shot prefix defined below; only # the question and example axis differ per feature type. _FEW_SHOT_RIVER = ( "You name rivers on alien planets. Reply with ONLY the name, 1-3 words, " "no brackets, no quotes, no markdown, no label.\n" "\n" "Style: British/Australian. Answer: Wolcott Beck\n" "Style: Korean/Japanese. Answer: Nakamura Stream\n" "Style: Portuguese/Swahili. Answer: Ribeiro do Sal\n" "\n" "Style: {inflection}. Answer:" ) _FEW_SHOT_OCEAN = ( "You name oceans on alien planets. Reply with ONLY the name, 1-3 words, " "no brackets, no quotes, no markdown, no label.\n" "\n" "Style: British/Australian. Answer: Tarnsea\n" "Style: Korean/Japanese. Answer: Aomine Deep\n" "Style: German/Dutch/Nordic. Answer: Nordhav\n" "\n" "Style: {inflection}. Answer:" ) _FEW_SHOT_SEA = ( "You name seas on alien planets. Reply with ONLY the name, 1-3 words, " "no brackets, no quotes, no markdown, no label.\n" "\n" "Style: British/Australian. Answer: Harven Sea\n" "Style: Portuguese/Swahili. Answer: Mar de Quelim\n" "Style: institutional Latin. Answer: Mare Ardens\n" "\n" "Style: {inflection}. Answer:" ) _FEW_SHOT_LAKE = ( "You name lakes on alien planets. Reply with ONLY the name, 1-3 words, " "no brackets, no quotes, no markdown, no label.\n" "\n" "Style: British/Australian. Answer: Kelstern Mere\n" "Style: Korean/Japanese. Answer: Shiromizu\n" "Style: German/Dutch/Nordic. Answer: Eikmeer\n" "\n" "Style: {inflection}. Answer:" ) _FEW_SHOT_MOUNTAIN = ( "You name mountain ranges on alien planets. Reply with ONLY the name, " "1-3 words, no brackets, no quotes, no markdown, no label.\n" "\n" "Style: British/Australian. Answer: Drayton Spine\n" "Style: Korean/Japanese. Answer: Takamine Ridge\n" "Style: German/Dutch/Nordic. Answer: Drachenberg\n" "\n" "Style: {inflection}. Answer:" ) _FEW_SHOT_CITY_CAPITAL = ( "You name capital cities on alien planets. Reply with ONLY the name, " "1-2 words, no brackets, no quotes, no markdown, no label.\n" "\n" "Style: British/Australian. Answer: Holmwood\n" "Style: Korean/Japanese. Answer: Seungmun\n" "Style: Portuguese/Swahili. Answer: Porto Keli\n" "\n" "Style: {inflection}. Planet: {planet_class}. Answer:" ) _FEW_SHOT_CITY_SECONDARY = ( "You name secondary cities on alien planets. Reply with ONLY the name, " "1-2 words, no brackets, no quotes, no markdown, no label.\n" "\n" "Style: British/Australian. Answer: Carberry\n" "Style: Korean/Japanese. Answer: Yurigawa\n" "Style: German/Dutch/Nordic. Answer: Wolfsturm\n" "\n" "Style: {inflection}. Planet: {planet_class}. Answer:" ) _FEW_SHOT_POI_TRANSIT = ( "You name gate terminals on alien planets. Reply with ONLY the name, " "2-3 words ending in 'Gate Terminal', 'Transit', 'Exchange', or " "'Concourse'. No brackets, no quotes, no markdown, no label.\n" "\n" "Style: British/Australian. Answer: Holmwood Gate Terminal\n" "Style: Korean/Japanese. Answer: Seungmun Transit\n" "Style: institutional Latin. Answer: Meridian Concourse\n" "\n" "Style: {inflection}. Answer:" ) _FEW_SHOT_POI_INSTITUTIONAL = ( "You name institutional landmarks on alien planets. Reply with ONLY " "the name, 2-4 words, no brackets, no quotes, no markdown, no label.\n" "\n" "Style: British/Australian. Answer: Holmwood Assembly Hall\n" "Style: Korean/Japanese. Answer: Seungmun Archive\n" "Style: institutional Latin. Answer: Meridian Registry\n" "\n" "Style: {inflection}. Answer:" ) _FEW_SHOT_POI_CULTURAL = ( "You name cultural landmarks on alien planets. Reply with ONLY the " "name, 2-4 words, no brackets, no quotes, no markdown, no label.\n" "\n" "Style: British/Australian. Answer: Holmwood Commons\n" "Style: Korean/Japanese. Answer: Yurigawa Grounds\n" "Style: Portuguese/Swahili. Answer: Praça do Vento\n" "\n" "Style: {inflection}. Answer:" ) FEATURE_PROMPTS: dict[str, str] = { "river": _FEW_SHOT_RIVER, "ocean": _FEW_SHOT_OCEAN, "sea": _FEW_SHOT_SEA, "lake": _FEW_SHOT_LAKE, "mountain_range": _FEW_SHOT_MOUNTAIN, "city_capital": _FEW_SHOT_CITY_CAPITAL, "city_secondary": _FEW_SHOT_CITY_SECONDARY, "poi_transit": _FEW_SHOT_POI_TRANSIT, "poi_institutional": _FEW_SHOT_POI_INSTITUTIONAL, "poi_cultural": _FEW_SHOT_POI_CULTURAL, } # --------------------------------------------------------------------------- # 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: dict[str, list[str]] = { "north_reach": ["Wolcott", "Mildern", "Ashbourne", "Briarfell", "Tarndale", "Kelston", "Threlwood", "Harford", "Rowanmoor", "Pennhowe"], "south_reach": ["Vargas", "Inhaca", "Monteforte", "Ribeiro", "Kilimi", "Serravale", "Cabo", "Ilhabela", "Moçambo", "Ngola"], "east_reach": ["Hanyang", "Takamine", "Seorak", "Ginoza", "Baektu", "Tsukuri", "Naruhan", "Saeyeon", "Morimine", "Taegong"], "west_reach": ["Vanebach", "Kloosterdam", "Bergfjord", "Hellekade", "Straend", "Eikhof", "Viskrans", "Nordhölm", "Lindeborg", "Drachenberg"], "inner_corridor": ["Meridian", "Concord", "Prefecture", "Cardinal", "Lumen", "Foro", "Tabula", "Vox", "Axis", "Senatus"], "inner_orbit": ["Meridian", "Concord", "Prefecture", "Cardinal", "Lumen", "Foro", "Tabula", "Vox", "Axis", "Senatus"], "frontier": ["Okafor", "Stenner", "Weller", "Pruitt", "Hale", "Kettle", "Bowman", "Alder", "Risher", "Vickery"], } _FALLBACK_SUFFIXES: dict[str, list[str]] = { "river": ["Run", "Water", "Beck", "Rill", "Course"], "ocean": ["Sea", "Expanse", "Deep", "Reach"], "sea": ["Sea", "Gulf", "Basin"], "lake": ["Lake", "Mere", "Tarn", "Pool"], "mountain_range": ["Range", "Ridge", "Spine", "Heights", "Scarp"], "city_capital": ["Hold", "Prime", "Seat", "Court"], "city_secondary": ["Cross", "Reach", "Hollow", "Fields", "Stand"], "poi_transit": ["Gate Terminal", "Transit", "Concourse", "Exchange"], "poi_institutional": ["Archive", "Hall", "Assembly", "Registry"], "poi_cultural": ["Commons", "Grounds", "Circle", "Square"], } 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, ): 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.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"] cmd = [str(self.sr_voice_bin), "serve", "--stdio"] if self.model_path is not None: cmd += ["--model", str(self.model_path)] return 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+)" _STEM_WORD_RE = re.compile(r"[A-Za-z][A-Za-z'\-]{2,}") # Tokens so generic they should never count toward stem-dominance — they # just describe the feature type and don't carry cultural identity. _IGNORED_STEMS = { "the", "of", "a", "an", "and", "or", "river", "sea", "lake", "ocean", "bay", "gulf", "range", "peak", "peaks", "ridge", "spine", "scarp", "heights", "hollow", "run", "beck", "water", "course", "flow", "basin", "reach", "hold", "cross", "prime", "city", "town", "capital", "gate", "terminal", "transit", "exchange", "concourse", "assembly", "archive", "commons", "grounds", "square", "circle", "mountain", "mountains", "stream", "brook", "spring", "tarn", "mere", "pool", "deep", "expanse", "crest", "summit", "fells", "series", "sea", "maris", "mare", "aquae", "fluvius", "terrae", } def _extract_stems(name: str) -> list[str]: """Return the lowercase 'interesting' stems of a name — cultural root tokens only, with 'the', 'of', 'river', 'peaks' etc. dropped. Used to enforce per-stem dominance caps across the full run so no single root (e.g. 'Arcturus') can appear in hundreds of names across 3240 bodies. """ return [ t.lower() for t in _STEM_WORD_RE.findall(name) if t.lower() not in _IGNORED_STEMS ] def name_feature( voice: VoiceSubprocess, feature_type: str, ctx: dict, blocklist: set[str], corpus: dict[tuple[str, str], set[str]], body_used: set[str], stem_counts: dict[str, int], stem_cap: int, world_seed: int, body_id: str, local_id: str, log: "Logger", verbose: bool, max_attempts: int = 3, ) -> str: """Request a name from Gemma, enforce blocklist + corridor dedup + per-body cross-type dedup, fall back to the palette generator on persistent failure. Dedup scopes: - `corpus[(corridor, feature_type)]` — cross-body dedup within the same corridor and feature type. Two rivers in the north_reach should not share a name; two rivers on opposite arcs can. - `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, which reads as ridiculous even when the types differ. """ template = FEATURE_PROMPTS.get(feature_type) corridor = ctx.get("cultural_corridor") or "core" if template is None: return fallback_name(corridor, feature_type, _seed_for(world_seed, body_id, local_id, 0)) palette = palette_for(corridor) prompt = template.format( inflection=palette["inflection"], planet_class=ctx.get("planet_class") or "habitable", ) dedup_key = (corridor, feature_type) used = corpus.setdefault(dedup_key, set()) 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 _exceeds_stem_cap(candidate: str) -> str | None: """Return the first stem in `candidate` that would exceed the cap after this accept, or None if all stems are under the cap.""" if stem_cap <= 0: return None for stem in _extract_stems(candidate): if stem_counts.get(stem, 0) >= stem_cap: return stem return None def _commit_name(final: str) -> None: used.add(final) body_used.add(final) for stem in _extract_stems(final): stem_counts[stem] = stem_counts.get(stem, 0) + 1 for attempt in range(max_attempts): seed = _seed_for(world_seed, body_id, local_id, attempt) try: raw = voice.request(prompt, seed) except RuntimeError as e: if verbose: log(f" subprocess error on {body_id}/{local_id} attempt " f"{attempt}: {e} — retrying") continue cleaned = post_process(raw) if not cleaned: continue if is_placeholder(cleaned): if verbose: log(f" placeholder '{cleaned}' ({body_id}/{local_id}) — retrying") continue if is_blocked(cleaned, blocklist): if verbose: log(f" blocklist hit '{cleaned}' ({body_id}/{local_id}) — retrying") continue if _is_duplicate(cleaned): if verbose: log(f" dedup hit '{cleaned}' ({body_id}/{local_id}) — retrying") continue over_stem = _exceeds_stem_cap(cleaned) if over_stem is not None: if verbose: log(f" stem cap hit '{cleaned}' (stem '{over_stem}' at " f"cap {stem_cap}) {body_id}/{local_id} — retrying") continue _commit_name(cleaned) return cleaned # All attempts exhausted — deterministic palette fallback, then dedup. # Fallback names draw from the palette stems and do NOT count against # the stem cap (the palette is intentionally narrow and would trigger # infinite rejection loops otherwise). salt = _seed_for(world_seed, body_id, local_id, max_attempts) & 0xFFFFFF fallback = fallback_name(corridor, feature_type, salt) bump = 0 while _is_duplicate(fallback) and bump < 100: bump += 1 fallback = fallback_name(corridor, feature_type, salt + bump) used.add(fallback) body_used.add(fallback) log(f" fallback: {body_id}/{local_id} → '{fallback}'") return fallback # --------------------------------------------------------------------------- # 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 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", } planet_class, settlement_pattern, corridor, population, economic_role = 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), } 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[str, str], set[str]], stem_counts: dict[str, int], stem_cap: int, world_seed: 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 # canonical identifiers and new features don't collide with them. # Hand-authored names also count against the stem cap so those # anchors take priority over generator output. 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()) for stem in _extract_stems(name): stem_counts[stem] = stem_counts.get(stem, 0) + 1 # Cities for city in markers.get("cities") or []: if not _is_blank(city.get("name")): corpus.setdefault((corridor, _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, stem_counts, stem_cap, world_seed, body_id, city.get("id") or "city_?", log, verbose, ) 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((corridor, "river"), set()).add(river["name"]) counts["preserved"] += 1 continue name = name_feature( voice, "river", ctx, blocklist, corpus, body_used, stem_counts, stem_cap, world_seed, body_id, river.get("id") or "river_?", log, verbose, ) 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((corridor, _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, stem_counts, stem_cap, world_seed, body_id, water.get("id") or "water_?", log, verbose, ) 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((corridor, "mountain_range"), set()).add( rng_feat["name"] ) counts["preserved"] += 1 continue name = name_feature( voice, "mountain_range", ctx, blocklist, corpus, body_used, stem_counts, stem_cap, world_seed, body_id, rng_feat.get("id") or "range_?", log, verbose, ) 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((corridor, _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, stem_counts, stem_cap, world_seed, body_id, poi.get("id") or "poi_?", log, verbose, ) 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( "--refresh", type=int, default=200, 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( "--stem-cap", type=int, default=20, help="Max times any single cultural stem (e.g. 'Arcturus', " "'Meridian') may appear across the full run before dedup " "starts rejecting it. 0 = disabled. Default: 20.", ) parser.add_argument( "--shard", default="0/1", help="Process a slice of the body list for parallel runs. Format " "N/M: worker N of M total. Each worker walks bodies in hop " "order, picking every Mth body starting at offset N. Example: " "run `--shard 0/2` in one terminal and `--shard 1/2` in " "another to split the work in half. Default: 0/1 (all bodies).", ) parser.add_argument( "--log", default=None, help="Path to a log file. Every status line is written to both " "stdout and the log. Defaults to " ".tmp/gemma_naming.shard{N}of{M}.log when --shard is not " "trivially 0/1; disabled otherwise. Pass '-' to disable.", ) parser.add_argument( "--verbose", action="store_true", help="Print per-feature retry detail (noisy) and subprocess " "lifecycle events", ) args = parser.parse_args() # Parse shard spec. try: shard_n_str, shard_m_str = args.shard.split("/", 1) shard_n = int(shard_n_str) shard_m = int(shard_m_str) if shard_m < 1 or not (0 <= shard_n < shard_m): raise ValueError except ValueError: print( f"error: invalid --shard '{args.shard}'. " "Expected N/M with 0 <= N < M, e.g. 0/2.", file=sys.stderr, ) sys.exit(1) # Default log path: enable for non-trivial shards, disable for 0/1. if args.log is None: if shard_m > 1: log_path = REPO_ROOT / ".tmp" / f"gemma_naming.shard{shard_n}of{shard_m}.log" else: log_path = None elif args.log == "-": log_path = None else: log_path = 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) all_markers = discover_bodies(args.body, args.limit, hop_order) if not all_markers: log(f"error: no markers.json found (body={args.body})") conn.close() sys.exit(1) # Apply shard slicing AFTER discovery + sort. Shard 0/2 gets indices # [0, 2, 4, …], shard 1/2 gets [1, 3, 5, …] — interleaved in hop # order so both shards advance core → frontier in parallel. markers_paths = all_markers[shard_n::shard_m] if not markers_paths: log(f"error: shard {shard_n}/{shard_m} contains no bodies " f"(full set = {len(all_markers)}).") conn.close() sys.exit(1) # Count distinct systems the shard will touch so the progress lines # can report `systems X/Y done` alongside `bodies X/Y done`. shard_systems = sorted({_body_id_from_path(p)[1] for p in markers_paths}) total_shard_systems = len(shard_systems) 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] banner_shard = f"{shard_n}/{shard_m}" if shard_m > 1 else "single" 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" Shard: {banner_shard}") 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 " f"stem-cap: {args.stem_cap}") log.raw(f" {len(markers_paths)} markers.json files in this shard " f"(full set: {len(all_markers)})") log.raw(f" hop {first_hop} → hop {last_hop}, core-first ordering") log.raw(f" {total_shard_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("") corpus: dict[tuple[str, str], set[str]] = {} stem_counts: dict[str, int] = {} 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, ) as voice: for i, markers_path in enumerate(markers_paths): body_id, system_id = _body_id_from_path(markers_path) seen_systems.add(system_id) t0 = time.time() counts = process_body( body_id=body_id, system_id=system_id, markers_path=markers_path, voice=voice, conn=conn, blocklist=blocklist, corpus=corpus, stem_counts=stem_counts, stem_cap=args.stem_cap, world_seed=args.seed, 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_shard_systems}" hop = hop_order.get(body_id, (99, ""))[0] progress = f"{body_progress} {sys_progress} hop={hop}" if "error" in counts: totals["errors"] += 1 log(f" [{progress}] {body_id:14s} 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_id:14s} +{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"] if args.verbose: log( f" [{progress}] {body_id:14s} " f"no blanks ({counts['preserved']} preserved)" ) # 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_shard_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" shard: {banner_shard}") log.raw(f" bodies processed: {len(markers_paths)}") log.raw(f" bodies touched: {bodies_touched}") log.raw(f" systems seen: {len(seen_systems)}/{total_shard_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()