let domains author their own pass-2 estimator as formulas, not Python

Pass 2 previously only knew one estimator: a hardcoded physics model that
matches dimensions literally named platform/actuator/energy_storage. Any
domain outside that shape (e.g. archery) got all-zero estimates and failed
every combo. Domains can now declare free variables and per-metric formulas
as data instead; a safe AST-based evaluator (engine/formula.py, no eval())
resolves declared entity properties via dep(key, constraint_type) and
generalizes the existing hand-nested mass-budget search into an N-variable
recursive optimizer. Fully additive -- the legacy platform/actuator/
energy_storage path is untouched and still runs unchanged for every domain
that declares no formulas.

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

CLAUDE.md

@@ -60,7 +60,7 @@ tests/                     # pytest, uses seeded_repo fixture from conftest.py
 ## Data flow (pipeline passes)
 
 1. **Pass 1 — Constraints**: `ConstraintResolver.resolve()` → blocked/conditional/valid. Blocked combos get a result row and `continue`.
-2. **Pass 2 — Estimation**: LLM or `_stub_estimate()` → raw metric values. Saved immediately via `save_raw_estimates()` (normalized_score=NULL).
+2. **Pass 2 — Estimation**: `_estimate_physics()` (deterministic physics engine; estimator-only, no LLM) → raw metric values. Saved immediately via `save_raw_estimates()` (normalized_score=NULL).
 3. **Pass 3 — Scoring**: `Scorer.score_combination()` → log-normalized scores + weighted geometric mean composite. Saves via `save_scores()` + `save_result()`.
 4. **Pass 4 — LLM Review**: Only for above-threshold combos with an LLM provider. No real provider yet (only `MockLLMProvider`).
 5. **Pass 5 — Human Review**: Manual via web UI results page.

LOGIC DOCS/003-gpu-batching-for-scaled-optimizer.md

@@ -0,0 +1,34 @@
+# GPU batching for the mass-allocation optimizer — not yet, here's the threshold
+
+## Context
+
+`Pipeline._decide_masses`'s joint platform/actuator/storage optimizer (coarse-to-fine grid search, see `_search_best_allocation`) calls its objective function roughly 11,700 times per combo. Profiling confirmed this dominates pipeline runtime: for 50 combos, 582,920 objective-function calls, each doing scalar arithmetic (power_density, the drag cubic solve, normalize, composite_score) on one `(platform, actuator, storage)` triple. The cost is Python's per-call overhead (bytecode dispatch, refcounting, attribute lookups), not the arithmetic itself — the individual formulas are cheap.
+
+## Why GPU doesn't fit today
+
+A single combo's grid is only ~169 points per round (13×13). GPUs pay off when there's enough independent parallel work to amortize kernel-launch and host↔device transfer overhead (each typically tens of microseconds to low milliseconds); 169 elements doesn't come close, and that overhead would be paid repeatedly — once per grid round, ~6-10 rounds per combo.
+
+The parallelism that actually exists is **across combos**, not within one combo's grid — every combo's optimization is fully independent of every other's. At current scale (~180 combos reach the optimizer per domain after Pass 1 filtering), batching every combo's grid into one array gives ~180×169 ≈ 30K elements per round — borderline, probably a wash against plain CPU numpy.
+
+## The actual threshold
+
+Combo count scales **multiplicatively** with added dimensions or entities per dimension (today: 11 platforms × 15 actuators × 18 storages ≈ 2,970 combos, ~180 of which reach the optimizer). Add a 4th dimension with even 10 options and total combos scale to ~30,000, with optimizer-eligible combos likely growing roughly proportionally to ~1,800/domain — batched grid size ≈ 300K elements/round. A 5th dimension does it again, into the low millions. That's the regime where a GPU's thousands of cores start meaningfully outrunning a CPU's 4-16-wide SIMD lanes.
+
+So: not "more dimensions" directly, but the combo×grid batch size those dimensions produce. Rough rule of thumb from this discussion:
+
+- **Tens of thousands of elements/round** (current scale, or a modest one-dimension addition): plain CPU numpy vectorization is enough, no GPU.
+- **Hundreds of thousands to low millions**: GPU batching across combos starts being worth evaluating.
+
+## What GPU batching would actually require
+
+Not just "swap numpy for cupy." It means restructuring `_process_pass2` from combo-first (one combo through the optimizer at a time) to batch-first (a chunk of N combos' grids evaluated together as one array with a "combo" axis, broadcasting each combo's own constants — `k_act`, `k_med`, `e_dens`, drag coefficients, mass bounds — across that axis). That's a real architectural change, not a drop-in acceleration:
+
+- **CLAUDE.md documents the pipeline as deliberately combo-first**: "each combo goes through all requested passes before the next combo starts... Progress is persisted per-combo (crash-safe, resumable)." Batching means checkpointing per-*batch*, not per-combo — a real (if manageable) tradeoff against that resumability guarantee, not something that comes free alongside the speedup.
+- Every branch in `_raw_physics_from_masses` / `_solve_achievable_speed_mps` (biological floors, ambient energy forms, degenerate fallbacks, the cubic's edge cases) needs to become `np.where(condition, a, b)` instead of `if/else` — careful, error-prone translation work, not mechanical.
+
+## Decision
+
+Don't build this now — no current need at ~180 combos/domain. If dimension count grows enough to matter, do it in two steps:
+
+1. **CPU numpy vectorization first** (batch one combo's grid into arrays, evaluate with vectorized ops instead of a Python double-loop). This is needed regardless of GPU or not, since it's the same rewrite either way, and profiling suggests it could plausibly give 10-50x on its own by replacing ~11,700 Python calls/combo with a couple dozen numpy batch calls.
+2. **Re-profile at the new scale.** Only reach for GPU batching-across-combos if CPU numpy is still the dominant cost after step 1, and only once the batched element count is actually in GPU-favorable territory (see thresholds above) — this is a "measure, then decide" call, not something to build ahead of need.

src/physcom/db/repository.py

@@ -9,7 +9,7 @@ from datetime import datetime, timezone
 from typing import Sequence
 
 from physcom.models.entity import Dependency, Entity
-from physcom.models.domain import Domain, DomainConstraint, MetricBound
+from physcom.models.domain import Domain, DomainConstraint, FreeVariable, MetricBound, MetricFormula
 from physcom.models.combination import Combination
 
 
@@ -286,6 +286,10 @@ class Repository:
                     "INSERT OR IGNORE INTO domain_constraints (domain_id, key, value) VALUES (?, ?, ?)",
                     (domain.id, dc.key, val),
                 )
+        for fv in domain.free_variables:
+            self.add_free_variable(domain.id, fv, commit=False)
+        for mf in domain.metric_formulas:
+            self.add_metric_formula(domain.id, mf, commit=False)
         self.conn.commit()
         return domain
 
@@ -299,6 +303,31 @@ class Repository:
             by_key.setdefault(r["key"], []).append(r["value"])
         return [DomainConstraint(key=k, allowed_values=v) for k, v in by_key.items()]
 
+    def _load_free_variables(self, domain_id: int) -> list[FreeVariable]:
+        rows = self.conn.execute(
+            """SELECT id, name, sort_order, floor_formula, ceiling_formula
+               FROM domain_free_variables WHERE domain_id = ? ORDER BY sort_order""",
+            (domain_id,),
+        ).fetchall()
+        return [
+            FreeVariable(
+                id=r["id"], name=r["name"], sort_order=r["sort_order"],
+                floor_formula=r["floor_formula"], ceiling_formula=r["ceiling_formula"],
+            )
+            for r in rows
+        ]
+
+    def _load_metric_formulas(self, domain_id: int) -> list[MetricFormula]:
+        rows = self.conn.execute(
+            """SELECT id, metric_name, formula
+               FROM domain_metric_formulas WHERE domain_id = ? ORDER BY metric_name""",
+            (domain_id,),
+        ).fetchall()
+        return [
+            MetricFormula(id=r["id"], metric_name=r["metric_name"], formula=r["formula"])
+            for r in rows
+        ]
+
     def _load_domain(self, where: str, param: str | int) -> Domain | None:
         row = self.conn.execute(f"SELECT * FROM domains WHERE {where} = ?", (param,)).fetchone()
         if not row:
@@ -325,6 +354,8 @@ class Repository:
                 for w in weights
             ],
             constraints=self._load_domain_constraints(row["id"]),
+            free_variables=self._load_free_variables(row["id"]),
+            metric_formulas=self._load_metric_formulas(row["id"]),
         )
 
     def get_domain(self, name: str) -> Domain | None:
@@ -376,12 +407,63 @@ class Repository:
         )
         self.conn.commit()
 
