harn-vm 0.10.169

//! Bounds use the full catalog context, never a request tokenizer estimate or
//! prior output/cache averages. Unsupported billing shapes are refused.
use rust_decimal::Decimal;

use super::{error, money, DenialKind};
use crate::llm::api::{LlmRequestPayload, LlmResult, PromptCacheTtl};
use crate::value::VmError;

pub(super) struct AttemptBound {
    catalog_id: String,
    input_limit: i64,
    output_limit: i64,
    input_rate: Decimal,
    output_rate: Decimal,
    fresh_input_basis: bool,
}

impl AttemptBound {
    pub(super) fn for_request(request: &LlmRequestPayload) -> Result<Self, VmError> {
        let anthropic = request.provider == "anthropic";
        if !(request.provider == "openai" || anthropic)
            || request.fast
            || request
                .reasoning_mode
                .as_deref()
                .is_some_and(|mode| mode != crate::llm::reasoning_modes::STANDARD_MODE_ID)
            || request.vision
            || !request.provider_tools.is_empty()
            || request.provider_overrides.is_some()
            || request.previous_response_id.is_some()
            || request.background == Some(true)
            || request.compact == Some(true)
            || request.prediction.is_some()
            || !request.anthropic_beta_features.is_empty()
            || request
                .native_tools
                .as_ref()
                .is_some_and(|tools| tools.iter().any(|tool| !function_tool(tool, anthropic)))
            || !request
                .messages
                .iter()
                .all(|message| text_message(message, anthropic))
            || (anthropic
                && (request.prompt_cache_ttl == Some(PromptCacheTtl::OneHour)
                    || !request.messages.iter().all(ordinary_message_cache_control)
                    || request
                        .native_tools
                        .as_ref()
                        .is_some_and(|tools| !tools.iter().all(ordinary_cache_control))))
        {
            return Err(error(
                DenialKind::UnsupportedBillingShape,
                "conservative admission does not support this provider or billing shape",
            ));
        }
        let catalog_id =
            crate::llm_config::model_catalog_id_for_route(&request.provider, &request.model)
                .ok_or_else(|| {
                    error(
                        DenialKind::UnknownPricing,
                        "conservative admission requires an exact catalog route",
                    )
                })?;
        let row = crate::llm_config::model_catalog_entry(&catalog_id)
            .ok_or_else(|| error(DenialKind::UnknownPricing, "missing catalog model"))?;
        let input_limit = i64::try_from(
            row.context_window
                .max(row.runtime_context_window.unwrap_or(0)),
        )
        .ok()
        .filter(|value| *value > 0)
        .ok_or_else(|| {
            error(
                DenialKind::UnsupportedBillingShape,
                "conservative admission requires a catalog context limit",
            )
        })?;
        let pricing = row.pricing.ok_or_else(|| {
            error(
                DenialKind::UnknownPricing,
                "conservative admission requires known model pricing",
            )
        })?;
        if anthropic && pricing.cache_write_per_mtok.is_none() {
            return Err(error(
                DenialKind::UnknownPricing,
                "conservative Anthropic admission requires a known cache-write rate",
            ));
        }
        // An expiring discount must not reduce a reservation. Until the rate
        // contract exposes a temporal upper bound, these rows are unsupported.
        if !pricing.promotions.is_empty() || request.max_tokens <= 0 {
            return Err(error(
                DenialKind::UnsupportedBillingShape,
                "conservative admission requires stable pricing and an explicit output limit",
            ));
        }
        let mut input_rate = Decimal::ZERO;
        let mut output_rate = Decimal::ZERO;
        // Inspect every whole-request band; do not assume multipliers increase.
        let thresholds =
            std::iter::once(0).chain(pricing.input_token_bands.iter().filter_map(|b| {
                i64::try_from(b.minimum_input_tokens)
                    .ok()
                    .filter(|n| *n <= input_limit)
            }));
        for threshold in thresholds {
            let band = pricing.for_input_tokens(threshold);
            for rate in [
                Some(band.input_per_mtok),
                band.cache_read_per_mtok,
                band.cache_write_per_mtok,
            ]
            .into_iter()
            .flatten()
            {
                input_rate = input_rate.max(money(rate)?);
            }
            output_rate = output_rate.max(money(band.output_per_mtok)?);
        }
        Ok(Self {
            catalog_id,
            input_limit,
            output_limit: request.max_tokens,
            input_rate,
            output_rate,
            fresh_input_basis: anthropic,
        })
    }

