mirror of
https://github.com/razzant/ouroboros.git
synced 2026-10-02 19:58:46 +00:00
168 lines
9.1 KiB
Python
168 lines
9.1 KiB
Python
"""Processing qualifiers survive reservation, native observations and settlement."""
|
|
|
|
import json
|
|
import copy
|
|
|
|
import pytest
|
|
|
|
from ouroboros import usage_accounting as ua
|
|
from ouroboros._usage_response import processing_receipt, usage_from_response
|
|
from tests.test_usage_accounting import data_root as _usage_data_root
|
|
|
|
data_root = _usage_data_root
|
|
|
|
|
|
def rows(root):
|
|
return [json.loads(line) for line in (root / ua.LEDGER_REL).read_text().splitlines()]
|
|
|
|
|
|
def test_native_mode_reaches_pricing_before_send_and_observation_before_settlement(data_root, monkeypatch):
|
|
priced = []
|
|
|
|
def price(*args, **kwargs):
|
|
priced.append(kwargs.get("processing_mode"))
|
|
return {"priority": 2.0, "default": 1.0}.get(kwargs.get("processing_mode"))
|
|
|
|
monkeypatch.setattr(ua, "estimate_cost_optional", price)
|
|
request = ua.AttemptRequest("openai/model", "openai", drive_root=data_root,
|
|
prompt_tokens_estimate=10, max_completion_tokens=20,
|
|
processing_preference="fast", submitted_processing_mode="priority")
|
|
|
|
def send():
|
|
assert priced == ["priority"]
|
|
assert rows(data_root)[0]["reservation_upper_bound_usd"] == 2.0
|
|
return {"service_tier": "default", "usage": {"prompt_tokens": 10, "completion_tokens": 2}}
|
|
|
|
ua.execute_physical_attempt(request, send)
|
|
assert priced == ["priority", "default"]
|
|
ledger = rows(data_root)
|
|
assert all(row["processing_preference"] == "fast" for row in ledger)
|
|
assert all(row["submitted_processing_mode"] == "priority" for row in ledger)
|
|
assert ledger[-1]["cost_usd"] == 1.0 and not ledger[-1]["cost_final"]
|
|
assert ledger[-1]["processing"]["observed"] == "standard"
|
|
assert ua.last_physical_attempt_capture().submitted_processing_mode == "priority"
|
|
|
|
|
|
def test_missing_observation_never_borrows_requested_fast_or_ordinary_tariff(data_root, monkeypatch):
|
|
priced = []
|
|
monkeypatch.setattr(ua, "estimate_cost_optional",
|
|
lambda *args, **kwargs: priced.append(kwargs.get("processing_mode")))
|
|
request = ua.AttemptRequest("openai/model", "openai", drive_root=data_root,
|
|
processing_preference="fast", submitted_processing_mode="priority")
|
|
ua.execute_physical_attempt(request, lambda: {"usage": {"prompt_tokens": 10, "completion_tokens": 1}})
|
|
assert priced == ["priority", "unknown"]
|
|
assert rows(data_root)[-1]["cost_usd"] is None
|
|
assert rows(data_root)[-1]["processing"]["observed"] == "unknown"
|
|
|
|
|
|
def test_anthropic_fast_cache_does_not_warm_standard_reservation(data_root, monkeypatch):
|
|
from ouroboros._usage_cache_splits import last_task_cache_split
|
|
|
|
monkeypatch.setattr(ua, "estimate_cost_optional", lambda *args, **kwargs: 1.0)
|
|
with ua.usage_scope(ua.UsageScope(drive_root=data_root, task_id="cache-task")):
|
|
request = ua.AttemptRequest("anthropic/model", "anthropic", processing_preference="fast",
|
|
submitted_processing_mode="fast")
|
|
ua.execute_physical_attempt(request, lambda: {"usage": {
|
|
"input_tokens": 10, "output_tokens": 2, "cache_read_input_tokens": 90, "speed": "fast"}})
|
|
assert last_task_cache_split("cache-task", "anthropic/model", provider="anthropic", processing_mode="fast") == 90
|
|
assert last_task_cache_split("cache-task", "anthropic/model", provider="anthropic", processing_mode="standard") is None
|
|
|
|
|
|
def test_anthropic_capacity_tier_is_not_speed_and_engine_mixed_stays_mixed():
|
|
usage, _, _ = usage_from_response({"usage": {"service_tier": "standard", "speed": "fast"}})
|
|
assert processing_receipt("anthropic", usage)["observed"] == "fast"
|
|
assert processing_receipt("anthropic", {"service_tier": "standard"}) is None
|
|
receipt = {"requested": "fast", "submitted": "fast", "submittedNative": "fast",
|
|
"observed": "mixed", "observedNative": ["fast", "default"],
|
|
"reason": None, "source": "native_session"}
|
|
assert processing_receipt("claudexor", {"processing": receipt}) == receipt
|
|
|
|
|
|
@pytest.mark.parametrize("attempts,expected", [
|
|
([{"usageCost": {"cashUsd": 0, "cashKnowledge": "unknown", "valuationUsd": 5}}], (None, False)),
|
|
([{"usageCost": {"cashUsd": 0, "cashKnowledge": "exact", "valuationUsd": 5}}], (0.0, False)),
|
|
([{"usageCost": {"cashUsd": 2, "cashKnowledge": "exact", "valuationUsd": 5}}], (2.0, False)),
|
|
([{"usageCost": {"cashUsd": 2, "cashKnowledge": "estimated"}}, {}], (2.0, True)),
|
|
([{}, {"usageCost": {"cashUsd": 2, "cashKnowledge": "exact"}}], (2.0, True)),
|
|
])
|
|
def test_session_actual_cost_components_preserve_unknown_and_cash_without_valuation(attempts, expected):
|
|
from ouroboros.delegate_custody_usage import disclosed_spend
|
|
|
|
assert disclosed_spend({"spendUsd": 99}, attempt_execution=attempts) == expected
|
|
|
|
|
|
def test_session_processing_and_components_are_retained_once_without_repricing(data_root):
|
|
from ouroboros.delegate_custody_usage import disclosed_spend
|
|
|
|
evidence = [{"attemptId": "a1", "harnessId": "claude",
|
|
"processing": {"requested": "fast", "submitted": "fast", "submittedNative": "fast",
|
|
"observed": "mixed", "observedNative": ["fast", "standard"],
|
|
"reason": None, "source": "native_telemetry"},
|
|
"processingCostBasis": {"nativeMode": "fast", "kind": "paid_credits", "source": "native_policy"},
|
|
"usageCost": {"cashUsd": 0, "valuationUsd": 5, "unknownUsd": 2,
|
|
"cashKnowledge": "unknown", "valuationKnowledge": "exact"}}]
|
|
original = copy.deepcopy(evidence)
|
|
spend, estimated = disclosed_spend({}, attempt_execution=evidence)
|
|
ua.record_subscription_session("session", drive_root=data_root, route="claude", spend_usd=spend,
|
|
spend_estimated=estimated, attempt_execution=evidence)
|
|
before = (data_root / ua.LEDGER_REL).read_bytes()
|
|
evidence[0]["usageCost"]["cashUsd"] = 90
|
|
ua.record_subscription_session("session", drive_root=data_root, route="claude", spend_usd=90,
|
|
attempt_execution=evidence)
|
|
assert (data_root / ua.LEDGER_REL).read_bytes() == before
|
|
assert rows(data_root)[-1]["attempt_execution"] == original
|
|
assert rows(data_root)[-1]["cost_usd"] is None
|
|
assert ua.usage_projection(data_root)["unknown_unmetered"] == 1
|
|
|
|
|
|
|
|
def test_review_wave_prices_each_captured_processing_choice_before_dispatch(data_root, monkeypatch):
|
|
monkeypatch.setenv("OUROBOROS_PROCESSING_PREFERENCE", "fast")
|
|
seen = []
|
|
monkeypatch.setattr(ua, "_reservation_cost", lambda request: seen.append(
|
|
(request.processing_preference, request.submitted_processing_mode)) or 0.1)
|
|
result = ua.review_wave_admission(data_root, root_task_id="review", models=["openai::same"] * 3,
|
|
prompt_chars=100, remaining_usd_override=1,
|
|
processing_preferences=["standard", "economy", ""])
|
|
assert seen == [("standard", "default"), ("economy", "flex"), ("", "")]
|
|
assert result["fits"] and result["slot_bounds"] == [0.1, 0.1, 0.1]
|
|
|
|
|
|
@pytest.mark.parametrize("observed,modes,expected", [("unknown", [], "unknown"),
|
|
("mixed", ["fast", "default"], "unknown"), ("standard", ["default"], "default")])
|
|
def test_loop_and_helper_display_keep_the_observed_price_qualifier(monkeypatch, observed, modes, expected):
|
|
from types import SimpleNamespace
|
|
from ouroboros import loop_llm_call
|
|
from ouroboros.tools import search
|
|
|
|
calls = []
|
|
def price(*args, **kwargs):
|
|
calls.append(kwargs.get("processing_mode"))
|
|
return None if kwargs.get("processing_mode") == "unknown" else 0.1
|
|
monkeypatch.setattr(loop_llm_call, "estimate_cost_optional", price)
|
|
monkeypatch.setattr(search, "estimate_cost_optional", price)
|
|
usage = {"cost": None, "provider": "openrouter", "prompt_tokens": 10, "completion_tokens": 2,
|
|
"processing": {"observed": observed, "observedNative": modes}}
|
|
cost, _, _, _ = loop_llm_call._normalize_usage_cost(usage, model="openai/same", use_local=False)
|
|
ctx = SimpleNamespace(pending_events=[], task_metadata={}, task_id="helper")
|
|
search._emit_simple_usage(ctx, provider="openrouter", model="openai/same", usage={**usage, "cost": None})
|
|
assert calls == [expected, expected]
|
|
assert ctx.pending_events[0]["cost"] == cost == (None if expected == "unknown" else 0.1)
|
|
|
|
|
|
def test_host_view_callback_is_invoked_but_never_recorded_as_a_model_argument(tmp_path, monkeypatch):
|
|
from ouroboros import llm_observability as observed
|
|
|
|
recorded, views = [], []
|
|
monkeypatch.setattr(observed, "persist_call", lambda *args, **kwargs: recorded.append(kwargs["payload"]) or {})
|
|
callback = lambda messages, schemas: views.append(messages)
|
|
|
|
class Client:
|
|
def chat(self, **kwargs):
|
|
kwargs["model_context_observer"](kwargs["messages"], kwargs.get("tools"))
|
|
return {"role": "assistant", "content": "done"}, {}
|
|
|
|
observed.chat_observed(Client(), drive_root=tmp_path, messages=[{"role": "user", "content": "exact"}],
|
|
model="model", model_context_observer=callback)
|
|
assert len(views) == 1 and recorded
|
|
assert all("model_context_observer" not in payload.get("kwargs", {}) for payload in recorded)
|