+    # ── Free variables & metric formulas ──────────────────────────
+
+    def add_free_variable(self, domain_id: int, fv: FreeVariable, commit: bool = True) -> FreeVariable:
+        cur = self.conn.execute(
+            """INSERT INTO domain_free_variables
+               (domain_id, name, sort_order, floor_formula, ceiling_formula)
+               VALUES (?, ?, ?, ?, ?)""",
+            (domain_id, fv.name, fv.sort_order, fv.floor_formula, fv.ceiling_formula),
+        )
+        fv.id = cur.lastrowid
+        if commit:
+            self.conn.commit()
+        return fv
+
+    def update_free_variable(self, fv_id: int, fv: FreeVariable) -> None:
+        self.conn.execute(
+            """UPDATE domain_free_variables
+               SET name = ?, sort_order = ?, floor_formula = ?, ceiling_formula = ?
+               WHERE id = ?""",
+            (fv.name, fv.sort_order, fv.floor_formula, fv.ceiling_formula, fv_id),
+        )
+        self.conn.commit()
+
+    def delete_free_variable(self, fv_id: int) -> None:
+        self.conn.execute("DELETE FROM domain_free_variables WHERE id = ?", (fv_id,))
+        self.conn.commit()
+
+    def add_metric_formula(self, domain_id: int, mf: MetricFormula, commit: bool = True) -> MetricFormula:
+        cur = self.conn.execute(
+            """INSERT OR REPLACE INTO domain_metric_formulas (domain_id, metric_name, formula)
+               VALUES (?, ?, ?)""",
+            (domain_id, mf.metric_name, mf.formula),
+        )
+        mf.id = cur.lastrowid
+        if commit:
+            self.conn.commit()
+        return mf
+
+    def update_metric_formula(self, mf_id: int, mf: MetricFormula) -> None:
+        self.conn.execute(
+            "UPDATE domain_metric_formulas SET metric_name = ?, formula = ? WHERE id = ?",
+            (mf.metric_name, mf.formula, mf_id),
+        )
+        self.conn.commit()
+
+    def delete_metric_formula(self, mf_id: int) -> None:
+        self.conn.execute("DELETE FROM domain_metric_formulas WHERE id = ?", (mf_id,))
+        self.conn.commit()
+
     def delete_domain(self, domain_id: int) -> None:
         self.conn.execute("DELETE FROM pipeline_runs WHERE domain_id = ?", (domain_id,))
         self.conn.execute("DELETE FROM combination_results WHERE domain_id = ?", (domain_id,))
         self.conn.execute("DELETE FROM combination_scores WHERE domain_id = ?", (domain_id,))
         self.conn.execute("DELETE FROM domain_metric_weights WHERE domain_id = ?", (domain_id,))
         self.conn.execute("DELETE FROM domain_constraints WHERE domain_id = ?", (domain_id,))
+        self.conn.execute("DELETE FROM domain_free_variables WHERE domain_id = ?", (domain_id,))
+        self.conn.execute("DELETE FROM domain_metric_formulas WHERE domain_id = ?", (domain_id,))
         self.conn.execute("DELETE FROM domains WHERE id = ?", (domain_id,))
         self.conn.commit()
 
@@ -475,7 +557,7 @@ class Repository:
         self, combo_id: int, status: str, block_reason: str | None = None, commit: bool = True
     ) -> None:
         # Don't downgrade from higher pass states — preserves human/LLM review data
-        if status in ("scored", "llm_reviewed") or status.endswith("_fail"):
+        if status in ("scored", "llm_reviewed", "valid") or status.endswith("_fail"):
             row = self.conn.execute(
                 "SELECT status FROM combinations WHERE id = ?", (combo_id,)
             ).fetchone()
@@ -488,6 +570,13 @@ class Repository:
                     return
                 if status == "llm_reviewed" and cur == "reviewed":
                     return
