lm15 1.0.0-rc.7

//! The lm15 vet shim (harness/PROTOCOL.md): newline-delimited JSON on
//! stdin/stdout, one reply per request, same `id`. Thin by rule: it parses
//! the protocol line, calls the public functions users call, and
//! serializes. No network.

#[path = "vet/managed.rs"]
#[allow(clippy::result_large_err)]
mod vet_managed;
use std::io::{self, BufRead, Write};

use serde_json::{json, Map, Value};

use lm15::auth::{parse_rfc3339, Credential};
use lm15::cloud::sigv4::{self, AwsKeys, SigningRequest};
use lm15::errors::{normalize_error, Lm15Error};
use lm15::serde::{roundtrip, validate};
use lm15::stream::materialize_response;
use lm15::types::{
    BatchRequest, FileUploadRequest, ImageGenerationRequest, LiveClientEvent, LiveConfig, Request,
    Response, SpeechGenerationRequest, ValidationError, VideoGenerationRequest,
};
use lm15::wire::{FixedClock, TransportRequest};
use lm15::{Canonical, HostSettings};

const LANGUAGE: &str = "rust";
const IMPL_VERSION: &str = env!("CARGO_PKG_VERSION");

/// Ops this shim answers (`capabilities.ops`).
const OPS: &[&str] = &[
    "build_request",
    "plan",
    "surface_dump",
    "ingest_openai_chat",
    "build_models_request",
    "batch_op_build",
    "cache_op_build",
    "cache_op_parse",
    "batch_op_parse",
    "file_op_build",
    "generation_build",
    "generation_parse",
    "file_op_parse",
    "capabilities",
    "explain_auth",
    "managed_run",
    "normalize_error",
    "parse_response",
    "parse_models_response",
    "replay_live",
    "replay_stream",
    "resolve_model",
    "resolve_openai_chat_model",
    "serde_roundtrip",
    "sigv4_sign",
    "token_exchange_build",
    "token_exchange_parse",
    "validate",
    "video_op_build",
    "video_op_parse",
];

enum Failure {
    Validation(ValidationError),
    Lm15(Box<Lm15Error>),
    /// A typed error plus extra reply fields (`replay_stream`: the event
    /// trace that parsed before assembly refused, MAP-9).
    Lm15Extra(Box<Lm15Error>, Map<String, Value>),
}

impl From<ValidationError> for Failure {
    fn from(err: ValidationError) -> Self {
        Failure::Validation(err)
    }
}

impl From<Lm15Error> for Failure {
    fn from(err: Lm15Error) -> Self {
        Failure::Lm15(Box::new(err))
    }
}

fn field<'a>(msg: &'a Map<String, Value>, key: &str) -> Result<&'a Value, Failure> {
    msg.get(key)
        .ok_or_else(|| ValidationError::type_error(format!("missing field: {key}")).into())
}

fn field_str(msg: &Map<String, Value>, key: &str) -> Result<String, Failure> {
    match field(msg, key)? {
        Value::String(s) => Ok(s.clone()),
        other => Ok(other.to_string()),
    }
}

fn op_capabilities() -> Value {
    json!({
        "language": LANGUAGE,
        "ops": OPS,
        "impl_version": IMPL_VERSION,
    })
}

fn op_serde_roundtrip(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let kind = field_str(msg, "kind")?;
    let value = roundtrip(&kind, field(msg, "value")?)?;
    Ok(json!({ "value": value }))
}

fn op_validate(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let kind = field_str(msg, "kind")?;
    let normalized = validate(&kind, field(msg, "value")?)?;
    Ok(json!({ "ok": true, "normalized": normalized }))
}

fn op_resolve_model(msg: &Map<String, Value>) -> Result<Value, Failure> {
    // PROTOCOL.md resolve_model: the router's pure `resolve` over
    // harness-supplied inputs only — the env map (never the process env)
    // and, when given, an explicit catalog of canonical ModelInfo values.
    let model = field_str(msg, "model")?;
    let mut config = lm15::router::RouterConfig::new();
    let mut env: Vec<(String, String)> = Vec::new();
    if let Some(map) = msg.get("env").and_then(Value::as_object) {
        for (k, v) in map {
            if let Some(s) = v.as_str() {
                env.push((k.clone(), s.to_string()));
            }
        }
    }
    config = config.env(env);
    if let Some(entries) = msg.get("catalog").and_then(Value::as_array) {
        let mut catalog = Vec::with_capacity(entries.len());
        for entry in entries {
            catalog.push(lm15::types::ModelInfo::from_json(entry)?);
        }
        config = config.catalog(catalog);
    }
    let resolution = lm15::router::LMRouter::with_config(config)?.resolve(&model)?;
    Ok(json!({
        "provider": resolution.provider,
        "model": resolution.model,
        "source": resolution.source.as_str(),
    }))
}