    fn cost(&self, input: i64, output: i64) -> Decimal {
        (Decimal::from(input) * self.input_rate + Decimal::from(output) * self.output_rate)
            / Decimal::from(1_000_000)
    }

    pub(super) fn total(&self) -> Decimal {
        self.cost(self.input_limit, self.output_limit)
    }

    pub(super) fn known_contract_violation(&self, result: &LlmResult) -> bool {
        crate::llm_config::model_catalog_id_for_route(&result.provider, &result.model).as_deref()
            != Some(self.catalog_id.as_str())
            || result
                .telemetry
                .server_prompt_tokens
                .is_some_and(|input| input > self.input_limit)
            || result
                .telemetry
                .server_output_tokens
                .is_some_and(|output| output > self.output_limit)
            || (self.fresh_input_basis && result.input_tokens > self.input_limit)
    }

    /// Settle at the conservative rates, without claiming a billing receipt or
    /// applying cache discounts. Missing or inconsistent wire usage keeps the
    /// full reservation. Native OpenAI counters include reasoning output.
    pub(super) fn observed_upper(&self, result: &LlmResult) -> Option<(Decimal, bool)> {
        let input = result.telemetry.server_prompt_tokens?;
        let output = result.telemetry.server_output_tokens?;
        let route = crate::llm_config::model_catalog_id_for_route(&result.provider, &result.model)?;
        if (result.provider != "openai" && result.provider != "anthropic")
            || route != self.catalog_id
            || result.served_fast
            || input < 0
            || output < 0
            || (!self.fresh_input_basis && input != result.input_tokens)
            || output != result.output_tokens
        {
            return None;
        }
        // Anthropic's wire counter is fresh input. Only complete, consistent
        // cache categories can release the unused input reservation.
        let input_upper = if self.fresh_input_basis {
            match result.telemetry.reported_cache_usage.as_deref() {
                Some(cache) if cache.read_tokens.is_some() && cache.write_tokens.is_some() => {
                    let total = cache.complete_prompt_tokens(input)?;
                    if total != result.input_tokens
                        || cache.read_tokens != Some(result.cache_read_tokens)
                        || cache.write_tokens != Some(result.cache_write_tokens)
                    {
                        return None;
                    }
                    total
                }
                _ => self.input_limit.max(result.input_tokens).max(input),
            }
        } else {
            input
        };
        Some((
            self.cost(input_upper, output),
            input_upper > self.input_limit || output > self.output_limit,
        ))
    }
}

fn text_message(message: &serde_json::Value, anthropic: bool) -> bool {
    let Some(fields) = message.as_object() else {
        return false;
    };
    if !supported_tool_result_error_flag(message) {
        return false;
    }
    if !fields.keys().all(|key| {
        matches!(
            key.as_str(),
            "role" | "content" | "name" | "tool_calls" | "tool_call_id" | "is_error"
        ) || (anthropic
            && matches!(
                key.as_str(),
                "cache_control" | "provider_continuation" | "reasoning"
            ))
    }) {
        return false;
    }
    let is_assistant = message.get("role").and_then(serde_json::Value::as_str) == Some("assistant");
    if fields.contains_key("provider_continuation")
        && (!anthropic
            || !is_assistant
            || !crate::llm::reasoning_history::is_supported_anthropic_continuation(
                &fields["provider_continuation"],
            ))
    {
        return false;
    }
    // Durable `reasoning` is private transcript text. Anthropic egress drops
    // it from the message object; the signed provider continuation above is the
    // separately typed wire data. The full-context admission bound already
    // covers the provider-visible signed blocks and generated thinking.
    if fields.contains_key("reasoning")
        && (!anthropic || !is_assistant || !fields["reasoning"].is_string())
    {
        return false;
    }
    // Function arguments/results are text; image/audio/file payloads require a
    // separate pricing contract. Reject even unknown content part kinds.
    match message.get("content") {
        None | Some(serde_json::Value::Null) | Some(serde_json::Value::String(_)) => true,
        Some(serde_json::Value::Array(parts)) => parts.iter().all(|part| {
            matches!(
                part.get("type").and_then(|v| v.as_str()),
                Some("text" | "input_text" | "output_text")
            ) || (anthropic && is_assistant && supported_anthropic_content_part(part))
        }),
        _ => false,
    }
}

