feat(routing): free-first price preference among equal-capability agents

README.md

@@ -129,12 +129,13 @@ curl -s http://127.0.0.1:8000/api/v1/spend_analytics/latest \
 ```
 
 - **Tokens.** `by_model[].output_tokens` uses the provider-reported `usage.completion_tokens` when a real worker returns it, and falls back to a `~4 chars/token` estimate otherwise. Each row carries `usage_source`: `reported` (all steps reported), `mixed`, or `estimated`. `estimated_output_tokens` is always the estimate, kept alongside for comparison. `measurement_status` is `local_runtime_estimate`, not production telemetry.
-- **Cost.** Supply a price table to turn tokens into money — `TaskOrchestrator(price_per_million={"gpt-5.5": 10.0})` (USD per 1M output tokens). Models without a price appear under `unpriced_models` with `estimated_cost_usd: null`. No prices are assumed or fabricated.
+- **Cost.** Supply a price table to turn tokens into money — `TaskOrchestrator(price_per_million={"gpt-5.5": 10.0})` or CLI `--price-per-million '{"gpt-5.5": 10.0}'` (USD per 1M output tokens). Live routing prefers an explicit free rate (`0`) over paid among equally capable agents; missing prices are never treated as free. Models without a price appear under `unpriced_models` with `estimated_cost_usd: null`. No prices are assumed or fabricated.
 - **Budget cap.** Set an operator cap to refuse runaway spend (default: no cap):
 
   ```bash
   python -m contextual_orchestrator --serve --agents examples/agents.mock.json \