fn op_resolve_openai_chat_model(msg: &Map<String, Value>) -> Result<Value, Failure> {
    // PROTOCOL.md resolve_openai_chat_model: the router's OpenAI-SDK /
    // litellm door over harness-supplied inputs only.
    let model = field_str(msg, "model")?;
    let mut env: Vec<(String, String)> = Vec::new();
    if let Some(map) = msg.get("env").and_then(Value::as_object) {
        for (k, v) in map {
            if let Some(s) = v.as_str() {
                env.push((k.clone(), s.to_string()));
            }
        }
    }
    let config = lm15::router::RouterConfig::new().env(env);
    let resolution = lm15::router::LMRouter::with_config(config)?.resolve_openai_chat(&model)?;
    Ok(json!({
        "provider": resolution.provider,
        "model": resolution.model,
        "source": resolution.source.as_str(),
    }))
}

fn op_normalize_error(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let provider = field_str(msg, "provider")?;
    let status = field(msg, "status")?
        .as_u64()
        .and_then(|s| u16::try_from(s).ok())
        .ok_or_else(|| ValidationError::type_error("status must be an int"))?;
    let body_text = field_str(msg, "body_text")?;
    let err = normalize_error(&provider, status, &body_text)?;
    Ok(json!({
        "class": err.class_name(),
        "code": err.code().as_str(),
        "provider_code": err.provider_code(),
        "message": err.message(),
    }))
}

fn op_explain_auth(msg: &Map<String, Value>) -> Result<Value, Failure> {
    // PROTOCOL.md explain_auth: the harness owns every input. The env map is
    // the whole environment (never the process env), api_keys_providers get
    // the sentinel planted, credentials_path is the harness-written file.
    let provider = field_str(msg, "provider")?;
    let mut env = std::collections::HashMap::new();
    if let Some(map) = msg.get("env").and_then(Value::as_object) {
        for (k, v) in map {
            if let Some(s) = v.as_str() {
                env.insert(k.clone(), s.to_string());
            }
        }
    }
    let options = lm15::auth::ExplainOptions {
        env: Some(env),
        api_key_providers: msg
            .get("api_keys_providers")
            .and_then(Value::as_array)
            .map(|a| {
                a.iter()
                    .filter_map(Value::as_str)
                    .map(str::to_string)
                    .collect()
            })
            .unwrap_or_default(),
        credentials_path: msg
            .get("credentials_path")
            .and_then(Value::as_str)
            .map(std::path::PathBuf::from),
        files: msg.get("files").and_then(Value::as_object).map(|m| {
            m.iter()
                .filter_map(|(k, v)| v.as_str().map(|s| (k.clone(), s.to_string())))
                .collect()
        }),
        settings: settings_of(msg),
        credential: msg
            .get("credential")
            .and_then(Value::as_str)
            .map(str::to_string),
        base_url: msg
            .get("base_url")
            .and_then(Value::as_str)
            .map(str::to_string),
        auth: None,
        callable_providers: msg
            .get("callable_providers")
            .and_then(Value::as_array)
            .map(|v| {
                v.iter()
                    .filter_map(Value::as_str)
                    .map(str::to_string)
                    .collect()
            })
            .unwrap_or_default(),
    };
    let report = lm15::auth::explain_auth(&provider, &options).map_err(Lm15Error::from)?;
    let steps: Vec<Value> = report
        .steps
        .iter()
        .map(|s| json!({ "kind": s.kind, "state": s.state.as_str() }))
        .collect();
    let report_text = format!("{}\n{}\n{:?}", report.describe(), report, report);
    // PROTOCOL.md explain_auth `settings` (2026-09-26): value and origin per
    // host setting; unprobed and missing ones have a null value.
    let values: Map<String, Value> = report
        .settings
        .iter()
        .filter(|(k, _)| k != "error")
        .map(|(k, v)| (k.clone(), Value::String(v.clone())))
        .collect();
    let settings: Map<String, Value> = report
        .setting_sources
        .iter()
        .map(|(name, origin)| {
            let entry = if origin == "missing" {
                json!({"value": null, "from": null})
            } else if let Some(from) = origin.strip_prefix("unprobed:") {
                json!({"value": null, "from": from, "state": "unprobed"})
            } else {
                json!({"value": values.get(name).cloned().unwrap_or(Value::Null), "from": origin})
            };
            (name.clone(), entry)
        })
        .collect();
    Ok(
        json!({ "configured": report.configured, "steps": steps, "report_text": report_text,
        "base_url":report.base_url,"settings":settings,
        "named_credential":report.named_credential,"endpoint_source":report.endpoint_source }),
    )
}

/// PROTOCOL.md: `credential` (an AUTH-2 value) wins over the `api_key`
/// shorthand. The shim never reads environment keys.
fn credential_of(msg: &Map<String, Value>) -> Result<Credential, Failure> {
    if let Some(value @ Value::Object(_)) = msg.get("credential") {
        return Ok(Credential::from_json(value)?);
    }
    let api_key = field_str(msg, "api_key")?;
    Ok(Credential::api_key(api_key).map_err(Lm15Error::from)?)
}