+                # "valid" is pass 1's domain-agnostic result -- a combo
+                # already at any later pass state (or a fail state) has
+                # progressed past pass 1 already, in this domain or
+                # another one sharing the same combo. Pass 1 re-running
+                # for a different domain must not silently revert that.
+                if status == "valid" and cur not in (None, "valid"):
+                    return
         self.conn.execute(
             "UPDATE combinations SET status = ?, block_reason = ? WHERE id = ?",
             (status, block_reason, combo_id),
@@ -899,6 +988,8 @@ class Repository:
         self.conn.execute("DELETE FROM entities")
         self.conn.execute("DELETE FROM domain_metric_weights")
         self.conn.execute("DELETE FROM domain_constraints")
+        self.conn.execute("DELETE FROM domain_free_variables")
+        self.conn.execute("DELETE FROM domain_metric_formulas")
         self.conn.execute("DELETE FROM domains")
         self.conn.execute("DELETE FROM metrics")
         self.conn.execute("DELETE FROM dimensions")

src/physcom/db/schema.py

@@ -120,6 +120,24 @@ CREATE TABLE IF NOT EXISTS domain_constraints (
     UNIQUE(domain_id, key, value)
 );
 
+CREATE TABLE IF NOT EXISTS domain_free_variables (
+    id              INTEGER PRIMARY KEY AUTOINCREMENT,
+    domain_id       INTEGER NOT NULL REFERENCES domains(id),
+    name            TEXT NOT NULL,
+    sort_order      INTEGER NOT NULL,
+    floor_formula   TEXT NOT NULL,
+    ceiling_formula TEXT NOT NULL,
+    UNIQUE(domain_id, name)
+);
+
+CREATE TABLE IF NOT EXISTS domain_metric_formulas (
+    id          INTEGER PRIMARY KEY AUTOINCREMENT,
+    domain_id   INTEGER NOT NULL REFERENCES domains(id),
+    metric_name TEXT NOT NULL,
+    formula     TEXT NOT NULL,
+    UNIQUE(domain_id, metric_name)
+);
+
 CREATE INDEX IF NOT EXISTS idx_deps_entity ON dependencies(entity_id);
 CREATE INDEX IF NOT EXISTS idx_deps_category_key ON dependencies(category, key);
 CREATE INDEX IF NOT EXISTS idx_combo_status ON combinations(status);

src/physcom/engine/constraint_resolver.py

@@ -32,6 +32,13 @@ CATEGORY_SEVERITY: dict[str, str] = {
     "energy": "block",
     "environment": "block",
     "infrastructure": "skip",
+    # Safety-critical physical necessities (radiation shielding, containment,
+    # etc.) -- same severity as energy/environment, not the softer default
+    # "warn" every other category falls through to. Missing this entry meant
+    # Nuclear Thermal Drive/Nuclear Fuel's "material" requires (radiation_
+    # shielding) defaulted to a non-blocking warning nothing in the catalog
+    # ever satisfies -- see LOGIC DOCS/002's "silent guardrail hole" pattern.
+    "material": "block",
 }
 
 # For provides-vs-range_min: deficit > this ratio = hard block, else warning
@@ -53,6 +60,40 @@ KEY_AGGREGATION: dict[str, str] = {
 OVERRUN_TOLERANCE: float = 0.10
 
 
+def aggregate_dependency_value(
+    combination: Combination,
+    key: str,
+    constraint_type: str,
+    key_aggregation: dict[str, str] | None = None,
+) -> float | None:
+    """Collapse every numeric dependency matching (key, constraint_type)
+    across a combination's entities into one system-level number: summed for
+    extensive keys (KEY_AGGREGATION says "sum", e.g. mass/footprint --
+    independent components sharing one physical vehicle), otherwise the
+    strongest single value wins (today's default pairwise behavior). Returns
+    None if no entity declares a matching numeric dependency. Shared by
+    ConstraintResolver._check_provides_vs_range and the formula evaluator's
+    injected dep() builtin (see engine/formula.py, engine/pipeline.py) so
+    there's one implementation of "how do these entities' declared numbers
+    combine," not two.
+    """
+    key_aggregation = KEY_AGGREGATION if key_aggregation is None else key_aggregation
+    values: list[float] = []
+    for entity in combination.entities:
+        for dep in entity.dependencies:
+            if dep.key != key or dep.constraint_type != constraint_type:
+                continue
+            try:
+                values.append(float(dep.value))
+            except (ValueError, TypeError):
+                continue
+    if not values:
+        return None
+    if key_aggregation.get(key) == "sum":
+        return sum(values)
+    return max(values)
+
+
 @dataclass
 class ConstraintResult:
     """Outcome of constraint resolution for a combination."""
@@ -243,11 +284,13 @@ class ConstraintResolver:
                     required.setdefault(dep.key, []).append((entity.name, val))
 
         for key in set(provided) & set(required):
+            prov_val = aggregate_dependency_value(
+                combination, key, "provides", self.key_aggregation
+            )
             if self.key_aggregation.get(key) == "sum":
                 prov_name = " + ".join(name for name, _ in provided[key])
-                prov_val = sum(val for _, val in provided[key])
             else:
-                prov_name, prov_val = max(provided[key], key=lambda t: t[1])
+                prov_name, _ = max(provided[key], key=lambda t: t[1])
 
             for req_name, req_val in required[key]:
                 if prov_val < req_val * self.deficit_threshold:

src/physcom/engine/formula.py

@@ -0,0 +1,139 @@
+"""Safe arithmetic expression language for domain-authored estimator formulas.
+
+No eval()/exec() anywhere -- compile_formula validates every AST node against
+a fixed whitelist (arithmetic, numeric/string constants, name lookups, calls
+to an explicitly supplied function table) before evaluate_formula ever walks
+it, so a formula can express "how much drawback_force does this combo
+produce" but never anything with attribute/subscript/import access.
+"""
+
+from __future__ import annotations
+
+import ast
+import math
+import operator
+from dataclasses import dataclass
+from typing import Callable
+
+
+class FormulaError(Exception):
+    """Raised for invalid formula syntax/structure or a failed evaluation."""
+
+
+_ALLOWED_NODES = (
+    ast.Expression, ast.BinOp, ast.UnaryOp, ast.Constant, ast.Name, ast.Load,
+    ast.Call, ast.keyword,
+    ast.Add, ast.Sub, ast.Mult, ast.Div, ast.Pow, ast.USub, ast.UAdd,
+)
+
+_BINOPS: dict[type, Callable[[float, float], float]] = {
+    ast.Add: operator.add,
+    ast.Sub: operator.sub,
+    ast.Mult: operator.mul,
+    ast.Div: operator.truediv,
+    ast.Pow: operator.pow,
+}
+
+_UNARYOPS: dict[type, Callable[[float], float]] = {
+    ast.USub: operator.neg,
+    ast.UAdd: operator.pos,
+}
+
+DEFAULT_FUNCTIONS: dict[str, Callable[..., float]] = {
+    "min": min,
+    "max": max,
+    "abs": abs,
+    "sqrt": math.sqrt,
+    "log": math.log,
+    "log1p": math.log1p,
+    "exp": math.exp,
+}
+
+
+@dataclass(frozen=True)
+class CompiledFormula:
+    source: str
+    _tree: ast.Expression
+
+
+def _validate(tree: ast.AST) -> None:
+    for node in ast.walk(tree):
+        if not isinstance(node, _ALLOWED_NODES):
+            raise FormulaError(
+                f"disallowed expression element: {type(node).__name__}"
+            )
+        if isinstance(node, ast.Constant):
+            if isinstance(node.value, bool) or not isinstance(node.value, (int, float, str)):
+                raise FormulaError(
+                    f"disallowed constant type: {type(node.value).__name__}"
+                )
+        if isinstance(node, ast.Name) and node.id.startswith("__"):
+            raise FormulaError(f"disallowed name: {node.id}")
+        if isinstance(node, ast.Call) and not isinstance(node.func, ast.Name):
+            raise FormulaError("only direct function calls are allowed")
+
+
+def compile_formula(source: str) -> CompiledFormula:
+    try:
+        tree = ast.parse(source, mode="eval")
+    except SyntaxError as exc:
+        raise FormulaError(f"invalid syntax in '{source}': {exc}") from exc
+    _validate(tree)
+    return CompiledFormula(source=source, _tree=tree)
+
+
+def _eval(node: ast.AST, variables: dict[str, float], functions: dict[str, Callable]):
+    if isinstance(node, ast.Expression):
+        return _eval(node.body, variables, functions)
+    if isinstance(node, ast.Constant):
+        # Numeric constants are cast to float (not left as int) so a formula
+        # like a**b**c can't build an arbitrary-precision giant int before
+        # ever raising -- float exponentiation overflows to inf/OverflowError
+        # quickly instead. String constants (dep() key/constraint_type args)
+        # pass through unchanged.
+        return node.value if isinstance(node.value, str) else float(node.value)
+    if isinstance(node, ast.Name):
+        if node.id not in variables:
+            raise FormulaError(f"unknown variable '{node.id}'")
+        return variables[node.id]
+    if isinstance(node, ast.BinOp):
+        op = _BINOPS.get(type(node.op))
+        if op is None:
+            raise FormulaError(f"unsupported operator: {type(node.op).__name__}")
+        return op(
+            _eval(node.left, variables, functions),
+            _eval(node.right, variables, functions),
+        )
+    if isinstance(node, ast.UnaryOp):
+        op = _UNARYOPS.get(type(node.op))
+        if op is None:
+            raise FormulaError(f"unsupported operator: {type(node.op).__name__}")
+        return op(_eval(node.operand, variables, functions))
+    if isinstance(node, ast.Call):
+        fname = node.func.id  # validated as ast.Name by _validate
+        func = functions.get(fname)
+        if func is None:
+            raise FormulaError(f"unknown function '{fname}'")
+        args = [_eval(a, variables, functions) for a in node.args]
+        kwargs = {kw.arg: _eval(kw.value, variables, functions) for kw in node.keywords}
+        return func(*args, **kwargs)
+    raise FormulaError(f"unsupported expression element: {type(node).__name__}")
+
+
+def evaluate_formula(
+    compiled: CompiledFormula,
+    variables: dict[str, float],
+    functions: dict[str, Callable] | None = None,
+) -> float:
+    effective_functions = {**DEFAULT_FUNCTIONS, **(functions or {})}
+    try:
+        result = _eval(compiled._tree, variables, effective_functions)
+    except FormulaError:
+        raise
+    except (TypeError, ValueError, ArithmeticError) as exc:
+        raise FormulaError(f"error evaluating '{compiled.source}': {exc}") from exc
+    if not isinstance(result, (int, float)) or isinstance(result, bool):
+        raise FormulaError(
+            f"formula '{compiled.source}' did not evaluate to a number"
+        )
+    return float(result)

src/physcom/engine/pipeline.py

@@ -278,19 +278,33 @@ SPECIFIC_ENERGY_CONSUMPTION_J_PER_KG_M: dict[str, float] = {
 #
 # DRAG_POWER_COEFF_BY_MEDIUM below adds that missing term: a mass-INDEPENDENT
 # drag power coefficient (0.5 * air_density * drag_coefficient * frontal_area,
-# W per (m/s)^3) added on top of the existing mass-proportional term. Ground
+# W per (m/s)^3) added on top of the existing mass-proportional term.
-# only for now (the diagnosed case, and where a car-like reference
+#
-# cross-section is a defensible categorical estimate the way the rest of
+# "air" was originally left out of this table on the reasoning that its
-# this file's constants are); air's existing L/D-based model is already a
+# L/D-based cruise model was already a reasonable velocity-roughly-linear
-# reasonable velocity-roughly-linear cruise approximation and doesn't have
+# approximation -- true for computing energy per meter during a normal
-# this problem, and water hull drag would need its own (different) treatment
+# cruise, but WRONG for the exact same reason ground was wrong: L/D-based
-# rather than reusing a car's frontal area, so it's left as a known
+# drag force is also roughly velocity-independent within a design cruise
-# remaining gap rather than guessed at here.
+# band, so it's still just "resistance = mass-proportional constant" with
+# no v^2 term, and inverting power/resistance for achieved speed still had
+# no ceiling. Confirmed live: a Rotorcraft + Gas Turbine combo showed a
+# "speed" of 3,127 m/s (Mach 9) with phi4's pass-4 review flagging it
+# directly ("unrealistic for urban commuting, likely indicating an
+# error"). Added below with an aircraft-like reference cross-section.
+# Water hull drag would need its own (different) treatment rather than
+# reusing either reference, so it's left as a known remaining gap.
 DRAG_POWER_COEFF_BY_MEDIUM: dict[str, float] = {
     # 0.5 * rho_air(1.225 kg/m^3) * Cd(~0.3) * frontal_area(~2.2 m^2, small
     # car reference) -- sanity check: at 30 m/s (108 km/h) this alone costs
     # ~11kW, in the right ballpark for real highway cruise power.
     "ground": 0.5 * 1.225 * 0.3 * 2.2,
+    # 0.5 * rho_air(1.225) * Cd(~0.2, streamlined fuselage) * frontal_area
+    # (~1.74 m^2, small aircraft/rotorcraft reference) -- sanity check: at
+    # 60 m/s (a fast urban rotorcraft cruise) this alone costs ~46kW, a
+    # plausible fraction of a light helicopter's real cruise power (most
+    # of the rest goes to induced/rotor drag, not modeled here -- this
+    # coefficient only covers fuselage parasite drag).
+    "air": 0.5 * 1.225 * 0.2 * 1.74,
 }
 
 # Structural manufacturing cost, $ per kg of platform mass -- certification
@@ -660,7 +674,7 @@ class Pipeline:
             result.pass2_estimated += 1
             return
 
-        raw_metrics, feasible = self._stub_estimate(combo, domain.metric_bounds)
+        raw_metrics, feasible = self._estimate_physics(combo, domain)
 
         if not feasible:
             # No platform mass within its own declared ceiling could
@@ -693,7 +707,7 @@ class Pipeline:
                 estimate_dicts.append({
                     "metric_id": mb.metric_id,
                     "raw_value": rval,
-                    "estimation_method": "stub",
+                    "estimation_method": "physics_calc",
                     "confidence": 1.0,
                 })
         if estimate_dicts:
@@ -831,7 +845,7 @@ class Pipeline:
             self._wait_for_rate_limit(run_id, exc.retry_after)
             try:
                 review_result = self.llm.review_plausibility(
-                    description, raw_dict, score_dict, domain.metric_bounds
+                    description, raw_dict, score_dict, domain
                 )
             except LLMRateLimitError:
                 return  # still limited; skip, retry next run
@@ -876,9 +890,9 @@ class Pipeline:
         self, combo: Combination, bounds_by_name: dict[str, MetricBound]
     ) -> "_PhysicsContext | None":
         """Derive the entity-level physics inputs that don't depend on a
-        mass allocation choice -- shared by _stub_estimate (which picks the
+        mass allocation choice -- shared by _estimate_physics (which picks
-        allocation via solve or a special case) and _optimize_allocation
+        the allocation via _decide_masses) and evaluate_allocation (the
-        (which searches over candidate allocations). Returns None if the
+        explore-panel's direct evaluation). Returns None if the
         combo doesn't have the platform/actuator/storage shape this whole
         formula assumes (shouldn't happen for real combos, but a domain
         without all three dimensions requested would hit this)."""
@@ -964,17 +978,29 @@ class Pipeline:
         # scores is just which metric_name it declares. Floored at 0: a
         # storage mass bigger than the whole allowance leaves no cargo
         # room, not negative room.
-        lightship_mass = p_mass + actuator_mass
+        # Skip the arithmetic entirely for domains that don't score either
-        cargo_capacity_2_5x = max(0.0, lightship_mass * CARGO_KG_PER_STRUCTURAL_KG - storage_mass)
+        # cargo convention and don't need it as cost_efficiency's $/(kg·m)
-        cargo_capacity_0_3x = max(0.0, lightship_mass * 0.3 - storage_mass)
+        # denominator either -- this runs on every one of the ~11,000 grid
-        if "cargo_capacity" in bounds_by_name:
+        # points the search below tries per combo, so a domain like
-            out["cargo_capacity"] = cargo_capacity_2_5x
+        # interplanetary_travel (scores neither) shouldn't pay for it.
-        if "cargo_capacity_kg" in bounds_by_name:
+        needs_cargo = (
-            out["cargo_capacity_kg"] = cargo_capacity_0_3x
+            "cargo_capacity" in bounds_by_name
+            or "cargo_capacity_kg" in bounds_by_name
+            or units_by_name.get("cost_efficiency") == "$/(kg·m)"
+        )
+        cargo_capacity_2_5x = 0.0
+        if needs_cargo:
+            lightship_mass = p_mass + actuator_mass
+            cargo_capacity_2_5x = max(0.0, lightship_mass * CARGO_KG_PER_STRUCTURAL_KG - storage_mass)
+            cargo_capacity_0_3x = max(0.0, lightship_mass * 0.3 - storage_mass)
+            if "cargo_capacity" in bounds_by_name:
+                out["cargo_capacity"] = cargo_capacity_2_5x
+            if "cargo_capacity_kg" in bounds_by_name:
+                out["cargo_capacity_kg"] = cargo_capacity_0_3x
 
         # Achieved steady-state cruise speed, DERIVED from this specific
         # build's actual power_density, the medium's mass-proportional
-        # resistance, and (ground only, see DRAG_POWER_COEFF_BY_MEDIUM) a
+        # resistance, and (ground/air, see DRAG_POWER_COEFF_BY_MEDIUM) a
         # mass-independent aerodynamic drag term -- not a platform-declared
         # constant. A build with more power than the platform's bare
         # target_velocity requires achieves a genuinely higher speed here;
@@ -1018,11 +1044,20 @@ class Pipeline:
             )
             amortized_per_m = upfront_cost / lifetime_m
 
-            fuel_price_per_mj = FUEL_PRICE_PER_MJ.get(ctx.storage_energy_form, 0.04)
+            if ctx.k_med is not None:
-            energy_per_m_mj = (
+                fuel_price_per_mj = FUEL_PRICE_PER_MJ.get(ctx.storage_energy_form, 0.04)
-                (effective_k_med or SPECIFIC_ENERGY_CONSUMPTION_J_PER_KG_M["ground"]) * floor_total
+                energy_per_m_mj = (effective_k_med * floor_total) / 1e6
-            ) / 1e6
+                operating_per_m = energy_per_m_mj * fuel_price_per_mj
-            operating_per_m = energy_per_m_mj * fuel_price_per_mj
+            else:
+                # No resistance model for this medium (space -- real range
+                # is governed by the rocket equation, not implemented
+                # here, see the comment above
+                # SPECIFIC_ENERGY_CONSUMPTION_J_PER_KG_M). Don't fabricate
+                # an operating cost from ground physics the way an earlier
+                # version of this did (`effective_k_med or ...["ground"]`)
+                # -- report upfront/amortized cost only, an honest partial
+                # answer, rather than a wrong number for an unmodeled term.
+                operating_per_m = 0.0
 
             cost_per_m = amortized_per_m + operating_per_m
             if units_by_name.get("cost_efficiency") == "$/(kg·m)":
@@ -1099,7 +1134,12 @@ class Pipeline:
         range_bounds = bounds_by_name.get("range_fuel")
         target_range = range_bounds.norm_max if range_bounds else None
 
-        if min_accel and specific_thrust:
+        # `is not None`, not truthy -- dep_value() legitimately returns 0.0
+        # for a declared floor of zero (Spaceship declares
+        # min_effective_accel=0, a real "no acceleration floor" value, not
+        # "undeclared"). A truthy check would silently treat that the same
+        # as an absent requirement and fall through to the wrong branch.
+        if min_accel is not None and specific_thrust is not None:
             c1, r1 = specific_thrust, min_accel
         elif target_velocity and ctx.k_med:
             # Resistance alone (k_med) only covers steady-state cruise --
@@ -1283,21 +1323,32 @@ class Pipeline:
 
         return best_a, best_s, best_score
 
-    def _stub_estimate(
+    def _estimate_physics(
-        self, combo: Combination, metric_bounds: list[MetricBound]
+        self, combo: Combination, domain: Domain
     ) -> tuple[dict[str, float], bool]:
-        """Deterministic estimation from declared entity attributes (no LLM).
+        """Deterministic physics-based estimation from declared entity
+        attributes (no LLM) -- pass 2's estimator.
 
-        power_density, range_fuel, and cost_efficiency are computed from the
+        Domains that declare their own metric_formulas (see
-        platform's declared mass envelope treated as a combo-wide budget —
+        _estimate_via_formulas) are estimated entirely by those
-        see the module-level comment above BIOLOGICAL_OPERATOR_MASS_KG for
+        domain-authored formulas instead -- this method's hardcoded
-        the full formula rationale.
+        platform/actuator/energy_storage physics model below is additive,
+        not the only path: it's untouched and still runs unchanged for
+        every domain that declares no formulas (every existing transport
+        domain today).
 
-        safety/availability/reliability/cargo_capacity/environmental_impact
+        power_density, speed, range_fuel, cost_efficiency, and
-        are untouched — these are judgment calls (regulatory, economic,
+        cargo_capacity/cargo_capacity_kg are all computed from the
-        qualitative), not physics, and stay on the categorical lookup-table
+        platform's declared mass envelope treated as a combo-wide budget,
-        heuristics below (actuator's thrust_profile and energy_form and the
+        jointly optimized by _decide_masses -- see the module-level
-        combo's infrastructure requirements).
+        comment above BIOLOGICAL_OPERATOR_MASS_KG for the full formula
+        rationale.
+
+        safety/availability/reliability/environmental_impact are untouched
+        — these are judgment calls (regulatory, economic, qualitative),
+        not physics, and stay on the categorical lookup-table heuristics
+        below (actuator's thrust_profile and energy_form and the combo's
+        infrastructure requirements).
 
         cost_efficiency additionally checks the domain's declared unit:
         "$/(kg·m)" (freight-style domains) isn't a rescaling of "$/m" — it's
@@ -1313,6 +1364,10 @@ class Pipeline:
         whose absolute mass tramples its own platform's declared ceiling
         can still produce perfectly plausible-looking per-kg ratios.
         """
+        if domain.metric_formulas:
+            return self._estimate_via_formulas(combo, domain)
+
+        metric_bounds = domain.metric_bounds
         metric_names = [mb.metric_name for mb in metric_bounds]
         units_by_name = {mb.metric_name: mb.unit for mb in metric_bounds}
         bounds_by_name = {mb.metric_name: mb for mb in metric_bounds}
@@ -1384,6 +1439,169 @@ class Pipeline:
 
         return raw, feasible
 
+    def _estimate_via_formulas(
+        self, combo: Combination, domain: Domain
+    ) -> tuple[dict[str, float], bool]:
+        """Generic estimator for domains that declare their own
+        metric_formulas/free_variables instead of matching the
+        platform/actuator/energy_storage shape _estimate_physics assumes.
+
+        Every formula (free-variable floor/ceiling, and each metric) is
+        evaluated through the same safe engine.formula evaluator, given two
+        kinds of names to resolve: dep(key, constraint_type="provides",
+        agg=None) -- a declared entity property, aggregated across the
+        combo's entities via constraint_resolver.aggregate_dependency_value
+        (the exact sum-vs-max rule ConstraintResolver itself already
+        applies to mass/footprint) -- and any free variable already chosen
+        earlier in its declared sort_order, as a bare name. Free variables
+        are resolved by _search_free_variables, a generalization of
+        _search_best_allocation/_decide_masses's hand-nested 2-3-variable
+        search to an arbitrary domain-declared list; with zero free
+        variables (most new domains -- nothing needs sizing) it degenerates
+        to a single direct formula evaluation, no search at all.
+
+        An invalid formula (bad syntax, unknown name, division by zero for
+        this combo's values) degrades that one metric to 0.0 / that one free
+        variable to being skipped, rather than failing the whole combo --
+        symmetric with how a metric_bounds entry with no matching raw value
+        already defaults to 0.0 elsewhere in this pipeline.
+        """
+        from physcom.engine.constraint_resolver import aggregate_dependency_value
+        from physcom.engine.formula import FormulaError, compile_formula, evaluate_formula
+
+        def dep(key: str, constraint_type: str = "provides", agg: str | None = None) -> float:
+            key_aggregation = None if agg is None else {key: agg}
+            value = aggregate_dependency_value(combo, key, constraint_type, key_aggregation)
+            return value if value is not None else 0.0
+
+        functions = {"dep": dep}
+
+        compiled_metrics: dict[str, object] = {}
+        for mf in domain.metric_formulas:
+            try:
+                compiled_metrics[mf.metric_name] = compile_formula(mf.formula)
+            except FormulaError:
+                compiled_metrics[mf.metric_name] = None
+
+        compiled_vars: list[tuple[str, object, object]] = []
+        for fv in sorted(domain.free_variables, key=lambda v: v.sort_order):
+            try:
+                compiled_vars.append((
+                    fv.name, compile_formula(fv.floor_formula), compile_formula(fv.ceiling_formula),
+                ))
+            except FormulaError:
+                continue
+
+        bounds_by_name = {mb.metric_name: mb for mb in domain.metric_bounds}
+
+        def evaluate_metrics(resolved: dict[str, float]) -> dict[str, float]:
+            raw: dict[str, float] = {}
+            for name, compiled in compiled_metrics.items():
+                if compiled is None:
+                    raw[name] = 0.0
+                    continue
+                try:
+                    raw[name] = evaluate_formula(compiled, resolved, functions)
+                except FormulaError:
+                    raw[name] = 0.0
+            return raw
+
+        def objective(resolved: dict[str, float]) -> float:
+            raw = evaluate_metrics(resolved)
+            scores, weights = [], []
+            for mb in bounds_by_name.values():
+                val = raw.get(mb.metric_name)
+                if val is None:
+                    continue
+                n = normalize(val, mb.norm_min, mb.norm_max)
+                if mb.lower_is_better:
+                    n = 1.0 - n
+                scores.append(n)
+                weights.append(mb.weight)
+            return composite_score(scores, weights)
+
+        resolved, _score, feasible = self._search_free_variables(
+            compiled_vars, {}, functions, objective
+        )
+        return evaluate_metrics(resolved), feasible
+
+    def _search_free_variables(
+        self,
+        var_specs: list[tuple[str, object, object]],
+        resolved: dict[str, float],
+        functions: dict,
+        objective,
+        grid: int = 10,
+        rounds: int = 5,
+    ) -> tuple[dict[str, float], float, bool]:
+        """Recursive generalization of _search_best_allocation/_decide_masses's
+        hand-nested outer-platform/inner-(actuator,storage) search to an
+        arbitrary ordered list of free variables: resolve var_specs[0]'s
+        floor/ceiling against what's already been resolved plus dep(...),
+        coarse-to-fine grid-search it, and recurse into the rest for each
+        candidate -- exactly the same nesting the hardcoded 3-variable
+        search already does by hand, just generalized to N. grid/rounds
+        shrink with remaining depth (cost is grid**depth * rounds**depth)
+        for the same reason the hardcoded search's outer loop already uses
+        a cheaper inner search per candidate; tuned for a handful of free
+        variables (transport's own 3 is the reference point), not domains
+        declaring dozens. Returns (resolved, score, feasible); feasible is
+        False only when no candidate at some depth has floor <= ceiling.
+        """
+        from physcom.engine.formula import FormulaError, evaluate_formula
+
+        if not var_specs:
+            return dict(resolved), objective(resolved), True
+
+        name, floor_f, ceiling_f = var_specs[0]
+        rest = var_specs[1:]
+        try:
+            floor = evaluate_formula(floor_f, resolved, functions)
+            ceiling = evaluate_formula(ceiling_f, resolved, functions)
+        except FormulaError:
+            result = dict(resolved)
+            result[name] = 0.0
+            return result, -1.0, False
+
+        if ceiling < floor:
+            result = dict(resolved)
+            result[name] = floor
+            return result, -1.0, False
+
+        inner_grid = max(4, grid - 2 * len(rest))
+        inner_rounds = max(2, rounds - len(rest))
+
+        if ceiling == floor:
+            candidate = dict(resolved)
+            candidate[name] = floor
+            return self._search_free_variables(
+                rest, candidate, functions, objective, inner_grid, inner_rounds
+            )
+
+        win_lo, win_hi = floor, ceiling
+        best_resolved, best_score, best_feasible = None, -1.0, False
+        for _round in range(rounds):
+            for i in range(grid + 1):
+                v = win_lo + (win_hi - win_lo) * i / grid
+                candidate = dict(resolved)
+                candidate[name] = v
+                r_resolved, r_score, r_feasible = self._search_free_variables(
+                    rest, candidate, functions, objective, inner_grid, inner_rounds
+                )
+                if r_score > best_score:
+                    best_resolved, best_score, best_feasible = r_resolved, r_score, r_feasible
+            if best_resolved is None:
+                break
+            span = max((win_hi - win_lo) / grid * 2, 1e-9)
+            center = best_resolved[name]
+            win_lo, win_hi = max(floor, center - span), min(ceiling, center + span)
+
+        if best_resolved is None:
+            result = dict(resolved)
+            result[name] = floor
+            return result, -1.0, False
+        return best_resolved, best_score, best_feasible
+
     def evaluate_allocation(
         self,
         combo: Combination,
@@ -1401,7 +1619,7 @@ class Pipeline:
         is ever persisted.
 
         Any mass left as None defaults to what the real requirement-based
-        solve already picked (see _decide_masses / _stub_estimate), so a
+        solve already picked (see _decide_masses / _estimate_physics), so a
         slider opens on today's actual build, not an arbitrary point.
         Explicit values are floor-clamped to each component's own declared
         minimum (platform is also ceiling-clamped to its declared max) --

src/physcom/models/domain.py

@@ -26,6 +26,32 @@ class DomainConstraint:
     allowed_values: list[str] = field(default_factory=list)  # e.g. ["ground", "air"]
 
 
+@dataclass
+class FreeVariable:
+    """A domain-declared quantity pass 2's estimator searches to maximize
+    the composite score (see Pipeline._estimate_via_formulas), e.g.
+    "actuator_mass". floor_formula/ceiling_formula are evaluated per-combo
+    and may reference dep(...) and any free variable declared at a lower
+    sort_order (mirrors the outer/inner nesting the built-in transport
+    physics model already does by hand)."""
+
+    name: str
+    floor_formula: str
+    ceiling_formula: str
+    sort_order: int = 0
+    id: int | None = None
+
+
+@dataclass
+class MetricFormula:
+    """A domain-declared formula computing one metric's raw value, evaluated
+    against dep(...) lookups and the domain's resolved free variables."""
+
+    metric_name: str
+    formula: str
+    id: int | None = None
+
+
 @dataclass
 class Domain:
     """A context frame that defines what 'good' means (e.g., urban_commuting)."""
@@ -34,4 +60,6 @@ class Domain:
     description: str = ""
     metric_bounds: list[MetricBound] = field(default_factory=list)
     constraints: list[DomainConstraint] = field(default_factory=list)
+    free_variables: list[FreeVariable] = field(default_factory=list)
+    metric_formulas: list[MetricFormula] = field(default_factory=list)
     id: int | None = None

src/physcom/seed/transport_example.py

@@ -95,6 +95,7 @@ WATER_PLATFORMS: list[Entity] = [
             Dependency("environment", "gravity", "true", None, "provides"),
             Dependency("physical", "footprint", "200", "m²", "range_max"),
             Dependency("physical", "footprint", "20", "m²", "range_min"),
+            Dependency("physical", "mass", "20000000", "kg", "range_max"),
             Dependency("physical", "mass", "10000", "kg", "range_min"),
             Dependency("environment", "medium", "water", None, "requires"),
             Dependency("physical", "energy_density", "720000", "J/kg", "range_min"),
@@ -198,6 +199,20 @@ MULTI_PLATFORMS: list[Entity] = [
             Dependency("physical", "footprint", "5", "m²", "range_min"),
             Dependency("physical", "mass", "10000", "kg", "range_max"),
             Dependency("physical", "mass", "1500", "kg", "range_min"),
+            # No requires here previously -- vacuously satisfied every
+            # domain's medium DomainConstraint (check_domain_constraints
+            # only flags a violation when an entity DECLARES a requires
+            # for the constrained key), including space-only
+            # interplanetary_travel. The current requires/domain-constraint
+            # model only supports one value per key -- there's no OR
+            # mechanism for "ground or water" -- so this picks ground
+            # (its primary, most-common domain) rather than leaving it
+            # undeclared. Real tradeoff: it can no longer participate in
+            # maritime_shipping (water-only) either, losing the water half
+            # of "amphibious." Closes the vacuous-pass hole; genuine
+            # multi-medium support would need OR semantics added to
+            # check_domain_constraints, a separate, bigger change.
+            Dependency("environment", "medium", "ground", None, "requires"),
         ],
     ),
 ]

src/physcom/snapshot.py

@@ -6,7 +6,7 @@ from datetime import datetime, timezone
 
 from physcom.db.repository import Repository
 from physcom.models.entity import Entity, Dependency
-from physcom.models.domain import Domain, DomainConstraint, MetricBound
+from physcom.models.domain import Domain, DomainConstraint, FreeVariable, MetricBound, MetricFormula
 from physcom.models.combination import Combination
 
 
@@ -70,11 +70,27 @@ def export_snapshot(repo: Repository) -> dict:
                 "key": dc.key,
                 "allowed_values": dc.allowed_values,
             })
+        fvs = []
+        for fv in d.free_variables:
+            fvs.append({
+                "name": fv.name,
+                "sort_order": fv.sort_order,
+                "floor_formula": fv.floor_formula,
+                "ceiling_formula": fv.ceiling_formula,
+            })
+        mfs = []
+        for mf in d.metric_formulas:
+            mfs.append({
+                "metric_name": mf.metric_name,
+                "formula": mf.formula,
+            })
         domain_list.append({
             "name": d.name,
             "description": d.description,
             "metric_bounds": mbs,
             "constraints": dcs,
+            "free_variables": fvs,
+            "metric_formulas": mfs,
         })
 
     # Export combinations
@@ -207,11 +223,26 @@ def import_snapshot(repo: Repository, data: dict, *, clear: bool = False) -> dic
                 )
                 for dc in d_data.get("constraints", [])
             ]