fn supported_anthropic_content_part(part: &serde_json::Value) -> bool {
    match part.get("type").and_then(serde_json::Value::as_str) {
        Some("thinking" | "redacted_thinking") => {
            crate::llm::reasoning_history::is_supported_anthropic_continuation(
                &serde_json::json!({"anthropic": {"content_blocks": [part]}}),
            )
        }
        Some("tool_use") => {
            let Some(fields) = part.as_object() else {
                return false;
            };
            fields.len() == 4
                && fields.contains_key("id")
                && fields.contains_key("name")
                && fields.contains_key("input")
                && part
                    .get("id")
                    .and_then(serde_json::Value::as_str)
                    .is_some_and(|value| !value.is_empty())
                && part
                    .get("name")
                    .and_then(serde_json::Value::as_str)
                    .is_some_and(|value| !value.is_empty())
                && part.get("input").is_some_and(serde_json::Value::is_object)
        }
        _ => false,
    }
}

/// The host records a typed result outcome beside native tool-result messages.
/// Providers that understand the flag must receive the same outcome the host
/// observed; arbitrary values or assistant/user-authored variants are not part
/// of this message contract. Text-channel tool results use the ordinary user
/// role, so their boolean remains admissible there as well.
pub(super) fn supported_tool_result_error_flag(message: &serde_json::Value) -> bool {
    let Some(is_error) = message.get("is_error") else {
        return true;
    };
    if !is_error.is_boolean() {
        return false;
    }
    match message.get("role").and_then(serde_json::Value::as_str) {
        Some("tool" | "tool_result") => message
            .get("tool_call_id")
            .or_else(|| message.get("tool_use_id"))
            .or_else(|| message.get("call_id"))
            .and_then(serde_json::Value::as_str)
            .is_some_and(|call_id| !call_id.is_empty()),
        Some("user") => true,
        _ => false,
    }
}

fn function_tool(tool: &serde_json::Value, anthropic: bool) -> bool {
    if anthropic {
        // The native Anthropic client-function schema has no `type` field.
        // Server tools carry their own typed billing contract and stay refused.
        tool.get("type").is_none()
            && tool.get("name").is_some_and(|value| value.is_string())
            && tool
                .get("input_schema")
                .is_some_and(|value| value.is_object())
    } else {
        tool.get("type").and_then(|value| value.as_str()) == Some("function")
    }
}

fn ordinary_message_cache_control(message: &serde_json::Value) -> bool {
    ordinary_cache_control(message)
        && message
            .get("content")
            .and_then(|value| value.as_array())
            .is_none_or(|parts| parts.iter().all(ordinary_cache_control))
}

fn ordinary_cache_control(value: &serde_json::Value) -> bool {
    // Only wire-level breakpoints affect billing. Function argument/schema
    // properties named `cache_control` are user data, not cache directives.
    value.get("cache_control").is_none_or(|control| {
        control.get("type").and_then(|value| value.as_str()) == Some("ephemeral")
            && control
                .get("ttl")
                .is_none_or(|ttl| ttl.as_str() == Some("5m"))
    })
}