/// `now` (RFC 3339, UTC) as Unix seconds: the clock for every
/// time-dependent byte. Absent means the wall clock.
fn clock_of(msg: &Map<String, Value>) -> Result<Option<FixedClock>, Failure> {
    match msg.get("now") {
        None | Some(Value::Null) => Ok(None),
        Some(Value::String(text)) => parse_rfc3339(text)
            .map(|unix| Some(FixedClock(unix)))
            .ok_or_else(|| ValidationError::value(format!("now is not RFC 3339: {text:?}")).into()),
        Some(other) => {
            Err(ValidationError::type_error(format!("now must be a string, got {other}")).into())
        }
    }
}

fn settings_of(msg: &Map<String, Value>) -> Option<HostSettings> {
    let settings = msg.get("settings")?.as_object()?;
    Some(
        settings
            .iter()
            .map(|(k, v)| {
                let value = match v {
                    Value::String(s) => s.clone(),
                    other => other.to_string(),
                };
                (k.clone(), value)
            })
            .collect(),
    )
}

/// The protocol's `build_request` shape: `url` without its query, decoded
/// `params`, lowercase header names, the JSON body or `null`.
fn transport_request_json(request: &TransportRequest) -> Value {
    let params: Map<String, Value> = request
        .params
        .iter()
        .map(|(k, v)| (k.clone(), Value::String(v.clone())))
        .collect();
    let headers: Map<String, Value> = request
        .headers
        .iter()
        .map(|(k, v)| (k.to_ascii_lowercase(), Value::String(v.clone())))
        .collect();
    let mut out = json!({
        "method": request.method,
        "url": request.url,
        "params": params,
        "headers": headers,
        "body": request.body.clone().unwrap_or(Value::Null),
    });
    if let Some(raw) = &request.raw {
        out["body_b64"] = Value::String(lm15::types::base64_encode(raw));
    }
    out
}

fn adaptation_policy(msg: &Map<String, Value>) -> Result<lm15::AdaptationPolicy, Failure> {
    Ok(lm15::AdaptationPolicy::parse(
        msg.get("adaptations")
            .and_then(Value::as_str)
            .unwrap_or("note"),
    )?)
}

fn op_plan(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let provider = field_str(msg, "provider")?;
    let definition = lm15::registry::lookup(&provider)
        .ok_or_else(|| Lm15Error::not_configured("unknown provider"))?;
    let builder = lm15::LmBuilder::for_entry(definition).adaptations(adaptation_policy(msg)?);
    let records = builder.plan(&Request::from_json(field(msg, "canonical_request")?)?)?;
    Ok(json!({"adaptations":records.iter().map(Canonical::to_json).collect::<Vec<_>>()}))
}

fn op_build_request(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let provider = field_str(msg, "provider")?;
    let credential = credential_of(msg)?;
    let base_url = msg.get("base_url").and_then(Value::as_str);
    let clock = clock_of(msg)?;
    let lm = lm15::registry::adapter_for(
        &provider,
        credential,
        base_url,
        settings_of(msg),
        clock.map(|c| Box::new(c) as Box<dyn lm15::wire::Clock + Send + Sync>),
    )?;
    let request = Request::from_json(field(msg, "canonical_request")?)?;
    let stream = msg.get("stream").and_then(Value::as_bool).unwrap_or(false);
    let lm = lm.with_adaptations(adaptation_policy(msg)?);
    let transport = lm.build_request(&request, stream)?;
    let mut output = transport_request_json(&transport);
    let records = if stream {
        lm.plan_stream(&request)?
    } else {
        lm.plan(&request)?
    };
    if !records.is_empty() && lm.adaptation_policy() != lm15::AdaptationPolicy::Silent {
        output["adaptations"] = Value::Array(records.iter().map(Canonical::to_json).collect());
    }
    Ok(output)
}

/// PROTOCOL.md § build_models_request: the shim constructs `openai-codex`
/// with account id `test-account` (a non-JWT key carries none).
fn surface_adapter(
    msg: &Map<String, Value>,
    credential: Credential,
) -> Result<lm15::ProviderLM, Failure> {
    let provider = field_str(msg, "provider")?;
    let definition = lm15::registry::lookup(&provider)
        .ok_or_else(|| Lm15Error::not_configured(format!("unknown provider {provider:?}")))?;
    let mut builder = lm15::LmBuilder::for_entry(definition)
        .api_key(credential)
        .adaptations(adaptation_policy(msg)?);
    if let Some(base_url) = msg.get("base_url").and_then(Value::as_str) {
        builder = builder.base_url(base_url);
    }
    if let Some(settings) = settings_of(msg) {
        builder = builder.settings(settings);
    }
    if let Some(clock) = clock_of(msg)? {
        builder = builder.clock(clock);
    }
    if definition.id == "openai-codex" {
        builder = builder.account_id("test-account");
    }
    Ok(builder.build()?)
}