+            fvs = [
+                FreeVariable(
+                    name=fv["name"],
+                    floor_formula=fv["floor_formula"],
+                    ceiling_formula=fv["ceiling_formula"],
+                    sort_order=fv.get("sort_order", 0),
+                )
+                for fv in d_data.get("free_variables", [])
+            ]
+            mfs = [
+                MetricFormula(metric_name=mf["metric_name"], formula=mf["formula"])
+                for mf in d_data.get("metric_formulas", [])
+            ]
             domain = Domain(
                 name=d_data["name"],
                 description=d_data.get("description", ""),
                 metric_bounds=mbs,
                 constraints=dcs,
+                free_variables=fvs,
+                metric_formulas=mfs,
             )
             repo.add_domain(domain)
         counts["domains"] += 1

src/physcom_web/routes/domains.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 from flask import Blueprint, flash, redirect, render_template, request, url_for
 
-from physcom.models.domain import Domain, MetricBound
+from physcom.models.domain import Domain, FreeVariable, MetricBound, MetricFormula
 from physcom_web.app import get_repo
 
 bp = Blueprint("domains", __name__, url_prefix="/domains")
@@ -117,3 +117,96 @@ def metric_delete(domain_id: int, metric_id: int):
     flash("Metric removed.", "success")
     domain = repo.get_domain_by_id(domain_id)
     return render_template("domains/_metrics_table.html", domain=domain)
