close guardrail gaps and fix the scoring pipeline top to bottom

Constraint resolver: aggregate mass/footprint across a combo instead of
pairwise-only checks, treat medium/atmosphere as agreement not supply/demand,
reduce multi-provider checks by best/sum instead of AND-ing every provider,
fail closed on unrecognized mutex values, add a propulsion-viability
(thrust-to-weight) rule. Seed data updated to match (nuclear/solar-sail
footprint floors, water-medium exclusions, explicit ground/gravity providers).

Domain metric units were stored globally per metric name instead of
per-domain, silently corrupting cost_efficiency for every domain but the
first one seeded — fixed with a schema migration.

Stub estimator's cost_efficiency/safety/availability/reliability were a
backwards formula and flat constants; replaced with heuristics grounded in
each entity's thrust_profile/energy_form/infrastructure.

LLM estimate_physics() now receives each metric's unit and expected range
instead of a bare name, fixing wildly miscalibrated estimates traced back to
the prompt's own hardcoded example anchoring the model to the wrong order of
magnitude. Sharpened the safety-estimation and plausibility-review prompts.
Deduped provider parsing logic into llm/parsing.py.

Web pipeline form can now pick an LLM provider per run instead of only via
server env var.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-26 00:13:16 -05:00
parent 63295ab80e
commit 434df718d7
18 changed files with 836 additions and 175 deletions

View File

