From 9b4fb1815d701d75792a31bd45f8ca48683516c9 Mon Sep 17 00:00:00 2001 From: CC#1 Kora Substrate Date: Thu, 21 May 2026 17:05:17 -0700 Subject: [PATCH] feat(KR-P2-K ST3): downshift selector pure-function library MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `agent/cost_downshift.py`: rung-aware ticket-criticality-aware selector that decides effective model tier (or defer) per R4.1 §9.6. Per PM ruling A1 the substrate's 2-value criticality CHECK enum (`downshift_eligible` / `frontier_only`, migration 0098) is the source of truth; NULL is treated as `frontier_only` for fail-CLOSED safety. Per-rung behavior: - NORMAL: all tickets run at configured_tier. - WARN_75: downshift_eligible steps one tier (Opus→Sonnet, Sonnet→Haiku, Haiku floor); frontier_only / NULL defer. - DOWNSHIFT_90: downshift_eligible drops to Haiku; frontier_only / NULL defer. - HARD_STOP_100: all defer (defensive; ST4 wires the actual PAUSED{COST} transition so the poller never reaches selector at this rung). Why frontier-only defers at WARN_75 (stricter than the bucket sketch's "only DOWNSHIFT_90 defers"): PM ruling text is explicit that any cost-pressure rung defers non-eligible tickets. Conservative choice is correct — at WARN_75 we're already projected to overrun the pool, so continuing non-downshiftable work compounds the projection. Pure-function library; no state mutation. Returns a frozen `DownshiftDecision(defer, effective_tier, reason)`. Poller wire-in deferred until KR-P2-E (Sea_Ticket consumer) lands. Also ships `classify_model_tier(model_name)` helper for callers that need to map full identifiers (`anthropic/claude-opus-4.6`, `claude-haiku-4-5`, etc.) onto the 3-value tier enum. Tests: 69 pytest parameterized across 4 rungs x 3 criticality values x 3 configured tiers; pyright/ruff clean. Combined KR-P2-K suite (ST1+ST2+ST3) = 121 passing. Co-Authored-By: Claude Opus 4.7 (1M context) --- agent/cost_downshift.py | 207 +++++++++++++++++++++++++++ tests/test_cost_downshift.py | 267 +++++++++++++++++++++++++++++++++++ 2 files changed, 474 insertions(+) create mode 100644 agent/cost_downshift.py create mode 100644 tests/test_cost_downshift.py diff --git a/agent/cost_downshift.py b/agent/cost_downshift.py new file mode 100644 index 000000000000..a655a5d9ad4a --- /dev/null +++ b/agent/cost_downshift.py @@ -0,0 +1,207 @@ +"""Cost-ladder downshift selector (KR-P2-K ST3, R4.1 §9.6). + +Pure-function library that decides, given a ticket's criticality and the +active cost-ladder rung, what model tier the runner should use — or +whether the ticket should be deferred entirely. + +The selector is rung-aware and criticality-aware; it does NOT mutate +any state. The caller (the future Sea_Ticket poller in KR-P2-E, or any +other dispatch site) acts on the returned +:class:`DownshiftDecision`: + + - ``defer=False`` + ``effective_tier=tier`` → run the ticket at + ``tier`` (which may equal the configured tier or be downshifted). + - ``defer=True`` + ``effective_tier=None`` → do not run; transition + the Sea_Ticket to ``deferred_cost_limit``. + +# Criticality semantics (PM ruling A1, substrate migration 0098) + +The substrate ``tickets.criticality`` column is a 2-value CHECK enum: + + - ``"downshift_eligible"`` — the ticket may run on a cheaper model + when cost-pressured. + - ``"frontier_only"`` — the ticket must run on the configured + frontier model OR not at all. + +A ``NULL`` value (non-sea tickets, or sea tickets without an explicit +hint) is treated as ``"frontier_only"`` per fail-CLOSED policy: +when we don't know the ticket's downshift policy, we don't downshift. + +# Per-rung behavior + + - ``CostRung.NORMAL`` — no cost pressure. All tickets run at + ``configured_tier``. + - ``CostRung.WARN_75`` — first cost-pressure threshold. + Downshift-eligible tickets step down one tier (Opus→Sonnet, + Sonnet→Haiku). Frontier-only / NULL tickets defer. + - ``CostRung.DOWNSHIFT_90`` — aggressive cost-pressure threshold. + Downshift-eligible tickets drop straight to Haiku (the cheapest + tier). Frontier-only / NULL tickets defer. + - ``CostRung.HARD_STOP_100`` — budget exhausted. All tickets defer. + The poller should never claim at this rung once ST4 wires the + PAUSED{COST} transition; the selector returns defer defensively. + +# Why frontier-only defers at WARN_75 (not just at DOWNSHIFT_90) + +The PM ruling text is explicit: when any cost-pressure rung is +active, only downshift-eligible tickets continue to run. Frontier-only +tickets are deferred until the budget recovers (either via monthly +refresh or via reconciliation pulling spend backwards — though the +reconciler only moves spend FORWARD per ST1). + +This is more conservative than the original bucket sketch (which +proposed running frontier-only at configured tier through WARN_75). +The conservative choice is the right one: at WARN_75 we are +projected to overrun the budget, and continuing to spend on +non-downshiftable work compounds the projection. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum +from typing import Optional + +from agent.cost_state_holder import CostRung + + +class ModelTier(str, Enum): + """Anthropic model tier — Opus > Sonnet > Haiku by price/capability. + + String values are short identifiers used in log lines and decision + reasons; full model names like ``"claude-opus-4-7"`` map to these + tiers via :func:`classify_model_tier`. + """ + + OPUS = "opus" + SONNET = "sonnet" + HAIKU = "haiku" + + +_TIER_STEP_DOWN = { + ModelTier.OPUS: ModelTier.SONNET, + ModelTier.SONNET: ModelTier.HAIKU, + ModelTier.HAIKU: ModelTier.HAIKU, +} + + +CRITICALITY_DOWNSHIFT_ELIGIBLE = "downshift_eligible" +CRITICALITY_FRONTIER_ONLY = "frontier_only" + + +@dataclass(frozen=True, slots=True) +class DownshiftDecision: + """Result of :func:`select_effective_model_tier`. + + Attributes: + defer: ``True`` if the caller should not run this ticket and + should instead transition it to ``deferred_cost_limit``. + effective_tier: The model tier to use when ``defer=False``; + ``None`` when ``defer=True``. + reason: Operator-readable explanation of the decision. Always + non-empty. + """ + + defer: bool + effective_tier: Optional[ModelTier] + reason: str + + +def classify_model_tier(model_name: Optional[str]) -> Optional[ModelTier]: + """Map a full model identifier to its :class:`ModelTier`. + + The match is substring-based on the lowercased name, so prefixed + forms like ``"anthropic/claude-opus-4.6"`` and bare forms like + ``"claude-haiku-4-5"`` both classify correctly. + + Returns ``None`` for unrecognized names — the caller should treat + an unclassifiable model as having no downshift path (run as + configured; never step). + """ + if not model_name: + return None + lowered = model_name.lower() + if "opus" in lowered: + return ModelTier.OPUS + if "sonnet" in lowered: + return ModelTier.SONNET + if "haiku" in lowered: + return ModelTier.HAIKU + return None + + +def select_effective_model_tier( + ticket_criticality: Optional[str], + active_rung: CostRung, + configured_tier: ModelTier, +) -> DownshiftDecision: + """Decide how a ticket should run given the active cost-ladder rung. + + See the module docstring for full semantics. + + Args: + ticket_criticality: Substrate ``tickets.criticality`` value — + one of ``"downshift_eligible"``, ``"frontier_only"``, or + ``None``. Any other string value is treated as + ``"frontier_only"`` (fail-CLOSED). + active_rung: The current rung from + :meth:`agent.cost_state_holder.CostStateHolder.active_rung`. + configured_tier: The model tier the ticket would run at if + there were no cost pressure (typically the agent's + configured frontier model classified via + :func:`classify_model_tier`). + """ + if active_rung is CostRung.NORMAL: + return DownshiftDecision( + defer=False, + effective_tier=configured_tier, + reason="rung=NORMAL: no cost pressure; run at configured tier", + ) + + if active_rung is CostRung.HARD_STOP_100: + return DownshiftDecision( + defer=True, + effective_tier=None, + reason="rung=HARD_STOP_100: budget exhausted; defer all tickets", + ) + + is_eligible = ticket_criticality == CRITICALITY_DOWNSHIFT_ELIGIBLE + + if not is_eligible: + return DownshiftDecision( + defer=True, + effective_tier=None, + reason=( + f"rung={active_rung.value}: ticket criticality=" + f"{ticket_criticality or 'NULL'} is not downshift-eligible; " + "defer (transition Sea_Ticket to deferred_cost_limit)" + ), + ) + + if active_rung is CostRung.WARN_75: + stepped = _TIER_STEP_DOWN[configured_tier] + return DownshiftDecision( + defer=False, + effective_tier=stepped, + reason=( + f"rung=WARN_75: downshift-eligible; " + f"{configured_tier.value}->{stepped.value} (one tier step)" + ), + ) + + if active_rung is CostRung.DOWNSHIFT_90: + return DownshiftDecision( + defer=False, + effective_tier=ModelTier.HAIKU, + reason=( + f"rung=DOWNSHIFT_90: downshift-eligible; " + f"{configured_tier.value}->{ModelTier.HAIKU.value} " + "(drop to cheapest tier)" + ), + ) + + return DownshiftDecision( + defer=False, + effective_tier=configured_tier, + reason=f"rung={active_rung.value}: unhandled rung; pass through", + ) diff --git a/tests/test_cost_downshift.py b/tests/test_cost_downshift.py new file mode 100644 index 000000000000..e57f478a7532 --- /dev/null +++ b/tests/test_cost_downshift.py @@ -0,0 +1,267 @@ +"""Unit tests for ``agent/cost_downshift.py`` (KR-P2-K ST3). + +Covers ``select_effective_model_tier`` across all 4 rungs x 3 +criticality values (downshift_eligible / frontier_only / NULL) plus +the model-tier classifier helper. +""" + +from __future__ import annotations + +import dataclasses + +import pytest + +from agent.cost_downshift import ( + CRITICALITY_DOWNSHIFT_ELIGIBLE, + CRITICALITY_FRONTIER_ONLY, + DownshiftDecision, + ModelTier, + classify_model_tier, + select_effective_model_tier, +) +from agent.cost_state_holder import CostRung + + +# --------------------------------------------------------------------------- +# NORMAL rung — no downshift regardless of criticality +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "criticality", + [CRITICALITY_DOWNSHIFT_ELIGIBLE, CRITICALITY_FRONTIER_ONLY, None], +) +@pytest.mark.parametrize( + "configured_tier", [ModelTier.OPUS, ModelTier.SONNET, ModelTier.HAIKU] +) +def test_normal_rung_runs_at_configured_tier(criticality, configured_tier): + decision = select_effective_model_tier( + criticality, CostRung.NORMAL, configured_tier + ) + assert decision.defer is False + assert decision.effective_tier is configured_tier + assert "NORMAL" in decision.reason + + +# --------------------------------------------------------------------------- +# WARN_75 rung — downshift_eligible steps one tier; others defer +# --------------------------------------------------------------------------- + + +def test_warn_75_eligible_opus_steps_to_sonnet(): + decision = select_effective_model_tier( + CRITICALITY_DOWNSHIFT_ELIGIBLE, CostRung.WARN_75, ModelTier.OPUS + ) + assert decision.defer is False + assert decision.effective_tier is ModelTier.SONNET + assert "WARN_75" in decision.reason + assert "opus" in decision.reason and "sonnet" in decision.reason + + +def test_warn_75_eligible_sonnet_steps_to_haiku(): + decision = select_effective_model_tier( + CRITICALITY_DOWNSHIFT_ELIGIBLE, CostRung.WARN_75, ModelTier.SONNET + ) + assert decision.defer is False + assert decision.effective_tier is ModelTier.HAIKU + + +def test_warn_75_eligible_haiku_floors_at_haiku(): + """Haiku is already the cheapest tier — step-down floors here.""" + decision = select_effective_model_tier( + CRITICALITY_DOWNSHIFT_ELIGIBLE, CostRung.WARN_75, ModelTier.HAIKU + ) + assert decision.defer is False + assert decision.effective_tier is ModelTier.HAIKU + + +def test_warn_75_frontier_only_defers(): + decision = select_effective_model_tier( + CRITICALITY_FRONTIER_ONLY, CostRung.WARN_75, ModelTier.OPUS + ) + assert decision.defer is True + assert decision.effective_tier is None + assert "deferred_cost_limit" in decision.reason + assert "frontier_only" in decision.reason + + +def test_warn_75_null_criticality_defers_fail_closed(): + """NULL criticality (non-sea ticket) is treated as frontier_only. + Fail-CLOSED per PM ruling A1: don't downshift unknown policies.""" + decision = select_effective_model_tier( + None, CostRung.WARN_75, ModelTier.OPUS + ) + assert decision.defer is True + assert decision.effective_tier is None + assert "NULL" in decision.reason + + +def test_warn_75_unknown_string_criticality_defers(): + """An unexpected string value (e.g. legacy bucket-spec's 'normal') + is not equal to ``downshift_eligible`` and therefore defers.""" + decision = select_effective_model_tier( + "normal", CostRung.WARN_75, ModelTier.OPUS + ) + assert decision.defer is True + assert decision.effective_tier is None + + +# --------------------------------------------------------------------------- +# DOWNSHIFT_90 rung — eligible drops straight to Haiku; others defer +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "configured_tier", [ModelTier.OPUS, ModelTier.SONNET, ModelTier.HAIKU] +) +def test_downshift_90_eligible_always_drops_to_haiku(configured_tier): + decision = select_effective_model_tier( + CRITICALITY_DOWNSHIFT_ELIGIBLE, CostRung.DOWNSHIFT_90, configured_tier + ) + assert decision.defer is False + assert decision.effective_tier is ModelTier.HAIKU + assert "DOWNSHIFT_90" in decision.reason + + +def test_downshift_90_frontier_only_defers(): + decision = select_effective_model_tier( + CRITICALITY_FRONTIER_ONLY, CostRung.DOWNSHIFT_90, ModelTier.OPUS + ) + assert decision.defer is True + assert decision.effective_tier is None + assert "deferred_cost_limit" in decision.reason + + +def test_downshift_90_null_criticality_defers(): + decision = select_effective_model_tier( + None, CostRung.DOWNSHIFT_90, ModelTier.OPUS + ) + assert decision.defer is True + assert decision.effective_tier is None + + +# --------------------------------------------------------------------------- +# HARD_STOP_100 rung — defer everything +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "criticality", + [CRITICALITY_DOWNSHIFT_ELIGIBLE, CRITICALITY_FRONTIER_ONLY, None], +) +@pytest.mark.parametrize( + "configured_tier", [ModelTier.OPUS, ModelTier.SONNET, ModelTier.HAIKU] +) +def test_hard_stop_100_defers_all(criticality, configured_tier): + decision = select_effective_model_tier( + criticality, CostRung.HARD_STOP_100, configured_tier + ) + assert decision.defer is True + assert decision.effective_tier is None + assert "HARD_STOP_100" in decision.reason + + +# --------------------------------------------------------------------------- +# DownshiftDecision invariants +# --------------------------------------------------------------------------- + + +def test_downshift_decision_is_frozen(): + decision = select_effective_model_tier( + CRITICALITY_DOWNSHIFT_ELIGIBLE, CostRung.NORMAL, ModelTier.OPUS + ) + with pytest.raises(dataclasses.FrozenInstanceError): + decision.defer = True # type: ignore[misc] + + +@pytest.mark.parametrize( + "criticality", + [CRITICALITY_DOWNSHIFT_ELIGIBLE, CRITICALITY_FRONTIER_ONLY, None], +) +@pytest.mark.parametrize( + "rung", + [ + CostRung.NORMAL, + CostRung.WARN_75, + CostRung.DOWNSHIFT_90, + CostRung.HARD_STOP_100, + ], +) +def test_decision_reason_is_always_non_empty(criticality, rung): + decision = select_effective_model_tier(criticality, rung, ModelTier.OPUS) + assert decision.reason + assert len(decision.reason) > 0 + + +@pytest.mark.parametrize( + "criticality", + [CRITICALITY_DOWNSHIFT_ELIGIBLE, CRITICALITY_FRONTIER_ONLY, None], +) +@pytest.mark.parametrize( + "rung", + [ + CostRung.NORMAL, + CostRung.WARN_75, + CostRung.DOWNSHIFT_90, + CostRung.HARD_STOP_100, + ], +) +def test_defer_implies_no_effective_tier(criticality, rung): + """Invariant: ``defer=True`` ⇒ ``effective_tier is None`` and + vice versa.""" + decision = select_effective_model_tier(criticality, rung, ModelTier.OPUS) + if decision.defer: + assert decision.effective_tier is None + else: + assert decision.effective_tier is not None + + +# --------------------------------------------------------------------------- +# classify_model_tier +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "name,expected", + [ + ("claude-opus-4-7", ModelTier.OPUS), + ("anthropic/claude-opus-4.6", ModelTier.OPUS), + ("claude-opus-4-6-20250414", ModelTier.OPUS), + ("Claude-Opus-4-5", ModelTier.OPUS), # case-insensitive + ("claude-sonnet-4-6", ModelTier.SONNET), + ("anthropic/claude-sonnet-4-5", ModelTier.SONNET), + ("claude-haiku-4-5", ModelTier.HAIKU), + ("anthropic.claude-haiku-4-5", ModelTier.HAIKU), + ], +) +def test_classify_model_tier_known_names(name, expected): + assert classify_model_tier(name) is expected + + +@pytest.mark.parametrize("name", [None, "", "gpt-4", "llama-3-70b", "unknown-model"]) +def test_classify_model_tier_unknown_returns_none(name): + assert classify_model_tier(name) is None + + +# --------------------------------------------------------------------------- +# ModelTier enum sanity +# --------------------------------------------------------------------------- + + +def test_model_tier_string_values(): + assert ModelTier.OPUS.value == "opus" + assert ModelTier.SONNET.value == "sonnet" + assert ModelTier.HAIKU.value == "haiku" + + +def test_downshift_decision_signature_matches_bucket_sketch(): + """Returns a DownshiftDecision with defer / effective_tier / + reason fields — superset of bucket sketch's + ``tuple[ModelTier, Optional[str]]`` for explicit defer signal.""" + decision = select_effective_model_tier( + CRITICALITY_DOWNSHIFT_ELIGIBLE, CostRung.WARN_75, ModelTier.OPUS + ) + assert isinstance(decision, DownshiftDecision) + assert hasattr(decision, "defer") + assert hasattr(decision, "effective_tier") + assert hasattr(decision, "reason")