+
+
+# ── Free variable CRUD (HTMX partials) ───────────────────────
+
+
+@bp.route("/<int:domain_id>/free-vars/add", methods=["POST"])
+def free_var_add(domain_id: int):
+    repo = get_repo()
+    name = request.form["name"].strip()
+    floor_formula = request.form.get("floor_formula", "").strip()
+    ceiling_formula = request.form.get("ceiling_formula", "").strip()
+    try:
+        sort_order = int(request.form.get("sort_order", "0"))
+    except ValueError:
+        sort_order = 0
+    if not name or not floor_formula or not ceiling_formula:
+        flash("Name, floor formula, and ceiling formula are required.", "error")
+    else:
+        fv = FreeVariable(
+            name=name, floor_formula=floor_formula, ceiling_formula=ceiling_formula,
+            sort_order=sort_order,
+        )
+        repo.add_free_variable(domain_id, fv)
+        flash(f"Free variable '{name}' added.", "success")
+    domain = repo.get_domain_by_id(domain_id)
+    return render_template("domains/_free_vars_table.html", domain=domain)
+
+
+@bp.route("/<int:domain_id>/free-vars/<int:fv_id>/edit", methods=["POST"])
+def free_var_edit(domain_id: int, fv_id: int):
+    repo = get_repo()
+    try:
+        sort_order = int(request.form.get("sort_order", "0"))
+    except ValueError:
+        sort_order = 0
+    fv = FreeVariable(
+        name=request.form["name"].strip(),
+        floor_formula=request.form.get("floor_formula", "").strip(),
+        ceiling_formula=request.form.get("ceiling_formula", "").strip(),
+        sort_order=sort_order,
+    )
+    repo.update_free_variable(fv_id, fv)
+    flash("Free variable updated.", "success")
+    domain = repo.get_domain_by_id(domain_id)
+    return render_template("domains/_free_vars_table.html", domain=domain)
+
+
+@bp.route("/<int:domain_id>/free-vars/<int:fv_id>/delete", methods=["POST"])
+def free_var_delete(domain_id: int, fv_id: int):
+    repo = get_repo()
+    repo.delete_free_variable(fv_id)
+    flash("Free variable removed.", "success")
+    domain = repo.get_domain_by_id(domain_id)
+    return render_template("domains/_free_vars_table.html", domain=domain)
+
+
+# ── Metric formula CRUD (HTMX partials) ──────────────────────
+
+
+@bp.route("/<int:domain_id>/formulas/add", methods=["POST"])
+def formula_add(domain_id: int):
+    repo = get_repo()
+    metric_name = request.form["metric_name"].strip()
+    formula = request.form.get("formula", "").strip()
+    if not metric_name or not formula:
+        flash("Metric name and formula are required.", "error")
+    else:
+        repo.add_metric_formula(domain_id, MetricFormula(metric_name=metric_name, formula=formula))
+        flash(f"Formula for '{metric_name}' added.", "success")
+    domain = repo.get_domain_by_id(domain_id)
+    return render_template("domains/_formulas_table.html", domain=domain)
+
+
+@bp.route("/<int:domain_id>/formulas/<int:formula_id>/edit", methods=["POST"])
+def formula_edit(domain_id: int, formula_id: int):
+    repo = get_repo()
+    mf = MetricFormula(
+        metric_name=request.form["metric_name"].strip(),
+        formula=request.form.get("formula", "").strip(),
+    )
+    repo.update_metric_formula(formula_id, mf)
+    flash("Formula updated.", "success")
+    domain = repo.get_domain_by_id(domain_id)
+    return render_template("domains/_formulas_table.html", domain=domain)
+
+
+@bp.route("/<int:domain_id>/formulas/<int:formula_id>/delete", methods=["POST"])
+def formula_delete(domain_id: int, formula_id: int):
+    repo = get_repo()
+    repo.delete_metric_formula(formula_id)
+    flash("Formula removed.", "success")
+    domain = repo.get_domain_by_id(domain_id)
+    return render_template("domains/_formulas_table.html", domain=domain)