@@ -12,7 +12,113 @@ from physcom.engine.constraint_resolver import ConstraintResolver, ConstraintRes
from physcom.engine.scorer import Scorer
from physcom.llm.base import LLMProvider, LLMRateLimitError
from physcom.models.combination import Combination, ScoredResult
from physcom.models.domain import Domain
from physcom.models.domain import Domain, MetricBound
# Stub-estimator heuristics (used only when no LLM provider is configured).
# Keyed by the same categorical vocabulary already used in seed data — never
# by entity name, so new entities inherit sensible behavior automatically.
# How controllable a thrust delivery profile is — bursty/extreme profiles are
# harder to control and cost more per use (ammunition, propellant, wear) than
# steady ones. Missing values fall back to a neutral 1.0/0.6.
THRUST_PROFILE_COST_MULTIPLIER: dict[str, float] = {
"low_continuous": 1.0,
"continuous_low": 1.0,
"moderate_continuous": 1.1,
"high_continuous": 1.3,
"extreme_continuous": 1.6,
"high_burst": 2.5,
"extreme_burst": 4.0,
}
THRUST_PROFILE_SAFETY: dict[str, float] = {
"low_continuous": 0.9,
"continuous_low": 0.9,
"moderate_continuous": 0.75,
"high_continuous": 0.6,
"extreme_continuous": 0.45,
"high_burst": 0.3,
"extreme_burst": 0.15,
}
# Baseline hazard of the energy form itself, independent of delivery profile.
ENERGY_FORM_SAFETY: dict[str, float] = {
"biological": 0.9,
"electrical": 0.85,
"wind": 0.9,
"gravitational": 0.9,
"radiation_pressure": 0.85,
"kinetic_stored": 0.7,
"pneumatic": 0.75,
"chemical_combustible": 0.6,
"ion_propellant": 0.75,
"chemical_propellant": 0.4,
"chemical_explosive": 0.3,
"nuclear_thermal": 0.35,
}
# Rough $/m base cost by energy form — renewables/muscle power are ~free,
# consumables (propellant, ammunition, nuclear fuel) cost real money per use.
# This is a categorical placeholder, not a physics formula — energy_density
# (J/kg) can't give a $/m figure on its own since it says nothing about price.
ENERGY_FORM_BASE_COST: dict[str, float] = {
"wind": 1e-6,
"gravitational": 1e-6,
"radiation_pressure": 1e-6,
"electrical": 3e-5,
"kinetic_stored": 2e-5,
"biological": 5e-5,
"pneumatic": 4e-5,
"chemical_combustible": 8e-5,
"nuclear_thermal": 1e-3,
"ion_propellant": 2e-3,
"chemical_propellant": 5e-3,
"chemical_explosive": 8e-3,
}
# How available the required infrastructure/fuel supply chain is today.
# Multiple matches in one combo (e.g. a platform's road_network requirement
# plus a storage's fuel_infrastructure requirement) are averaged.
INFRASTRUCTURE_AVAILABILITY: dict[tuple[str, str], float] = {
("road_network", "true"): 0.95,
("rail_network", "true"): 0.8,
("runway", "true"): 0.5,
("tow_or_winch", "true"): 0.5,
("hyperloop_tube", "true"): 0.1,
("launch_facility", "true"): 0.05,
("fuel_infrastructure", "none"): 1.0,
("fuel_infrastructure", "fuel_station"): 0.95,
("fuel_infrastructure", "charging_station"): 0.85,
("fuel_infrastructure", "cng_station"): 0.5,
("fuel_infrastructure", "coal_supply"): 0.5,
("fuel_infrastructure", "hydrogen_station"): 0.25,
("fuel_infrastructure", "compressed_air_station"): 0.3,
("fuel_infrastructure", "ammunition"): 0.3,
("fuel_infrastructure", "jet_fuel"): 0.6,
("fuel_infrastructure", "solid_propellant"): 0.15,
("fuel_infrastructure", "nuclear_fuel"): 0.05,
("fuel_infrastructure", "xenon_propellant"): 0.05,
}
# Crude freight-capacity proxy: kg of cargo per kg of vehicle structural mass.
CARGO_KG_PER_STRUCTURAL_KG: float = 500
# How mechanically proven/predictable an energy form is in practice — distinct
# from safety (risk when something goes wrong) and thrust_profile (delivery
# smoothness). Missing values fall back to a neutral 0.6.
ENERGY_FORM_RELIABILITY: dict[str, float] = {
"chemical_combustible": 0.85,
"electrical": 0.85,
"biological": 0.8,
"gravitational": 0.7,
"pneumatic": 0.7,
"kinetic_stored": 0.65,
"wind": 0.6,
"ion_propellant": 0.6,
"nuclear_thermal": 0.55,
"chemical_propellant": 0.5,
"radiation_pressure": 0.5,
"chemical_explosive": 0.45,
}
@dataclass
@@ -124,7 +230,6 @@ class Pipeline:
self.repo.update_pipeline_run(run_id, total_combos=len(combos))
# Prepare metric lookup
metric_names = [mb.metric_name for mb in domain.metric_bounds]
bounds_by_name = {mb.metric_name: mb for mb in domain.metric_bounds}
# ── Combo-first loop ─────────────────────────────────────
@@ -212,10 +317,10 @@ class Pipeline:
description = _describe_combination(combo)
if self.llm:
raw_metrics = self.llm.estimate_physics(
description, metric_names
description, domain.metric_bounds
)
else:
raw_metrics = self._stub_estimate(combo, metric_names)
raw_metrics = self._stub_estimate(combo, domain.metric_bounds)
# Save raw estimates immediately (crash-safe)
estimate_dicts = []
@@ -435,15 +540,32 @@ class Pipeline:
self.repo.update_pipeline_run(run_id, status="running")
def _stub_estimate(
self, combo: Combination, metric_names: list[str]
self, combo: Combination, metric_bounds: list[MetricBound]
) -> dict[str, float]:
"""Simple heuristic estimation from dependency data (all values in SI base units)."""
"""Simple heuristic estimation from dependency data (all values in SI base units).
cost_efficiency/safety/availability/reliability are driven by the
actuator's thrust_profile and energy_form and the combo's
infrastructure requirements — categorical properties every entity
already declares — rather than flat constants or a formula that
conflates power_density (W/kg, intensive) with cost.
cost_efficiency additionally checks the domain's declared unit:
"$/(kg·m)" (freight-style domains) isn't a rescaling of "$/m" — it's
a different quantity that needs dividing by cargo mass, not a
conversion factor.
"""
metric_names = [mb.metric_name for mb in metric_bounds]
units_by_name = {mb.metric_name: mb.unit for mb in metric_bounds}
raw: dict[str, float] = {m: 0.0 for m in metric_names}
# Extract intrinsic properties from entities
power_density = 0.0 # W/kg
energy_density = 0.0 # J/kg
mass = 100.0 # kg, default
mass_total = 0.0 # kg, extensive — components share one vehicle
thrust_profile: str | None = None
energy_form: str | None = None
infra_matches: list[float] = []
for entity in combo.entities:
for dep in entity.dependencies:
if dep.key == "power_density" and dep.constraint_type == "provides":
@@ -451,19 +573,44 @@ class Pipeline:
if dep.key == "energy_density" and dep.constraint_type == "provides":
energy_density = max(energy_density, float(dep.value))
if dep.key == "mass" and dep.constraint_type == "range_min":
mass = max(mass, float(dep.value))
mass_total += float(dep.value)
if dep.key == "thrust_profile" and dep.constraint_type == "provides":
thrust_profile = dep.value
if dep.key == "energy_form" and dep.constraint_type == "requires":
energy_form = dep.value
if dep.category == "infrastructure" and dep.constraint_type == "requires":
match = INFRASTRUCTURE_AVAILABILITY.get((dep.key, dep.value))
if match is not None:
infra_matches.append(match)
mass = mass_total if mass_total > 0 else 100.0 # kg, default if undeclared
cargo_capacity_kg = mass * CARGO_KG_PER_STRUCTURAL_KG
if "power_density" in raw:
raw["power_density"] = power_density
if "cost_efficiency" in raw:
raw["cost_efficiency"] = max(1e-5, 2e-3 - power_density / 1e6)
base_cost = ENERGY_FORM_BASE_COST.get(energy_form, 5e-4)
cost_mult = THRUST_PROFILE_COST_MULTIPLIER.get(thrust_profile, 1.0)
cost_per_meter = base_cost * cost_mult
if units_by_name.get("cost_efficiency") == "$/(kg·m)":
raw["cost_efficiency"] = cost_per_meter / max(cargo_capacity_kg, 1.0)
else:
raw["cost_efficiency"] = cost_per_meter
if "safety" in raw:
raw["safety"] = 0.5
candidates = [
v for v in (
THRUST_PROFILE_SAFETY.get(thrust_profile),
ENERGY_FORM_SAFETY.get(energy_form),
)
if v is not None
]
raw["safety"] = min(candidates) if candidates else 0.6
if "availability" in raw:
raw["availability"] = 0.5
raw["availability"] = (
sum(infra_matches) / len(infra_matches) if infra_matches else 0.5
)
if "range_fuel" in raw:
raw["range_fuel"] = min(energy_density * 2.78, 1e13)
@@ -472,7 +619,7 @@ class Pipeline:
raw["range_degradation"] = 365 * 86400
if "cargo_capacity" in raw:
raw["cargo_capacity"] = mass * 500
raw["cargo_capacity"] = cargo_capacity_kg
if "cargo_capacity_kg" in raw:
raw["cargo_capacity_kg"] = mass * 0.3
@@ -481,6 +628,6 @@ class Pipeline:
raw["environmental_impact"] = max(0.0, power_density * 2e-7)
if "reliability" in raw:
raw["reliability"] = 0.5
raw["reliability"] = ENERGY_FORM_RELIABILITY.get(energy_form, 0.6)
return raw