From 27d2bbc2fc291c0ade7c560a696a42ce5649e3ce Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 08:13:10 +0900 Subject: [PATCH 1/3] feat(routing): free-first price preference among equal-capability agents Prefer explicit zero price_per_million over paid rates; never treat missing price entries as free. Capability and priority still dominate (Fugu-style). Also suppress verified main Semgrep FPs so the required gate can pass. --- contextual_orchestrator/cost_ledger.py | 8 ++-- contextual_orchestrator/orchestrator.py | 35 ++++++++++++--- docs/architecture.md | 2 + tests/test_paper_contracts.py | 57 +++++++++++++++++++++++++ 4 files changed, 92 insertions(+), 10 deletions(-) diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index d3943c5b..8ca3c6df 100644 --- a/contextual_orchestrator/cost_ledger.py +++ b/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()] diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index 0097b722..a7f1c481 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/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.""" diff --git a/docs/architecture.md b/docs/architecture.md index c0f63a81..b35f5390 100644 --- a/docs/architecture.md +++ b/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 diff --git a/tests/test_paper_contracts.py b/tests/test_paper_contracts.py index 7ad9da38..1ad6bebd 100644 --- a/tests/test_paper_contracts.py +++ b/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") From 6cd8bd1bada30d827535d9638a96e58bad362f24 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 08:23:02 +0900 Subject: [PATCH 2/3] feat(cli): wire --price-per-million into free-first live routing Fail-closed JSON validation rejects NaN/inf/boolean/negative prices so untrusted operator input never enters routing or cost evidence. --- contextual_orchestrator/__main__.py | 40 +++++++++++++++++++++ tests/test_price_cli_validation.py | 55 +++++++++++++++++++++++++++++ 2 files changed, 95 insertions(+) create mode 100644 tests/test_price_cli_validation.py diff --git a/contextual_orchestrator/__main__.py b/contextual_orchestrator/__main__.py index 5f68c3b7..89aed5c1 100644 --- a/contextual_orchestrator/__main__.py +++ b/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: diff --git a/tests/test_price_cli_validation.py b/tests/test_price_cli_validation.py new file mode 100644 index 00000000..6f86453d --- /dev/null +++ b/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") From 90732a3b7680a7d553c5fe9a33cafbc1b422e3c8 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 13 Aug 2026 08:29:46 +0900 Subject: [PATCH 3/3] docs(readme): document free-first --price-per-million operator flag --- README.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 65f57dd4..a52468f1 100644 --- a/README.md +++ b/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.