fn op_file_op_build(msg: &Map<String, Value>) -> Result<Value, Failure> {
    use lm15::adapter::FileOp;
    let lm = surface_adapter(msg, credential_of(msg)?)?;
    let op = field_str(msg, "file_op")?;
    let file_id = msg.get("file_id").and_then(Value::as_str).unwrap_or("");
    let built = match op.as_str() {
        "upload" => {
            let request = FileUploadRequest::from_json(field(msg, "upload_request")?)?;
            lm.file_request(&FileOp::Upload(&request))?
        }
        "get" => lm.file_request(&FileOp::Get(file_id))?,
        "delete" => lm.file_request(&FileOp::Delete(file_id))?,
        "download" => lm.file_request(&FileOp::Download(file_id))?,
        "list" => lm.file_request(&FileOp::List {
            limit: msg.get("limit").and_then(Value::as_u64).unwrap_or(20),
            cursor: msg.get("cursor").and_then(Value::as_str),
        })?,
        other => return Err(ValidationError::value(format!("unknown file_op {other:?}")).into()),
    };
    Ok(transport_request_json(&built))
}

fn op_file_op_parse(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let lm = surface_adapter(
        msg,
        Credential::api_key("vet-parse-only").map_err(Lm15Error::from)?,
    )?;
    let status = msg.get("status").and_then(Value::as_u64).unwrap_or(200) as u16;
    let body = body_of(msg)?;
    Ok(match field_str(msg, "kind")?.as_str() {
        "info" => json!({ "file": lm.parse_file_info(status, &body)?.to_json() }),
        "page" => json!({ "page": lm.parse_file_page(status, &body)?.to_json() }),
        other => return Err(ValidationError::value(format!("unknown kind {other:?}")).into()),
    })
}

fn op_batch_op_build(msg: &Map<String, Value>) -> Result<Value, Failure> {
    use lm15::adapter::BatchAction;
    let lm = surface_adapter(msg, credential_of(msg)?)?;
    let action = field_str(msg, "action")?;
    let batch_id = msg.get("batch_id").and_then(Value::as_str).unwrap_or("");
    let request = match msg.get("batch_request") {
        Some(value @ Value::Object(_)) => Some(BatchRequest::from_json(value)?),
        _ => None,
    };
    let need = || -> Result<&BatchRequest, Failure> {
        request
            .as_ref()
            .ok_or_else(|| ValidationError::value("batch_request is required").into())
    };
    let upload_body = msg.get("upload_body").and_then(Value::as_object);
    let status_body = msg.get("status_body").and_then(Value::as_object);
    let built = match action.as_str() {
        "upload" => lm.batch_requests(&BatchAction::Upload(need()?))?,
        "submit" => lm.batch_requests(&BatchAction::Submit {
            request: need()?,
            upload_body,
        })?,
        "status" => lm.batch_requests(&BatchAction::Status(batch_id))?,
        "cancel" => lm.batch_requests(&BatchAction::Cancel(batch_id))?,
        "result_fetches" => lm.batch_requests(&BatchAction::ResultFetches(
            status_body.ok_or_else(|| ValidationError::value("status_body is required"))?,
        ))?,
        "list" => lm.batch_requests(&BatchAction::List(
            msg.get("limit").and_then(Value::as_u64).unwrap_or(20),
        ))?,
        other => return Err(ValidationError::value(format!("unknown action {other:?}")).into()),
    };
    Ok(json!({ "requests": built.iter().map(transport_request_json).collect::<Vec<_>>() }))
}

fn op_batch_op_parse(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let lm = surface_adapter(
        msg,
        Credential::api_key("vet-parse-only").map_err(Lm15Error::from)?,
    )?;
    let status = msg.get("status").and_then(Value::as_u64).unwrap_or(200) as u16;
    Ok(match field_str(msg, "kind")?.as_str() {
        "job" => json!({ "job": lm.parse_batch_job(status, &body_of(msg)?)?.to_json() }),
        "list" => {
            json!({ "jobs": lm.parse_batch_jobs(status, &body_of(msg)?)?.iter().map(|j| j.to_json()).collect::<Vec<_>>() })
        }
        "entries" => {
            let status_body = msg
                .get("status_body")
                .and_then(Value::as_object)
                .ok_or_else(|| ValidationError::value("status_body is required"))?;
            let fetched: Vec<Vec<u8>> = msg
                .get("fetched_b64")
                .and_then(Value::as_array)
                .into_iter()
                .flatten()
                .filter_map(Value::as_str)
                .map(|b64| lm15::types::base64_decode(b64).map_err(Failure::from))
                .collect::<Result<_, _>>()?;
            let entries = lm.parse_batch_entries(status_body, &fetched)?;
            json!({ "entries": entries.iter().map(|e| e.to_json()).collect::<Vec<_>>() })
        }
        other => return Err(ValidationError::value(format!("unknown kind {other:?}")).into()),
    })
}

fn op_cache_op_build(msg: &Map<String, Value>) -> Result<Value, Failure> {
    use lm15::adapter::CacheOp;
    let lm = surface_adapter(msg, credential_of(msg)?)?;
    let cache_id = msg.get("cache_id").and_then(Value::as_str).unwrap_or("");
    let ttl = msg.get("ttl_seconds").and_then(Value::as_u64);
    let built = match field_str(msg, "cache_op")?.as_str() {
        "create" => {
            let prefix = Request::from_json(field(msg, "prefix_request")?)?;
            lm.cache_request(&CacheOp::Create {
                prefix: &prefix,
                ttl_seconds: ttl,
                label: msg.get("label").and_then(Value::as_str),
            })?
        }
        "get" => lm.cache_request(&CacheOp::Get(cache_id))?,
        "delete" => lm.cache_request(&CacheOp::Delete(cache_id))?,
        "update" => lm.cache_request(&CacheOp::Update {
            cache_id,
            ttl_seconds: ttl.ok_or_else(|| ValidationError::value("ttl_seconds is required"))?,
        })?,
        "list" => lm.cache_request(&CacheOp::List {
            limit: msg.get("limit").and_then(Value::as_u64).unwrap_or(20),
            cursor: msg.get("cursor").and_then(Value::as_str),
        })?,
        other => return Err(ValidationError::value(format!("unknown cache_op {other:?}")).into()),
    };
    Ok(transport_request_json(&built))
}