src/physcom_web/routes/pipeline.py

@@ -32,6 +32,8 @@ def _run_pipeline_in_background(
     from physcom.engine.scorer import Scorer
     from physcom.engine.pipeline import Pipeline
 
+    conn = None
+    repo = None
     try:
         conn = init_db(db_path)
         repo = Repository(conn)
@@ -58,18 +60,27 @@ def _run_pipeline_in_background(
             run_id=run_id,
         )
     except Exception as exc:
-        try:
+        if repo is not None:
-            repo.update_pipeline_run(
+            try:
-                run_id, status="failed",
+                repo.update_pipeline_run(
-                error_message=str(exc)[:500],
+                    run_id, status="failed",
-            )
+                    error_message=str(exc)[:500],
-        except Exception:
+                )
-            pass
+            except Exception:
+                pass
+        else:
+            # Couldn't even open the DB to record the failure (bad
+            # PHYSCOM_DB path, locked/corrupt file) -- the pipeline_runs
+            # row will stay "pending" forever with no way to write an
+            # error_message to it, so at least don't let that swallow the
+            # real cause silently. Server logs are the only trace left.
+            print(f"pipeline run {run_id} failed before DB was reachable: {exc!r}")
     finally:
-        try:
+        if conn is not None:
-            conn.close()
+            try:
-        except Exception:
+                conn.close()
-            pass
+            except Exception:
+                pass
 
 
 @bp.route("/")

src/physcom_web/templates/domains/_formulas_table.html

@@ -0,0 +1,56 @@
+<table id="formulas-table">
+  <thead>
+    <tr>
+      <th>Metric</th>
+      <th>Formula</th>
+      <th></th>
+    </tr>
+  </thead>
+  <tbody>
+    {% for mf in domain.metric_formulas %}
+    <tr>
+      <td>{{ mf.metric_name }}</td>
+      <td><code>{{ mf.formula }}</code></td>
+      <td class="actions">
+        <button class="btn btn-sm"
+                onclick="this.closest('tr').nextElementSibling.style.display='table-row'; this.closest('tr').style.display='none'">
+          Edit
+        </button>
+        <form method="post"
+              hx-post="{{ url_for('domains.formula_delete', domain_id=domain.id, formula_id=mf.id) }}"
+              hx-target="#formulas-section" hx-swap="innerHTML"
+              class="inline-form">
+          <button type="submit" class="btn btn-sm btn-danger">Del</button>
+        </form>
+      </td>
+    </tr>
+    <tr class="edit-row" style="display:none">
+      <form method="post"
+            hx-post="{{ url_for('domains.formula_edit', domain_id=domain.id, formula_id=mf.id) }}"
+            hx-target="#formulas-section" hx-swap="innerHTML">
+        <td><input name="metric_name" value="{{ mf.metric_name }}" required></td>
+        <td><input name="formula" value="{{ mf.formula }}" required></td>
+        <td>
+          <button type="submit" class="btn btn-sm btn-primary">Save</button>
+          <button type="button" class="btn btn-sm"
+                  onclick="this.closest('tr').style.display='none'; this.closest('tr').previousElementSibling.style.display=''">
+            Cancel
+          </button>
+        </td>
+      </form>
+    </tr>
+    {% endfor %}
+  </tbody>
+</table>
+
+<h3>Add Formula</h3>
+<form method="post"
+      hx-post="{{ url_for('domains.formula_add', domain_id=domain.id) }}"
+      hx-target="#formulas-section" hx-swap="innerHTML"
+      class="dep-add-form">
+  <div class="form-row">
+    <input name="metric_name" placeholder="metric name" required>
+    <input name="formula" placeholder='formula, e.g. draw_weight_chosen * 2' required>
+    <button type="submit" class="btn btn-primary">Add</button>
+  </div>
+</form>

src/physcom_web/templates/domains/_free_vars_table.html

@@ -0,0 +1,64 @@
+<table id="free-vars-table">
+  <thead>
+    <tr>
+      <th>Name</th>
+      <th>Order</th>
+      <th>Floor formula</th>
+      <th>Ceiling formula</th>
+      <th></th>
+    </tr>
+  </thead>
+  <tbody>
+    {% for fv in domain.free_variables %}
+    <tr>
+      <td>{{ fv.name }}</td>
+      <td>{{ fv.sort_order }}</td>
+      <td><code>{{ fv.floor_formula }}</code></td>
+      <td><code>{{ fv.ceiling_formula }}</code></td>
+      <td class="actions">
+        <button class="btn btn-sm"
+                onclick="this.closest('tr').nextElementSibling.style.display='table-row'; this.closest('tr').style.display='none'">
+          Edit
+        </button>
+        <form method="post"
+              hx-post="{{ url_for('domains.free_var_delete', domain_id=domain.id, fv_id=fv.id) }}"
+              hx-target="#free-vars-section" hx-swap="innerHTML"
+              class="inline-form">
+          <button type="submit" class="btn btn-sm btn-danger">Del</button>
+        </form>
+      </td>
+    </tr>
+    <tr class="edit-row" style="display:none">
+      <form method="post"
+            hx-post="{{ url_for('domains.free_var_edit', domain_id=domain.id, fv_id=fv.id) }}"
+            hx-target="#free-vars-section" hx-swap="innerHTML">
+        <td><input name="name" value="{{ fv.name }}" required></td>
+        <td><input name="sort_order" type="number" step="1" value="{{ fv.sort_order }}"></td>
+        <td><input name="floor_formula" value="{{ fv.floor_formula }}" required></td>
+        <td><input name="ceiling_formula" value="{{ fv.ceiling_formula }}" required></td>
+        <td>
+          <button type="submit" class="btn btn-sm btn-primary">Save</button>
+          <button type="button" class="btn btn-sm"
+                  onclick="this.closest('tr').style.display='none'; this.closest('tr').previousElementSibling.style.display=''">
+            Cancel
+          </button>
+        </td>
+      </form>
+    </tr>
+    {% endfor %}
+  </tbody>
+</table>
+
+<h3>Add Free Variable</h3>
+<form method="post"
+      hx-post="{{ url_for('domains.free_var_add', domain_id=domain.id) }}"
+      hx-target="#free-vars-section" hx-swap="innerHTML"
+      class="dep-add-form">
+  <div class="form-row">
+    <input name="name" placeholder="name, e.g. draw_weight_chosen" required>
+    <input name="sort_order" type="number" step="1" placeholder="order" value="0">
+    <input name="floor_formula" placeholder='floor formula, e.g. dep("draw_weight", "range_min")' required>
+    <input name="ceiling_formula" placeholder='ceiling formula, e.g. dep("draw_weight", "range_max")' required>
+    <button type="submit" class="btn btn-primary">Add</button>
+  </div>
+</form>

src/physcom_web/templates/domains/detail.html

@@ -33,4 +33,18 @@
 <div id="metrics-section">
   {% include "domains/_metrics_table.html" %}
 </div>
+
+<h2>Free Variables</h2>
+<p class="hint">Quantities pass 2's estimator searches to maximize this domain's composite score. Leave empty for domains where nothing needs sizing.</p>
+
+<div id="free-vars-section">
+  {% include "domains/_free_vars_table.html" %}
+</div>
+
+<h2>Metric Formulas</h2>
+<p class="hint">How each metric's raw value is computed from declared entity properties (via <code>dep(key, constraint_type="provides")</code>) and any free variable above. If this domain declares any formulas, pass 2 uses them instead of the built-in vehicle physics model.</p>
+
+<div id="formulas-section">
+  {% include "domains/_formulas_table.html" %}
+</div>
 {% endblock %}

src/physcom_web/templates/results/_explore_result.html

@@ -25,9 +25,11 @@ platform has no declared mass ceiling to bound the sliders.</p>
 
 <div class="mass-bar-container" title="platform {{ '%.1f'|format(r.platform_mass) }}kg / actuator {{ '%.1f'|format(r.actuator_mass) }}kg / storage {{ '%.1f'|format(r.storage_mass) }}kg">
   {% set total = r.total_mass %}
+  {% if total > 0 %}
   <div class="mass-bar-seg mass-bar-platform" style="width: {{ (r.platform_mass / total * 100)|round(1) }}%"></div>
   <div class="mass-bar-seg mass-bar-actuator" style="width: {{ (r.actuator_mass / total * 100)|round(1) }}%"></div>
   <div class="mass-bar-seg mass-bar-storage" style="width: {{ (r.storage_mass / total * 100)|round(1) }}%"></div>
+  {% endif %}
 </div>
 <div class="mass-bar-legend">
   <span><span class="mass-swatch mass-bar-platform"></span>platform {{ "%.1f"|format(r.platform_mass) }}kg</span>

tests/test_formula.py

@@ -0,0 +1,108 @@
+"""Tests for the safe formula evaluator."""
+
+import math
+
+import pytest
+
+from physcom.engine.formula import FormulaError, compile_formula, evaluate_formula
+
+
+class TestArithmetic:
+    def test_constant(self):
+        assert evaluate_formula(compile_formula("42"), {}) == 42.0
+
+    def test_basic_ops(self):
+        assert evaluate_formula(compile_formula("2 + 3 * 4"), {}) == 14.0
+        assert evaluate_formula(compile_formula("(2 + 3) * 4"), {}) == 20.0
+        assert evaluate_formula(compile_formula("10 / 4"), {}) == 2.5
+        assert evaluate_formula(compile_formula("2 ** 3"), {}) == 8.0
+
+    def test_unary_minus(self):
+        assert evaluate_formula(compile_formula("-5 + 2"), {}) == -3.0
+
+    def test_variable_lookup(self):
+        result = evaluate_formula(compile_formula("mass * 2"), {"mass": 3.0})
+        assert result == 6.0
+
+    def test_unknown_variable_raises(self):
+        with pytest.raises(FormulaError):
+            evaluate_formula(compile_formula("unknown_var"), {})
+
+    def test_division_by_zero_raises_formula_error(self):
+        with pytest.raises(FormulaError):
+            evaluate_formula(compile_formula("1 / 0"), {})
+
+
+class TestFunctions:
+    def test_default_math_functions(self):
+        assert evaluate_formula(compile_formula("sqrt(16)"), {}) == 4.0
+        assert evaluate_formula(compile_formula("max(1, 2, 3)"), {}) == 3.0
+        assert evaluate_formula(compile_formula("min(1, 2, 3)"), {}) == 1.0
+        assert evaluate_formula(compile_formula("abs(-5)"), {}) == 5.0
+        assert evaluate_formula(compile_formula("exp(0)"), {}) == 1.0
+        assert math.isclose(evaluate_formula(compile_formula("log(exp(1))"), {}), 1.0)
+
+    def test_custom_injected_function(self):
+        formula = compile_formula('dep("power_density", "provides")')
+        result = evaluate_formula(
+            formula, {}, functions={"dep": lambda key, constraint_type: 99.0}
+        )
+        assert result == 99.0
+
+    def test_unknown_function_raises(self):
+        with pytest.raises(FormulaError):
+            evaluate_formula(compile_formula("unknown_fn(1)"), {})
+
+    def test_string_constant_passthrough_to_function(self):
+        formula = compile_formula('dep("mass")')
+        result = evaluate_formula(formula, {}, functions={"dep": lambda key: len(key)})
+        assert result == 4.0
+
+
+class TestSecurity:
+    @pytest.mark.parametrize("source", [
+        "__import__('os').system('echo hi')",
+        "().__class__",
+        "[1, 2, 3]",
+        "{1: 2}",
+        "{1, 2}",
+        "(x for x in [1])",
+        "lambda: 1",
+        "1 if True else 0",
+        "1 == 1",
+        "x.__class__",
+        "x[0]",
+        "(lambda: 1)()",
+        "1; 2",
+    ])
+    def test_disallowed_constructs_rejected(self, source):
+        with pytest.raises(FormulaError):
+            compile_formula(source)
+
+    def test_unregistered_function_name_never_executes(self):
+        """exec/eval/__import__ etc. parse as ordinary Call nodes -- the
+        actual guarantee is that no function name is callable unless it's
+        explicitly in DEFAULT_FUNCTIONS or caller-supplied, checked at
+        evaluate time, not that the bare name is rejected at compile time."""
+        with pytest.raises(FormulaError):
+            evaluate_formula(compile_formula("exec('1')"), {})
+
+    def test_dunder_name_rejected(self):
+        with pytest.raises(FormulaError):
+            compile_formula("__builtins__")
+
+    def test_indirect_call_rejected(self):
+        with pytest.raises(FormulaError):
+            compile_formula("(a + b)(1)")
+
+    def test_invalid_syntax_raises_formula_error(self):
+        with pytest.raises(FormulaError):
+            compile_formula("2 +")
+
+    def test_boolean_constant_rejected(self):
+        with pytest.raises(FormulaError):
+            compile_formula("True")
+
+    def test_large_exponent_overflows_cleanly_not_hangs(self):
+        with pytest.raises(FormulaError):
+            evaluate_formula(compile_formula("9 ** 9 ** 9 ** 9"), {})

tests/test_pipeline_formulas.py

@@ -0,0 +1,113 @@
+"""End-to-end test that pass 2 can estimate a non-transport domain entirely
+from domain-authored formulas, without touching the platform/actuator/
+energy_storage physics model in Pipeline._estimate_physics."""
+
+import pytest
+
+from physcom.engine.constraint_resolver import ConstraintResolver
+from physcom.engine.scorer import Scorer
+from physcom.engine.pipeline import Pipeline
+from physcom.models.domain import Domain, FreeVariable, MetricBound, MetricFormula
+from physcom.models.entity import Dependency, Entity
+
+
+def _build_archery_domain(repo):
+    repo.add_entity(Entity(
+        name="Recurve",
+        dimension="bow",
+        dependencies=[
+            Dependency("physical", "draw_weight", "20", None, "range_min"),
+            Dependency("physical", "draw_weight", "50", None, "range_max"),
+        ],
+    ))
+    repo.add_entity(Entity(
+        name="Carbon",
+        dimension="arrow",
+        dependencies=[
+            Dependency("physical", "arrow_mass", "0.02", None, "provides"),
+        ],
+    ))
+    return repo.add_domain(Domain(
+        name="archery_test",
+        metric_bounds=[
+            MetricBound("drawback_force", weight=0.6, norm_min=0, norm_max=100),
+            MetricBound("range", weight=0.4, norm_min=0, norm_max=300),
+        ],
+        free_variables=[
+            FreeVariable(
+                name="draw_weight_chosen",
+                floor_formula='dep("draw_weight", "range_min")',
+                ceiling_formula='dep("draw_weight", "range_max")',
+                sort_order=0,
+            ),
+        ],
+        metric_formulas=[
+            MetricFormula(metric_name="drawback_force", formula="draw_weight_chosen * 2"),
+            MetricFormula(
+                metric_name="range",
+                formula='draw_weight_chosen * 5 / dep("arrow_mass")',
+            ),
+        ],
+    ))
+
+
+def test_formula_domain_scores_without_platform_actuator_shape(repo):
+    domain = _build_archery_domain(repo)
+    resolver = ConstraintResolver()
+    scorer = Scorer(domain)
+    pipeline = Pipeline(repo, resolver, scorer)
+
+    result = pipeline.run(
+        domain, ["bow", "arrow"], score_threshold=0.01, passes=[1, 2, 3, 5],
+    )
+
+    assert result.total_generated == 1
+    assert result.pass1_failed == 0
+    assert result.pass2_estimated == 1
+    assert result.pass3_above_threshold == 1
+
+    combos = repo.list_combinations()
+    assert len(combos) == 1
+    combo = combos[0]
+    scores = {
+        s["metric_name"]: s["raw_value"]
+        for s in repo.get_combination_scores(combo.id, domain.id)
+    }
+    # Both metrics increase monotonically with draw_weight_chosen and nothing
+    # trades off against it, so the optimizer should push to the declared
+    # ceiling (50) -- confirms _search_free_variables is actually searching,
+    # not just evaluating at the floor.
+    assert scores["drawback_force"] == pytest.approx(100.0, rel=0.02)
+
+
+def test_formula_domain_zero_free_variables_direct_evaluation(repo):
+    """A domain with metric_formulas but no free_variables should evaluate
+    each formula once directly -- no search loop at all."""
+    repo.add_entity(Entity(
+        name="Recurve",
+        dimension="bow",
+        dependencies=[Dependency("physical", "draw_weight", "30", None, "provides")],
+    ))
+    repo.add_entity(Entity(name="Carbon", dimension="arrow"))
+    domain = repo.add_domain(Domain(
+        name="archery_direct_test",
+        metric_bounds=[MetricBound("drawback_force", weight=1.0, norm_min=0, norm_max=100)],
+        metric_formulas=[
+            MetricFormula(metric_name="drawback_force", formula='dep("draw_weight") * 2'),
+        ],
+    ))
+    resolver = ConstraintResolver()
+    scorer = Scorer(domain)
+    pipeline = Pipeline(repo, resolver, scorer)
+
+    result = pipeline.run(
+        domain, ["bow", "arrow"], score_threshold=0.01, passes=[1, 2, 3, 5],
+    )
+
+    assert result.pass2_estimated == 1
+    combos = repo.list_combinations()
+    scores = {
+        s["metric_name"]: s["raw_value"]
+        for s in repo.get_combination_scores(combos[0].id, domain.id)
+    }
+    assert scores["drawback_force"] == 60.0

tests/test_repository.py

@@ -1,7 +1,7 @@
 """Tests for the database repository."""
 
 from physcom.models.entity import Entity, Dependency
-from physcom.models.domain import Domain, MetricBound
+from physcom.models.domain import Domain, FreeVariable, MetricBound, MetricFormula
 
 
 def test_ensure_dimension(repo):
@@ -63,6 +63,62 @@ def test_add_and_get_domain(repo):
     assert loaded.metric_bounds[0].metric_name == "speed"
 
 
+def test_add_domain_with_free_variables_and_formulas(repo):
+    domain = Domain(
+        name="archery_test",
+        metric_bounds=[MetricBound("drawback_force", weight=1.0, norm_min=0, norm_max=500)],
+        free_variables=[
+            FreeVariable(
+                name="draw_weight",
+                floor_formula='dep("draw_weight", "range_min")',
+                ceiling_formula='dep("draw_weight", "range_max")',
+                sort_order=0,
+            ),
+        ],
+        metric_formulas=[
+            MetricFormula(metric_name="drawback_force", formula="draw_weight * 1.5"),
+        ],
+    )
+    saved = repo.add_domain(domain)
+    assert saved.id is not None
+
+    loaded = repo.get_domain("archery_test")
+    assert loaded is not None
+    assert len(loaded.free_variables) == 1
+    assert loaded.free_variables[0].name == "draw_weight"
+    assert loaded.free_variables[0].id is not None
+    assert len(loaded.metric_formulas) == 1
+    assert loaded.metric_formulas[0].formula == "draw_weight * 1.5"
+
+
+def test_free_variable_and_formula_crud(repo):
+    domain = repo.add_domain(Domain(name="crud_test"))
+
+    fv = repo.add_free_variable(
+        domain.id,
+        FreeVariable(name="x", floor_formula="0", ceiling_formula="100", sort_order=0),
+    )
+    mf = repo.add_metric_formula(
+        domain.id, MetricFormula(metric_name="m", formula="x * 2")
+    )
+
+    repo.update_free_variable(
+        fv.id, FreeVariable(name="x", floor_formula="1", ceiling_formula="200", sort_order=0)
+    )
+    repo.update_metric_formula(mf.id, MetricFormula(metric_name="m", formula="x * 3"))
+
+    loaded = repo.get_domain_by_id(domain.id)
+    assert loaded.free_variables[0].floor_formula == "1"
+    assert loaded.free_variables[0].ceiling_formula == "200"
+    assert loaded.metric_formulas[0].formula == "x * 3"
+
+    repo.delete_free_variable(fv.id)
+    repo.delete_metric_formula(mf.id)
+    loaded = repo.get_domain_by_id(domain.id)
+    assert loaded.free_variables == []
+    assert loaded.metric_formulas == []
+
+
 def test_combination_save_and_dedup(repo):
     e1 = repo.add_entity(Entity(name="A", dimension="platform"))
     e2 = repo.add_entity(Entity(name="B", dimension="actuator"))

tests/test_snapshot.py

@@ -7,7 +7,7 @@ import pytest
 from physcom.db.schema import init_db
 from physcom.db.repository import Repository
 from physcom.models.entity import Entity, Dependency
-from physcom.models.domain import Domain, DomainConstraint, MetricBound
+from physcom.models.domain import Domain, DomainConstraint, FreeVariable, MetricBound, MetricFormula
 from physcom.models.combination import Combination
 from physcom.snapshot import export_snapshot, import_snapshot
 
@@ -169,6 +169,46 @@ def test_import_with_combinations(seeded_repo, tmp_path):
     assert len(fresh_combos) == len(data["combinations"])
 
 
+def test_export_import_roundtrip_free_variables_and_formulas(repo, tmp_path):
+    domain = Domain(
+        name="archery_snapshot_test",
+        metric_bounds=[MetricBound("drawback_force", weight=1.0, norm_min=0, norm_max=100)],
+        free_variables=[
+            FreeVariable(
+                name="draw_weight_chosen",
+                floor_formula='dep("draw_weight", "range_min")',
+                ceiling_formula='dep("draw_weight", "range_max")',
+                sort_order=0,
+            ),
+        ],
+        metric_formulas=[
+            MetricFormula(metric_name="drawback_force", formula="draw_weight_chosen * 2"),
+        ],
+    )
+    repo.add_domain(domain)
+
+    data = export_snapshot(repo)
+    exported = next(d for d in data["domains"] if d["name"] == "archery_snapshot_test")
+    assert exported["free_variables"] == [{
+        "name": "draw_weight_chosen", "sort_order": 0,
+        "floor_formula": 'dep("draw_weight", "range_min")',
+        "ceiling_formula": 'dep("draw_weight", "range_max")',
+    }]
+    assert exported["metric_formulas"] == [
+        {"metric_name": "drawback_force", "formula": "draw_weight_chosen * 2"},
+    ]
+
+    conn = init_db(tmp_path / "fresh.db")
+    fresh = Repository(conn)
+    import_snapshot(fresh, data, clear=True)
+
+    loaded = fresh.get_domain("archery_snapshot_test")
+    assert len(loaded.free_variables) == 1
+    assert loaded.free_variables[0].floor_formula == 'dep("draw_weight", "range_min")'
+    assert len(loaded.metric_formulas) == 1
+    assert loaded.metric_formulas[0].formula == "draw_weight_chosen * 2"
+
+
 def test_import_merge_skips_existing_domain(repo):
     """Merge import skips domains that already exist."""
     domain = Domain(