Files
settled-reach/tooling/economy-db/economy_import/traits.py
T
jpmschweitzerandClaude Fable 5 b8e0b1b660 fix(db): harden architecture_zone_bias axis-value validation (PR #174 review)
H1: guard that each axis value is a {token=weight} table before .items() —
a scalar (wall = 15000) or the array shape (wall = ["steel_frame"], a
plausible copy-paste from the sibling catalog's visual_bundle) now yields a
clean V-TT-06 error instead of a bare AttributeError. Mirrors the isinstance
guards already on zone_map and axes.
H2: exclude bool from the positive-integer weight check (weight = true is an
int subclass, previously slipped through as 1) — matches the guard
populate_color_register_bands already applies to its own values.
Two new ZoneBiasValidationTests cover both branches.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-11 13:22:40 +02:00

523 lines
24 KiB
Python

"""Architecture-flavor trait templates (D-232, #993): catalog baker + hero bias."""
import json
import sqlite3
import tomllib
from .errors import ImportAborted
from .paths import (
ARCHITECTURE_TRAIT_BIAS_TOML,
ARCHITECTURE_TRAIT_CATALOG_TOML,
ARCHITECTURE_ZONE_BIAS_TOML,
COLOR_REGISTER_BANDS_TOML,
OBJECT_TAG_VOCABULARY_TOML,
)
_TRAIT_CORRIDOR_POOLS: set[str] = {"baseline", "heritage", "cross_corridor"}
_TRAIT_BIAS_KINDS: set[str] = {"pin", "boost", "suppress"}
# V-TT-05: pins count toward K (D-232) and the largest K is 5
# (ComplexityTier::Full) — a body can never use more pins than that.
_MAX_PINS_PER_BODY: int = 5
# JSON-encoded list/map columns on trait_templates (TOML inline arrays/tables ->
# JSON text the generator parses).
_TRAIT_JSON_LIST: tuple[str, ...] = ("bulk_class_gate", "production_ubiquity_gate", "allow_tags", "block_tags")
_TRAIT_JSON_MAP: tuple[str, ...] = ("weight_mods", "zone_affinity", "visual_bundle")
# ObjectTag vocabulary (T-995, Q-049, D-235): the four visual_bundle axes the
# catalog keys on. `color_register` is a free-form palette cue, not an
# ObjectTag, and is intentionally not one of these.
_TAG_AXES: tuple[str, ...] = ("wall", "roof", "facade", "street")
# The four always-present fallback-terminal placeholders (D-232/D-235).
_EXPECTED_GENERIC_TAGS: frozenset[str] = frozenset(
{"generic_wall", "generic_roof", "generic_facade", "generic_street"}
)
def _resolve_tag_fallback_chain(tag: str, registry: dict[str, dict], errors: list[str]) -> None:
"""Walk one registry tag's fallback chain to a generic parent (V-TT-04).
Appends a finding to `errors` if the chain references an unknown tag,
cycles back on itself, or a non-generic tag never reaches a generic
parent. Generics are terminal by construction (checked at load time, not
here) so walking stops the moment a generic entry is reached.
"""
seen: list[str] = []
cur = tag
while True:
entry = registry.get(cur)
if entry is None:
errors.append(
f"V-TT-04: object_tag_vocabulary '{tag}': fallback chain references "
f"unknown tag '{cur}'"
)
return
if entry["generic"]:
return
if cur in seen:
chain = " -> ".join((*seen, cur))
errors.append(f"V-TT-04: object_tag_vocabulary '{tag}': fallback chain cycles ({chain})")
return
seen.append(cur)
nxt = entry.get("fallback")
if not nxt:
errors.append(
f"V-TT-04: object_tag_vocabulary '{tag}': non-generic tag has no 'fallback' "
"and never reaches a generic parent"
)
return
cur = nxt
def _load_object_tag_vocabulary(errors: list[str]) -> dict[str, dict]:
"""Parse + self-validate the ObjectTag registry (T-995, Q-049, D-235).
Returns a flat {tag_name: {"axis", "fallback", "generic"}} lookup spanning
every axis (a tag name is unique across axes). Appends malformed-registry
findings (unknown axis, duplicate tag, missing description, a generic
declaring a fallback, a non-generic missing one, or a broken fallback
chain) to the shared `errors` list — same accumulate-then-abort pattern as
the trait_templates checks below. Absent registry with a present catalog
is itself a hard error (the catalog now depends on this file to validate
against); returns {} in that case so downstream per-tag lookups no-op
rather than raising a second, redundant error.
"""
if not OBJECT_TAG_VOCABULARY_TOML.exists():
errors.append(
"V-TT-03: object_tag_vocabulary.toml not found — required to validate "
"architecture_trait_catalog.toml's ObjectTag references (T-995)"
)
return {}
with open(OBJECT_TAG_VOCABULARY_TOML, "rb") as f:
data = tomllib.load(f)
axes = data.get("tags", {})
registry: dict[str, dict] = {}
for axis, tags in axes.items():
if axis not in _TAG_AXES:
errors.append(f"V-TT-03: object_tag_vocabulary axis '{axis}' not one of {_TAG_AXES}")
continue
for tag, entry in tags.items():
if tag in registry:
errors.append(
f"V-TT-03: object_tag_vocabulary tag '{tag}' declared in both "
f"'{registry[tag]['axis']}' and '{axis}' axes"
)
continue
if not entry.get("description"):
errors.append(f"V-TT-03: object_tag_vocabulary '{tag}': missing 'description'")
is_generic = bool(entry.get("generic", False))
fallback = entry.get("fallback")
if is_generic and fallback:
errors.append(
f"V-TT-04: object_tag_vocabulary '{tag}': generic tag must not declare "
f"'fallback' (got '{fallback}') — generics are fallback-terminal"
)
elif not is_generic and not fallback:
errors.append(
f"V-TT-04: object_tag_vocabulary '{tag}': non-generic tag must declare "
"a 'fallback' parent"
)
registry[tag] = {"axis": axis, "fallback": fallback, "generic": is_generic}
found_generics = {t for t, e in registry.items() if e["generic"]}
if found_generics != _EXPECTED_GENERIC_TAGS:
errors.append(
f"V-TT-04: object_tag_vocabulary generic placeholders {sorted(found_generics)} "
f"!= expected {sorted(_EXPECTED_GENERIC_TAGS)}"
)
for tag in registry:
_resolve_tag_fallback_chain(tag, registry, errors)
return registry
def populate_trait_templates(conn: sqlite3.Connection, dry_run: bool) -> int:
"""Bake the D-232 architecture-flavor catalog into trait_templates (#993).
Reads ARCHITECTURE_TRAIT_CATALOG_TOML (`[templates.<tag>]` stanzas) and
rebuilds the table. List/map fields are stored as JSON text; numeric
eligibility is integer basis-points (D-010). The catalog *content* is
authored in #1005 — this baker is the mechanism. Absent source -> 0 rows
(the table still exists for the downstream pipeline). Deterministic rebuild:
clears trait_templates (cascading atlas_body_trait_bias) first.
"""
if not ARCHITECTURE_TRAIT_CATALOG_TOML.exists():
if not dry_run:
conn.execute("DELETE FROM atlas_body_trait_bias")
conn.execute("DELETE FROM trait_templates")
return 0
with open(ARCHITECTURE_TRAIT_CATALOG_TOML, "rb") as f:
data = tomllib.load(f)
templates = data.get("templates", {})
errors: list[str] = []
# ObjectTag registry (T-995, Q-049): loaded once, self-validated (V-TT-04
# fallback-graph checks happen inside), then used below to check every
# template's tag references actually exist (V-TT-03).
tag_registry = _load_object_tag_vocabulary(errors)
rows: list[tuple] = []
for tag, t in templates.items():
pool = t.get("corridor_pool", "baseline")
if pool not in _TRAIT_CORRIDOR_POOLS:
errors.append(f"trait_templates '{tag}': corridor_pool '{pool}' invalid")
if "label" not in t:
errors.append(f"trait_templates '{tag}': missing required 'label'")
for k in (*_TRAIT_JSON_LIST, *_TRAIT_JSON_MAP):
# any provided list/map field must JSON-encode cleanly
if k in t:
try:
json.dumps(t[k])
except (TypeError, ValueError):
errors.append(f"trait_templates '{tag}': field '{k}' not JSON-serialisable")
# V-TT-03: every ObjectTag the template references must exist in the
# object_tag_vocabulary.toml registry. Skipped when the registry
# itself failed to load (one clear error above beats N spurious ones).
if tag_registry:
for used in (*(t.get("allow_tags") or []), *(t.get("block_tags") or [])):
if used not in tag_registry:
errors.append(
f"V-TT-03: trait_templates '{tag}' ({t.get('label')}): tag '{used}' "
"(allow_tags/block_tags) not in object_tag_vocabulary.toml"
)
vb = t.get("visual_bundle") or {}
for axis in _TAG_AXES:
for used in vb.get(axis) or []:
entry = tag_registry.get(used)
if entry is None:
errors.append(
f"V-TT-03: trait_templates '{tag}' ({t.get('label')}): tag '{used}' "
f"(visual_bundle.{axis}) not in object_tag_vocabulary.toml"
)
elif entry["axis"] != axis:
errors.append(
f"V-TT-03: trait_templates '{tag}' ({t.get('label')}): tag '{used}' "
f"used as visual_bundle.{axis} but registered under axis "
f"'{entry['axis']}' in object_tag_vocabulary.toml"
)
rows.append((
tag, t.get("label", ""), t.get("cultural_description"),
pool, t.get("geographic_sector"),
json.dumps(t["bulk_class_gate"]) if t.get("bulk_class_gate") else None,
json.dumps(t["production_ubiquity_gate"]) if t.get("production_ubiquity_gate") else None,
int(t.get("min_prosperity_bps", 0)),
int(t.get("base_weight", 10000)),
json.dumps(t["weight_mods"]) if t.get("weight_mods") else None,
json.dumps(t["zone_affinity"]) if t.get("zone_affinity") else None,
json.dumps(t["allow_tags"]) if t.get("allow_tags") else None,
json.dumps(t["block_tags"]) if t.get("block_tags") else None,
t.get("era_scope"),
json.dumps(t["visual_bundle"]) if t.get("visual_bundle") else None,
))
if errors:
print(f" TRAIT TEMPLATE ERRORS ({len(errors)}):")
for e in errors:
print(f" - {e}")
raise ImportAborted()
# CI guardrails (D-232, Nigel): >=5 templates eligible per BulkClass (a
# template with no bulk_class_gate is eligible for all); no single template
# may exceed 60% of its eligible pool's base weight. Skipped when the catalog
# is empty (the bootstrap/absent-source case). Integer math (no float).
if templates:
gerrors: list[str] = []
for bc in ("BulkSolid", "BulkLiquid", "PrecisionDense", "Perishable", "NonPhysical"):
elig = [t for t in templates.values()
if not t.get("bulk_class_gate") or bc in t["bulk_class_gate"]]
if len(elig) < 5:
gerrors.append(
f"V-TT-01: only {len(elig)} template(s) eligible for BulkClass "
f"{bc} (need >=5)"
)
total = sum(int(t.get("base_weight", 10000)) for t in elig)
for t in elig:
w = int(t.get("base_weight", 10000))
if total and w * 5 > total * 3: # w/total > 0.60
gerrors.append(
f"V-TT-02: template '{t.get('label')}' is "
f"{w * 100 // total}% of the {bc} pool weight (>60%)"
)
if gerrors:
print(f" TRAIT TEMPLATE GUARDRAIL FAILURES ({len(gerrors)}):")
for e in gerrors:
print(f" - {e}")
raise ImportAborted()
# Validation/guardrails run on dry-run too; only mutate when committing.
if not dry_run:
conn.execute("DELETE FROM atlas_body_trait_bias")
conn.execute("DELETE FROM trait_templates")
conn.executemany(
"""INSERT INTO trait_templates
(tag, label, cultural_description, corridor_pool, geographic_sector,
bulk_class_gate, production_ubiquity_gate, min_prosperity_bps,
base_weight, weight_mods, zone_affinity, allow_tags, block_tags,
era_scope, visual_bundle)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
rows,
)
return len(rows)
def populate_architecture_zone_bias(conn: sqlite3.Connection, dry_run: bool) -> int:
"""Bake the D-235 step-2 zone-type bias table (T-988, resolves the ticket's
"REAL authored table" requirement).
Reads ARCHITECTURE_ZONE_BIAS_TOML (`[bias.<template_tag>.<zone_type_id>]`
stanzas; each is an optional `{axis: {token: weight_bps}}` sub-table per
axis — wall/roof/facade/street). Sparse/fallback model: a (template,
zone_type) pair absent here — or a token absent within a listed entry —
uses a uniform draw across the template's own `visual_bundle.<axis>` (the
generator's job, not this baker's).
V-TT-06: every referenced token must already appear in THAT template's own
`visual_bundle.<axis>` (checked against the just-baked `trait_templates`
row, not the catalog TOML in memory, so the check can never silently pass
against a stale in-process parse) — this file never expands a template's
palette, only re-weights within it. Must run AFTER populate_trait_templates
(FK + visual_bundle lookup). Absent source -> 0 rows.
"""
if not ARCHITECTURE_ZONE_BIAS_TOML.exists():
if not dry_run:
conn.execute("DELETE FROM architecture_zone_bias")
return 0
with open(ARCHITECTURE_ZONE_BIAS_TOML, "rb") as f:
data = tomllib.load(f)
bias_root = data.get("bias", {})
known_tags: set[str] = set()
template_axis_tokens: dict[str, dict[str, set[str]]] = {}
for tag, vb_json in conn.execute("SELECT tag, visual_bundle FROM trait_templates"):
known_tags.add(tag)
vb = json.loads(vb_json) if vb_json else {}
template_axis_tokens[tag] = {axis: set(vb.get(axis) or []) for axis in _TAG_AXES}
errors: list[str] = []
rows: list[tuple] = []
for template_tag, zone_map in bias_root.items():
if template_tag not in known_tags:
errors.append(f"architecture_zone_bias '{template_tag}': not in trait_templates")
continue
if not isinstance(zone_map, dict):
errors.append(
f"architecture_zone_bias '{template_tag}': expected a table of zone_type_id entries"
)
continue
for zone_type_id, axes in zone_map.items():
if not isinstance(axes, dict):
errors.append(
f"architecture_zone_bias '{template_tag}.{zone_type_id}': expected a table of axis entries"
)
continue
for axis, weights in axes.items():
if axis not in _TAG_AXES:
errors.append(
f"architecture_zone_bias '{template_tag}.{zone_type_id}': axis '{axis}' "
f"not one of {_TAG_AXES}"
)
continue
if not isinstance(weights, dict):
errors.append(
f"V-TT-06: architecture_zone_bias '{template_tag}.{zone_type_id}.{axis}': "
f"each axis value must be a {{token = weight_bps}} table, got "
f"{type(weights).__name__}"
)
continue
allowed = template_axis_tokens.get(template_tag, {}).get(axis, set())
for token, weight in weights.items():
if token not in allowed:
errors.append(
f"V-TT-06: architecture_zone_bias '{template_tag}.{zone_type_id}.{axis}': "
f"token '{token}' not in this template's visual_bundle.{axis} {sorted(allowed)}"
)
# bool is an int subclass — exclude it explicitly, matching
# populate_color_register_bands' guard on its own values.
if isinstance(weight, bool) or not isinstance(weight, int) or weight <= 0:
errors.append(
f"architecture_zone_bias '{template_tag}.{zone_type_id}.{axis}.{token}': "
f"weight_bps must be a positive integer, got {weight!r}"
)
try:
bias_json = json.dumps(axes)
except (TypeError, ValueError):
errors.append(
f"architecture_zone_bias '{template_tag}.{zone_type_id}': not JSON-serialisable"
)
continue
rows.append((template_tag, zone_type_id, bias_json))
if errors:
print(f" ARCHITECTURE ZONE BIAS ERRORS ({len(errors)}):")
for e in errors:
print(f" - {e}")
raise ImportAborted()
if not dry_run:
conn.execute("DELETE FROM architecture_zone_bias")
conn.executemany(
"INSERT INTO architecture_zone_bias (template_tag, zone_type_id, bias) VALUES (?,?,?)",
rows,
)
return len(rows)
# D-235 numeric HSV axis bounds (T-988): hue is centidegrees (0..36000 = 0-360
# deg x 100); sat/val are basis points (0..10000 = 0-100.00%).
_COLOR_AXIS_BOUNDS: dict[str, int] = {"hue": 36000, "sat": 10000, "val": 10000}
def populate_color_register_bands(conn: sqlite3.Connection, dry_run: bool) -> int:
"""Bake the D-235 numeric color-register HSV sampling bands (T-988).
Reads COLOR_REGISTER_BANDS_TOML (`[register.<name>]` stanzas; each of
`hue`/`sat`/`val` a 2-element `[min, max]` integer array). The fill seed
samples a single (hue, sat, val) point uniformly within a template's
register band per building (D-235) — this baker only validates + stores
the bands.
V-TT-07: (1) every band's ranges are in-bounds (hue 0..36000, sat/val
0..10000) with `min < max`; (2) every `color_register` referenced by
`trait_templates.visual_bundle.color_register` (checked against the
just-baked table, same self-consistency reasoning as V-TT-06) has a band
here — coverage, not just shape. Must run AFTER populate_trait_templates.
Absent source -> 0 rows.
"""
if not COLOR_REGISTER_BANDS_TOML.exists():
if not dry_run:
conn.execute("DELETE FROM color_register_bands")
return 0
with open(COLOR_REGISTER_BANDS_TOML, "rb") as f:
data = tomllib.load(f)
registers = data.get("register", {})
errors: list[str] = []
rows: list[tuple] = []
for name, band in registers.items():
values: dict[str, tuple[int, int]] = {}
for axis, bound in _COLOR_AXIS_BOUNDS.items():
pair = band.get(axis) if isinstance(band, dict) else None
if not (isinstance(pair, list) and len(pair) == 2):
errors.append(
f"V-TT-07: color_register_bands '{name}': '{axis}' must be a 2-element "
"[min, max] array"
)
continue
lo, hi = pair
if isinstance(lo, bool) or isinstance(hi, bool) or not (
isinstance(lo, int) and isinstance(hi, int)
):
errors.append(
f"V-TT-07: color_register_bands '{name}.{axis}': min/max must be integers, "
f"got {pair!r}"
)
continue
if not (0 <= lo < hi <= bound):
errors.append(
f"V-TT-07: color_register_bands '{name}.{axis}': range [{lo}, {hi}] must "
f"satisfy 0 <= min < max <= {bound}"
)
continue
values[axis] = (lo, hi)
if len(values) == 3:
rows.append((
name,
values["hue"][0], values["hue"][1],
values["sat"][0], values["sat"][1],
values["val"][0], values["val"][1],
))
# Coverage half of V-TT-07: every color_register the catalog actually
# references must have a band.
referenced: set[str] = set()
for (vb_json,) in conn.execute("SELECT visual_bundle FROM trait_templates"):
vb = json.loads(vb_json) if vb_json else {}
reg = vb.get("color_register")
if reg:
referenced.add(reg)
banded = {r[0] for r in rows}
for reg in sorted(referenced - banded):
errors.append(f"V-TT-07: color_register '{reg}' referenced by trait_templates but has no band")
if errors:
print(f" COLOR REGISTER BAND ERRORS ({len(errors)}):")
for e in errors:
print(f" - {e}")
raise ImportAborted()
if not dry_run:
conn.execute("DELETE FROM color_register_bands")
conn.executemany(
"""INSERT INTO color_register_bands
(color_register, hue_min, hue_max, sat_min, sat_max, val_min, val_max)
VALUES (?,?,?,?,?,?,?)""",
rows,
)
return len(rows)
def populate_atlas_body_trait_bias(conn: sqlite3.Connection, dry_run: bool) -> int:
"""Bake the sparse per-body hero pins into atlas_body_trait_bias (#993).
Reads ARCHITECTURE_TRAIT_BIAS_TOML (`[[bias]]` array). FK-validates body_id
against bodies and template_tag against trait_templates (which must be baked
first), and validates bias_kind + the basis-point multiplier ranges
(boost 10001..30000 = <=3x; suppress 3300..9999 = >=0.33x never 0; pin: no
multiplier), plus V-TT-05: at most _MAX_PINS_PER_BODY pins per body (pins
count toward K, D-232). Hero-pin *content* is authored in #1017. Absent
source -> 0 rows. Must run AFTER populate_trait_templates.
"""
if not ARCHITECTURE_TRAIT_BIAS_TOML.exists():
if not dry_run:
conn.execute("DELETE FROM atlas_body_trait_bias")
return 0
with open(ARCHITECTURE_TRAIT_BIAS_TOML, "rb") as f:
data = tomllib.load(f)
entries = data.get("bias", [])
body_ids = {r[0] for r in conn.execute("SELECT body_id FROM bodies")}
tags = {r[0] for r in conn.execute("SELECT tag FROM trait_templates")}
errors: list[str] = []
rows: list[tuple] = []
seen: set[tuple] = set()
for i, b in enumerate(entries):
bid = b.get("body_id")
tag = b.get("template_tag")
kind = b.get("bias_kind")
mult = b.get("weight_multiplier_bps")
loc = f"bias[{i}] ({bid}/{tag})"
if bid not in body_ids:
errors.append(f"{loc}: body_id not in bodies")
if tag not in tags:
errors.append(f"{loc}: template_tag not in trait_templates")
if kind not in _TRAIT_BIAS_KINDS:
errors.append(f"{loc}: bias_kind '{kind}' invalid (pin|boost|suppress)")
if (bid, tag) in seen:
errors.append(f"{loc}: duplicate (body_id, template_tag)")
seen.add((bid, tag))
if kind == "boost" and not (mult and 10001 <= mult <= 30000):
errors.append(f"{loc}: boost weight_multiplier_bps must be 10001..30000 (<=3x), got {mult}")
if kind == "suppress" and not (mult and 3300 <= mult <= 9999):
errors.append(f"{loc}: suppress weight_multiplier_bps must be 3300..9999 (>=0.33x, never 0), got {mult}")
if kind == "pin" and mult is not None:
errors.append(f"{loc}: pin is mandatory and must not carry weight_multiplier_bps (got {mult})")
rows.append((bid, tag, kind, mult, b.get("note")))
# V-TT-05 (PR #173 review H4): pins count toward K (D-232) and the largest
# possible K is 5 (ComplexityTier::Full) — more pins than that can never
# fit any body's vocabulary budget and would force the draw past K.
pin_counts: dict[str, int] = {}
for bid, _tag, kind, _mult, _note in rows:
if kind == "pin":
pin_counts[bid] = pin_counts.get(bid, 0) + 1
for bid, count in sorted(pin_counts.items()):
if count > _MAX_PINS_PER_BODY:
errors.append(
f"V-TT-05: body '{bid}' has {count} pins — more than the maximum "
f"K of {_MAX_PINS_PER_BODY} (ComplexityTier::Full); pins count toward K (D-232)"
)
if errors:
print(f" TRAIT BIAS ERRORS ({len(errors)}):")
for e in errors:
print(f" - {e}")
raise ImportAborted()
if not dry_run:
conn.execute("DELETE FROM atlas_body_trait_bias")
conn.executemany(
"""INSERT INTO atlas_body_trait_bias
(body_id, template_tag, bias_kind, weight_multiplier_bps, note)
VALUES (?,?,?,?,?)""",
rows,
)
return len(rows)