fn op_cache_op_parse(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let lm = surface_adapter(
        msg,
        Credential::api_key("vet-parse-only").map_err(Lm15Error::from)?,
    )?;
    let status = msg.get("status").and_then(Value::as_u64).unwrap_or(200) as u16;
    let body = body_of(msg)?;
    Ok(match field_str(msg, "kind")?.as_str() {
        "info" => json!({ "cache": lm.parse_cache_info(status, &body)?.to_json() }),
        "page" => json!({ "page": lm.parse_cache_page(status, &body)?.to_json() }),
        other => return Err(ValidationError::value(format!("unknown kind {other:?}")).into()),
    })
}

fn headers_of(msg: &Map<String, Value>) -> Vec<(String, String)> {
    msg.get("headers")
        .and_then(Value::as_object)
        .map(|h| {
            h.iter()
                .map(|(k, v)| {
                    (
                        k.clone(),
                        v.as_str()
                            .map(str::to_string)
                            .unwrap_or_else(|| v.to_string()),
                    )
                })
                .collect()
        })
        .unwrap_or_default()
}

fn op_generation_build(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let lm = surface_adapter(msg, credential_of(msg)?)?;
    let request = field(msg, "generation_request")?;
    let built = match field_str(msg, "kind")?.as_str() {
        "image" => lm.image_generate_request(&ImageGenerationRequest::from_json(request)?)?,
        "speech" => lm.speech_generate_request(&SpeechGenerationRequest::from_json(request)?)?,
        other => return Err(ValidationError::value(format!("unknown kind {other:?}")).into()),
    };
    Ok(transport_request_json(&built))
}

fn op_generation_parse(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let lm = surface_adapter(
        msg,
        Credential::api_key("vet-parse-only").map_err(Lm15Error::from)?,
    )?;
    let request = field(msg, "generation_request")?;
    let status = msg.get("status").and_then(Value::as_u64).unwrap_or(200) as u16;
    let headers = headers_of(msg);
    let body = body_of(msg)?;
    Ok(match field_str(msg, "kind")?.as_str() {
        "image" => lm
            .parse_image_generation(
                &ImageGenerationRequest::from_json(request)?,
                status,
                &headers,
                &body,
            )?
            .to_json(),
        "speech" => lm
            .parse_speech_generation(
                &SpeechGenerationRequest::from_json(request)?,
                status,
                &headers,
                &body,
            )?
            .to_json(),
        other => return Err(ValidationError::value(format!("unknown kind {other:?}")).into()),
    })
}

fn op_video_op_build(msg: &Map<String, Value>) -> Result<Value, Failure> {
    use lm15::adapter::VideoAction;
    let lm = surface_adapter(msg, credential_of(msg)?)?;
    let video_id = msg.get("video_id").and_then(Value::as_str).unwrap_or("");
    let built = match field_str(msg, "action")?.as_str() {
        "submit" => {
            let request = VideoGenerationRequest::from_json(field(msg, "video_request")?)?;
            lm.video_requests(&VideoAction::Submit(&request))?
        }
        "status" => lm.video_requests(&VideoAction::Status(video_id))?,
        "result_fetch" => lm.video_requests(&VideoAction::ResultFetch(
            msg.get("status_body")
                .and_then(Value::as_object)
                .ok_or_else(|| ValidationError::value("status_body is required"))?,
        ))?,
        "list" => lm.video_requests(&VideoAction::List {
            limit: msg.get("limit").and_then(Value::as_u64).unwrap_or(20),
            model: msg.get("model").and_then(Value::as_str),
        })?,
        other => return Err(ValidationError::value(format!("unknown action {other:?}")).into()),
    };
    Ok(json!({ "requests": built.iter().map(transport_request_json).collect::<Vec<_>>() }))
}