-    --budget-max-output-tokens 2000000 --budget-max-cost-usd 50
+    --budget-max-output-tokens 2000000 --budget-max-cost-usd 50 \
+    --price-per-million '{"mock-generalist": 0, "gpt-example": 2.0}'
   ```
 
   Or in code: `TaskOrchestrator(budget_max_output_tokens=..., budget_max_cost_usd=...)`. Once spend reaches a cap, the next run is refused — `run()` raises `BudgetExceededError` and `/v1/chat/completions` returns HTTP `429 budget_exceeded`. Current state is in `spend_analytics()["budget"]` (`enabled`, limits, `spent_*`, `remaining_*`, `exceeded`). Cost caps require a price table; token caps do not.

contextual_orchestrator/__main__.py

@@ -4,6 +4,7 @@
 
 import argparse
 import json
+import math
 import os
 import sys
 
@@ -12,6 +13,34 @@
 from .server import SecurityConfig, serve
 
 
+def _json_object(raw: str) -> dict[str, float]:
+    """Parse a CLI JSON object of finite non-negative USD-per-million prices.
+
+    Used by ``--price-per-million``. Rejects non-objects, non-numeric values,
+    booleans, NaN, infinities, and negatives so untrusted operator input fails at
+    argparse rather than entering routing or cost evidence.
+    """
+    try:
+        value = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise argparse.ArgumentTypeError(f"invalid JSON for --price-per-million: {exc.msg}") from exc
+    if not isinstance(value, dict):
+        raise argparse.ArgumentTypeError("--price-per-million must be a JSON object mapping model names to prices")
+    parsed: dict[str, float] = {}
+    for key, price in value.items():
+        if not isinstance(key, str) or not key.strip():
+            raise argparse.ArgumentTypeError("--price-per-million keys must be non-empty model name strings")
+        if isinstance(price, bool) or not isinstance(price, (int, float)):
+            raise argparse.ArgumentTypeError(f"--price-per-million[{key!r}] must be a finite non-negative number")
+        number = float(price)
+        if not math.isfinite(number):
+            raise argparse.ArgumentTypeError(f"--price-per-million[{key!r}] must be finite")
+        if number < 0:
+            raise argparse.ArgumentTypeError(f"--price-per-million[{key!r}] must be non-negative")
+        parsed[key] = number
+    return parsed
+
+
 def _register_credential_command(argv: list[str]) -> None:
     """Bootstrap: read a deploy-time secret and store it in the KV credential registry.
 
@@ -88,6 +117,16 @@ def main() -> None:
                         help="Refuse new runs once estimated/reported output tokens reach this cap (default: no cap).")
     parser.add_argument("--budget-max-cost-usd", type=float, default=None,
                         help="Refuse new runs once estimated cost reaches this USD cap (needs a price table; default: no cap).")
+    parser.add_argument(
+        "--price-per-million",
+        type=_json_object,
+        default={},
+        help=(
+            "USD price per 1M output tokens by model, e.g. "
+            "'{\"gpt-5.5-mini\": 0.5, \"local-free\": 0}'. Live routing prefers free "
+            "then cheapest among equally-capable agents (default: no price data)."
+        ),
+    )
     parser.add_argument("--cache-ttl", type=float, default=0.0,
                         help="Seconds to cache identical requests (default 0 = disabled).")
     parser.add_argument("--eval", nargs="+", metavar="PROMPT",
@@ -103,6 +142,7 @@ def main() -> None:
         budget_max_output_tokens=args.budget_max_output_tokens,
         budget_max_cost_usd=args.budget_max_cost_usd,
         cache_ttl=args.cache_ttl,
+        price_per_million=args.price_per_million,
     )
 
     if args.eval:

contextual_orchestrator/cost_ledger.py

@@ -583,12 +583,12 @@ def _seed_dimension_catalog(self) -> None:
         ph = self._placeholder()
         cur = self._conn.cursor()
         for order, (name, label, _column) in enumerate(ATTRIBUTION_DIMENSION_CATALOG):
-            cur.execute(
+            cur.execute(  # nosemgrep -- sqlalchemy-execute-raw-query FP: only the DB-API placeholder char is interpolated; the value is bound.
                 f"SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = {ph}",  # nosec B608 - ph is a DB-API placeholder.
                 (name,),
             )
             if cur.fetchone() is None:
-                cur.execute(
+                cur.execute(  # nosemgrep -- sqlalchemy-execute-raw-query FP: only DB-API placeholder chars are interpolated; values are bound.
                     "INSERT INTO cost_attribution_dimensions "
                     f"(dimension_name, dimension_label, dimension_order) VALUES ({ph}, {ph}, {ph})",  # nosec B608 - ph is a DB-API placeholder.
                     (name, label, order),
@@ -602,7 +602,7 @@ def append(self, record: UsageRecord) -> None:
         placeholders = ", ".join(ph for _ in _USAGE_COLUMNS)
         columns = ", ".join(_USAGE_COLUMNS)
         cur = self._conn.cursor()
-        cur.execute(
+        cur.execute(  # nosemgrep -- sqlalchemy-execute-raw-query FP: columns are the fixed _USAGE_COLUMNS constant; values are bound.
             f"INSERT INTO llm_usage_records ({columns}) VALUES ({placeholders})",  # nosec B608 - columns are fixed _USAGE_COLUMNS.
             tuple(row.get(column) for column in _USAGE_COLUMNS),
         )
@@ -622,7 +622,7 @@ def query(self, start: Optional[int] = None, end: Optional[int] = None) -> List[
         where = f" WHERE {' AND '.join(clauses)}" if clauses else ""
         columns = ", ".join(_USAGE_COLUMNS)
         cur = self._conn.cursor()
-        cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params))  # nosec B608 - columns and clauses are fixed.
+        cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params))  # nosec B608 - columns and clauses are fixed.  # nosemgrep -- sqlalchemy-execute-raw-query FP: fixed columns and clause templates; all values are bound.
         return [dict(zip(_USAGE_COLUMNS, values)) for values in cur.fetchall()]
 
 

contextual_orchestrator/orchestrator.py

@@ -230,7 +230,7 @@ def __init__(
     @staticmethod
     def _build_ssl_context(ca_bundle: str | None, verify_tls: bool) -> ssl.SSLContext:
         if not verify_tls:
-            return ssl._create_unverified_context()  # nosec B323 - explicit dev-only provider TLS opt-out.
+            return ssl._create_unverified_context()  # nosec B323 - explicit dev-only provider TLS opt-out.  # nosemgrep -- unverified-ssl-context: intentional, default-secure (verify_tls defaults True) dev-only opt-out for self-signed endpoints.
         if ca_bundle:
             if not os.path.isfile(ca_bundle):
                 raise ValueError(f"provider CA bundle does not exist: {ca_bundle}")
@@ -307,7 +307,7 @@ def _send(self, agent: ModelAgent, payload: dict[str, Any]) -> str:
 
     def _open_provider(self, request: urllib.request.Request) -> Any:
         """Open a provider request built from a validated provider URL."""
-        return urllib.request.urlopen(  # nosec B310 - request URL comes from _provider_url after provider validation.
+        return urllib.request.urlopen(  # nosec B310 - request URL comes from _provider_url after provider validation.  # nosemgrep -- dynamic-urllib-use: URL is built by _provider_url after scheme/host validation; egress to loopback/private/reserved is blocked.
             request,
             timeout=self.timeout,
             context=self._ssl_context,
@@ -1512,17 +1512,40 @@ def _plan(self, task: str) -> list[WorkflowStep]:
             WorkflowStep(3, "synthesizer", synthesizer, "Produce the final answer, incorporating only verified work.", (0, 1, 2)),
         ]
 
-    def _score_agent(self, agent: ModelAgent, role: str, lowered: str) -> tuple[int, int, str]:
+    def _score_agent(
+        self, agent: ModelAgent, role: str, lowered: str
+    ) -> tuple[int, int, int, float, int, str]:
         if agent.disabled:
-            return (-20_000, len(agent.tags), agent.id)
+            return (-20_000, 0, 0, 0.0, len(agent.tags), agent.id)
         if role in agent.provider_exclusions:
-            return (-10_000, len(agent.tags), agent.id)
+            return (-10_000, 0, 0, 0.0, len(agent.tags), agent.id)
         role_score = sum(3 for tag in agent.tags if tag in self.ROLE_TAGS.get(role, ()))
         domain_score = 0
         for tag, hints in self.DOMAIN_HINTS.items():
             if tag in agent.tags and any(hint in lowered for hint in hints):
                 domain_score += 2
-        return (role_score + domain_score + agent.priority, len(agent.tags), agent.id)
+        # Capability first, then free-first price honesty among equals:
+        # 1) free (explicit USD/M token rate 0) beats any positive price and unpriced;
+        # 2) known positive prices: cheaper wins; 3) unpriced is not free — ranks last.
+        # Never invent a zero rate for missing price_per_million entries.
+        price = self.price_per_million.get(agent.model)
+        if price is None:
+            free_bonus = 0
+            price_known = 0
+            cheapness = 0.0
+        else:
+            rate = float(price)
+            free_bonus = 1 if rate == 0.0 else 0
+            price_known = 1
+            cheapness = -rate
+        return (
+            role_score + domain_score + agent.priority,
+            free_bonus,
+            price_known,
+            cheapness,
+            len(agent.tags),
+            agent.id,
+        )
 
     def _ranked_agents(self, text: str, role: str) -> list[ModelAgent]:
         """Agents sorted best-first for a role; the head is the primary, the tail are failovers."""

docs/architecture.md

@@ -40,6 +40,8 @@ This repository implements the interface and control plane, not the trained coor
 
 The deliberate simplification is the policy. The paper systems learn routing and topology from rewards; this lab uses deterministic keyword scoring so the repo runs without training data, GPUs, or vendor credentials.
 
+Live routing (`_score_agent`/`_ranked_agents`) maximizes capability-tag match and operator priority first (the honest, measurable performance proxy). Among agents tied on that ranking, free-first price preference applies: an explicit free rate (`price_per_million[model] == 0`) beats any positive rate; cheaper known paid rates beat more expensive ones; unpriced models (missing table entries) are never treated as free and rank after known prices. This never invents a quality score or a zero price. Set prices via `TaskOrchestrator(price_per_million=...)` or the CLI's `--price-per-million` flag when available.
+
 Add learned routing only when there is an evaluation set and logs proving the heuristic policy is the bottleneck.
 
 ## Product Planning Interpretation

tests/test_paper_contracts.py

@@ -60,8 +60,65 @@ def test_conductor_contract_uses_access_lists_to_control_context() -> None:
     assert "Step 1: builder_agent:2" in verifier_prompt
 
 
+def test_fugu_contract_prefers_cheapest_equally_capable_agent() -> None:
+    """Among equal capability, lower known price_per_million wins."""
+    orchestrator = TaskOrchestrator(
+        [
+            ModelAgent("pricey_agent", "model-pricey", tags=("coding",)),
+            ModelAgent("cheap_agent", "model-cheap", tags=("coding",)),
+        ],
+        price_per_million={"model-pricey": 10.0, "model-cheap": 1.0},
+    )
+    result = orchestrator.route_once([{"role": "user", "content": "fix this bug"}])
+    assert result["trace"][0]["agent_id"] == "cheap_agent"
+
+
+def test_fugu_contract_price_is_only_a_tie_break_not_a_priority_override() -> None:
+    """Capability/priority dominate price; free/cheap cannot override a better match."""
+    orchestrator = TaskOrchestrator(
+        [
+            ModelAgent("higher_priority_pricey_agent", "model-pricey", tags=("coding",), priority=5),
+            ModelAgent("lower_priority_free_agent", "model-free", tags=("coding",), priority=0),
+        ],
+        price_per_million={"model-pricey": 50.0, "model-free": 0.0},
+    )
+    result = orchestrator.route_once([{"role": "user", "content": "fix this bug"}])
+    assert result["trace"][0]["agent_id"] == "higher_priority_pricey_agent"
+
+
+def test_fugu_contract_prefers_free_over_paid_when_equally_capable() -> None:
+    """Explicit free rate (0) wins free-first tie-break; unpriced is not free."""
+    orchestrator = TaskOrchestrator(
+        [
+            ModelAgent("paid_agent", "model-paid", tags=("coding",)),
+            ModelAgent("free_agent", "model-free", tags=("coding",)),
+            ModelAgent("unpriced_agent", "model-unpriced", tags=("coding",)),
+        ],
+        price_per_million={"model-paid": 2.0, "model-free": 0.0},
+    )
+    result = orchestrator.route_once([{"role": "user", "content": "fix this bug"}])
+    assert result["trace"][0]["agent_id"] == "free_agent"
+
+
+def test_fugu_contract_prefers_priced_over_unpriced_when_equally_capable() -> None:
+    """Missing price_per_million is not treated as free; known paid still ranks above it."""
+    orchestrator = TaskOrchestrator(
+        [
+            ModelAgent("unpriced_agent", "model-unpriced", tags=("coding",)),
+            ModelAgent("paid_agent", "model-paid", tags=("coding",)),
+        ],
+        price_per_million={"model-paid": 5.0},
+    )
+    result = orchestrator.route_once([{"role": "user", "content": "fix this bug"}])
+    assert result["trace"][0]["agent_id"] == "paid_agent"
+
+
 if __name__ == "__main__":  # pragma: no cover
     test_fugu_contract_fuses_fast_route_and_deep_workflow()
     test_trinity_contract_has_explicit_thinker_worker_verifier_roles()
     test_conductor_contract_uses_access_lists_to_control_context()
+    test_fugu_contract_prefers_cheapest_equally_capable_agent()
+    test_fugu_contract_price_is_only_a_tie_break_not_a_priority_override()
+    test_fugu_contract_prefers_free_over_paid_when_equally_capable()
+    test_fugu_contract_prefers_priced_over_unpriced_when_equally_capable()
     print("ok")

tests/test_price_cli_validation.py

@@ -0,0 +1,55 @@
+"""CLI price-table validation for fail-closed cost-aware routing."""
+
+from __future__ import annotations
+
+import argparse
+
+import pytest
+
+from contextual_orchestrator.__main__ import _json_object
+
+
+def test_price_table_accepts_finite_non_negative_values() -> None:
+    """Finite zero and positive prices remain valid operator evidence."""
+    assert _json_object('{"model_a": 0, "model_b": 1.25}') == {
+        "model_a": 0.0,
+        "model_b": 1.25,
+    }
+
+
+@pytest.mark.parametrize(
+    "raw",
+    [
+        '{"model_a": NaN}',
+        '{"model_a": Infinity}',
+        '{"model_a": -Infinity}',
+        '{"model_a": 1e999}',
+    ],
+)
+def test_price_table_rejects_non_finite_values(raw: str) -> None:
+    """NaN and infinities must never enter routing or cost evidence."""
+    with pytest.raises(argparse.ArgumentTypeError, match="finite"):
+        _json_object(raw)
+
+
+def test_price_table_rejects_boolean_and_negative_values() -> None:
+    """JSON booleans and negative prices remain invalid despite Python scalar coercion."""
+    with pytest.raises(argparse.ArgumentTypeError):
+        _json_object('{"model_a": true}')
+    with pytest.raises(argparse.ArgumentTypeError):
+        _json_object('{"model_a": -0.01}')
+
+
+def test_price_table_rejects_non_object_payloads() -> None:
+    """Only a JSON object of model→price mappings is accepted."""
+    with pytest.raises(argparse.ArgumentTypeError, match="JSON object"):
+        _json_object("[1, 2]")
+    with pytest.raises(argparse.ArgumentTypeError):
+        _json_object("not-json")
+
+
+if __name__ == "__main__":  # pragma: no cover
+    test_price_table_accepts_finite_non_negative_values()
+    test_price_table_rejects_boolean_and_negative_values()
+    test_price_table_rejects_non_object_payloads()
+    print("ok")