diff --git a/crates/mesh-llm-host-runtime/src/network/openai/ingress.rs b/crates/mesh-llm-host-runtime/src/network/openai/ingress.rs index dc9e8266b6..64c9c81322 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/ingress.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/ingress.rs @@ -9,7 +9,8 @@ use crate::network::openai::client_stream::ClientStream; use crate::network::openai::transport as proxy; use crate::network::router; use crate::plugin::openai_exchange::{ - OpenAiExchangeChannel, OpenAiExchangeDispatchPath, OpenAiExchangeEnvelope, + ExchangeUsage, OpenAiExchangeChannel, OpenAiExchangeDispatchPath, OpenAiExchangeEnvelope, + ServingProvenance, request_body_digest, }; use mesh_llm_events::audit::{audit_events, emit_audit}; use mesh_llm_events::{OutputEvent, emit_event}; @@ -27,6 +28,194 @@ fn plugin_route_status(outcome: &proxy::RouteDispatchOutcome) -> Option { } } +/// Gather the maximum inference provenance the host *actually knows* for the +/// exchange just served — what ran, at what fidelity, on whose hardware — from +/// state the local node already holds: the served-model descriptor (quant, +/// architecture, context length, identity hash, revision) and this host's +/// startup hardware survey (gpu, vram, soc, hostname). Every field is a real +/// value or omitted; nothing is invented. Returned as a plain data struct so +/// the wire event stays independent of the node internals. +async fn serving_provenance_for_model(node: &mesh::Node, model_name: &str) -> ServingProvenance { + // The served-model descriptor for exactly this model, if the node has one. + // We match on the served identity's `model_name`; a miss (peer-served or + // not-yet-described) leaves every model field `None` rather than guessing. + let descriptor = node + .served_model_descriptors() + .await + .into_iter() + .find(|d| d.identity.model_name == model_name); + + let (identity, metadata) = match descriptor { + Some(d) => (Some(d.identity), d.metadata), + None => (None, None), + }; + + ServingProvenance { + served_by_node_id: node.id().to_string(), + hostname: node.hostname.clone(), + quantization: metadata.as_ref().and_then(|m| m.quant.clone()), + architecture: metadata.as_ref().and_then(|m| m.architecture.clone()), + context_length: metadata.as_ref().and_then(|m| m.native_context_length), + parameter_size: metadata.as_ref().and_then(|m| m.parameter_size.clone()), + layer_count: metadata.as_ref().and_then(|m| m.layer_count), + model_identity_hash: identity.as_ref().and_then(|i| i.identity_hash.clone()), + model_canonical_ref: identity.as_ref().and_then(|i| i.canonical_ref.clone()), + model_revision: identity.as_ref().and_then(|i| i.revision.clone()), + gpu: node.gpu_name.clone(), + // `advertised_memory.total_bytes` is 0 when no accelerator memory was + // enumerated; surface it only when it is a real, non-zero figure. + vram_bytes: (node.advertised_memory.total_bytes != 0) + .then_some(node.advertised_memory.total_bytes), + is_soc: node.is_soc, + } +} + +/// Extract the served backend's real token usage from a dispatch outcome, when +/// it carried one. Only `RespondedWithUsage` — the outcome the host-served +/// `route_model_request` returns after reading the backend's `usage` object — +/// yields counts; every other outcome (plugin stub, status-only, error, drop) +/// yields `None`, so the terminal envelope omits `usage` rather than reporting +/// fabricated zeros. +fn exchange_usage_from_outcome(outcome: &proxy::RouteDispatchOutcome) -> Option { + let proxy::RouteDispatchOutcome::RespondedWithUsage { usage, .. } = outcome else { + return None; + }; + // All three or none: a backend that reports two counts but not the + // third has told us something is missing, and this mirrors the + // invariant `mesh_llm_events::logging::events::TokenUsage::from_counts` + // already documents elsewhere -- "missing, overflowing, or internally + // inconsistent usage must not be estimated." Never a derived total (a + // backend's real total can legitimately disagree with + // prompt+completion, e.g. reasoning tokens folded into `total`), and + // never a zero standing in for an absent count. `cached_prompt_tokens` + // rides along when the backend reported it -- for billing + // reconciliation, dropping it is usually the difference between a right + // and a wrong number. + let (Some(prompt_tokens), Some(completion_tokens), Some(total_tokens)) = ( + usage.prompt_tokens, + usage.completion_tokens, + usage.total_tokens, + ) else { + return None; + }; + Some(ExchangeUsage { + prompt_tokens, + cached_prompt_tokens: usage.cached_prompt_tokens, + completion_tokens, + total_tokens, + }) +} + +/// Whether a dispatch outcome actually means inference ran and a response +/// body was returned to the client — the only case `ServingProvenance`'s +/// contract ("what ran, at what fidelity, on whose hardware") can honestly +/// describe. A 503/`Failed`/`Dropped` outcome served nothing, so attaching +/// provenance there would be exactly the fabrication the envelope promises +/// never to do. +fn outcome_was_served(outcome: &proxy::RouteDispatchOutcome) -> bool { + matches!( + outcome, + proxy::RouteDispatchOutcome::Responded(200..=299) + | proxy::RouteDispatchOutcome::RespondedWithUsage { + status_code: 200..=299, + .. + } + ) +} + +/// Publish the raw-proxy path's terminal event for a served exchange, enriched +/// with the serving provenance the host resolved from the node (what ran / at +/// what fidelity / on whose hardware), the real token usage the dispatch +/// outcome carried, and the canonical digest of the real request body. Only +/// real values ride along; anything the node doesn't know for this model, or +/// the dispatch didn't carry, stays omitted (never fabricated). No +/// `X-Capsule-Id` marker exists on this path (it never runs through +/// `openai-frontend`'s `OpenAiHookPolicy`, the only place a marker is minted), +/// so nonce/nonce_source are `None`. +/// +/// `served_locally` distinguishes the host-served path (this node's own +/// weights, via `route_model_request`) from the plugin-served path (a plugin +/// endpoint that may proxy anywhere, including a third-party service). The +/// whole `serving_provenance` block — not just the hardware fields — is +/// attached only when `served_locally` is true: a plugin endpoint can return +/// 200 for an exchange this node's GPU/VRAM/hostname never touched, and the +/// model-name-keyed descriptor lookup itself is a second, independent source +/// of staleness (the descriptor list and the routing target table update on +/// separate schedules, so a teardown window can hand a plugin-served exchange +/// this node's own served-model quant/architecture/identity_hash). Omitting +/// the whole block on the plugin path closes both windows at once rather than +/// leaving the model-identity half open. +async fn publish_raw_proxy_terminal( + node: &mesh::Node, + channel: &dyn OpenAiExchangeChannel, + exchange_id: &str, + model_name: &str, + final_outcome: &proxy::RouteDispatchOutcome, + served_locally: bool, + request_digest: Option<&str>, +) { + let mut envelope = OpenAiExchangeEnvelope::terminal( + exchange_id.to_string(), + OpenAiExchangeDispatchPath::RawProxy, + model_name, + plugin_route_status(final_outcome), + None, + None, + ); + // Nothing was served on a 503 / `Failed` / `Dropped` outcome, so there is + // no hardware or model identity to report — and no reason to pay the + // `served_model_descriptors()` lock for a lookup whose result would be + // thrown away. On the plugin-served path, skip it regardless of outcome: + // see the `served_locally` doc above. + if served_locally && outcome_was_served(final_outcome) { + let provenance = serving_provenance_for_model(node, model_name).await; + envelope = envelope.with_serving_provenance(provenance); + } + // The host-served (real-weights) branch reaches this via + // `route_model_request`, whose outcome carries the served backend's own + // `usage` object. Attach it so a downstream plugin can seal the REAL token + // counts of the exchange, not a zeroed stub. A plugin-served exchange has + // no such usage on the outcome, so this stays absent for it — never zeroed. + if let Some(usage) = exchange_usage_from_outcome(final_outcome) { + envelope = envelope.with_usage(usage); + } + // The canonical digest of the REAL request body the host dispatched, so a + // downstream capsule can bind its `agent_input_digest` to the real bytes. + // Computed by the caller (which holds the parsed request); absent only when + // the host held no JSON body to digest — never fabricated. + if let Some(digest) = request_digest { + envelope = envelope.with_request_digest(digest.to_string()); + } + channel.publish(&envelope).await; +} + +/// Path 2's own "effective request" moment: the plugin/endpoint is resolved +/// and dispatch is about to happen. There is no typed `ChatCompletionRequest` +/// on this path (see the #1331 design note), so the envelope carries only +/// the model — the same narrow route fact path 1's `ChatExchangeRoute` +/// carries. Mints the exchange id here, at admission, so it can pair this +/// effective event with its terminal event even when concurrent raw-proxy +/// requests share the same model. Returns `None` — no id minted, no publish +/// — when nobody declares `openai.exchange.v1`: nothing downstream would +/// ever see it. +async fn mint_and_publish_effective_raw_proxy( + plugin_manager: &crate::plugin::PluginManager, + model_name: &str, +) -> Option { + if !plugin_manager.has_subscriber().await { + return None; + } + let exchange_id = uuid::Uuid::new_v4().to_string(); + plugin_manager + .publish(&OpenAiExchangeEnvelope::effective( + exchange_id.clone(), + OpenAiExchangeDispatchPath::RawProxy, + model_name, + )) + .await; + Some(exchange_id) +} + enum AutoRouteResolution { Continue { effective_model: Option, @@ -635,22 +824,8 @@ async fn try_route_plugin_model( .await { Ok(Some(endpoint)) => { - // Path 2's own "effective request" moment: the plugin/endpoint - // is resolved and dispatch is about to happen. There is no typed - // `ChatCompletionRequest` on this path (see the #1331 design - // note), so the envelope carries only the model — the same - // narrow route fact path 1's `ChatExchangeRoute` carries. Mint - // the exchange id here, at admission, so it can pair this - // effective event with its terminal event below even when - // concurrent raw-proxy requests share the same model. - let exchange_id = uuid::Uuid::new_v4().to_string(); - plugin_manager - .publish(&OpenAiExchangeEnvelope::effective( - exchange_id.clone(), - OpenAiExchangeDispatchPath::RawProxy, - model_name, - )) - .await; + let exchange_id = + mint_and_publish_effective_raw_proxy(plugin_manager, model_name).await; let outcome = proxy::route_http_endpoint_request( ctx.node, Some(model_name), @@ -679,20 +854,28 @@ async fn try_route_plugin_model( } else { outcome }; - plugin_manager - .publish(&OpenAiExchangeEnvelope::terminal( - exchange_id, - OpenAiExchangeDispatchPath::RawProxy, - model_name, - plugin_route_status(&final_outcome), - // No X-Capsule-Id marker on this path: it never runs - // through `openai-frontend`'s `OpenAiHookPolicy`, the - // only place a marker is minted (see the design note). - None, - // No marker means no nonce, so no nonce_source either. - None, - )) - .await; + let Some(exchange_id) = exchange_id else { + return final_outcome; + }; + // Bind the real request body digest when the parsed body is already + // available (this path holds `request` by shared ref, so it does not + // force parsing); `None` otherwise, never fabricated. The + // plugin-served completion itself is a stub (zero usage), but the + // request digest is still the real request that was asked. + let request_digest = request + .body_json + .as_ref() + .and_then(|body| request_body_digest(body, request.body_bytes.as_deref())); + publish_raw_proxy_terminal( + ctx.node, + plugin_manager, + &exchange_id, + model_name, + &final_outcome, + false, // plugin-served: never this node's own hardware/weights + request_digest.as_deref(), + ) + .await; final_outcome } Ok(None) => { @@ -764,7 +947,55 @@ async fn route_request( } // Local candidates available — route normally. - proxy::route_model_request( + // + // Host-served (real-weights) exchange. This branch, unlike the + // plugin-served `try_route_plugin_model` path, previously published NO + // `openai.exchange.v1` terminal event — so a downstream capsule-emit + // plugin never saw the exchange that carried the host's REAL served-model + // descriptor (architecture / context / layers / params / identity) AND + // the backend's REAL token usage. Publish the same effective→terminal + // pair the plugin path does, resolving provenance by the actually-served + // model and attaching the real usage the dispatch outcome carries and the + // canonical digest of the real request body, so one sealed capsule can + // hold real model identity + real usage + real hardware + what was asked + // together. Tokenize requests are not chat exchanges, so they are not + // announced. `plugin_manager` is `None` when no plugin is loaded, and + // even with one loaded nothing may declare `openai.exchange.v1` — in + // either case there is no subscriber, so skip minting an exchange id + // and, below, the body digest and served-model provenance lookup + // that only exist to build an event nobody would receive. + let has_subscriber = match ctx.plugin_manager { + Some(plugin_manager) => plugin_manager.has_subscriber().await, + None => false, + }; + let announce = (!request.is_tokenize_request() && has_subscriber) + .then_some(ctx.plugin_manager) + .flatten() + .map(|plugin_manager| (plugin_manager, uuid::Uuid::new_v4().to_string())); + // Digest the REAL request body up front, while the parsed body is still + // in hand and before `route_model_request` streams it to the backend — + // this is the one binding a downstream capsule needs to tie its + // `agent_input_digest` to what was actually asked. `ensure_body_json` + // is idempotent; `None` when the request carried no JSON body (e.g. a + // non-chat proxy passthrough), in which case no digest is forwarded + // rather than a fabricated one. + let request_digest = announce.as_ref().and_then(|_| { + request.ensure_body_json(); + request + .body_json + .as_ref() + .and_then(|body| request_body_digest(body, request.body_bytes.as_deref())) + }); + if let Some((plugin_manager, exchange_id)) = announce.as_ref() { + plugin_manager + .publish(&OpenAiExchangeEnvelope::effective( + exchange_id.clone(), + OpenAiExchangeDispatchPath::RawProxy, + model_name, + )) + .await; + } + let outcome = proxy::route_model_request( ctx.node.clone(), tcp_stream, ctx.targets, @@ -776,7 +1007,20 @@ async fn route_request( route_observer, }, ) - .await + .await; + if let Some((plugin_manager, exchange_id)) = announce.as_ref() { + publish_raw_proxy_terminal( + ctx.node, + *plugin_manager, + exchange_id, + model_name, + &outcome, + true, // host-served: this node's own weights and hardware survey + request_digest.as_deref(), + ) + .await; + } + outcome } else { // No model specified — generic fallback routing to first available target. diff --git a/crates/mesh-llm-host-runtime/src/network/openai/ingress_tests/durable_artifacts.rs b/crates/mesh-llm-host-runtime/src/network/openai/ingress_tests/durable_artifacts.rs index 27c734689c..bcbd5b8e2f 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/ingress_tests/durable_artifacts.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/ingress_tests/durable_artifacts.rs @@ -1,6 +1,11 @@ +use mesh_llm_events::logging::events::TokenUsage; use mesh_llm_events::logging::identifiers::RequestId; -use super::{affinity, election, handle_api_proxy_connection, mesh}; +use crate::plugin::openai_exchange::{OpenAiExchangePhase, test_support::RecordingChannel}; + +use super::{ + affinity, election, handle_api_proxy_connection, mesh, proxy, publish_raw_proxy_terminal, +}; #[tokio::test] #[serial_test::serial] @@ -206,3 +211,308 @@ async fn ingress_body_parse_error_persists_a_response_only_after_complete_header .expect("response artifact content"); assert_eq!(content.bytes, wire[body_start..]); } + +/// A node with a real GPU/hostname/VRAM survey and a served-model descriptor +/// for `model_name`, so `publish_raw_proxy_terminal`'s served-2xx branch has +/// real hardware and model identity to attach. +async fn node_with_hardware_and_descriptor(model_name: &str) -> mesh::Node { + let mut node = mesh::Node::new_for_tests(crate::mesh::NodeRole::Worker) + .await + .expect("test node"); + node.gpu_name = Some("Test GPU".to_string()); + node.hostname = Some("test-host".to_string()); + node.is_soc = Some(false); + node.advertised_memory = mesh::AdvertisedMemory { + total_bytes: 16_000_000_000, + ..Default::default() + }; + node.set_served_model_descriptors(vec![mesh::ServedModelDescriptor { + identity: mesh::ServedModelIdentity { + model_name: model_name.to_string(), + identity_hash: Some("hash-abc123".to_string()), + canonical_ref: Some("org/repo@rev1".to_string()), + revision: Some("rev1".to_string()), + ..Default::default() + }, + metadata: Some(mesh::ServedModelMetadata { + quant: Some("Q4_K_M".to_string()), + architecture: Some("llama".to_string()), + native_context_length: Some(4096), + parameter_size: Some("7B".to_string()), + layer_count: Some(32), + ..Default::default() + }), + ..Default::default() + }]) + .await; + node +} + +/// A served 2xx outcome on the host-served path attaches the full serving +/// provenance block (hardware + model identity, both real), the real usage +/// the backend reported, and the real request digest — everything a +/// downstream capsule needs, and nothing fabricated. +#[tokio::test] +async fn publish_raw_proxy_terminal_attaches_full_provenance_and_usage_on_a_served_2xx_outcome() { + let node = node_with_hardware_and_descriptor("test-model").await; + let channel = RecordingChannel::default(); + let outcome = proxy::RouteDispatchOutcome::RespondedWithUsage { + status_code: 200, + usage: TokenUsage { + prompt_tokens: Some(10), + cached_prompt_tokens: Some(2), + completion_tokens: Some(5), + total_tokens: Some(15), + }, + }; + + publish_raw_proxy_terminal( + &node, + &channel, + "exchange-1", + "test-model", + &outcome, + true, + Some("digest-abc"), + ) + .await; + + let events = channel.events(); + assert_eq!(events.len(), 1); + let event = &events[0]; + assert_eq!(event.phase, OpenAiExchangePhase::Terminal); + assert_eq!(event.status, Some(200)); + assert_eq!(event.request_digest.as_deref(), Some("digest-abc")); + + let provenance = event + .serving_provenance + .as_ref() + .expect("served 2xx outcome carries provenance"); + assert_eq!(provenance.served_by_node_id, node.id().to_string()); + assert_eq!(provenance.hostname.as_deref(), Some("test-host")); + assert_eq!(provenance.gpu.as_deref(), Some("Test GPU")); + assert_eq!(provenance.vram_bytes, Some(16_000_000_000)); + assert_eq!(provenance.is_soc, Some(false)); + assert_eq!(provenance.quantization.as_deref(), Some("Q4_K_M")); + assert_eq!(provenance.architecture.as_deref(), Some("llama")); + assert_eq!(provenance.context_length, Some(4096)); + assert_eq!(provenance.parameter_size.as_deref(), Some("7B")); + assert_eq!(provenance.layer_count, Some(32)); + assert_eq!( + provenance.model_identity_hash.as_deref(), + Some("hash-abc123") + ); + assert_eq!( + provenance.model_canonical_ref.as_deref(), + Some("org/repo@rev1") + ); + assert_eq!(provenance.model_revision.as_deref(), Some("rev1")); + + let usage = event.usage.expect("served 2xx outcome carries real usage"); + assert_eq!(usage.prompt_tokens, 10); + assert_eq!(usage.cached_prompt_tokens, Some(2)); + assert_eq!(usage.completion_tokens, 5); + assert_eq!(usage.total_tokens, 15); +} + +/// A degraded 503 served nothing, so there is no hardware or model identity +/// to report — but the client did get a real status, and the terminal event +/// must carry it rather than leaving the outcome unaccounted for. +#[tokio::test] +async fn publish_raw_proxy_terminal_on_a_503_has_no_provenance_but_keeps_the_status() { + let node = node_with_hardware_and_descriptor("test-model").await; + let channel = RecordingChannel::default(); + let outcome = proxy::RouteDispatchOutcome::Responded(503); + + publish_raw_proxy_terminal( + &node, + &channel, + "exchange-1", + "test-model", + &outcome, + true, + None, + ) + .await; + + let events = channel.events(); + assert_eq!(events.len(), 1); + assert_eq!(events[0].status, Some(503)); + assert!(events[0].serving_provenance.is_none()); + assert!(events[0].usage.is_none()); +} + +/// A `Failed` outcome never produced an HTTP response at all, so unlike the +/// 503 case there is no status to report either — never a fabricated one. +#[tokio::test] +async fn publish_raw_proxy_terminal_on_a_failed_outcome_has_no_provenance_and_no_status() { + let node = node_with_hardware_and_descriptor("test-model").await; + let channel = RecordingChannel::default(); + let outcome = proxy::RouteDispatchOutcome::Failed("connect_timeout"); + + publish_raw_proxy_terminal( + &node, + &channel, + "exchange-1", + "test-model", + &outcome, + true, + None, + ) + .await; + + let events = channel.events(); + assert_eq!(events.len(), 1); + assert!(events[0].status.is_none()); + assert!(events[0].serving_provenance.is_none()); +} + +/// A `Dropped` outcome (the client disconnected mid-dispatch) is the same +/// shape as `Failed` for this envelope: no status, no provenance. +#[tokio::test] +async fn publish_raw_proxy_terminal_on_a_dropped_outcome_has_no_provenance_and_no_status() { + let node = node_with_hardware_and_descriptor("test-model").await; + let channel = RecordingChannel::default(); + let outcome = proxy::RouteDispatchOutcome::Dropped("client_disconnected"); + + publish_raw_proxy_terminal( + &node, + &channel, + "exchange-1", + "test-model", + &outcome, + true, + None, + ) + .await; + + let events = channel.events(); + assert_eq!(events.len(), 1); + assert!(events[0].status.is_none()); + assert!(events[0].serving_provenance.is_none()); +} + +/// The plugin-served path omits the WHOLE serving-provenance block on a 2xx +/// outcome, even when the node happens to have real hardware and a +/// served-model descriptor for the same model name — a plugin endpoint can +/// proxy anywhere, so none of this node's own hardware/identity is honest to +/// report for it. +#[tokio::test] +async fn publish_raw_proxy_terminal_on_the_plugin_served_path_omits_the_whole_block_even_on_2xx() { + let node = node_with_hardware_and_descriptor("test-model").await; + let channel = RecordingChannel::default(); + let outcome = proxy::RouteDispatchOutcome::Responded(200); + + publish_raw_proxy_terminal( + &node, + &channel, + "exchange-1", + "test-model", + &outcome, + false, // plugin-served + None, + ) + .await; + + let events = channel.events(); + assert_eq!(events.len(), 1); + assert_eq!(events[0].status, Some(200)); + assert!(events[0].serving_provenance.is_none()); +} + +/// A served 2xx outcome for a model this node has no served-model descriptor +/// for (a routing-table/descriptor-list staleness window) still reports the +/// real hardware survey, but every model-identity field stays `None` rather +/// than borrowing another model's descriptor. The node DOES have a +/// descriptor registered — just for a different model — so this actually +/// exercises the name match rather than an incidentally-empty list. +#[tokio::test] +async fn publish_raw_proxy_terminal_omits_model_identity_on_a_descriptor_miss() { + let mut node = mesh::Node::new_for_tests(crate::mesh::NodeRole::Worker) + .await + .expect("test node"); + node.gpu_name = Some("Test GPU".to_string()); + node.hostname = Some("test-host".to_string()); + node.advertised_memory = mesh::AdvertisedMemory { + total_bytes: 16_000_000_000, + ..Default::default() + }; + // A descriptor IS registered, but for a different model than the one + // being served — must not be borrowed for "test-model". + node.set_served_model_descriptors(vec![mesh::ServedModelDescriptor { + identity: mesh::ServedModelIdentity { + model_name: "other-model".to_string(), + identity_hash: Some("other-hash".to_string()), + ..Default::default() + }, + metadata: Some(mesh::ServedModelMetadata { + quant: Some("Q8_0".to_string()), + ..Default::default() + }), + ..Default::default() + }]) + .await; + let channel = RecordingChannel::default(); + let outcome = proxy::RouteDispatchOutcome::Responded(200); + + publish_raw_proxy_terminal( + &node, + &channel, + "exchange-1", + "test-model", + &outcome, + true, + None, + ) + .await; + + let events = channel.events(); + let provenance = events[0] + .serving_provenance + .as_ref() + .expect("hardware is still real and reportable on a descriptor miss"); + assert_eq!(provenance.gpu.as_deref(), Some("Test GPU")); + assert_eq!(provenance.vram_bytes, Some(16_000_000_000)); + assert!(provenance.quantization.is_none()); + assert!(provenance.architecture.is_none()); + assert!(provenance.context_length.is_none()); + assert!(provenance.parameter_size.is_none()); + assert!(provenance.layer_count.is_none()); + assert!(provenance.model_identity_hash.is_none()); + assert!(provenance.model_canonical_ref.is_none()); + assert!(provenance.model_revision.is_none()); +} + +/// `advertised_memory.total_bytes == 0` means nothing was actually enumerated +/// (a bare CPU host with no accelerator), so `vram_bytes` is omitted rather +/// than reporting a fabricated zero. +#[tokio::test] +async fn publish_raw_proxy_terminal_omits_vram_bytes_when_advertised_total_is_zero() { + // node_with_hardware_and_descriptor sets a nonzero total; override it. + let mut node = node_with_hardware_and_descriptor("test-model").await; + node.advertised_memory = mesh::AdvertisedMemory::default(); + let channel = RecordingChannel::default(); + let outcome = proxy::RouteDispatchOutcome::Responded(200); + + publish_raw_proxy_terminal( + &node, + &channel, + "exchange-1", + "test-model", + &outcome, + true, + None, + ) + .await; + + let events = channel.events(); + let provenance = events[0] + .serving_provenance + .as_ref() + .expect("served 2xx outcome still carries provenance"); + assert!(provenance.vram_bytes.is_none()); + // The rest of the hardware/model survey is unaffected by the VRAM figure + // being unavailable. + assert_eq!(provenance.gpu.as_deref(), Some("Test GPU")); + assert_eq!(provenance.quantization.as_deref(), Some("Q4_K_M")); +} diff --git a/crates/mesh-llm-host-runtime/src/network/openai/ingress_tests/tests.rs b/crates/mesh-llm-host-runtime/src/network/openai/ingress_tests/tests.rs index 688e00c587..1ca5a42182 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/ingress_tests/tests.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/ingress_tests/tests.rs @@ -909,3 +909,104 @@ fn disconnect_is_dropped_and_cannot_audit_model_access_as_success() { proxy::RouteDispatchOutcome::Responded(200) )); } + +/// The host-served path's usage extraction turns a `RespondedWithUsage` outcome +/// into the real `ExchangeUsage` the terminal envelope carries, and yields +/// `None` for every non-usage-bearing outcome so no all-zero record is ever +/// fabricated. +#[test] +fn exchange_usage_from_outcome_extracts_real_counts_and_omits_otherwise() { + use mesh_llm_events::logging::events::TokenUsage; + + let with_usage = proxy::RouteDispatchOutcome::RespondedWithUsage { + status_code: 200, + usage: TokenUsage { + prompt_tokens: Some(42), + cached_prompt_tokens: None, + completion_tokens: Some(6), + total_tokens: Some(48), + }, + }; + let usage = exchange_usage_from_outcome(&with_usage).expect("real usage present"); + assert_eq!(usage.prompt_tokens, 42); + assert_eq!(usage.cached_prompt_tokens, None); + assert_eq!(usage.completion_tokens, 6); + assert_eq!(usage.total_tokens, 48); + + // A backend total that disagrees with prompt+completion (e.g. reasoning + // tokens folded into `total`) must ride through as reported — never + // silently replaced by a derived sum. + let disagreeing_total = proxy::RouteDispatchOutcome::RespondedWithUsage { + status_code: 200, + usage: TokenUsage { + prompt_tokens: Some(10), + cached_prompt_tokens: None, + completion_tokens: Some(5), + total_tokens: Some(23), + }, + }; + assert_eq!( + exchange_usage_from_outcome(&disagreeing_total) + .expect("real total present") + .total_tokens, + 23 + ); + + // The backend omitted `total_tokens` — never derive it (prompt+completion + // is not necessarily the real total), so this yields None entirely. + let missing_total = proxy::RouteDispatchOutcome::RespondedWithUsage { + status_code: 200, + usage: TokenUsage { + prompt_tokens: Some(10), + cached_prompt_tokens: None, + completion_tokens: Some(5), + total_tokens: None, + }, + }; + assert!(exchange_usage_from_outcome(&missing_total).is_none()); + + // A backend reporting prompt but not completion (or vice versa) must + // never surface a fabricated zero for the missing count. + let missing_completion = proxy::RouteDispatchOutcome::RespondedWithUsage { + status_code: 200, + usage: TokenUsage { + prompt_tokens: Some(42), + cached_prompt_tokens: None, + completion_tokens: None, + total_tokens: Some(42), + }, + }; + assert!(exchange_usage_from_outcome(&missing_completion).is_none()); + + // The real cached-token count rides through when the backend reports it. + let with_cache = proxy::RouteDispatchOutcome::RespondedWithUsage { + status_code: 200, + usage: TokenUsage { + prompt_tokens: Some(42), + cached_prompt_tokens: Some(30), + completion_tokens: Some(6), + total_tokens: Some(48), + }, + }; + assert_eq!( + exchange_usage_from_outcome(&with_cache) + .expect("real usage present") + .cached_prompt_tokens, + Some(30) + ); + + // A status-only response, an error, and a wholly-empty usage object all + // yield None — never a fabricated all-zero record. + assert!(exchange_usage_from_outcome(&proxy::RouteDispatchOutcome::Responded(200)).is_none()); + assert!(exchange_usage_from_outcome(&proxy::RouteDispatchOutcome::Failed("x")).is_none()); + let empty = proxy::RouteDispatchOutcome::RespondedWithUsage { + status_code: 200, + usage: TokenUsage { + prompt_tokens: None, + cached_prompt_tokens: None, + completion_tokens: None, + total_tokens: None, + }, + }; + assert!(exchange_usage_from_outcome(&empty).is_none()); +} diff --git a/crates/mesh-llm-host-runtime/src/plugin/channel_broadcast.rs b/crates/mesh-llm-host-runtime/src/plugin/channel_broadcast.rs index f1a4f6c3b9..a4459ea4d2 100644 --- a/crates/mesh-llm-host-runtime/src/plugin/channel_broadcast.rs +++ b/crates/mesh-llm-host-runtime/src/plugin/channel_broadcast.rs @@ -48,6 +48,21 @@ impl PluginManager { .await .is_some_and(|manifest| manifest_declares_mesh_channel(&manifest, channel)) } + + /// Whether any currently loaded plugin declares `channel` — a cached + /// manifest lookup per plugin, cheap next to the work a caller typically + /// gates behind it (canonicalizing and hashing a request body, cloning a + /// served-model descriptor). Lets a publisher skip that work entirely + /// when nobody on [`Self::broadcast_channel_message`]'s per-plugin filter + /// would have kept the message anyway. + pub async fn any_plugin_declares_mesh_channel(&self, channel: &str) -> bool { + for plugin_id in self.inner.plugins.keys() { + if self.plugin_declares_mesh_channel(plugin_id, channel).await { + return true; + } + } + false + } } /// Turn per-plugin delivery outcomes from [`PluginManager::broadcast_channel_message`] @@ -166,4 +181,24 @@ mod tests { manager.shutdown().await; } + + #[tokio::test] + async fn any_plugin_declares_mesh_channel_is_false_with_no_plugins_loaded() { + let specs = ResolvedPlugins { + externals: Vec::new(), + inactive: Vec::new(), + }; + let (mesh_tx, _mesh_rx) = mpsc::channel(1); + let manager = PluginManager::start(&specs, private_host_mode(), mesh_tx) + .await + .expect("empty plugin set starts cleanly"); + + assert!( + !manager + .any_plugin_declares_mesh_channel("openai.exchange.v1") + .await + ); + + manager.shutdown().await; + } } diff --git a/crates/mesh-llm-host-runtime/src/plugin/openai_exchange.rs b/crates/mesh-llm-host-runtime/src/plugin/openai_exchange.rs index 73f7ffa7c9..94d0433572 100644 --- a/crates/mesh-llm-host-runtime/src/plugin/openai_exchange.rs +++ b/crates/mesh-llm-host-runtime/src/plugin/openai_exchange.rs @@ -56,6 +56,102 @@ pub enum ClientNonceSource { SidecarGeneratedFallback, } +/// What the host actually knows, at serve time, about *what ran, at what +/// fidelity, on whose hardware* for one exchange — the proof-of-inference +/// provenance a downstream capsule attests over (advances #1233's digest +/// advertisement). Every field is either a real value the host holds for the +/// served model/node, or omitted (serialized only `if Some`) when the host +/// genuinely does not know it for this exchange — never a fabricated string. +/// +/// Sourced entirely from state the local [`mesh::Node`](crate::mesh::Node) +/// already holds for the served model and this host's hardware survey (see +/// the raw-proxy dispatch callsite in `network/openai/ingress.rs`): model +/// metadata comes from the served-model descriptor (`ServedModelMetadata`: +/// `quant`, `architecture`, `native_context_length`, `identity_hash`, +/// revision/repository), and hardware comes from the node's startup hardware +/// survey (`gpu_name`, `hostname`, `is_soc`, `advertised_memory`). No raw +/// prompt or response text is carried — provenance only. +#[derive(Debug, Clone, PartialEq, Serialize)] +pub struct ServingProvenance { + /// The node that actually served the inference — this host's own mesh + /// endpoint id. On a plugin-served (raw-proxy) exchange this is the node + /// whose plugin endpoint produced the response. + pub served_by_node_id: String, + /// Serving host name, when the hardware survey resolved one. + #[serde(skip_serializing_if = "Option::is_none")] + pub hostname: Option, + /// Model quantization format as the served-model descriptor reports it + /// (e.g. `"Q4_K_M"`), from `ServedModelMetadata.quant`. Omitted when the + /// descriptor carries no quant (unquantized weights, or metadata absent). + #[serde(skip_serializing_if = "Option::is_none")] + pub quantization: Option, + /// Model architecture / family (e.g. `"llama"`), from + /// `ServedModelMetadata.architecture`. + #[serde(skip_serializing_if = "Option::is_none")] + pub architecture: Option, + /// Native context length (n_ctx) the served weights advertise, from + /// `ServedModelMetadata.native_context_length`. + #[serde(skip_serializing_if = "Option::is_none")] + pub context_length: Option, + /// Human-readable parameter size (e.g. `"7B"`), from + /// `ServedModelMetadata.parameter_size`. + #[serde(skip_serializing_if = "Option::is_none")] + pub parameter_size: Option, + /// Transformer layer count, from `ServedModelMetadata.layer_count`. + #[serde(skip_serializing_if = "Option::is_none")] + pub layer_count: Option, + /// Content-addressed identity hash of the served model artifact, from + /// `ServedModelIdentity.identity_hash` — a digest of the actual model + /// identity (not a hash of the model *name* string). Omitted when the + /// descriptor did not resolve one. + #[serde(skip_serializing_if = "Option::is_none")] + pub model_identity_hash: Option, + /// Canonical model reference (e.g. `repo@rev/file`), from + /// `ServedModelIdentity.canonical_ref`. + #[serde(skip_serializing_if = "Option::is_none")] + pub model_canonical_ref: Option, + /// Source revision (git commit / tag) of the served model, from + /// `ServedModelIdentity.revision`. + #[serde(skip_serializing_if = "Option::is_none")] + pub model_revision: Option, + /// GPU display name from this host's startup hardware survey + /// (`Node.gpu_name`). Omitted on CPU-only hosts or where no accelerator + /// was enumerated. + #[serde(skip_serializing_if = "Option::is_none")] + pub gpu: Option, + /// Enumerated accelerator VRAM capacity in bytes this host advertised + /// (`Node.advertised_memory.total_bytes`) — the sum of device VRAM, or + /// the unified working set on SoCs. Omitted when nothing was enumerated + /// (a bare CPU host advertising only via an explicit cap has no real + /// enumerated figure to report). + #[serde(skip_serializing_if = "Option::is_none")] + pub vram_bytes: Option, + /// Whether the serving host is a unified-memory SoC (Apple Silicon and + /// similar), from the hardware survey (`Node.is_soc`) — the honest + /// device signal this host has (it does not carry a separate cpu/cuda/ + /// metal enum on the served-model path). + #[serde(skip_serializing_if = "Option::is_none")] + pub is_soc: Option, +} + +/// The real token accounting the host observed for a served exchange, from +/// the dispatch outcome's +/// [`RespondedWithUsage`](crate::network::openai::transport::RouteDispatchOutcome::RespondedWithUsage) +/// (the served backend's own OpenAI-shaped `usage` object). Present on a +/// terminal envelope only when the served response actually carried usage; +/// omitted (never zeroed) when the dispatch produced no usage — so a +/// downstream plugin can seal the REAL token counts of a host-served +/// real-weights exchange rather than a stub's zeros. Every field is a real +/// count the host read off the wire; nothing is fabricated. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +pub struct ExchangeUsage { + pub prompt_tokens: u64, + #[serde(skip_serializing_if = "Option::is_none")] + pub cached_prompt_tokens: Option, + pub completion_tokens: u64, + pub total_tokens: u64, +} + /// The wire shape both dispatch paths publish on [`OPENAI_EXCHANGE_CHANNEL`]. /// Deliberately independent of `openai_frontend`'s typed request/response — /// the raw-proxy path never has one — so one shape covers both paths without @@ -86,6 +182,33 @@ pub struct OpenAiExchangeEnvelope { /// exactly when `nonce` is `None` (no marker minted). #[serde(skip_serializing_if = "Option::is_none")] pub nonce_source: Option, + /// What ran, at what fidelity, on whose hardware — see [`ServingProvenance`]. + /// Present on a `Terminal` envelope only when the dispatch outcome was an + /// actual 2xx response (`Responded`/`RespondedWithUsage`); `None` on + /// effective-request envelopes and on any non-2xx terminal envelope (a + /// denial/error before dispatch, a 503, or a dropped/failed connection) — + /// those served nothing, so there is nothing this field can honestly + /// report. + #[serde(skip_serializing_if = "Option::is_none")] + pub serving_provenance: Option, + /// The real token usage the served backend reported for this exchange (see + /// [`ExchangeUsage`]). Present on a terminal envelope for a host-served + /// exchange whose response carried a `usage` object; `None` on + /// effective-request envelopes and wherever the dispatch produced no usage + /// (a plugin-served stub, a denial, or a non-usage-bearing backend). + #[serde(skip_serializing_if = "Option::is_none")] + pub usage: Option, + /// The canonical JSON-DIGEST (`HEX(SHA-256(JCS(stringify_floats(body))))` + /// — see [`request_body_digest`]) of the REAL request body this host + /// actually dispatched. This is the one fact a downstream capsule needs + /// to bind its `agent_input_digest` to the real bytes: the terminal event + /// otherwise carries provenance and usage but nothing tying the sealed + /// capsule to *what was asked*. Present on a terminal envelope on either + /// dispatch path (host-served or plugin-served) whenever the host held a + /// parsed JSON request body; `None` when it did not (never a fabricated + /// digest). No raw prompt text is carried — only its digest. + #[serde(skip_serializing_if = "Option::is_none")] + pub request_digest: Option, } impl OpenAiExchangeEnvelope { @@ -103,6 +226,9 @@ impl OpenAiExchangeEnvelope { capsule_id: None, nonce: None, nonce_source: None, + serving_provenance: None, + usage: None, + request_digest: None, } } @@ -123,10 +249,51 @@ impl OpenAiExchangeEnvelope { capsule_id: marker.as_ref().map(|marker| marker.capsule_id.clone()), nonce: marker.as_ref().map(|marker| marker.nonce.clone()), nonce_source, + serving_provenance: None, + usage: None, + request_digest: None, } } + + /// Attach the serving provenance the host resolved for this exchange. A + /// small builder rather than a wider constructor so the two existing + /// callsites that already pass six positional args aren't churned, and so + /// the raw-proxy path can add provenance in one readable line after it has + /// gathered it from the node. + #[must_use] + pub fn with_serving_provenance(mut self, provenance: ServingProvenance) -> Self { + self.serving_provenance = Some(provenance); + self + } + + /// Attach the real token usage the served backend reported. Mirrors + /// [`Self::with_serving_provenance`] — a small builder so the host-served + /// raw-proxy path can add the REAL counts it read off the dispatch outcome + /// in one readable line, without churning the positional `terminal` + /// constructor. Only ever called with real usage; the field stays `None` + /// when the dispatch produced none. + #[must_use] + pub fn with_usage(mut self, usage: ExchangeUsage) -> Self { + self.usage = Some(usage); + self + } + + /// Attach the canonical JSON-DIGEST of the REAL request body this host + /// dispatched, so a downstream capsule can bind its `agent_input_digest` to + /// the real bytes. Mirrors the other builders — a small one-liner the + /// raw-proxy host-served path calls after it has the request body in hand. + /// Only ever called with a real digest computed by [`request_body_digest`]; + /// the field stays `None` when the host held no parsed body. + #[must_use] + pub fn with_request_digest(mut self, digest: String) -> Self { + self.request_digest = Some(digest); + self + } } +mod canonical_digest; +pub use canonical_digest::request_body_digest; + /// Publishes [`OpenAiExchangeEnvelope`]s to whatever is subscribed on /// [`OPENAI_EXCHANGE_CHANNEL`] — an out-of-process plugin in production, a /// recording double in tests. Fire-and-forget by design, mirroring @@ -135,6 +302,16 @@ impl OpenAiExchangeEnvelope { #[async_trait] pub trait OpenAiExchangeChannel: Send + Sync + 'static { async fn publish(&self, event: &OpenAiExchangeEnvelope); + + /// Whether anything is actually listening on [`OPENAI_EXCHANGE_CHANNEL`] + /// right now. Lets a caller skip the work that only exists to build an + /// event (canonicalizing and hashing a request body, cloning a + /// served-model descriptor) before finding out `publish` had nowhere to + /// send it. Defaults to `true` — a test double with no subscriber + /// concept (e.g. a recording channel) should behave as it always has. + async fn has_subscriber(&self) -> bool { + true + } } #[async_trait] @@ -159,6 +336,11 @@ impl OpenAiExchangeChannel for PluginManager { tracing::warn!(%error, "failed to publish openai exchange event to plugins"); } } + + async fn has_subscriber(&self) -> bool { + self.any_plugin_declares_mesh_channel(OPENAI_EXCHANGE_CHANNEL) + .await + } } /// Bridges path 1 (`openai-frontend`'s typed hook seam) to @@ -276,25 +458,45 @@ fn client_nonce_source(request: &ChatCompletionRequest) -> ClientNonceSource { } } +/// A publish sink that records every envelope it receives instead of +/// delivering it anywhere — shared by this module's own tests (path 1, the +/// typed frontend hook bridge) and `network::openai::ingress`'s tests (path +/// 2, the raw-proxy terminal builder), so both dispatch paths can assert on +/// exactly what a subscribing plugin would have seen without spinning one +/// up. #[cfg(test)] -mod tests { +pub(crate) mod test_support { use std::sync::Mutex; - use openai_frontend::{ChatCompletionOutcome, HookedOpenAiBackend, OpenAiBackend, Usage}; + use async_trait::async_trait; - use super::*; + use super::{OpenAiExchangeChannel, OpenAiExchangeEnvelope}; #[derive(Default)] - struct RecordingChannel { + pub(crate) struct RecordingChannel { events: Mutex>, } + impl RecordingChannel { + pub(crate) fn events(&self) -> Vec { + self.events.lock().unwrap().clone() + } + } + #[async_trait] impl OpenAiExchangeChannel for RecordingChannel { async fn publish(&self, event: &OpenAiExchangeEnvelope) { self.events.lock().unwrap().push(event.clone()); } } +} + +#[cfg(test)] +mod tests { + use openai_frontend::{ChatCompletionOutcome, HookedOpenAiBackend, OpenAiBackend, Usage}; + + use super::test_support::RecordingChannel; + use super::*; struct EchoBackend; @@ -349,7 +551,7 @@ mod tests { .await .expect("backend call succeeds"); - let events = channel.events.lock().unwrap(); + let events = channel.events(); assert_eq!(events.len(), 2, "one effective-request, one terminal"); assert_eq!( @@ -393,7 +595,7 @@ mod tests { .await .expect("backend call succeeds"); - let events = channel.events.lock().unwrap(); + let events = channel.events(); assert_eq!(events[1].nonce.as_deref(), Some("abc123")); assert_eq!( events[1].nonce_source, @@ -416,7 +618,7 @@ mod tests { .await .expect("backend call succeeds"); - let events = channel.events.lock().unwrap(); + let events = channel.events(); assert!( events[1] .nonce @@ -446,7 +648,7 @@ mod tests { .on_chat_completion_terminal(&request, "exchange-1", &denial) .await; - let events = channel.events.lock().unwrap(); + let events = channel.events(); assert_eq!(events.len(), 1); assert_eq!(events[0].exchange_id, "exchange-1"); assert_eq!(events[0].status, Some(400)); @@ -470,7 +672,7 @@ mod tests { .on_chat_completion_terminal(&request, "exchange-1", &ChatCompletionOutcome::Cancelled) .await; - let events = channel.events.lock().unwrap(); + let events = channel.events(); assert_eq!(events.len(), 1); assert_eq!(events[0].phase, OpenAiExchangePhase::Terminal); assert!(events[0].status.is_none()); @@ -539,7 +741,7 @@ mod tests { slow_result.expect("slow exchange succeeds"); fast_result.expect("fast exchange succeeds"); - let events = channel.events.lock().unwrap(); + let events = channel.events(); assert_eq!(events.len(), 4, "two effective + two terminal events"); assert_eq!(events[0].phase, OpenAiExchangePhase::EffectiveRequest); assert_eq!(events[1].phase, OpenAiExchangePhase::EffectiveRequest); @@ -564,4 +766,140 @@ mod tests { "slow exchange's terminal event pairs with its own effective event" ); } + + /// A terminal envelope with serving provenance serializes the known fields + /// and OMITS the unknown ones (never a fabricated `null` or empty string) — + /// this is the honesty contract a downstream capsule relies on: a field + /// that is present is a real host fact, a field that is absent is genuinely + /// unknown, not zeroed. + #[test] + fn terminal_carries_serving_provenance_and_omits_unknown_fields() { + let envelope = OpenAiExchangeEnvelope::terminal( + "exch-1", + OpenAiExchangeDispatchPath::RawProxy, + "hermes-2-pro-mistral-7b", + Some(200), + None, + None, + ) + .with_serving_provenance(ServingProvenance { + served_by_node_id: "node-abc".to_string(), + hostname: Some("host-1".to_string()), + quantization: Some("Q4_K_M".to_string()), + architecture: Some("llama".to_string()), + context_length: Some(8192), + parameter_size: Some("7B".to_string()), + layer_count: Some(32), + model_identity_hash: Some("abc123".to_string()), + model_canonical_ref: None, + model_revision: None, + gpu: None, + vram_bytes: None, + is_soc: Some(true), + }); + + let value = serde_json::to_value(&envelope).expect("serialize"); + let prov = &value["serving_provenance"]; + assert_eq!(prov["served_by_node_id"], "node-abc"); + assert_eq!(prov["quantization"], "Q4_K_M"); + assert_eq!(prov["architecture"], "llama"); + assert_eq!(prov["context_length"], 8192); + assert_eq!(prov["layer_count"], 32); + assert_eq!(prov["is_soc"], true); + // Unknown facts are ABSENT (omitted), not fabricated as null/empty. + assert!(prov.get("model_canonical_ref").is_none()); + assert!(prov.get("model_revision").is_none()); + assert!(prov.get("gpu").is_none()); + assert!(prov.get("vram_bytes").is_none()); + } + + /// An effective-request envelope carries NO serving provenance (the field + /// is omitted entirely), so the block is a terminal-only, served-exchange + /// fact — never claimed before the exchange actually ran. + #[test] + fn effective_envelope_has_no_serving_provenance() { + let envelope = + OpenAiExchangeEnvelope::effective("exch-1", OpenAiExchangeDispatchPath::RawProxy, "m"); + assert!(envelope.serving_provenance.is_none()); + let value = serde_json::to_value(&envelope).expect("serialize"); + assert!(value.get("serving_provenance").is_none()); + } + + /// The real token usage the host-served path reads off its dispatch outcome + /// rides the terminal envelope, so a downstream plugin can seal the REAL + /// counts of a host-served real-weights exchange instead of a stub's zeros. + #[test] + fn terminal_carries_real_usage_when_attached() { + let envelope = OpenAiExchangeEnvelope::terminal( + "exch-usage", + OpenAiExchangeDispatchPath::RawProxy, + "llama-3.2-3b-instruct", + Some(200), + None, + None, + ) + .with_usage(ExchangeUsage { + prompt_tokens: 42, + cached_prompt_tokens: Some(10), + completion_tokens: 6, + total_tokens: 48, + }); + + let value = serde_json::to_value(&envelope).expect("serialize"); + assert_eq!(value["usage"]["prompt_tokens"], 42); + assert_eq!(value["usage"]["cached_prompt_tokens"], 10); + assert_eq!(value["usage"]["completion_tokens"], 6); + assert_eq!(value["usage"]["total_tokens"], 48); + } + + /// A terminal envelope with no usage attached OMITS the `usage` key entirely + /// (never a fabricated all-zero object) — the same honesty contract the + /// serving-provenance fields hold: absent means genuinely unknown. + #[test] + fn terminal_omits_usage_when_none_attached() { + let envelope = OpenAiExchangeEnvelope::terminal( + "exch-no-usage", + OpenAiExchangeDispatchPath::RawProxy, + "some-plugin-model", + Some(200), + None, + None, + ); + let value = serde_json::to_value(&envelope).expect("serialize"); + assert!(value.get("usage").is_none()); + } + + /// A terminal envelope carrying a real request digest serializes it, and it + /// survives a round-trip — the one fact a downstream capsule binds its + /// `agent_input_digest` to. + #[test] + fn terminal_carries_request_digest_when_attached() { + let envelope = OpenAiExchangeEnvelope::terminal( + "exch-rd", + OpenAiExchangeDispatchPath::RawProxy, + "llama-3.2-3b-instruct", + Some(200), + None, + None, + ) + .with_request_digest("deadbeef".to_string()); + let value = serde_json::to_value(&envelope).expect("serialize"); + assert_eq!(value["request_digest"], "deadbeef"); + } + + /// No request digest attached -> the key is omitted entirely (never a + /// fabricated empty digest), same honesty contract as usage/provenance. + #[test] + fn terminal_omits_request_digest_when_none_attached() { + let envelope = OpenAiExchangeEnvelope::terminal( + "exch-no-rd", + OpenAiExchangeDispatchPath::RawProxy, + "m", + Some(200), + None, + None, + ); + let value = serde_json::to_value(&envelope).expect("serialize"); + assert!(value.get("request_digest").is_none()); + } } diff --git a/crates/mesh-llm-host-runtime/src/plugin/openai_exchange/canonical_digest.rs b/crates/mesh-llm-host-runtime/src/plugin/openai_exchange/canonical_digest.rs new file mode 100644 index 0000000000..00491271a9 --- /dev/null +++ b/crates/mesh-llm-host-runtime/src/plugin/openai_exchange/canonical_digest.rs @@ -0,0 +1,510 @@ +//! Canonical digest of an OpenAI request body: float-stringify, RFC 8785 JCS, +//! SHA-256. Pulled out of `openai_exchange.rs` (pure move, no behavior +//! change) because it is a self-contained RFC 8785 port with zero coupling to +//! the envelope types that file otherwise defines. + +/// The canonical JSON-DIGEST of a request body: `HEX(SHA-256(JCS(v)))` over +/// the float-stringified body. Byte-for-byte identical to the current +/// `agent_action_capsule.canonical.json_digest` reference and the Python +/// `capsule_sidecar.digest_json` sidecar wrapper it backs +/// (`digest_json(v) = json_digest(_stringify_floats(v))`) — cross-verified +/// against a live run of that reference, not just read off its source. +/// +/// Deliberately does **not** apply the profile's absent-field `normalize` +/// step: `agent_action_capsule.canonical` reserves normalization for the +/// vintage format-2 Capsule-ID path only — its current (non-vintage) +/// `json_digest` is plain `JCS(v)`, no normalize — so an OpenAI request body +/// that explicitly carries `null` for an unset optional field (common OpenAI +/// client behavior, e.g. `"stop": null`) must still digest to a DIFFERENT +/// value than the same body with that field omitted; normalizing here would +/// silently collapse the two. (Note for the record: `capsule-emit-mesh`'s own +/// Rust `canonical_body_digest` currently still routes through +/// `capsule_producer::jcs::json_digest`, which normalizes — a drift from the +/// current Python reference that this host digest does not replicate. Flagged +/// separately; out of scope for this crate to fix.) +/// +/// It is `HEX(SHA-256(JCS(stringify_floats(body))))`: +/// 1. `stringify_floats` — every JSON float becomes its exact decimal string +/// (JCS refuses floats in a digest-bearing value; OpenAI chat bodies are +/// full of them: temperature, top_p, penalties); +/// 2. JCS — RFC 8785 canonical serialization (sorted keys, minimal form); +/// 3. SHA-256, lowercase hex. +/// +/// A self-contained port kept in this crate (the host cannot depend on the +/// plugin's `capsule-producer`), verified against the Python reference on the +/// frozen fixture in the tests below. +/// +/// Returns `None` — never a digest of a body the reference would refuse — +/// when `body` contains a JSON integer literal outside the reference's safe +/// range; see [`MAX_SAFE_INTEGER`]. +pub fn request_body_digest(body: &serde_json::Value, source_json: Option<&[u8]>) -> Option { + if contains_unsafe_integer(body) || source_json.is_some_and(contains_oversized_integer_literal) + { + return None; + } + use sha2::{Digest, Sha256}; + let canonical = jcs_bytes(&stringify_floats(body)); + Some(hex::encode(Sha256::digest(&canonical))) +} + +/// The reference's safe-integer boundary (`agent_action_capsule.canonical` +/// §5.1): a JSON integer literal outside `+/-(2^53-1)` cannot round-trip +/// through the reference's digest (it raises `UnsafeIntegerError` rather than +/// digest a value it cannot represent losslessly). +const MAX_SAFE_INTEGER: u64 = (1u64 << 53) - 1; + +/// Whether `value` contains a JSON integer literal outside the reference's +/// safe range — see [`MAX_SAFE_INTEGER`]. Floats are exempt: a float outside +/// this range already loses precision in the source JSON itself, and +/// [`stringify_floats`] renders it via [`float_repr`], not this check. +fn contains_unsafe_integer(value: &serde_json::Value) -> bool { + use serde_json::Value; + match value { + Value::Number(n) => { + if let Some(i) = n.as_i64() { + i.unsigned_abs() > MAX_SAFE_INTEGER + } else if let Some(u) = n.as_u64() { + u > MAX_SAFE_INTEGER + } else { + false + } + } + Value::Object(map) => map.values().any(contains_unsafe_integer), + Value::Array(arr) => arr.iter().any(contains_unsafe_integer), + _ => false, + } +} + +/// Detect an integer token that serde_json would otherwise round to `f64` +/// after it exceeds `u64`. This small lexer runs over the original request +/// body so exponent/decimal floats remain distinguishable from integer +/// literals; strings are skipped, including escaped quotes. +fn contains_oversized_integer_literal(source: &[u8]) -> bool { + let mut index = 0; + let mut in_string = false; + let mut escaped = false; + while index < source.len() { + let byte = source[index]; + if in_string { + if escaped { + escaped = false; + } else if byte == b'\\' { + escaped = true; + } else if byte == b'"' { + in_string = false; + } + index += 1; + continue; + } + if byte == b'"' { + in_string = true; + index += 1; + continue; + } + if byte == b'-' || byte.is_ascii_digit() { + let start = index; + index += 1; + while index < source.len() + && matches!( + source[index], + b'0'..=b'9' | b'.' | b'e' | b'E' | b'+' | b'-' + ) + { + index += 1; + } + let token = &source[start..index]; + if !token.contains(&b'.') + && !token.contains(&b'e') + && !token.contains(&b'E') + && integer_token_exceeds_safe_range(token) + { + return true; + } + continue; + } + index += 1; + } + false +} + +fn integer_token_exceeds_safe_range(token: &[u8]) -> bool { + let digits = token.strip_prefix(b"-").unwrap_or(token); + let digits = digits + .iter() + .skip_while(|digit| **digit == b'0') + .copied() + .collect::>(); + let safe = MAX_SAFE_INTEGER.to_string(); + digits.len() > safe.len() || (digits.len() == safe.len() && digits.as_slice() > safe.as_bytes()) +} + +/// Replace every JSON float with its exact decimal-string form via +/// [`float_repr`] (mirrors the Python reference's `_stringify_floats`, which +/// stringifies via `repr(float)`). +/// +/// This is a property of the digest context, not a bug this function +/// introduces: once a float and its stringified form are both JSON strings, +/// JCS can no longer tell them apart, so `{"temperature": 0.7}` and +/// `{"temperature": "0.7"}` digest identically. It is inherited unchanged +/// from `agent_action_capsule.canonical`/`capsule_sidecar.digest_json`, the +/// same reference [`request_body_digest`] ports. The current declaration +/// point acknowledging this collision (rather than callers discovering it +/// silently) is the `x-mesh-poc-v1` PoC-only extension block in +/// `capsule-emit-mesh`'s `capsule_sidecar.py`; registering it as a real +/// profile-level property is a separate spec-lane item. +fn stringify_floats(value: &serde_json::Value) -> serde_json::Value { + use serde_json::Value; + match value { + // Without serde_json's `arbitrary_precision` feature, `is_f64()` is + // exclusive with in-range integer literals. Oversized integer syntax + // is checked against the original source before this traversal. + Value::Number(n) if n.is_f64() => match n.as_f64() { + Some(f) => Value::String(float_repr(f)), + // Unreachable given the guard above; keep the original value + // rather than a panic site over attacker-supplied JSON. + None => value.clone(), + }, + Value::Number(_) => value.clone(), + Value::Object(map) => Value::Object( + map.iter() + .map(|(k, v)| (k.clone(), stringify_floats(v))) + .collect(), + ), + Value::Array(arr) => Value::Array(arr.iter().map(stringify_floats).collect()), + other => other.clone(), + } +} + +/// Port of Python's `repr(float)` / `str(float)` / `json.dumps` float +/// formatting: the shortest decimal digit sequence that round-trips to `f`, +/// rendered in fixed-point form when the decimal exponent is in `[-4, 16)` +/// and in exponent form (`d[.ddd]e+NN`/`e-NN`, sign always present, at least +/// two exponent digits) otherwise. Rust's own `Display` for `f64` already +/// computes the shortest round-trip digit sequence but — unlike Python — +/// never switches to exponent form, so this reuses `Display`'s fixed-point +/// string purely as a source of those digits and re-derives the placement. +fn float_repr(f: f64) -> String { + if f == 0.0 { + return if f.is_sign_negative() { + "-0.0".to_string() + } else { + "0.0".to_string() + }; + } + let sign = if f.is_sign_negative() { "-" } else { "" }; + let fixed = format!("{}", f.abs()); + let (int_part, frac_part) = match fixed.split_once('.') { + Some((i, f)) => (i, f), + None => (fixed.as_str(), ""), + }; + // The decimal exponent of the most significant digit, and the + // significant digits themselves (no leading zeros). + let (digits, exponent): (String, i32) = if int_part != "0" { + // A double this large (>= 10^16, once `exponent` disqualifies fixed + // form below) has no fractional precision left — it is exactly an + // integer — so `int_part` alone carries every real digit. Whatever + // follows the last nonzero digit is a positional placeholder, not a + // significant digit, and must be trimmed for the mantissa (`2e17`, + // not `2.00000000000000000e+17`). + let exponent = int_part.len() as i32 - 1; + let trimmed = int_part.trim_end_matches('0'); + let trimmed = if trimmed.is_empty() { "0" } else { trimmed }; + (trimmed.to_string(), exponent) + } else { + let leading_zeros = frac_part.chars().take_while(|c| *c == '0').count(); + ( + frac_part[leading_zeros..].to_string(), + -(leading_zeros as i32) - 1, + ) + }; + if (-4..16).contains(&exponent) { + // Fixed notation: the source string already places the digits + // correctly; just guarantee the trailing `.0` Python always shows for + // a whole number. + let s = if fixed.contains('.') { + fixed + } else { + format!("{fixed}.0") + }; + format!("{sign}{s}") + } else { + let mantissa = if digits.len() == 1 { + digits + } else { + format!("{}.{}", &digits[..1], &digits[1..]) + }; + let exp_sign = if exponent < 0 { '-' } else { '+' }; + format!("{sign}{mantissa}e{exp_sign}{:02}", exponent.abs()) + } +} + +/// RFC 8785 JCS serialization (mirror of `agent_action_capsule.canonical.jcs`). +/// Floats are already stringified before this runs, so a bare float here is a +/// programmer error, serialized via serde's default rather than panicking. +fn jcs_bytes(v: &serde_json::Value) -> Vec { + let mut out = String::new(); + jcs_value(v, &mut out); + out.into_bytes() +} + +fn jcs_value(v: &serde_json::Value, out: &mut String) { + use serde_json::Value; + match v { + Value::Null => out.push_str("null"), + Value::Bool(true) => out.push_str("true"), + Value::Bool(false) => out.push_str("false"), + Value::String(s) => jcs_string(s, out), + Value::Number(n) => out.push_str(&n.to_string()), + Value::Array(arr) => { + out.push('['); + for (i, x) in arr.iter().enumerate() { + if i > 0 { + out.push(','); + } + jcs_value(x, out); + } + out.push(']'); + } + Value::Object(map) => { + // RFC 8785 §3.2.3: object members sorted by UTF-16 code-unit sequence. + let mut items: Vec<(&String, &Value)> = map.iter().collect(); + items.sort_by(|(a, _), (b, _)| { + let au: Vec = a.encode_utf16().collect(); + let bu: Vec = b.encode_utf16().collect(); + au.cmp(&bu).then_with(|| a.cmp(b)) + }); + out.push('{'); + for (i, (k, val)) in items.iter().enumerate() { + if i > 0 { + out.push(','); + } + jcs_string(k, out); + out.push(':'); + jcs_value(val, out); + } + out.push('}'); + } + } +} + +fn jcs_string(s: &str, out: &mut String) { + out.push('"'); + for ch in s.chars() { + let o = ch as u32; + match ch { + '"' => out.push_str("\\\""), + '\\' => out.push_str("\\\\"), + _ if o == 0x08 => out.push_str("\\b"), + _ if o == 0x09 => out.push_str("\\t"), + _ if o == 0x0A => out.push_str("\\n"), + _ if o == 0x0C => out.push_str("\\f"), + _ if o == 0x0D => out.push_str("\\r"), + _ if o < 0x20 => out.push_str(&format!("\\u{o:04x}")), + _ => out.push(ch), + } + } + out.push('"'); +} + +#[cfg(test)] +mod tests { + use super::*; + + fn digest(body: &serde_json::Value) -> Option { + request_body_digest(body, None) + } + + /// The host's `request_body_digest` is byte-for-byte the value a live run + /// of the Python reference, `capsule_sidecar.digest_json`, produces over + /// the identical JSON value: + /// + /// python3 -c " + /// from capsule_sidecar import digest_json + /// print(digest_json({ + /// 'model': 'hermes-2-pro-mistral-7b', + /// 'messages': [{'role': 'user', 'content': 'hello'}], + /// 'temperature': 0.7, + /// 'top_p': 1.0, + /// 'max_tokens': 512, + /// }))" + /// + /// `top_p: 1.0` exercises the whole-number-float edge case + /// (`stringify_floats` must emit "1.0", not "1"). This value has no + /// null/absent fields, so it does not by itself distinguish a + /// normalizing digest from a non-normalizing one — see the next test for + /// that. + #[test] + fn request_body_digest_matches_python_reference() { + let body = serde_json::json!({ + "model": "hermes-2-pro-mistral-7b", + "messages": [{"role": "user", "content": "hello"}], + "temperature": 0.7, + "top_p": 1.0, + "max_tokens": 512 + }); + let expected = "a6329c5ebb66562f38a8136a8d8511b6aeed166e4c7d889b9133ac96fc49a9d5"; + assert_eq!(digest(&body).as_deref(), Some(expected)); + } + + /// The same body, plus two explicit-`null` optional fields (as real + /// OpenAI clients routinely send, e.g. `"stop": null`), must digest to a + /// DIFFERENT value than the null-free body above — proving this digest + /// does NOT apply the profile's absent-field `normalize` step. Expected + /// value from the same live Python reference invocation with the two + /// extra `None` fields added to the dict. Pins the current + /// `agent_action_capsule.canonical.json_digest` contract (normalize is + /// vintage-format-2-only) against a future accidental reintroduction of + /// normalization here. + #[test] + fn request_body_digest_does_not_normalize_absent_fields() { + let body = serde_json::json!({ + "model": "hermes-2-pro-mistral-7b", + "messages": [{"role": "user", "content": "hello"}], + "temperature": 0.7, + "top_p": 1.0, + "max_tokens": 512, + "stop": null, + "user": null + }); + let expected = "ee8aeb450ccf8c8017caae0d3733d3dcd62ec88752053894118d28cea0d176fe"; + assert_eq!(digest(&body).as_deref(), Some(expected)); + } + + /// `float_repr`'s fixed-vs-exponent switch and digit sequence, each + /// checked directly against the Python `repr()` values ndizazzo's review + /// tabulated (`1e-05`, `1e-07`, `1e+16`, `1e+20`), plus the boundary the + /// table didn't cover (`1e-4` stays fixed) and `-0.0`. + #[test] + fn float_repr_matches_python_repr_at_and_around_the_exponent_boundaries() { + assert_eq!(float_repr(1e-5), "1e-05"); + assert_eq!(float_repr(1e-7), "1e-07"); + assert_eq!(float_repr(1e16), "1e+16"); + assert_eq!(float_repr(1e20), "1e+20"); + assert_eq!(float_repr(2e17), "2e+17"); + // exp == -4 is the last value that stays fixed; exp == -5 switches. + assert_eq!(float_repr(1e-4), "0.0001"); + assert_eq!(float_repr(-0.0), "-0.0"); + assert_eq!(float_repr(0.0), "0.0"); + assert_eq!(float_repr(0.7), "0.7"); + assert_eq!(float_repr(1.0), "1.0"); + } + + /// Each vector cross-checked against a live run of the Python reference, + /// `capsule_sidecar.digest_json`, over the identical JSON value: + /// + /// PYTHONPATH=.../agent-action-capsule/python python3 -c " + /// from capsule_sidecar import digest_json + /// print(digest_json({'v': }))" + /// + /// These are exactly the float values the previous digest silently got + /// wrong (Rust's bare `Display` never emits exponent notation), plus the + /// `1e-9` determinism trick and the `0.1 + 0.2` imprecision case. + #[test] + fn request_body_digest_matches_python_reference_for_float_edge_cases() { + assert_eq!( + digest(&serde_json::json!({"v": 1e-5_f64})).as_deref(), + Some("8edda8e740022353cd222db1ff9c56e6bef76663a10814bd6aaf579f8d0e2332") + ); + assert_eq!( + digest(&serde_json::json!({"v": 1e-7_f64})).as_deref(), + Some("bbe43d466a7e53136ff444be1b91529154b547b43774b99d7a6cf41bb4a3ae34") + ); + assert_eq!( + digest(&serde_json::json!({"v": 1e16_f64})).as_deref(), + Some("318fda488ff6a31c5710e73d0ad02340677be5ed8d553869596ff8b2e27b3b94") + ); + assert_eq!( + digest(&serde_json::json!({"v": 1e20_f64})).as_deref(), + Some("edf56ab854860723e8a400417d7605b1405964f0dbd4289bfe8c2373238496f5") + ); + assert_eq!( + digest(&serde_json::json!({"v": -0.0_f64})).as_deref(), + Some("7c9018d8078566c67ebff7a4fa6be32f7b4f0f1dbd4e5d3abcaca596e57d6831") + ); + assert_eq!( + digest(&serde_json::json!({"v": 1e-9_f64})).as_deref(), + Some("f253528520b4878440038edf53d7a32b00a08bd216c2a0c0b60ffbf5cda133b0") + ); + assert_eq!( + digest(&serde_json::json!({"v": 0.1_f64 + 0.2_f64})).as_deref(), + Some("5204642c42382100bd6fb098cb429a45a4f42c994b67f2de909274e597756b4b") + ); + } + + /// A JSON integer literal beyond the reference's safe range + /// (`+/-(2^53-1)`, `agent_action_capsule.canonical` §5.1) makes the + /// reference raise `UnsafeIntegerError` rather than digest a value it + /// cannot represent losslessly — verified live: + /// + /// PYTHONPATH=.../agent-action-capsule/python python3 -c " + /// from capsule_sidecar import digest_json + /// digest_json({'v': 9007199254740993})" + /// # -> UnsafeIntegerError: integer 9007199254740993 is outside the + /// # safe range +/-9007199254740991 (...) + /// + /// The host must omit `request_digest` entirely on such a body — never + /// digest what the reference would refuse. + #[test] + fn request_body_digest_omits_when_body_has_an_unsafe_integer() { + let safe = serde_json::json!({"v": 9_007_199_254_740_991_i64}); + assert!(digest(&safe).is_some(), "2^53-1 is safe"); + + let one_over = serde_json::json!({"v": 9_007_199_254_740_993_i64}); + assert!( + digest(&one_over).is_none(), + "2^53+1 exceeds the reference's safe-integer range" + ); + + let nested = serde_json::json!({"a": [1, {"b": 9_007_199_254_740_993_i64}]}); + assert!( + digest(&nested).is_none(), + "an unsafe integer nested inside an array/object must still be caught" + ); + + let negative_over = serde_json::json!({"v": -9_007_199_254_740_993_i64}); + assert!( + digest(&negative_over).is_none(), + "the safe range is symmetric" + ); + + let above_u64_source = br#"{"v":18446744073709551616}"#; + let above_u64: serde_json::Value = serde_json::from_slice(above_u64_source).unwrap(); + assert!( + request_body_digest(&above_u64, Some(above_u64_source)).is_none(), + "an integer literal above u64::MAX must not be rounded into a float digest" + ); + + let exponent_source = br#"{"v":1e20}"#; + let exponent: serde_json::Value = serde_json::from_slice(exponent_source).unwrap(); + assert!( + request_body_digest(&exponent, Some(exponent_source)).is_some(), + "a floating-point exponent remains digestible" + ); + + let quoted_source = br#"{"v":"18446744073709551616"}"#; + let quoted: serde_json::Value = serde_json::from_slice(quoted_source).unwrap(); + assert!( + request_body_digest("ed, Some(quoted_source)).is_some(), + "integer-looking text inside a string is not a numeric literal" + ); + } + + /// A nested array, a non-object top-level value, a non-BMP key, and a + /// control character in a string all digest without panicking, and match + /// the same live Python reference run as the other vectors here. + #[test] + fn request_body_digest_matches_python_reference_for_structural_edge_cases() { + assert_eq!( + digest(&serde_json::json!({"v": [1, [2, 3], {"a": 1.5}]})).as_deref(), + Some("1cb385d061fc163f6663b80217f5c262795c669da48c0b788dcc9b328b888226") + ); + assert_eq!( + digest(&serde_json::json!({"\u{1F600}": 1})).as_deref(), + Some("763606c9e0046348cc185a6e829e12ae6c0f3565b940f8184901a4d83dda6c33") + ); + assert_eq!( + digest(&serde_json::json!({"v": "a\tb"})).as_deref(), + Some("595711cef0e6e4d037e1fae2b1ece32702c442c0501d6362791e31cd1a6c866d") + ); + } +}