fn op_video_op_parse(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let lm = surface_adapter(
        msg,
        Credential::api_key("vet-parse-only").map_err(Lm15Error::from)?,
    )?;
    let status = msg.get("status").and_then(Value::as_u64).unwrap_or(200) as u16;
    let video_id = msg.get("video_id").and_then(Value::as_str);
    Ok(match field_str(msg, "kind")?.as_str() {
        "job" => json!({ "job": lm.parse_video_job(status, &body_of(msg)?, video_id)?.to_json() }),
        "list" => {
            json!({ "jobs": lm.parse_video_jobs(status, &body_of(msg)?)?.iter().map(|j| j.to_json()).collect::<Vec<_>>() })
        }
        "part" => {
            let status_body = msg
                .get("status_body")
                .and_then(Value::as_object)
                .ok_or_else(|| ValidationError::value("status_body is required"))?;
            let fetched = match msg.get("fetched_b64").and_then(Value::as_str) {
                Some(b64) => Some(lm15::types::base64_decode(b64).map_err(Failure::from)?),
                None => None,
            };
            let headers = headers_of(msg);
            let part = lm.parse_video_part(
                status_body,
                fetched.as_ref().map(|f| (headers.as_slice(), f.as_slice())),
            )?;
            json!({ "part": lm15::types::Part::Video(part).to_json() })
        }
        other => return Err(ValidationError::value(format!("unknown kind {other:?}")).into()),
    })
}

fn chain_context(msg: &Map<String, Value>) -> Result<lm15::cloud::chains::ChainContext, Failure> {
    let now = clock_of(msg)?
        .map(|c| c.0)
        .unwrap_or_else(lm15::auth::time_now);
    let input = msg.get("input").and_then(Value::as_object);
    let env = input
        .and_then(|i| i.get("env"))
        .and_then(Value::as_object)
        .map(|m| {
            m.iter()
                .filter_map(|(k, v)| v.as_str().map(|s| (k.clone(), s.to_string())))
                .collect()
        })
        .unwrap_or_default();
    let mut ctx = lm15::cloud::chains::ChainContext::offline(env, now);
    // The certificate vector: the harness supplies the PEM contents in
    // `input.certificate_pem` and `input.private_key_pem` for the path the
    // env names; both go into one overlay file.
    if let Some(path) = ctx.env.get("AZURE_CLIENT_CERTIFICATE_PATH").cloned() {
        let cert = input
            .and_then(|i| i.get("certificate_pem"))
            .and_then(Value::as_str)
            .unwrap_or("");
        let key = input
            .and_then(|i| i.get("private_key_pem"))
            .and_then(Value::as_str)
            .unwrap_or("");
        if !cert.is_empty() || !key.is_empty() {
            let mut files = std::collections::BTreeMap::new();
            files.insert(path, format!("{cert}\n{key}"));
            ctx = ctx.with_files(files);
        }
    }
    let mut settings = settings_of(msg).unwrap_or_default();
    if let Some(extra) = input
        .and_then(|i| i.get("settings"))
        .and_then(Value::as_object)
    {
        for (k, v) in extra {
            settings.insert(
                k.clone(),
                v.as_str()
                    .map(str::to_string)
                    .unwrap_or_else(|| v.to_string()),
            );
        }
    }
    Ok(ctx.with_settings(settings))
}

fn op_token_exchange_build(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let rung = field_str(msg, "rung")?;
    let ctx = chain_context(msg)?;
    let input = msg
        .get("input")
        .and_then(Value::as_object)
        .cloned()
        .unwrap_or_default();
    Ok(lm15::cloud::chains::token_exchange_build(&rung, &input, &ctx).map_err(Lm15Error::from)?)
}

fn op_token_exchange_parse(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let rung = field_str(msg, "rung")?;
    let ctx = chain_context(msg)?;
    let status = msg.get("status").and_then(Value::as_u64).unwrap_or(0) as u16;
    let body = match msg.get("body") {
        Some(Value::Object(map)) => map.clone(),
        Some(Value::String(text)) => serde_json::from_str::<Value>(text)
            .ok()
            .and_then(|v| v.as_object().cloned())
            .unwrap_or_default(),
        _ => Map::new(),
    };
    let credential = lm15::cloud::chains::token_exchange_parse(&rung, status, &body, &ctx)
        .map_err(Lm15Error::from)?;
    Ok(json!({ "ok": true, "credential": credential.to_json() }))
}

fn op_replay_live(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let lm = surface_adapter(
        msg,
        Credential::api_key("vet-replay-only").map_err(Lm15Error::from)?,
    )?;
    let config = LiveConfig::from_json(field(msg, "live_config")?)?;
    let codec = lm.live_codec(&config)?;
    let setup_frames = codec.setup_frames()?;
    let mut client_frames = Vec::new();
    for event in msg
        .get("client_events")
        .and_then(Value::as_array)
        .into_iter()
        .flatten()
    {
        let event = LiveClientEvent::from_json(event)?;
        client_frames.push(Value::Array(codec.encode(&event)?));
    }
    let mut events = Vec::new();
    for frame in msg
        .get("server_frames_b64")
        .and_then(Value::as_array)
        .into_iter()
        .flatten()
    {
        let bytes =
            lm15::types::base64_decode(frame.as_str().unwrap_or("")).map_err(Failure::from)?;
        let decoded = codec.decode(&bytes)?;
        events.push(Value::Array(decoded.iter().map(|e| e.to_json()).collect()));
    }
    Ok(json!({ "setup_frames": setup_frames, "client_frames": client_frames, "events": events }))
}

fn op_build_models_request(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let lm = surface_adapter(msg, credential_of(msg)?)?;
    Ok(transport_request_json(&lm.models_request()?))
}

/// PROTOCOL.md § parse_models_response: canonical `model_info` serde
/// INCLUDING `origin.provider_data`; a status of 400 or more is the
/// normalized error as an `ok: false` reply.
fn op_parse_models_response(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let lm = surface_adapter(
        msg,
        Credential::api_key("vet-parse-only").map_err(Lm15Error::from)?,
    )?;
    let status = msg.get("status").and_then(Value::as_u64).unwrap_or(200) as u16;
    let models = lm.parse_models(status, &body_of(msg)?)?;
    Ok(json!({ "models": models.iter().map(|m| m.to_json()).collect::<Vec<_>>() }))
}

/// A parse-side adapter: the harness gives no credential for these ops,
/// so a placeholder stands in (never sent; parsing reads no credential).
fn parse_adapter(msg: &Map<String, Value>) -> Result<lm15::ProviderLM, Failure> {
    let provider = field_str(msg, "provider")?;
    let base_url = msg.get("base_url").and_then(Value::as_str);
    let clock = clock_of(msg)?;
    Ok(lm15::registry::adapter_for(
        &provider,
        "vet-parse-only",
        base_url,
        settings_of(msg),
        clock.map(|c| Box::new(c) as Box<dyn lm15::wire::Clock + Send + Sync>),
    )?
    .with_adaptations(adaptation_policy(msg)?))
}

fn body_of(msg: &Map<String, Value>) -> Result<Vec<u8>, Failure> {
    let b64 = field_str(msg, "body_b64")?;
    lm15::types::base64_decode(&b64).map_err(Failure::from)
}

/// PROTOCOL.md: a Response WITHOUT provider_data, plus the `_lm15_unmapped`
/// canary surfaced as the top-level `unmapped` array.
fn response_result(response: &Response) -> Map<String, Value> {
    let mut result = Map::new();
    result.insert("canonical_response".into(), response.to_json());
    if let Some(unmapped) = response
        .provider_data
        .as_ref()
        .and_then(|pd| pd.get("_lm15_unmapped"))
    {
        result.insert("unmapped".into(), unmapped.clone());
    }
    result
}

fn op_parse_response(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let lm = parse_adapter(msg)?;
    let request = Request::from_json(field(msg, "canonical_request")?)?;
    let status = field(msg, "status")?
        .as_u64()
        .and_then(|s| u16::try_from(s).ok())
        .ok_or_else(|| ValidationError::type_error("status must be an int"))?;
    let body = body_of(msg)?;
    let response = lm.parse_response_with_headers(&request, status, &headers_of(msg), &body)?;
    Ok(Value::Object(response_result(&response)))
}

/// PROTOCOL.md § ingest_openai_chat (MAP-12): the case's provider binds
/// the compat; no credential is read.
fn op_ingest_openai_chat(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let lm = parse_adapter(msg)?;
    let request = lm.request_from_openai_chat(field(msg, "body")?)?;
    Ok(json!({"canonical_request": request.to_json()}))
}

fn op_replay_stream(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let lm = parse_adapter(msg)?;
    let request = Request::from_json(field(msg, "canonical_request")?)?;
    let body = body_of(msg)?;
    let mut decoder = lm.stream_decoder(&request);
    decoder.response_headers(headers_of(msg));
    let mut events = decoder.feed(&body)?;
    events.extend(decoder.finish()?);
    let event_json: Vec<Value> = events.iter().map(Canonical::to_json).collect();
    match materialize_response(events.iter(), &request) {
        Ok(response) => {
            let mut result = response_result(&response);
            result.insert("events".into(), Value::Array(event_json));
            Ok(Value::Object(result))
        }
        Err(err) if err.partial().is_some() => {
            // The trace parsed; assembly refused (MAP-9): report both.
            let mut extra = Map::new();
            extra.insert("events".into(), Value::Array(event_json));
            Err(Failure::Lm15Extra(Box::new(err), extra))
        }
        Err(err) => Err(err.into()),
    }
}

fn op_sigv4_sign(msg: &Map<String, Value>) -> Result<Value, Failure> {
    let request = field(msg, "request")?
        .as_object()
        .ok_or_else(|| ValidationError::type_error("request must be an object"))?;
    let method = field_str(request, "method")?;
    let url = field_str(request, "url")?;
    // A list value is the same name repeated on the wire, in that order.
    let mut headers: Vec<(String, String)> = Vec::new();
    if let Some(map) = request.get("headers").and_then(Value::as_object) {
        for (name, value) in map {
            match value {
                Value::Array(values) => {
                    for v in values {
                        headers.push((name.clone(), v.as_str().unwrap_or_default().to_string()));
                    }
                }
                Value::String(s) => headers.push((name.clone(), s.clone())),
                other => headers.push((name.clone(), other.to_string())),
            }
        }
    }
    let body = request
        .get("body")
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_string();
    let credential = Credential::from_json(field(msg, "credential")?)?;
    let keys = AwsKeys::from_credential(&credential)?;
    let region = field_str(msg, "region")?;
    let service = field_str(msg, "service")?;
    let now = clock_of(msg)?.ok_or_else(|| ValidationError::type_error("missing field: now"))?;
    let signing = SigningRequest {
        method: &method,
        url: &url,
        headers: &headers,
        payload: body.as_bytes(),
    };
    let signature = sigv4::sign(&signing, &keys, &region, &service, now.0);
    let headers: Map<String, Value> = signature
        .headers
        .iter()
        .map(|(k, v)| (k.clone(), Value::String(v.clone())))
        .collect();
    Ok(json!({
        "canonical_request": signature.canonical_request,
        "string_to_sign": signature.string_to_sign,
        "authorization": signature.authorization,
        "headers": headers,
    }))
}

fn dispatch(op: &str, msg: &Map<String, Value>) -> Result<Value, Failure> {
    match op {
        "capabilities" => Ok(op_capabilities()),
        "serde_roundtrip" => op_serde_roundtrip(msg),
        "validate" => op_validate(msg),
        "normalize_error" => op_normalize_error(msg),
        "explain_auth" => op_explain_auth(msg),
        "managed_run" => vet_managed::op_managed_run(msg).map_err(|e| Failure::Lm15(Box::new(e))),
        "resolve_model" => op_resolve_model(msg),
        "resolve_openai_chat_model" => op_resolve_openai_chat_model(msg),
        "build_request" => op_build_request(msg),
        "plan" => op_plan(msg),
        "surface_dump" => Ok(lm15::tooling::surface_dump()),
        "parse_response" => op_parse_response(msg),
        "ingest_openai_chat" => op_ingest_openai_chat(msg),
        "build_models_request" => op_build_models_request(msg),
        "file_op_build" => op_file_op_build(msg),
        "cache_op_build" => op_cache_op_build(msg),
        "generation_build" => op_generation_build(msg),
        "video_op_build" => op_video_op_build(msg),
        "replay_live" => op_replay_live(msg),
        "token_exchange_build" => op_token_exchange_build(msg),
        "token_exchange_parse" => op_token_exchange_parse(msg),
        "video_op_parse" => op_video_op_parse(msg),
        "generation_parse" => op_generation_parse(msg),
        "cache_op_parse" => op_cache_op_parse(msg),
        "batch_op_build" => op_batch_op_build(msg),
        "batch_op_parse" => op_batch_op_parse(msg),
        "file_op_parse" => op_file_op_parse(msg),
        "parse_models_response" => op_parse_models_response(msg),
        "replay_stream" => op_replay_stream(msg),
        "sigv4_sign" => op_sigv4_sign(msg),
        other => Err(Lm15Error::unsupported_feature(format!(
            "op {other:?} is not implemented by the Rust shim (ops: {})",
            OPS.join(", ")
        ))
        .into()),
    }
}

fn lm15_error_json(err: &Lm15Error, extra: Map<String, Value>) -> Value {
    let mut error = json!({
        "type": err.class_name(),
        "code": err.code().as_str(),
        "message": err.message(),
    });
    if let Some(feature) = &err.meta().feature {
        error["feature"] = feature.clone().into();
    }
    if let Some(status) = err.meta().status {
        error["status"] = status.into();
    }
    let evidence = err.meta().http_response();
    if !evidence.is_empty() {
        error["http_response"] = evidence.into();
    }
    if let Some(partial) = err.partial() {
        error["partial_response"] =
            Value::Object(response_result(partial))["canonical_response"].clone();
    }
    // The payload spec/vocabularies.md pins for the router codes.
    if let Some(model) = err.model() {
        error["model"] = Value::String(model.to_string());
    }
    if let Some(providers) = err.candidate_providers() {
        error["providers"] = json!(providers);
    }
    for (key, value) in extra {
        error[key] = value;
    }
    error
}

fn error_reply(id: Value, failure: Failure) -> Value {
    let error = match failure {
        Failure::Validation(err) => json!({
            "type": err.type_name(),
            "message": err.message,
        }),
        Failure::Lm15(err) => lm15_error_json(&err, Map::new()),
        Failure::Lm15Extra(err, extra) => lm15_error_json(&err, extra),
    };
    json!({ "id": id, "ok": false, "error": error })
}

fn handle_line(line: &str) -> Value {
    let msg: Value = match serde_json::from_str(line) {
        Ok(v) => v,
        Err(e) => {
            return error_reply(
                Value::Null,
                ValidationError::value(format!("invalid JSON request: {e}")).into(),
            )
        }
    };
    let Value::Object(msg) = msg else {
        return error_reply(
            Value::Null,
            ValidationError::value("request must be a JSON object").into(),
        );
    };
    let id = msg.get("id").cloned().unwrap_or(Value::Null);
    let op = msg.get("op").and_then(Value::as_str).unwrap_or_default();
    match dispatch(op, &msg) {
        Ok(result) => json!({ "id": id, "ok": true, "result": result }),
        Err(failure) => error_reply(id, failure),
    }
}

fn main() -> io::Result<()> {
    let stdin = io::stdin();
    let stdout = io::stdout();
    let mut out = stdout.lock();
    for line in stdin.lock().lines() {
        let line = line?;
        if line.trim().is_empty() {
            continue;
        }
        let reply = handle_line(&line);
        writeln!(out, "{reply}")?;
        out.flush()?;
    }
    Ok(())
}