diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..0a1d199 --- /dev/null +++ b/.gitattributes @@ -0,0 +1,2 @@ +schemas/*.json text eol=lf +crates/model-connector/tests/fixtures/wire_v3/* -text diff --git a/Cargo.lock b/Cargo.lock index b4e9546..3c3ff8d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -495,6 +495,7 @@ dependencies = [ "cultnet-rs", "rand_core 0.6.4", "rmp-serde", + "rmpv", "schemars", "serde", "serde_bytes", diff --git a/README.md b/README.md index 433f540..cbcccb5 100644 --- a/README.md +++ b/README.md @@ -11,11 +11,16 @@ Status: the `model-connector` crate is live. No connector binary exists yet Crate `crates/model-connector`: -- `src/contract.rs` - the v2 wire contract: `ProviderRequest`, `TransportInvocation`, - `TransportResult`, `TransportReceipt`, `Refusal`, and the canonical byte and digest - functions that bind a result to its request. -- `src/envelope.rs` - the AES-256-GCM `TransportEnvelope`, its identity-binding AAD, and - the length-prefixed framing. +- `src/contract.rs` - the v3 wire contract (`gamecult.model.*.v3`): `ProviderRequest`, `TransportInvocation`, + `TransportResult`, `TransportReceipt`, the typed `StopReason`, `FailureKind` and `Refusal`, + and the canonical byte and digest functions that bind a result to its request. Every + document is named MessagePack with unknown fields refused. +- `src/envelope.rs` - the AES-256-GCM `TransportEnvelope` (binary ciphertext), its + identity-binding AAD, the length-prefixed framing, and the CultNet carriage: each frame is a + `cultnet.operation_request.v0` / `operation_response.v0` for service + `gamecult.model.connector`, operation `invoke`, its `message_id` the request id and its + payload the base64 of the sealed envelope. +- `schemas/` - one JSON Schema per v3 document, generated from the types. - `src/client.rs` - `ConnectorClient`, the one-request caller side. - `src/service.rs` (feature `daemon`) - `TransportService`: caller admission, durable replay (a retried request is answered byte-identically and the provider runs once), @@ -24,16 +29,15 @@ Crate `crates/model-connector`: `ProviderBackend` trait. A connector binary implements `ProviderBackend` for its provider and calls `serve`; the kit names no provider. -Frame arithmetic is public in `src/envelope.rs`: `frame_bytes_for_payload(n)` is the frame a +Frame arithmetic is public in `src/envelope.rs`, measured on the carriage frame: `frame_bytes_for_payload(n)` is the frame a caller payload bound of `n` needs, `payload_bytes_for_frame(frame)` is the largest payload bound a frame allows, and `MAX_IDENTITY_BYTES` (256) bounds caller ids, request ids and model names. `serve` refuses a frame/payload pair that violates them (`ServeError::FrameBound` states the frame needed); the smallest legal frame is `frame_bytes_for_payload(MIN_PAYLOAD_BYTES)`. A request id past the bound is refused on the wire as `Refusal::Malformed`, the reply echoing the id (the connection is dropped only if that reply cannot fit the frame); `ConnectorClient` refuses such an id locally with `RequestIdTooLong`. -The default feature set compiles contract, envelope and client only. Schema ids are -still `gamecult.codex.*.v2`; renaming them changes the wire and belongs to a later -contract version. +The default feature set compiles contract, envelope and client only. The kit speaks v3 +only: a v2 frame gets no answer. ## Verify @@ -42,8 +46,11 @@ cargo test --locked --workspace cargo test --locked --workspace --features model-connector/daemon ``` -`crates/model-connector/tests/wire_golden_v2.rs` pins the kit byte-for-byte to the golden v2 -fixtures in `crates/model-connector/tests/fixtures/wire_v2`. +`crates/model-connector/tests/wire_golden_v3.rs` pins the v3 documents and one sealed carriage +frame to the fixtures in `crates/model-connector/tests/fixtures/wire_v3` (regenerate with +`WIRE_V3_REGENERATE=1`; see the fixtures' README), and +`crates/model-connector/tests/published_schemas.rs` fails when `schemas/` differs from the types +(`MODEL_CONNECTOR_REGENERATE_SCHEMAS=1` rewrites it). ## Provenance and license diff --git a/crates/model-connector/Cargo.toml b/crates/model-connector/Cargo.toml index f004be5..8bd884a 100644 --- a/crates/model-connector/Cargo.toml +++ b/crates/model-connector/Cargo.toml @@ -25,3 +25,6 @@ serde_json = "1" sha2 = "0.10" thiserror = "2" zeroize = "1" + +[dev-dependencies] +rmpv = "1" diff --git a/crates/model-connector/src/client.rs b/crates/model-connector/src/client.rs index ab1e3de..1421a5b 100644 --- a/crates/model-connector/src/client.rs +++ b/crates/model-connector/src/client.rs @@ -6,8 +6,8 @@ use zeroize::Zeroizing; use crate::contract::{TransportInvocation, TransportKey, TransportResult}; use crate::envelope::{ - MAX_IDENTITY_BYTES, ServiceError, TransportFrameError, decrypt_result, encrypt_invocation, - read_transport_frame, write_transport_frame, + MAX_IDENTITY_BYTES, ServiceError, TransportFrameError, decode_result_frame, decrypt_result, + encode_invocation_frame, encrypt_invocation, read_transport_frame, write_transport_frame, }; #[derive(Clone)] @@ -51,8 +51,7 @@ impl ConnectorClient { if invocation.request_id().len() > MAX_IDENTITY_BYTES { return Err(ConnectorClientError::RequestIdTooLong); } - let request = rmp_serde::to_vec(&encrypt_invocation(invocation, &self.security)?) - .map_err(|_| ConnectorClientError::Encoding)?; + let request = encode_invocation_frame(&encrypt_invocation(invocation, &self.security)?)?; let connect_timeout = self .response_timeout .unwrap_or(Duration::from_secs(10)) @@ -67,8 +66,7 @@ impl ConnectorClient { .map_err(client_frame_error)?; let response = read_transport_frame(&mut stream, self.max_frame_bytes).map_err(client_frame_error)?; - let envelope = - rmp_serde::from_slice(&response).map_err(|_| ConnectorClientError::Encoding)?; + let envelope = decode_result_frame(&response)?; decrypt_result(&envelope, &self.security, invocation).map_err(Into::into) } } @@ -120,8 +118,6 @@ pub enum ConnectorClientError { Connection(#[source] std::io::Error), #[error("connector frame exceeded its bound")] FrameSize, - #[error("connector MessagePack encoding failed")] - Encoding, #[error("request id is longer than the connector identity bound")] RequestIdTooLong, #[error(transparent)] @@ -151,7 +147,7 @@ mod fixture_tests { let server = std::thread::spawn(move || { let (mut stream, _) = listener.accept().unwrap(); let request = read_transport_frame(&mut stream, 64 * 1024).unwrap(); - let envelope: TransportEnvelope = rmp_serde::from_slice(&request).unwrap(); + let envelope = decode_invocation_frame(&request).unwrap(); assert_eq!(decrypt_invocation(&envelope, &security).unwrap(), expected); let response = encrypt_result( &TransportResult::refused(&expected, Refusal::ModelNotAllowed), @@ -160,7 +156,7 @@ mod fixture_tests { .unwrap(); write_transport_frame( &mut stream, - &rmp_serde::to_vec(&response).unwrap(), + &encode_result_frame(&response).unwrap(), 64 * 1024, ) .unwrap(); @@ -203,4 +199,31 @@ mod fixture_tests { "the client must not even connect" ); } + + #[test] + fn the_client_measures_a_request_id_in_bytes() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + let client = ConnectorClient::new( + listener.local_addr().unwrap(), + "epiphany-distinct-test-key", + 64 * 1024, + Some(Duration::from_secs(1)), + ) + .unwrap(); + // 86 characters, 258 bytes: under the bound by characters, over by bytes. + let id = "\u{20AC}".repeat(86); + assert_eq!((id.chars().count(), id.len()), (86, 258)); + let mut long = invocation(); + long.request.request_id = id; + assert!(matches!( + client.execute(&long), + Err(ConnectorClientError::RequestIdTooLong) + )); + assert_eq!( + listener.accept().unwrap_err().kind(), + std::io::ErrorKind::WouldBlock, + "the client must not even connect" + ); + } } diff --git a/crates/model-connector/src/contract.rs b/crates/model-connector/src/contract.rs index 5cfffa4..5796d19 100644 --- a/crates/model-connector/src/contract.rs +++ b/crates/model-connector/src/contract.rs @@ -1,5 +1,7 @@ use std::collections::HashSet; +use schemars::JsonSchema; +use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use thiserror::Error; @@ -7,11 +9,11 @@ use zeroize::Zeroize; use crate::envelope::ServiceError; -pub const PROVIDER_REQUEST_SCHEMA_ID: &str = "gamecult.codex.provider_request.v2"; -pub const INVOCATION_SCHEMA_ID: &str = "gamecult.codex.transport_invocation.v2"; -pub const RESULT_SCHEMA_ID: &str = "gamecult.codex.transport_result.v2"; -pub const RECEIPT_SCHEMA_ID: &str = "gamecult.codex.transport_receipt.v2"; -pub const ENVELOPE_SCHEMA_ID: &str = "gamecult.codex.transport_envelope.v2"; +pub const PROVIDER_REQUEST_SCHEMA_ID: &str = "gamecult.model.provider_request.v3"; +pub const INVOCATION_SCHEMA_ID: &str = "gamecult.model.transport_invocation.v3"; +pub const RESULT_SCHEMA_ID: &str = "gamecult.model.transport_result.v3"; +pub const RECEIPT_SCHEMA_ID: &str = "gamecult.model.transport_receipt.v3"; +pub const ENVELOPE_SCHEMA_ID: &str = "gamecult.model.transport_envelope.v3"; #[derive(Clone, PartialEq, Eq)] pub struct TransportKey(pub(crate) [u8; 32]); @@ -31,7 +33,8 @@ impl Drop for TransportKey { } } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] pub struct ProviderRequest { pub schema_id: String, pub request_id: String, @@ -127,14 +130,14 @@ impl ProviderRequest { } } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "snake_case")] pub enum ToolChoice { Auto, Required, } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "snake_case")] pub enum ReasoningEffort { Low, @@ -143,7 +146,7 @@ pub enum ReasoningEffort { Max, } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "snake_case")] pub enum CacheRetention { Short, @@ -151,7 +154,7 @@ pub enum CacheRetention { } /// Why a provider stopped producing output. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "snake_case")] pub enum StopReason { EndTurn, @@ -164,7 +167,7 @@ pub enum StopReason { /// What kind of failure a provider reported. `ResultSize` is the connector's own /// stand-in for a result whose sealed frame cannot be carried. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "snake_case")] pub enum FailureKind { ResultSize, @@ -179,7 +182,8 @@ pub enum FailureKind { /// Token accounting for one completed request. Cache reads and writes are /// counted apart from plain input; `reasoning_tokens` is absent when the /// provider does not report it. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] pub struct Usage { pub input_tokens: u64, pub cache_read_tokens: u64, @@ -188,7 +192,8 @@ pub struct Usage { pub reasoning_tokens: Option, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] pub enum InputItem { UserText { text: String, @@ -207,14 +212,16 @@ pub enum InputItem { }, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] pub struct ToolDefinition { pub name: String, pub description: String, pub parameters_json: String, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] pub struct TransportInvocation { pub schema_id: String, pub caller_runtime_id: String, @@ -267,7 +274,8 @@ impl TransportInvocation { } } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] pub struct TransportResult { pub schema_id: String, pub request_id: String, @@ -275,7 +283,8 @@ pub struct TransportResult { pub disposition: TransportDisposition, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] pub enum TransportDisposition { Refused(Refusal), Transported { @@ -333,7 +342,8 @@ impl TransportResult { } } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] pub enum TransportEventPayload { TextDelta { text: String, @@ -345,7 +355,8 @@ pub enum TransportEventPayload { }, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] pub struct TransportEvent { pub sequence: u64, pub payload: TransportEventPayload, @@ -368,7 +379,8 @@ impl TransportEvent { } } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] pub struct TransportReceipt { pub schema_id: String, pub request_id: String, @@ -415,7 +427,8 @@ impl TransportReceipt { } } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] pub enum TransportOutcome { Completed { provider_response_id: Option, @@ -428,7 +441,7 @@ pub enum TransportOutcome { }, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] pub enum Refusal { Expired, IdentitySubstitution, @@ -443,11 +456,21 @@ pub enum Refusal { Malformed, } +/// The one wire codec of every v3 document: named MessagePack (struct fields +/// are keyed by name), with unknown fields refused by the types themselves. +pub(crate) fn wire_encode(document: &T) -> Result, ContractError> { + rmp_serde::to_vec_named(document).map_err(|_| ContractError::Encoding) +} + +pub(crate) fn wire_decode(bytes: &[u8]) -> Result { + rmp_serde::from_slice(bytes).map_err(|_| ContractError::Encoding) +} + pub fn canonical_provider_request_bytes( request: &ProviderRequest, ) -> Result, ContractError> { request.validate()?; - rmp_serde::to_vec(request).map_err(|_| ContractError::Encoding) + wire_encode(request) } pub fn provider_request_sha256(request: &ProviderRequest) -> Result<[u8; 32], ContractError> { @@ -461,7 +484,7 @@ pub fn canonical_invocation_bytes( if invocation.provider_request_sha256 != provider_request_sha256(&invocation.request)? { return Err(ContractError::ProviderDigest); } - rmp_serde::to_vec(invocation).map_err(|_| ContractError::Encoding) + wire_encode(invocation) } pub(crate) fn require_id(value: &str, field: &'static str) -> Result<(), ContractError> { @@ -525,7 +548,7 @@ pub enum ContractError { Expiry, #[error("transport event sequence is not contiguous")] EventSequence, - #[error("canonical MessagePack encoding failed")] + #[error("wire encoding failed")] Encoding, } @@ -601,8 +624,8 @@ mod tests { #[test] fn typed_tool_round_survives_messagepack() { let invocation = invocation(); - let bytes = rmp_serde::to_vec(&invocation).unwrap(); - let decoded: TransportInvocation = rmp_serde::from_slice(&bytes).unwrap(); + let bytes = wire_encode(&invocation).unwrap(); + let decoded: TransportInvocation = wire_decode(&bytes).unwrap(); assert_eq!(decoded, invocation); assert!(matches!( decoded.request.tools.as_slice(), @@ -655,4 +678,189 @@ mod tests { Err(ContractError::NativeDigest) ); } + + #[test] + fn v3_documents_carry_their_v3_ids() { + assert_eq!( + [ + PROVIDER_REQUEST_SCHEMA_ID, + INVOCATION_SCHEMA_ID, + RESULT_SCHEMA_ID, + RECEIPT_SCHEMA_ID, + ENVELOPE_SCHEMA_ID + ], + [ + "gamecult.model.provider_request.v3", + "gamecult.model.transport_invocation.v3", + "gamecult.model.transport_result.v3", + "gamecult.model.transport_receipt.v3", + "gamecult.model.transport_envelope.v3" + ] + ); + let invocation = invocation(); + let result = TransportResult::transported(&invocation, Vec::new(), receipt(&invocation)); + let TransportDisposition::Transported { receipt: sent, .. } = &result.disposition else { + unreachable!() + }; + assert_eq!(invocation.request.schema_id, PROVIDER_REQUEST_SCHEMA_ID); + assert_eq!(invocation.schema_id, INVOCATION_SCHEMA_ID); + assert_eq!(result.schema_id, RESULT_SCHEMA_ID); + assert_eq!(sent.schema_id, RECEIPT_SCHEMA_ID); + + let v2 = |name: &str| format!("{}{name}.v2", concat!("gamecult.", "codex.")); + let mut request = invocation.request.clone(); + request.schema_id = v2("provider_request"); + assert_eq!(request.validate(), Err(ContractError::Schema)); + let mut stale = invocation.clone(); + stale.schema_id = v2("transport_invocation"); + assert_eq!(stale.validate(1_000, 2_000), Err(ContractError::Schema)); + let mut stale = result.clone(); + stale.schema_id = v2("transport_result"); + assert_eq!( + stale.validate_against(&invocation), + Err(ContractError::Schema) + ); + let mut stale = (**sent).clone(); + stale.schema_id = v2("transport_receipt"); + assert_eq!( + stale.validate_against(&invocation), + Err(ContractError::Schema) + ); + } + + /// `document` re-encoded as a named map with one more key at its top level. + fn with_extra_key(document: &T) -> Vec { + let bytes = wire_encode(document).unwrap(); + let mut value = rmpv::decode::read_value(&mut bytes.as_slice()).unwrap(); + let rmpv::Value::Map(entries) = &mut value else { + panic!("a v3 document is a named map"); + }; + entries.push((rmpv::Value::from("unexpected_field"), rmpv::Value::from(1))); + let mut out = Vec::new(); + rmpv::encode::write_value(&mut out, &value).unwrap(); + out + } + + #[test] + fn an_unknown_field_is_refused_not_dropped() { + let invocation = invocation(); + let result = TransportResult::transported(&invocation, Vec::new(), receipt(&invocation)); + let TransportDisposition::Transported { receipt: sent, .. } = &result.disposition else { + unreachable!() + }; + let envelope = crate::envelope::encrypt_invocation( + &invocation, + &security("epiphany-distinct-test-key"), + ) + .unwrap(); + assert!(wire_decode::(&wire_encode(&invocation.request).unwrap()).is_ok()); + assert!(wire_decode::(&with_extra_key(&invocation.request)).is_err()); + assert!(wire_decode::(&with_extra_key(&invocation)).is_err()); + assert!(wire_decode::(&with_extra_key(&result)).is_err()); + assert!(wire_decode::(&with_extra_key(&**sent)).is_err()); + assert!( + wire_decode::(&wire_encode(&envelope).unwrap()) + .is_ok() + ); + assert!( + wire_decode::(&with_extra_key(&envelope)).is_err() + ); + } + + #[test] + fn max_output_tokens_is_required() { + let mut request = request(); + assert_eq!(request.validate(), Ok(())); + request.max_output_tokens = 0; + assert_eq!( + request.validate(), + Err(ContractError::Invalid("max_output_tokens")) + ); + } + + #[test] + fn outcomes_are_typed() { + let invocation = invocation(); + for stop_reason in [ + StopReason::EndTurn, + StopReason::ToolUse, + StopReason::MaxOutputTokens, + StopReason::StopSequence, + StopReason::Refusal, + StopReason::ContextWindow, + ] { + let mut sent = receipt(&invocation); + sent.outcome = TransportOutcome::Completed { + provider_response_id: None, + stop_reason, + usage: Usage { + input_tokens: 1, + cache_read_tokens: 2, + cache_write_tokens: 3, + output_tokens: 4, + reasoning_tokens: None, + }, + }; + let decoded: TransportReceipt = wire_decode(&wire_encode(&sent).unwrap()).unwrap(); + assert_eq!(decoded, sent); + } + for kind in [ + FailureKind::ResultSize, + FailureKind::InvalidRequest, + FailureKind::Authentication, + FailureKind::RateLimited, + FailureKind::Unavailable, + FailureKind::Timeout, + FailureKind::Provider, + ] { + let mut sent = receipt(&invocation); + sent.outcome = TransportOutcome::Failed { + kind, + message: "failed".to_string(), + }; + assert_eq!(sent.validate_against(&invocation), Ok(())); + let decoded: TransportReceipt = wire_decode(&wire_encode(&sent).unwrap()).unwrap(); + assert_eq!(decoded, sent); + } + let mut blank = receipt(&invocation); + blank.outcome = TransportOutcome::Failed { + kind: FailureKind::Provider, + message: String::new(), + }; + assert_eq!( + blank.validate_against(&invocation), + Err(ContractError::Invalid("receipt.failure_message")) + ); + } + + #[test] + fn validate_against_refuses_each_identity_field() { + let invocation = invocation(); + let result = TransportResult::transported(&invocation, Vec::new(), receipt(&invocation)); + assert_eq!(result.validate_against(&invocation), Ok(())); + let mut other_request = result.clone(); + other_request.request_id = "request-2".to_string(); + let mut other_caller = result.clone(); + other_caller.caller_runtime_id = "ghostlight-dungeon-yggdrasil".to_string(); + for changed in [other_request, other_caller] { + assert_eq!( + changed.validate_against(&invocation), + Err(ContractError::Identity) + ); + } + + let good = receipt(&invocation); + let mut other_request = good.clone(); + other_request.request_id = "request-2".to_string(); + let mut other_caller = good.clone(); + other_caller.caller_runtime_id = "ghostlight-dungeon-yggdrasil".to_string(); + let mut other_model = good.clone(); + other_model.model = "another-model".to_string(); + for changed in [other_request, other_caller, other_model] { + assert_eq!( + changed.validate_against(&invocation), + Err(ContractError::Identity) + ); + } + } } diff --git a/crates/model-connector/src/envelope.rs b/crates/model-connector/src/envelope.rs index 5400eac..0ece953 100644 --- a/crates/model-connector/src/envelope.rs +++ b/crates/model-connector/src/envelope.rs @@ -2,83 +2,256 @@ use std::io::{Read, Write}; use aes_gcm::aead::{Aead, Payload}; use aes_gcm::{Aes256Gcm, KeyInit, Nonce}; +use base64::Engine; +use base64::engine::general_purpose::STANDARD; +use cultnet_rs::{ + CultNetMessage, CultNetWireContract, decode_cultnet_message_from_slice, + encode_cultnet_message_to_vec, +}; use rand_core::{OsRng, RngCore}; +use schemars::JsonSchema; use serde::{Deserialize, Serialize}; use thiserror::Error; use crate::contract::{ ContractError, ENVELOPE_SCHEMA_ID, TransportInvocation, TransportKey, TransportResult, - canonical_invocation_bytes, require_id, + canonical_invocation_bytes, require_id, wire_decode, wire_encode, }; -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] pub enum EnvelopeKind { Invocation, Result, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] pub struct TransportEnvelope { pub schema_id: String, pub caller_runtime_id: String, pub request_id: String, pub message_kind: EnvelopeKind, pub nonce: [u8; 12], + #[serde(with = "serde_bytes")] + #[schemars(with = "Vec")] pub ciphertext: Vec, } +/// The CultNet service every frame of this connector addresses. +pub const CONNECTOR_SERVICE_ID: &str = "gamecult.model.connector"; +/// The one operation of that service: run one sealed invocation. +pub const INVOKE_OPERATION: &str = "invoke"; + +const PAYLOAD_ENCODING: &str = "messagepack-base64"; +const OK_STATUS: &str = "ok"; + /// The longest caller id, request id or model name a service admits. The /// frame arithmetic below depends on it. A request whose id is longer is /// refused on the wire as malformed. pub const MAX_IDENTITY_BYTES: usize = 256; -/// The most bytes a serialized envelope holding `ciphertext_len` ciphertext -/// bytes can occupy, with ids at their longest: the frame a caller payload -/// bound (`CallerAdmission`'s `max_payload_bytes`, a bound on ciphertext -/// length) needs. The v2 wire serializes the ciphertext as a MessagePack array -/// of integers: a byte of 128 or more takes two bytes, so the worst case is -/// twice the ciphertext, plus the array header and the envelope's other -/// fields at their longest. Measured from the real encoding, not assumed. +fn carriage_bytes(message: &CultNetMessage) -> Result, ServiceError> { + encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0) + .map_err(|_| ServiceError::Carriage) +} + +fn seal_payload(envelope: &TransportEnvelope) -> Result { + Ok(STANDARD.encode(wire_encode(envelope)?)) +} + +/// The CultNet frame an invocation travels in: a `cultnet.operation_request.v0` +/// for [`CONNECTOR_SERVICE_ID`] and [`INVOKE_OPERATION`], its `message_id` the +/// request id and its payload the base64 of the named sealed envelope. +pub(crate) fn encode_invocation_frame( + envelope: &TransportEnvelope, +) -> Result, ServiceError> { + carriage_bytes(&CultNetMessage::OperationRequest { + message_id: envelope.request_id.clone(), + service_id: CONNECTOR_SERVICE_ID.to_string(), + operation: INVOKE_OPERATION.to_string(), + payload_schema: ENVELOPE_SCHEMA_ID.to_string(), + payload_encoding: PAYLOAD_ENCODING.to_string(), + payload: seal_payload(envelope)?, + source_runtime_id: None, + target_runtime_id: None, + }) +} + +/// The CultNet frame a result travels in: a `cultnet.operation_response.v0` +/// with status `ok`, otherwise shaped like the invocation's. +pub(crate) fn encode_result_frame(envelope: &TransportEnvelope) -> Result, ServiceError> { + carriage_bytes(&CultNetMessage::OperationResponse { + message_id: envelope.request_id.clone(), + service_id: CONNECTOR_SERVICE_ID.to_string(), + operation: INVOKE_OPERATION.to_string(), + status: OK_STATUS.to_string(), + payload_schema: ENVELOPE_SCHEMA_ID.to_string(), + payload_encoding: PAYLOAD_ENCODING.to_string(), + payload: seal_payload(envelope)?, + diagnostics: Vec::new(), + source_runtime_id: None, + }) +} + +/// Opens the carriage of one frame. Everything the carriage says is checked +/// here, before any decryption: the service, the operation, the payload schema, +/// and that the message id is the sealed envelope's request id. +fn open_carriage( + message_id: &str, + service_id: &str, + operation: &str, + payload_schema: &str, + payload: &str, +) -> Result { + if service_id != CONNECTOR_SERVICE_ID + || operation != INVOKE_OPERATION + || payload_schema != ENVELOPE_SCHEMA_ID + { + return Err(ServiceError::Carriage); + } + let bytes = STANDARD + .decode(payload) + .map_err(|_| ServiceError::Carriage)?; + let envelope: TransportEnvelope = wire_decode(&bytes).map_err(|_| ServiceError::Encoding)?; + if envelope.request_id != message_id { + return Err(ServiceError::OuterIdentity); + } + Ok(envelope) +} + +fn read_carriage(bytes: &[u8]) -> Result { + decode_cultnet_message_from_slice(bytes, CultNetWireContract::CultNetSchemaV0) + .map_err(|_| ServiceError::Carriage) +} + +pub(crate) fn decode_invocation_frame(bytes: &[u8]) -> Result { + match read_carriage(bytes)? { + CultNetMessage::OperationRequest { + message_id, + service_id, + operation, + payload_schema, + payload, + .. + } => open_carriage( + &message_id, + &service_id, + &operation, + &payload_schema, + &payload, + ), + _ => Err(ServiceError::Carriage), + } +} + +pub(crate) fn decode_result_frame(bytes: &[u8]) -> Result { + match read_carriage(bytes)? { + CultNetMessage::OperationResponse { + message_id, + service_id, + operation, + status, + payload_schema, + payload, + .. + } if status == OK_STATUS => open_carriage( + &message_id, + &service_id, + &operation, + &payload_schema, + &payload, + ), + _ => Err(ServiceError::Carriage), + } +} + +/// The most bytes one frame holding a sealed envelope of `ciphertext_len` +/// ciphertext bytes can occupy, with ids at their longest: the frame a caller +/// payload bound (`CallerAdmission`'s `max_payload_bytes`, a bound on +/// ciphertext length) needs. +/// +/// A frame is a CultNet operation message whose payload is the base64 of the +/// named envelope. The result frame is the larger of the two directions (it +/// also carries a status), so this bound holds for an invocation as well. The +/// envelope shell and the message shell are measured from their real +/// encodings; only the MessagePack length headers and base64 expansion that +/// grow with `ciphertext_len` are arithmetic. pub fn frame_bytes_for_payload(ciphertext_len: usize) -> usize { - let shell = rmp_serde::to_vec(&TransportEnvelope { + let long = "x".repeat(MAX_IDENTITY_BYTES); + let envelope_shell = wire_encode(&TransportEnvelope { schema_id: ENVELOPE_SCHEMA_ID.to_string(), - caller_runtime_id: "x".repeat(MAX_IDENTITY_BYTES), - request_id: "x".repeat(MAX_IDENTITY_BYTES), + caller_runtime_id: long.clone(), + request_id: long.clone(), message_kind: EnvelopeKind::Invocation, nonce: [0xff; 12], ciphertext: Vec::new(), }) .map_or(usize::MAX, |bytes| bytes.len()); - // The shell holds an empty array header of one byte. - let header = match ciphertext_len { - 0..=15 => 1, - 16..=65_535 => 3, + let message_shell = encode_cultnet_message_to_vec( + &CultNetMessage::OperationResponse { + message_id: long, + service_id: CONNECTOR_SERVICE_ID.to_string(), + operation: INVOKE_OPERATION.to_string(), + status: OK_STATUS.to_string(), + payload_schema: ENVELOPE_SCHEMA_ID.to_string(), + payload_encoding: PAYLOAD_ENCODING.to_string(), + payload: "A".to_string(), + diagnostics: Vec::new(), + source_runtime_id: None, + }, + CultNetWireContract::CultNetSchemaV0, + ) + .map_or(usize::MAX, |bytes| bytes.len()); + // The shell holds an empty `bin 8`: a marker and a zero length. The shell's + // one-character payload takes a one-byte string header and the character. + let envelope = envelope_shell + .saturating_sub(2) + .saturating_add(bin_header_bytes(ciphertext_len)) + .saturating_add(ciphertext_len); + let payload = envelope.div_ceil(3).saturating_mul(4); + message_shell + .saturating_sub(2) + .saturating_add(str_header_bytes(payload)) + .saturating_add(payload) +} + +/// MessagePack `bin` header width for a byte string of `len` bytes. +fn bin_header_bytes(len: usize) -> usize { + match len { + 0..=255 => 2, + 256..=65_535 => 3, _ => 5, - }; - shell - .saturating_sub(1) - .saturating_add(header) - .saturating_add(ciphertext_len.saturating_mul(2)) + } +} + +/// MessagePack `str` header width for a string of `len` bytes. +fn str_header_bytes(len: usize) -> usize { + match len { + 0..=31 => 1, + 32..=255 => 2, + 256..=65_535 => 3, + _ => 5, + } } /// The largest payload bound whose worst-case frame fits `max_frame_bytes`: /// the inverse of [`frame_bytes_for_payload`]. `None` when the frame cannot /// carry even an empty envelope. pub fn payload_bytes_for_frame(max_frame_bytes: usize) -> Option { - // frame_bytes_for_payload(0) is the shell plus a one-byte array header. - let fixed = frame_bytes_for_payload(0) - 1; - let room = max_frame_bytes.checked_sub(fixed)?; - // Each array header width bounds the payloads it can describe; the - // widest payload any width admits is the answer. - let by_header = |header: usize, cap: usize| room.checked_sub(header).map(|r| (r / 2).min(cap)); - [ - by_header(5, usize::MAX), - by_header(3, 65_535), - by_header(1, 15), - ] - .into_iter() - .flatten() - .max() + if frame_bytes_for_payload(0) > max_frame_bytes { + return None; + } + // A frame is longer than its payload, so the answer is below the bound. + let (mut fits, mut over) = (0, max_frame_bytes.saturating_add(1)); + while over - fits > 1 { + let middle = fits + (over - fits) / 2; + if frame_bytes_for_payload(middle) <= max_frame_bytes { + fits = middle; + } else { + over = middle; + } + } + Some(fits) } pub fn encrypt_invocation( @@ -102,7 +275,7 @@ pub fn decrypt_invocation( validate_envelope(envelope, EnvelopeKind::Invocation)?; let plaintext = decrypt_message(envelope, security)?; let invocation: TransportInvocation = - rmp_serde::from_slice(&plaintext).map_err(|_| ServiceError::Encoding)?; + wire_decode(&plaintext).map_err(|_| ServiceError::Encoding)?; if invocation.caller_runtime_id != envelope.caller_runtime_id || invocation.request_id() != envelope.request_id { @@ -115,7 +288,7 @@ pub fn encrypt_result( result: &TransportResult, security: &TransportKey, ) -> Result { - let plaintext = rmp_serde::to_vec(result).map_err(|_| ServiceError::Encoding)?; + let plaintext = wire_encode(result).map_err(|_| ServiceError::Encoding)?; encrypt_message( &result.caller_runtime_id, &result.request_id, @@ -132,8 +305,7 @@ pub fn decrypt_result( ) -> Result { validate_envelope(envelope, EnvelopeKind::Result)?; let plaintext = decrypt_message(envelope, security)?; - let result: TransportResult = - rmp_serde::from_slice(&plaintext).map_err(|_| ServiceError::Encoding)?; + let result: TransportResult = wire_decode(&plaintext).map_err(|_| ServiceError::Encoding)?; if result.caller_runtime_id != envelope.caller_runtime_id || result.request_id != envelope.request_id { @@ -242,13 +414,15 @@ pub(crate) fn decrypt_message( .map_err(|_| ServiceError::Encryption) } +/// The authenticated data of a sealed message: a positional tuple, a MAC input +/// and not a wire document (a tuple is positional under the named codec too). pub(crate) fn envelope_aad( schema_id: &str, caller_runtime_id: &str, request_id: &str, message_kind: EnvelopeKind, ) -> Result, ServiceError> { - rmp_serde::to_vec(&(schema_id, caller_runtime_id, request_id, message_kind)) + wire_encode(&(schema_id, caller_runtime_id, request_id, message_kind)) .map_err(|_| ServiceError::Encoding) } @@ -291,8 +465,10 @@ pub enum ServiceError { OuterIdentity, #[error("transport encryption or authentication failed")] Encryption, - #[error("transport MessagePack encoding failed")] + #[error("transport encoding failed")] Encoding, + #[error("frame is not a connector operation message")] + Carriage, #[error("completion did not carry a transported provider outcome")] InvalidCompletion, #[error("completion has no active request")] @@ -373,7 +549,7 @@ mod tests { let key = security("epiphany-distinct-test-key"); let invocation = invocation(); let result = TransportResult::transported(&invocation, Vec::new(), receipt(&invocation)); - let plaintext = rmp_serde::to_vec(&result).unwrap(); + let plaintext = wire_encode(&result).unwrap(); let seal = |caller: &str, request_id: &str| { encrypt_message(caller, request_id, EnvelopeKind::Result, &plaintext, &key).unwrap() }; @@ -401,7 +577,7 @@ mod tests { let good = encrypt_invocation(&invocation, &key).unwrap(); let mut schema = good.clone(); - schema.schema_id = "gamecult.codex.transport_envelope.v1".to_string(); + schema.schema_id = concat!("gamecult.", "codex.transport_envelope.v2").to_string(); let wrong_kind = sealed_invocation("epiphany-yggdrasil", "request-1", EnvelopeKind::Result); let mut empty = good.clone(); empty.ciphertext.clear(); @@ -452,20 +628,44 @@ mod tests { assert!(matches!(write(&[], 1_000), Err(TransportFrameError::Size))); } - #[test] - fn the_worst_frame_is_the_real_encoding_of_an_all_high_ciphertext() { + /// The real result frame for an envelope with ids at the longest admitted + /// and a ciphertext of `length` bytes. + fn real_worst_frame(length: usize) -> Vec { let long = "x".repeat(MAX_IDENTITY_BYTES); - for length in [0, 1, 15, 16, 17, 255, 256, 65_535, 65_536, 70_000] { - let envelope = TransportEnvelope { - schema_id: ENVELOPE_SCHEMA_ID.to_string(), - caller_runtime_id: long.clone(), - request_id: long.clone(), - message_kind: EnvelopeKind::Invocation, - nonce: [0xff; 12], - ciphertext: vec![0xff; length], - }; + encode_result_frame(&TransportEnvelope { + schema_id: ENVELOPE_SCHEMA_ID.to_string(), + caller_runtime_id: long.clone(), + request_id: long, + message_kind: EnvelopeKind::Invocation, + nonce: [0xff; 12], + ciphertext: vec![0xff; length], + }) + .unwrap() + } + + /// Ciphertext sizes either side of every length-header boundary the frame + /// crosses: bin 8 to 16 to 32, str 8 to 16 to 32, and each base64 padding. + fn boundary_sizes() -> Vec { + let mut sizes: Vec = (0..=40).collect(); + for center in [ + 100, 200, 250, 256, 300, 1000, 3000, 65_535, 65_536, 49_000, 49_100, 49_152, 49_200, + 70_000, + ] { + for delta in 0..=40 { + sizes.push(center + delta); + sizes.push(center.saturating_sub(delta)); + } + } + sizes.sort_unstable(); + sizes.dedup(); + sizes + } + + #[test] + fn the_frame_bound_is_the_real_carriage_encoding() { + for length in boundary_sizes() { assert_eq!( - rmp_serde::to_vec(&envelope).unwrap().len(), + real_worst_frame(length).len(), frame_bytes_for_payload(length), "{length}" ); @@ -480,16 +680,18 @@ mod tests { let sealed = encrypt_message(&id, &id, EnvelopeKind::Result, &vec![0x41; plaintext], &key) .unwrap(); - let real = rmp_serde::to_vec(&sealed).unwrap().len(); let worst = frame_bytes_for_payload(sealed.ciphertext.len()); - assert!(real <= worst, "{real} > {worst}"); - // Real ciphertext is random, so about half its bytes take two - // bytes: the worst case is not wildly above what is written. + for real in [ + encode_result_frame(&sealed).unwrap().len(), + encode_invocation_frame(&sealed).unwrap().len(), + ] { + assert!(real <= worst, "{real} > {worst}"); + } + // The frame is the base64 of the envelope: its worst case is + // within the base64 expansion of the real one. if id.len() == MAX_IDENTITY_BYTES { - assert!( - worst <= real + sealed.ciphertext.len() + 64, - "{worst} {real}" - ); + let real = encode_result_frame(&sealed).unwrap().len(); + assert!(worst <= real + 8, "{worst} {real}"); } } } @@ -502,7 +704,7 @@ mod tests { None ); assert_eq!(payload_bytes_for_frame(frame_bytes_for_payload(0)), Some(0)); - for bytes in [1, 15, 16, 4096, 65_535, 65_536, 100_000, 2_000_000] { + for bytes in boundary_sizes().into_iter().chain([2_000_000]) { let frame = frame_bytes_for_payload(bytes); assert_eq!(payload_bytes_for_frame(frame), Some(bytes), "{bytes}"); assert!( @@ -511,4 +713,199 @@ mod tests { ); } } + + #[test] + fn ciphertext_travels_as_binary() { + let key = security("epiphany-distinct-test-key"); + for plaintext in [10, 300, 70_000] { + let sealed = encrypt_message( + "caller", + "request", + EnvelopeKind::Invocation, + &vec![0x41; plaintext], + &key, + ) + .unwrap(); + let length = sealed.ciphertext.len(); + let mut expected = match length { + 0..=255 => vec![0xc4, length as u8], + 256..=65_535 => [&[0xc5][..], &(length as u16).to_be_bytes()].concat(), + _ => [&[0xc6][..], &(length as u32).to_be_bytes()].concat(), + }; + expected.extend(&sealed.ciphertext); + let encoded = wire_encode(&sealed).unwrap(); + assert!( + encoded + .windows(expected.len()) + .any(|window| window == expected), + "{plaintext}" + ); + } + } + + fn sealed_for_carriage() -> TransportEnvelope { + encrypt_invocation(&invocation(), &security("epiphany-distinct-test-key")).unwrap() + } + + fn request_message(edit: impl FnOnce(&mut CultNetMessage)) -> Vec { + let envelope = sealed_for_carriage(); + let mut message = decode_cultnet_message_from_slice( + &encode_invocation_frame(&envelope).unwrap(), + CultNetWireContract::CultNetSchemaV0, + ) + .unwrap(); + edit(&mut message); + carriage_bytes(&message).unwrap() + } + + fn response_message(edit: impl FnOnce(&mut CultNetMessage)) -> Vec { + let envelope = sealed_for_carriage(); + let mut message = decode_cultnet_message_from_slice( + &encode_result_frame(&envelope).unwrap(), + CultNetWireContract::CultNetSchemaV0, + ) + .unwrap(); + edit(&mut message); + carriage_bytes(&message).unwrap() + } + + #[test] + fn the_carriage_round_trips_both_directions() { + let envelope = sealed_for_carriage(); + assert_eq!( + decode_invocation_frame(&encode_invocation_frame(&envelope).unwrap()).unwrap(), + envelope + ); + assert_eq!( + decode_result_frame(&encode_result_frame(&envelope).unwrap()).unwrap(), + envelope + ); + // A request is not a response, in either direction. + assert_eq!( + decode_result_frame(&encode_invocation_frame(&envelope).unwrap()), + Err(ServiceError::Carriage) + ); + assert_eq!( + decode_invocation_frame(&encode_result_frame(&envelope).unwrap()), + Err(ServiceError::Carriage) + ); + } + + #[test] + fn the_carriage_refuses_another_service_operation_or_schema() { + fn edit_request(field: &str, value: &str) -> Vec { + request_message(|message| { + if let CultNetMessage::OperationRequest { + service_id, + operation, + payload_schema, + .. + } = message + { + match field { + "service" => *service_id = value.to_string(), + "operation" => *operation = value.to_string(), + _ => *payload_schema = value.to_string(), + } + } + }) + } + fn edit_response(field: &str, value: &str) -> Vec { + response_message(|message| { + if let CultNetMessage::OperationResponse { + service_id, + operation, + payload_schema, + status, + .. + } = message + { + match field { + "service" => *service_id = value.to_string(), + "operation" => *operation = value.to_string(), + "status" => *status = value.to_string(), + _ => *payload_schema = value.to_string(), + } + } + }) + } + for (field, value) in [ + ("service", "gamecult.model.other"), + ("operation", "cancel"), + ("schema", "gamecult.model.transport_envelope.v2"), + ] { + assert_eq!( + decode_invocation_frame(&edit_request(field, value)), + Err(ServiceError::Carriage), + "{field}" + ); + assert_eq!( + decode_result_frame(&edit_response(field, value)), + Err(ServiceError::Carriage), + "{field}" + ); + } + assert_eq!( + decode_result_frame(&edit_response("status", "rejected")), + Err(ServiceError::Carriage) + ); + // The unedited message decodes, so each refusal above is the edit's. + assert!(decode_invocation_frame(&edit_request("service", CONNECTOR_SERVICE_ID)).is_ok()); + assert!(decode_result_frame(&edit_response("status", "ok")).is_ok()); + } + + #[test] + fn the_carriage_refuses_a_message_id_that_is_not_the_envelopes_request_id() { + let request = request_message(|message| { + if let CultNetMessage::OperationRequest { message_id, .. } = message { + *message_id = "another-request".to_string(); + } + }); + let response = response_message(|message| { + if let CultNetMessage::OperationResponse { message_id, .. } = message { + *message_id = "another-request".to_string(); + } + }); + assert_eq!( + decode_invocation_frame(&request), + Err(ServiceError::OuterIdentity) + ); + assert_eq!( + decode_result_frame(&response), + Err(ServiceError::OuterIdentity) + ); + } + + #[test] + fn the_carriage_refuses_what_is_not_a_connector_message() { + let good = encode_invocation_frame(&sealed_for_carriage()).unwrap(); + // Not MessagePack of any CultNet message, nor a truncated one. + assert_eq!( + decode_invocation_frame(&[0xc1, 0x00, 0x01]), + Err(ServiceError::Carriage) + ); + assert_eq!( + decode_invocation_frame(&good[..good.len() / 2]), + Err(ServiceError::Carriage) + ); + // A payload that is not base64, and base64 of something else. + let not_base64 = request_message(|message| { + if let CultNetMessage::OperationRequest { payload, .. } = message { + *payload = "!!not base64!!".to_string(); + } + }); + assert_eq!( + decode_invocation_frame(¬_base64), + Err(ServiceError::Carriage) + ); + let not_an_envelope = request_message(|message| { + if let CultNetMessage::OperationRequest { payload, .. } = message { + *payload = STANDARD.encode(b"\x01\x02"); + } + }); + assert_eq!( + decode_invocation_frame(¬_an_envelope), + Err(ServiceError::Encoding) + ); + } } diff --git a/crates/model-connector/src/serve.rs b/crates/model-connector/src/serve.rs index 7ca0a3a..cef686a 100644 --- a/crates/model-connector/src/serve.rs +++ b/crates/model-connector/src/serve.rs @@ -12,8 +12,8 @@ use zeroize::Zeroizing; use crate::contract::{TransportInvocation, TransportResult}; use crate::envelope::{ - ServiceError, TransportEnvelope, TransportFrameError, frame_bytes_for_payload, - read_transport_frame, write_transport_frame, + ServiceError, TransportFrameError, decode_invocation_frame, encode_result_frame, + frame_bytes_for_payload, read_transport_frame, write_transport_frame, }; use crate::service::{CallerAdmission, TransportAdmission, TransportService}; @@ -222,8 +222,7 @@ fn execute_inference_frame( backend: &dyn ProviderBackend, limits: &ServeLimits, ) -> Result, ServeError> { - let envelope: TransportEnvelope = - rmp_serde::from_slice(request).map_err(|_| ServeError::FrameEncoding)?; + let envelope = decode_invocation_frame(request).map_err(|_| ServeError::FrameEncoding)?; let admission = service .lock() .map_err(|_| ServeError::ServicePoisoned)? @@ -238,7 +237,7 @@ fn execute_inference_frame( .complete(*claim, result, limits.max_frame_bytes)? } }; - rmp_serde::to_vec(&response).map_err(|_| ServeError::FrameEncoding) + encode_result_frame(&response).map_err(|_| ServeError::FrameEncoding) } fn frame_error(error: TransportFrameError) -> ServeError { @@ -314,7 +313,7 @@ pub enum ServeError { Connection(#[source] std::io::Error), #[error("connector frame exceeded its bound")] FrameSize, - #[error("connector frame was not valid MessagePack")] + #[error("connector frame was not a connector operation message")] FrameEncoding, #[error("system clock is before the Unix epoch")] Clock, @@ -506,7 +505,7 @@ mod tests { let backend = CountingBackend::new(true); let (_client, endpoint) = start(backend.clone(), &store); let security = crate::fixtures::security(KEY); - let request = rmp_serde::to_vec( + let request = crate::envelope::encode_invocation_frame( &crate::envelope::encrypt_invocation(&fresh("request-wire"), &security).unwrap(), ) .unwrap(); @@ -531,14 +530,14 @@ mod tests { // requires the result to name the request by its own id. let send = |id: &str| { let invocation = fresh(id); - let request = rmp_serde::to_vec( + let request = crate::envelope::encode_invocation_frame( &crate::envelope::encrypt_invocation(&invocation, &security).unwrap(), ) .unwrap(); let mut stream = TcpStream::connect(endpoint).unwrap(); write_transport_frame(&mut stream, &request, FRAME).unwrap(); let reply = read_transport_frame(&mut stream, FRAME).unwrap(); - let envelope: TransportEnvelope = rmp_serde::from_slice(&reply).unwrap(); + let envelope = crate::envelope::decode_result_frame(&reply).unwrap(); crate::envelope::decrypt_result(&envelope, &security, &invocation).unwrap() }; let served = send(&"i".repeat(crate::envelope::MAX_IDENTITY_BYTES)); @@ -684,7 +683,7 @@ mod tests { #[test] fn the_smallest_frame_also_carries_the_failure_that_stands_in_for_a_result() { let long = "x".repeat(crate::envelope::MAX_IDENTITY_BYTES); - let plaintext = rmp_serde::to_vec(&crate::service::result_size_failure( + let plaintext = crate::contract::wire_encode(&crate::service::result_size_failure( &long, &long, &long, [0xff; 32], [0xff; 32], )) .unwrap() @@ -963,7 +962,7 @@ mod tests { } fn encoded_len(result: &TransportResult) -> usize { - rmp_serde::to_vec(result).unwrap().len() + crate::contract::wire_encode(result).unwrap().len() } /// Two calls with one request id, through a live serve loop. @@ -1154,4 +1153,92 @@ mod tests { client.execute(&fresh("request-key")).unwrap(); assert_eq!(backend.executions.load(Ordering::SeqCst), 1); } + + #[test] + fn a_v2_frame_gets_no_answer() { + let store = TestStore::new(); + let backend = CountingBackend::new(true); + let (client, endpoint) = start(backend.clone(), &store); + // The v2 envelope: positional, an integer-array ciphertext, a v2 id. + let v2 = rmp_serde::to_vec(&( + concat!("gamecult.", "codex.transport_envelope.v2"), + CALLER, + "request-v2", + "Invocation", + [0_u8; 12], + vec![1_u8; 64], + )) + .unwrap(); + let mut stream = TcpStream::connect(endpoint).unwrap(); + write_transport_frame(&mut stream, &v2, FRAME).unwrap(); + assert!(matches!( + read_transport_frame(&mut stream, FRAME), + Err(TransportFrameError::Connection(_)) + )); + assert_eq!(backend.executions.load(Ordering::SeqCst), 0); + // The loop survives the v2 peer and serves a v3 one. + client.execute(&fresh("request-after-v2")).unwrap(); + assert_eq!(backend.executions.load(Ordering::SeqCst), 1); + } + + #[test] + fn a_reply_echoes_the_request_id() { + use cultnet_rs::{CultNetMessage, CultNetWireContract, decode_cultnet_message_from_slice}; + let store = TestStore::new(); + let backend = CountingBackend::new(true); + let (_client, endpoint) = start(backend.clone(), &store); + let security = crate::fixtures::security(KEY); + let send = |invocation: &TransportInvocation| { + let request = crate::envelope::encode_invocation_frame( + &crate::envelope::encrypt_invocation(invocation, &security).unwrap(), + ) + .unwrap(); + let mut stream = TcpStream::connect(endpoint).unwrap(); + write_transport_frame(&mut stream, &request, FRAME).unwrap(); + let reply = read_transport_frame(&mut stream, FRAME).unwrap(); + let CultNetMessage::OperationResponse { + message_id, status, .. + } = decode_cultnet_message_from_slice(&reply, CultNetWireContract::CultNetSchemaV0) + .unwrap() + else { + panic!("a reply is an operation response"); + }; + let envelope = crate::envelope::decode_result_frame(&reply).unwrap(); + let result = crate::envelope::decrypt_result(&envelope, &security, invocation).unwrap(); + (message_id, status, result) + }; + + let transported = fresh("request-echo-transported"); + let (message_id, status, result) = send(&transported); + assert_eq!( + ( + message_id.as_str(), + status.as_str(), + result.request_id.as_str() + ), + ("request-echo-transported", "ok", "request-echo-transported") + ); + assert!(matches!( + result.disposition, + TransportDisposition::Transported { .. } + )); + + let mut refused = fresh("request-echo-refused"); + refused.request.model = "unlisted-model".to_string(); + refused.provider_request_sha256 = + crate::contract::provider_request_sha256(&refused.request).unwrap(); + let (message_id, status, result) = send(&refused); + assert_eq!( + ( + message_id.as_str(), + status.as_str(), + result.request_id.as_str() + ), + ("request-echo-refused", "ok", "request-echo-refused") + ); + assert_eq!( + result.disposition, + TransportDisposition::Refused(Refusal::ModelNotAllowed) + ); + } } diff --git a/crates/model-connector/src/service.rs b/crates/model-connector/src/service.rs index 44c3fe1..75ac66f 100644 --- a/crates/model-connector/src/service.rs +++ b/crates/model-connector/src/service.rs @@ -13,7 +13,7 @@ use crate::contract::{ }; use crate::envelope::{ EnvelopeKind, MAX_IDENTITY_BYTES, ServiceError, TransportEnvelope, decrypt_invocation, - encrypt_result, validate_envelope, + encode_result_frame, encrypt_result, validate_envelope, }; pub struct CallerAdmission { @@ -90,7 +90,7 @@ enum ReplayState { } #[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] -#[cultcache(type = "gamecult.codex.replay_record.v1")] +#[cultcache(type = "gamecult.model.replay_record.v1")] struct ReplayRecord { #[cultcache(key = 0)] caller_runtime_id: String, @@ -313,8 +313,8 @@ impl TransportService { }))) } - /// Records a result and seals it. `max_frame_bytes` bounds the serialized - /// envelope, which is what a frame carries: a result whose sealed envelope + /// Records a result and seals it. `max_frame_bytes` bounds the frame the + /// sealed envelope travels in (its carriage): a result whose sealed envelope /// does not fit is recorded as the typed `result_size` failure instead, so /// no completed record is ever undeliverable and a retry replays bytes /// that fit. @@ -424,9 +424,7 @@ impl TransportService { } fn fits_frame(envelope: &TransportEnvelope, max_frame_bytes: usize) -> Result { - rmp_serde::to_vec(envelope) - .map(|bytes| bytes.len() <= max_frame_bytes) - .map_err(|_| ServiceError::Encoding) + encode_result_frame(envelope).map(|bytes| bytes.len() <= max_frame_bytes) } /// The provider named on the stand-in failure, fixed so its size is bounded. @@ -761,7 +759,7 @@ mod tests { &invocation.caller_runtime_id, invocation.request_id(), EnvelopeKind::Invocation, - &rmp_serde::to_vec(invocation).unwrap(), + &crate::contract::wire_encode(invocation).unwrap(), &security(KEY), ) .unwrap() @@ -1007,7 +1005,8 @@ mod tests { }; assert_eq!(reply.request_id, id, "{length}"); let result: TransportResult = - rmp_serde::from_slice(&decrypt_message(&reply, &security(KEY)).unwrap()).unwrap(); + crate::contract::wire_decode(&decrypt_message(&reply, &security(KEY)).unwrap()) + .unwrap(); assert_eq!( result.disposition, TransportDisposition::Refused(Refusal::Malformed) @@ -1053,7 +1052,7 @@ mod tests { fn a_frame_check_is_inclusive_of_the_serialized_envelope() { let envelope = encrypt_invocation(&invocation_for(CALLER, "sized"), &security(KEY)).unwrap(); - let size = rmp_serde::to_vec(&envelope).unwrap().len(); + let size = encode_result_frame(&envelope).unwrap().len(); assert_eq!(fits_frame(&envelope, size), Ok(true)); assert_eq!(fits_frame(&envelope, size - 1), Ok(false)); } @@ -1098,7 +1097,7 @@ mod tests { let (claim, invocation) = claim_for(&mut service, "stand-in"); let big = text_result(&invocation, &"y".repeat(4000), &"t".repeat(5000)); let response = service.complete(*claim, big, 2000).unwrap(); - assert!(rmp_serde::to_vec(&response).unwrap().len() <= 2000); + assert!(encode_result_frame(&response).unwrap().len() <= 2000); let stand_in = decrypt_result(&response, &security(KEY), &invocation).unwrap(); assert_eq!( stand_in, @@ -1172,7 +1171,7 @@ mod tests { let (claim, invocation) = claim_for(&mut service, "restart"); let big = text_result(&invocation, &"y".repeat(4000), "fixture-sse"); let recorded = service.complete(*claim, big, FRAME).unwrap(); - assert!(rmp_serde::to_vec(&recorded).unwrap().len() > 2000); + assert!(encode_result_frame(&recorded).unwrap().len() > 2000); drop(service); let mut service = open_service(&store, 8); @@ -1192,7 +1191,7 @@ mod tests { else { panic!("expected a replay"); }; - assert!(rmp_serde::to_vec(&smaller).unwrap().len() <= 2000); + assert!(encode_result_frame(&smaller).unwrap().len() <= 2000); assert_eq!( decrypt_result(&smaller, &security(KEY), &invocation).unwrap(), result_size_failure( @@ -1251,4 +1250,25 @@ mod tests { TransportDisposition::Refused(Refusal::Indeterminate) ); } + + #[test] + fn contract_refusal_maps_each_error() { + assert_eq!(contract_refusal(&ContractError::Expiry), Refusal::Expired); + assert_eq!( + contract_refusal(&ContractError::Identity), + Refusal::IdentitySubstitution + ); + assert_eq!( + contract_refusal(&ContractError::ProviderDigest), + Refusal::ProviderDigestSubstitution + ); + for other in [ + ContractError::Schema, + ContractError::Invalid("model"), + ContractError::NativeDigest, + ContractError::EventSequence, + ] { + assert_eq!(contract_refusal(&other), Refusal::Malformed, "{other:?}"); + } + } } diff --git a/crates/model-connector/tests/fixtures/wire_v3/README.md b/crates/model-connector/tests/fixtures/wire_v3/README.md new file mode 100644 index 0000000..918df33 --- /dev/null +++ b/crates/model-connector/tests/fixtures/wire_v3/README.md @@ -0,0 +1,15 @@ +# wire_v3 fixtures + +The named-MessagePack bytes of the v3 documents and one sealed carriage frame, +produced by the kit at the revision that added them. They pin drift; the types in +`src/contract.rs` own the shape, and `schemas/` publishes it. + +- `provider_request.mp`, `transport_invocation.mp`, `transport_result_transported.mp`, + `transport_result_refused.mp`, `transport_receipt_failed.mp`, `transport_envelope.mp`: + the plaintext documents. +- `carriage_invocation.frame`: a `cultnet.operation_request.v0` for service + `gamecult.model.connector`, operation `invoke`, whose payload is the base64 of an envelope + sealed under the fixed test key in `tests/wire_golden_v3.rs`. + +Generator: `tests/wire_golden_v3.rs` itself. Run it with `WIRE_V3_REGENERATE=1` to rewrite +the files from the kit, then review the diff: a changed byte is a wire change. diff --git a/crates/model-connector/tests/fixtures/wire_v3/carriage_invocation.frame b/crates/model-connector/tests/fixtures/wire_v3/carriage_invocation.frame new file mode 100644 index 0000000..6051b64 --- /dev/null +++ b/crates/model-connector/tests/fixtures/wire_v3/carriage_invocation.frame @@ -0,0 +1 @@ +‡©messageId®golden-request©operation¦invoke§payloadÚ$hqlzY2hlbWFfaWTZJGdhbWVjdWx0Lm1vZGVsLnRyYW5zcG9ydF9lbnZlbG9wZS52M7FjYWxsZXJfcnVudGltZV9pZK1nb2xkZW4tY2FsbGVyqnJlcXVlc3RfaWSuZ29sZGVuLXJlcXVlc3SsbWVzc2FnZV9raW5kqkludm9jYXRpb26lbm9uY2WcS2XMwjgRzMrM9QTM037MsczmqmNpcGhlcnRleHTFAy8eYmQGV5JCOLETwqQE+wIz/6yyPltHTPywbakMD/mgkJUxAvKWLNljXa4RAVVITAb3GxklnGF8zoOX/9qGSM9r0JZ126s+0LqvsRNhllG1S0pfG49fQJy99pevHReOSvX/cNWs2cHppYNdJUbVVAkDuioEXwtEs/vRCsmD/ab0bKUX0XRzrNbmjMU4+dHUCus8yt+iNTK9GdjXbEQAweeKNehirXEgnngLnwylo/MDoI4G237du3PmwhMzyIHztoIbwD0Bv9x7WqY+jj90vqEBsOHl+FmYc+F8+OiP6NKTDNRHypUtdo32G/IgfMfbsP/XHiNpPHtLxDqZVIcYdYqLAvhDfLBReLzQVLuNJnE0a2P8xLVAkfbmQZNsxRqS14RjYVtRoDcacxH1h0afo9OWMSK8Sp8eEBAhAJ7ovhCMklc7xWB3o4ny4cmDSExx5zWRLdI14rC1CZ7nq8k3V8fIIHctSp6am35A5yyiTwDH0m3h7/hvXrzhOqtAwHd0jkfNv+qOXZCEPdtbzOXOz0fMVZwvTAJWhoEYmhfKcGvXu8SmEtZcZe0WjF4xpFIFCFsSxh8HcrDFvxKDa58bnimo6DE2VeEMolpaBvnIGEst6qNZ9VQQnQIpCGbbtFd7jMrncnfcq+36NI10qW30r8sqAqgORp8cAGP5fy2X9SQDmH13SXgSfIdr7uxAHhK0X1r0iopJn4XtVfBiTd+IfoLFN+KO/JTZvLc48e6Jl9QRnxkL/UCgT/5wENd9VRxP+oO1CXG3DthI882rVbZDkhhwxupM9J/EURONajVyVSkeju+r9vzHCgR5odRW8cwqY1RXoq6L6mhj4ph01MBkUugz8MFwBsem+8QpnGatyVbZB99knWjK/01o5J6SZihcXNowRdon6BTydFByJaV5bRvgIK7IGRct3BmC8xidyxN9p7QAZYMPscxhQ0hAM615w76Cy82kx6eAzREBQX3COKApXSlPA/xlOKa/iUUMWHvFKyf/n0KrOlVvHjoDzPflHi3hpxxWcQGt8CxxNdSQMufeqce/5/BCeEwRIJ99+k7CEqIqpFIs7cAsI1GpF+NO5w==¯payloadEncoding²messagepack-base64­payloadSchemaÙ$gamecult.model.transport_envelope.v3­schemaVersion¼cultnet.operation_request.v0©serviceId¸gamecult.model.connector \ No newline at end of file diff --git a/crates/model-connector/tests/fixtures/wire_v3/provider_request.mp b/crates/model-connector/tests/fixtures/wire_v3/provider_request.mp new file mode 100644 index 0000000..9ff75e4 Binary files /dev/null and b/crates/model-connector/tests/fixtures/wire_v3/provider_request.mp differ diff --git a/crates/model-connector/tests/fixtures/wire_v3/transport_envelope.mp b/crates/model-connector/tests/fixtures/wire_v3/transport_envelope.mp new file mode 100644 index 0000000..cc0d7fe Binary files /dev/null and b/crates/model-connector/tests/fixtures/wire_v3/transport_envelope.mp differ diff --git a/crates/model-connector/tests/fixtures/wire_v3/transport_invocation.mp b/crates/model-connector/tests/fixtures/wire_v3/transport_invocation.mp new file mode 100644 index 0000000..9544786 Binary files /dev/null and b/crates/model-connector/tests/fixtures/wire_v3/transport_invocation.mp differ diff --git a/crates/model-connector/tests/fixtures/wire_v3/transport_receipt_failed.mp b/crates/model-connector/tests/fixtures/wire_v3/transport_receipt_failed.mp new file mode 100644 index 0000000..4e0f1b1 Binary files /dev/null and b/crates/model-connector/tests/fixtures/wire_v3/transport_receipt_failed.mp differ diff --git a/crates/model-connector/tests/fixtures/wire_v3/transport_result_refused.mp b/crates/model-connector/tests/fixtures/wire_v3/transport_result_refused.mp new file mode 100644 index 0000000..0c4f7b1 --- /dev/null +++ b/crates/model-connector/tests/fixtures/wire_v3/transport_result_refused.mp @@ -0,0 +1 @@ +„©schema_idÙ"gamecult.model.transport_result.v3ªrequest_id®golden-request±caller_runtime_id­golden-caller«disposition§Refused«OutputBound \ No newline at end of file diff --git a/crates/model-connector/tests/fixtures/wire_v3/transport_result_transported.mp b/crates/model-connector/tests/fixtures/wire_v3/transport_result_transported.mp new file mode 100644 index 0000000..1922073 Binary files /dev/null and b/crates/model-connector/tests/fixtures/wire_v3/transport_result_transported.mp differ diff --git a/crates/model-connector/tests/published_schemas.rs b/crates/model-connector/tests/published_schemas.rs new file mode 100644 index 0000000..1dabe18 --- /dev/null +++ b/crates/model-connector/tests/published_schemas.rs @@ -0,0 +1,66 @@ +//! `schemas/` at the repository root publishes one JSON Schema per v3 document, +//! generated from the types. This test regenerates them and fails on any +//! difference; run it with `MODEL_CONNECTOR_REGENERATE_SCHEMAS=1` to rewrite the +//! files from the types. + +use std::path::{Path, PathBuf}; + +use model_connector::*; +use schemars::{JsonSchema, schema_for}; + +const REGENERATE: &str = "MODEL_CONNECTOR_REGENERATE_SCHEMAS"; + +fn render() -> String { + let mut text = serde_json::to_string_pretty(&schema_for!(T)).unwrap(); + text.push('\n'); + text +} + +fn schemas_dir() -> PathBuf { + Path::new(env!("CARGO_MANIFEST_DIR")).join("../../schemas") +} + +#[test] +fn published_schemas_are_the_types() { + let documents = [ + ("provider_request", render::()), + ("transport_invocation", render::()), + ("transport_result", render::()), + ("transport_receipt", render::()), + ("transport_envelope", render::()), + ]; + let dir = schemas_dir(); + if std::env::var_os(REGENERATE).is_some() { + std::fs::create_dir_all(&dir).unwrap(); + for (name, text) in &documents { + std::fs::write(dir.join(format!("{name}.json")), text).unwrap(); + } + } + for (name, text) in &documents { + let committed = std::fs::read_to_string(dir.join(format!("{name}.json"))) + .unwrap_or_else(|_| panic!("schemas/{name}.json is missing")); + assert_eq!(&committed, text, "schemas/{name}.json is not the type's"); + } + let mut on_disk: Vec = std::fs::read_dir(&dir) + .unwrap() + .map(|entry| entry.unwrap().file_name().into_string().unwrap()) + .collect(); + on_disk.sort(); + let mut expected: Vec = documents + .iter() + .map(|(name, _)| format!("{name}.json")) + .collect(); + expected.sort(); + assert_eq!( + on_disk, expected, + "schemas/ holds exactly the five documents" + ); +} + +#[test] +fn a_published_schema_refuses_an_unknown_field() { + let schema = serde_json::to_value(schema_for!(ProviderRequest)).unwrap(); + assert_eq!(schema["additionalProperties"], serde_json::json!(false)); + let required = schema["required"].as_array().unwrap(); + assert!(required.contains(&serde_json::json!("max_output_tokens"))); +} diff --git a/crates/model-connector/tests/wire_golden_v3.rs b/crates/model-connector/tests/wire_golden_v3.rs new file mode 100644 index 0000000..5f91513 --- /dev/null +++ b/crates/model-connector/tests/wire_golden_v3.rs @@ -0,0 +1,214 @@ +//! Pins the bytes of the v3 wire so a drift is a failing test. The fixtures in +//! `fixtures/wire_v3` are produced by this file (see its README); they pin the +//! kit against itself and are not a second contract: the types in +//! `model_connector::contract` own the shape. + +use std::path::{Path, PathBuf}; + +use base64::Engine; +use base64::engine::general_purpose::STANDARD; +use cultnet_rs::{ + CultNetMessage, CultNetWireContract, decode_cultnet_message_from_slice, + encode_cultnet_message_to_vec, +}; +use model_connector::*; +use serde::Serialize; +use serde::de::DeserializeOwned; + +const KEY: &str = "golden-distinct-test-key"; +const CALLER: &str = "golden-caller"; +const REGENERATE: &str = "WIRE_V3_REGENERATE"; + +fn fixtures() -> PathBuf { + Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/wire_v3") +} + +fn named(value: &T) -> Vec { + rmp_serde::to_vec_named(value).unwrap() +} + +/// The value serializes to the committed bytes and the committed bytes decode +/// back to it. With `WIRE_V3_REGENERATE` set, the committed bytes are rewritten +/// from the value first. +fn assert_wire( + name: &str, + value: &T, +) { + let path = fixtures().join(name); + if std::env::var_os(REGENERATE).is_some() { + std::fs::create_dir_all(fixtures()).unwrap(); + std::fs::write(&path, named(value)).unwrap(); + } + let golden = std::fs::read(&path).unwrap_or_else(|_| panic!("{name}: fixture is missing")); + assert_eq!(named(value), golden, "{name}: serialized bytes"); + let decoded: T = rmp_serde::from_slice(&golden).unwrap(); + assert_eq!(&decoded, value, "{name}: decoded value"); +} + +fn request() -> ProviderRequest { + let mut request = ProviderRequest::new( + "golden-request", + "golden-conversation", + "claude-opus-4-5", + "Return the typed result.", + 8_192, + ); + request.input = vec![ + InputItem::UserText { + text: "Projected state".to_string(), + }, + InputItem::ToolCall { + call_id: "call-1".to_string(), + name: "read_source".to_string(), + arguments: "{}".to_string(), + }, + InputItem::ToolResult { + call_id: "call-1".to_string(), + output: "evidence".to_string(), + }, + ]; + request.reasoning_effort = Some(ReasoningEffort::High); + request.cache_retention = Some(CacheRetention::Short); + request.tools = vec![ToolDefinition { + name: "read_source".to_string(), + description: "Read one admitted source.".to_string(), + parameters_json: r#"{"type":"object"}"#.to_string(), + }]; + request.tool_choice = ToolChoice::Required; + request.parallel_tool_calls = true; + request.output_schema_json = Some(r#"{"type":"object"}"#.to_string()); + request +} + +fn invocation() -> TransportInvocation { + TransportInvocation::new(CALLER, 1_800_000_000_000, [7; 32], request()).unwrap() +} + +fn receipt(outcome: TransportOutcome) -> TransportReceipt { + let invocation = invocation(); + TransportReceipt { + schema_id: RECEIPT_SCHEMA_ID.to_string(), + request_id: invocation.request_id().to_string(), + caller_runtime_id: CALLER.to_string(), + native_request_sha256: invocation.native_request_sha256, + provider_request_sha256: invocation.provider_request_sha256, + model: invocation.request.model.clone(), + provider: "anthropic-messages".to_string(), + outcome, + } +} + +fn completed() -> TransportOutcome { + TransportOutcome::Completed { + provider_response_id: Some("msg-1".to_string()), + stop_reason: StopReason::ToolUse, + usage: Usage { + input_tokens: 120, + cache_read_tokens: 40, + cache_write_tokens: 8, + output_tokens: 33, + reasoning_tokens: Some(12), + }, + } +} + +#[test] +fn wire_golden_v3() { + let invocation = invocation(); + assert_wire("provider_request.mp", &invocation.request); + assert_wire("transport_invocation.mp", &invocation); + + let transported = TransportResult::transported( + &invocation, + vec![ + TransportEvent { + sequence: 0, + payload: TransportEventPayload::TextDelta { + text: "thinking".to_string(), + }, + }, + TransportEvent { + sequence: 1, + payload: TransportEventPayload::ToolCall { + call_id: "call-2".to_string(), + name: "read_source".to_string(), + arguments: "{}".to_string(), + }, + }, + ], + receipt(completed()), + ); + assert_eq!(transported.validate_against(&invocation), Ok(())); + assert_wire("transport_result_transported.mp", &transported); + + let refused = TransportResult::refused(&invocation, Refusal::OutputBound); + assert_wire("transport_result_refused.mp", &refused); + + let failed = receipt(TransportOutcome::Failed { + kind: FailureKind::RateLimited, + message: "provider asked to slow down".to_string(), + }); + assert_eq!(failed.validate_against(&invocation), Ok(())); + assert_wire("transport_receipt_failed.mp", &failed); + + let envelope = TransportEnvelope { + schema_id: ENVELOPE_SCHEMA_ID.to_string(), + caller_runtime_id: CALLER.to_string(), + request_id: "golden-request".to_string(), + message_kind: EnvelopeKind::Invocation, + nonce: [0x80, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 0xff], + ciphertext: (0..40_u8).map(|byte| byte.wrapping_mul(7)).collect(), + }; + assert_wire("transport_envelope.mp", &envelope); +} + +/// The committed carriage frame is a `cultnet.operation_request.v0` for the +/// connector, read here with CultNet itself and not the kit's own carriage code. +/// Its payload is the base64 of a sealed envelope that decrypts under the +/// fixed key to the golden invocation. +#[test] +fn the_golden_carriage_frame_decrypts_and_validates_under_the_fixed_key() { + let path = fixtures().join("carriage_invocation.frame"); + let key = TransportKey::from_connection_secret(KEY).unwrap(); + let invocation = invocation(); + if std::env::var_os(REGENERATE).is_some() { + let envelope = encrypt_invocation(&invocation, &key).unwrap(); + let message = CultNetMessage::OperationRequest { + message_id: envelope.request_id.clone(), + service_id: CONNECTOR_SERVICE_ID.to_string(), + operation: INVOKE_OPERATION.to_string(), + payload_schema: ENVELOPE_SCHEMA_ID.to_string(), + payload_encoding: "messagepack-base64".to_string(), + payload: STANDARD.encode(named(&envelope)), + source_runtime_id: None, + target_runtime_id: None, + }; + std::fs::write( + &path, + encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).unwrap(), + ) + .unwrap(); + } + let frame = std::fs::read(&path).expect("carriage_invocation.frame is missing"); + let CultNetMessage::OperationRequest { + message_id, + service_id, + operation, + payload_schema, + payload, + .. + } = decode_cultnet_message_from_slice(&frame, CultNetWireContract::CultNetSchemaV0).unwrap() + else { + panic!("the golden frame is a CultNet operation request"); + }; + assert_eq!(service_id, "gamecult.model.connector"); + assert_eq!(operation, "invoke"); + assert_eq!(payload_schema, "gamecult.model.transport_envelope.v3"); + assert_eq!(message_id, invocation.request_id()); + let envelope: TransportEnvelope = + rmp_serde::from_slice(&STANDARD.decode(payload).unwrap()).unwrap(); + assert_eq!(envelope.request_id, message_id); + let decrypted = decrypt_invocation(&envelope, &key).unwrap(); + assert_eq!(decrypted, invocation); + assert_eq!(decrypted.validate(1_799_999_999_000, 5_000), Ok(())); +} diff --git a/schemas/provider_request.json b/schemas/provider_request.json new file mode 100644 index 0000000..d4fb346 --- /dev/null +++ b/schemas/provider_request.json @@ -0,0 +1,229 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "ProviderRequest", + "type": "object", + "properties": { + "cache_retention": { + "anyOf": [ + { + "$ref": "#/$defs/CacheRetention" + }, + { + "type": "null" + } + ] + }, + "conversation_id": { + "type": "string" + }, + "input": { + "type": "array", + "items": { + "$ref": "#/$defs/InputItem" + } + }, + "instructions": { + "type": "string" + }, + "max_output_tokens": { + "type": "integer", + "format": "uint32", + "minimum": 0 + }, + "model": { + "type": "string" + }, + "output_schema_json": { + "type": [ + "string", + "null" + ] + }, + "parallel_tool_calls": { + "type": "boolean" + }, + "reasoning_effort": { + "anyOf": [ + { + "$ref": "#/$defs/ReasoningEffort" + }, + { + "type": "null" + } + ] + }, + "request_id": { + "type": "string" + }, + "schema_id": { + "type": "string" + }, + "tool_choice": { + "$ref": "#/$defs/ToolChoice" + }, + "tools": { + "type": "array", + "items": { + "$ref": "#/$defs/ToolDefinition" + } + } + }, + "additionalProperties": false, + "required": [ + "schema_id", + "request_id", + "conversation_id", + "model", + "instructions", + "input", + "tools", + "tool_choice", + "parallel_tool_calls", + "max_output_tokens" + ], + "$defs": { + "CacheRetention": { + "type": "string", + "enum": [ + "short", + "long" + ] + }, + "InputItem": { + "oneOf": [ + { + "type": "object", + "properties": { + "UserText": { + "type": "object", + "properties": { + "text": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "text" + ] + } + }, + "additionalProperties": false, + "required": [ + "UserText" + ] + }, + { + "type": "object", + "properties": { + "AssistantText": { + "type": "object", + "properties": { + "text": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "text" + ] + } + }, + "additionalProperties": false, + "required": [ + "AssistantText" + ] + }, + { + "type": "object", + "properties": { + "ToolCall": { + "type": "object", + "properties": { + "arguments": { + "type": "string" + }, + "call_id": { + "type": "string" + }, + "name": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "call_id", + "name", + "arguments" + ] + } + }, + "additionalProperties": false, + "required": [ + "ToolCall" + ] + }, + { + "type": "object", + "properties": { + "ToolResult": { + "type": "object", + "properties": { + "call_id": { + "type": "string" + }, + "output": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "call_id", + "output" + ] + } + }, + "additionalProperties": false, + "required": [ + "ToolResult" + ] + } + ] + }, + "ReasoningEffort": { + "type": "string", + "enum": [ + "low", + "medium", + "high", + "max" + ] + }, + "ToolChoice": { + "type": "string", + "enum": [ + "auto", + "required" + ] + }, + "ToolDefinition": { + "type": "object", + "properties": { + "description": { + "type": "string" + }, + "name": { + "type": "string" + }, + "parameters_json": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "name", + "description", + "parameters_json" + ] + } + } +} diff --git a/schemas/transport_envelope.json b/schemas/transport_envelope.json new file mode 100644 index 0000000..aee5469 --- /dev/null +++ b/schemas/transport_envelope.json @@ -0,0 +1,57 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "TransportEnvelope", + "type": "object", + "properties": { + "caller_runtime_id": { + "type": "string" + }, + "ciphertext": { + "type": "array", + "items": { + "type": "integer", + "format": "uint8", + "maximum": 255, + "minimum": 0 + } + }, + "message_kind": { + "$ref": "#/$defs/EnvelopeKind" + }, + "nonce": { + "type": "array", + "items": { + "type": "integer", + "format": "uint8", + "maximum": 255, + "minimum": 0 + }, + "maxItems": 12, + "minItems": 12 + }, + "request_id": { + "type": "string" + }, + "schema_id": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "schema_id", + "caller_runtime_id", + "request_id", + "message_kind", + "nonce", + "ciphertext" + ], + "$defs": { + "EnvelopeKind": { + "type": "string", + "enum": [ + "Invocation", + "Result" + ] + } + } +} diff --git a/schemas/transport_invocation.json b/schemas/transport_invocation.json new file mode 100644 index 0000000..9e899b1 --- /dev/null +++ b/schemas/transport_invocation.json @@ -0,0 +1,279 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "TransportInvocation", + "type": "object", + "properties": { + "caller_runtime_id": { + "type": "string" + }, + "expires_at_unix_ms": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "native_request_sha256": { + "type": "array", + "items": { + "type": "integer", + "format": "uint8", + "maximum": 255, + "minimum": 0 + }, + "maxItems": 32, + "minItems": 32 + }, + "provider_request_sha256": { + "type": "array", + "items": { + "type": "integer", + "format": "uint8", + "maximum": 255, + "minimum": 0 + }, + "maxItems": 32, + "minItems": 32 + }, + "request": { + "$ref": "#/$defs/ProviderRequest" + }, + "schema_id": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "schema_id", + "caller_runtime_id", + "expires_at_unix_ms", + "native_request_sha256", + "provider_request_sha256", + "request" + ], + "$defs": { + "CacheRetention": { + "type": "string", + "enum": [ + "short", + "long" + ] + }, + "InputItem": { + "oneOf": [ + { + "type": "object", + "properties": { + "UserText": { + "type": "object", + "properties": { + "text": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "text" + ] + } + }, + "additionalProperties": false, + "required": [ + "UserText" + ] + }, + { + "type": "object", + "properties": { + "AssistantText": { + "type": "object", + "properties": { + "text": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "text" + ] + } + }, + "additionalProperties": false, + "required": [ + "AssistantText" + ] + }, + { + "type": "object", + "properties": { + "ToolCall": { + "type": "object", + "properties": { + "arguments": { + "type": "string" + }, + "call_id": { + "type": "string" + }, + "name": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "call_id", + "name", + "arguments" + ] + } + }, + "additionalProperties": false, + "required": [ + "ToolCall" + ] + }, + { + "type": "object", + "properties": { + "ToolResult": { + "type": "object", + "properties": { + "call_id": { + "type": "string" + }, + "output": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "call_id", + "output" + ] + } + }, + "additionalProperties": false, + "required": [ + "ToolResult" + ] + } + ] + }, + "ProviderRequest": { + "type": "object", + "properties": { + "cache_retention": { + "anyOf": [ + { + "$ref": "#/$defs/CacheRetention" + }, + { + "type": "null" + } + ] + }, + "conversation_id": { + "type": "string" + }, + "input": { + "type": "array", + "items": { + "$ref": "#/$defs/InputItem" + } + }, + "instructions": { + "type": "string" + }, + "max_output_tokens": { + "type": "integer", + "format": "uint32", + "minimum": 0 + }, + "model": { + "type": "string" + }, + "output_schema_json": { + "type": [ + "string", + "null" + ] + }, + "parallel_tool_calls": { + "type": "boolean" + }, + "reasoning_effort": { + "anyOf": [ + { + "$ref": "#/$defs/ReasoningEffort" + }, + { + "type": "null" + } + ] + }, + "request_id": { + "type": "string" + }, + "schema_id": { + "type": "string" + }, + "tool_choice": { + "$ref": "#/$defs/ToolChoice" + }, + "tools": { + "type": "array", + "items": { + "$ref": "#/$defs/ToolDefinition" + } + } + }, + "additionalProperties": false, + "required": [ + "schema_id", + "request_id", + "conversation_id", + "model", + "instructions", + "input", + "tools", + "tool_choice", + "parallel_tool_calls", + "max_output_tokens" + ] + }, + "ReasoningEffort": { + "type": "string", + "enum": [ + "low", + "medium", + "high", + "max" + ] + }, + "ToolChoice": { + "type": "string", + "enum": [ + "auto", + "required" + ] + }, + "ToolDefinition": { + "type": "object", + "properties": { + "description": { + "type": "string" + }, + "name": { + "type": "string" + }, + "parameters_json": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "name", + "description", + "parameters_json" + ] + } + } +} diff --git a/schemas/transport_receipt.json b/schemas/transport_receipt.json new file mode 100644 index 0000000..68916ed --- /dev/null +++ b/schemas/transport_receipt.json @@ -0,0 +1,186 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "TransportReceipt", + "type": "object", + "properties": { + "caller_runtime_id": { + "type": "string" + }, + "model": { + "type": "string" + }, + "native_request_sha256": { + "type": "array", + "items": { + "type": "integer", + "format": "uint8", + "maximum": 255, + "minimum": 0 + }, + "maxItems": 32, + "minItems": 32 + }, + "outcome": { + "$ref": "#/$defs/TransportOutcome" + }, + "provider": { + "type": "string" + }, + "provider_request_sha256": { + "type": "array", + "items": { + "type": "integer", + "format": "uint8", + "maximum": 255, + "minimum": 0 + }, + "maxItems": 32, + "minItems": 32 + }, + "request_id": { + "type": "string" + }, + "schema_id": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "schema_id", + "request_id", + "caller_runtime_id", + "native_request_sha256", + "provider_request_sha256", + "model", + "provider", + "outcome" + ], + "$defs": { + "FailureKind": { + "description": "What kind of failure a provider reported. `ResultSize` is the connector's own\nstand-in for a result whose sealed frame cannot be carried.", + "type": "string", + "enum": [ + "result_size", + "invalid_request", + "authentication", + "rate_limited", + "unavailable", + "timeout", + "provider" + ] + }, + "StopReason": { + "description": "Why a provider stopped producing output.", + "type": "string", + "enum": [ + "end_turn", + "tool_use", + "max_output_tokens", + "stop_sequence", + "refusal", + "context_window" + ] + }, + "TransportOutcome": { + "oneOf": [ + { + "type": "object", + "properties": { + "Completed": { + "type": "object", + "properties": { + "provider_response_id": { + "type": [ + "string", + "null" + ] + }, + "stop_reason": { + "$ref": "#/$defs/StopReason" + }, + "usage": { + "$ref": "#/$defs/Usage" + } + }, + "additionalProperties": false, + "required": [ + "stop_reason", + "usage" + ] + } + }, + "additionalProperties": false, + "required": [ + "Completed" + ] + }, + { + "type": "object", + "properties": { + "Failed": { + "type": "object", + "properties": { + "kind": { + "$ref": "#/$defs/FailureKind" + }, + "message": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "kind", + "message" + ] + } + }, + "additionalProperties": false, + "required": [ + "Failed" + ] + } + ] + }, + "Usage": { + "description": "Token accounting for one completed request. Cache reads and writes are\ncounted apart from plain input; `reasoning_tokens` is absent when the\nprovider does not report it.", + "type": "object", + "properties": { + "cache_read_tokens": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "cache_write_tokens": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "input_tokens": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "output_tokens": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "reasoning_tokens": { + "type": [ + "integer", + "null" + ], + "format": "uint64", + "minimum": 0 + } + }, + "additionalProperties": false, + "required": [ + "input_tokens", + "cache_read_tokens", + "cache_write_tokens", + "output_tokens" + ] + } + } +} diff --git a/schemas/transport_result.json b/schemas/transport_result.json new file mode 100644 index 0000000..ceb1e42 --- /dev/null +++ b/schemas/transport_result.json @@ -0,0 +1,342 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "TransportResult", + "type": "object", + "properties": { + "caller_runtime_id": { + "type": "string" + }, + "disposition": { + "$ref": "#/$defs/TransportDisposition" + }, + "request_id": { + "type": "string" + }, + "schema_id": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "schema_id", + "request_id", + "caller_runtime_id", + "disposition" + ], + "$defs": { + "FailureKind": { + "description": "What kind of failure a provider reported. `ResultSize` is the connector's own\nstand-in for a result whose sealed frame cannot be carried.", + "type": "string", + "enum": [ + "result_size", + "invalid_request", + "authentication", + "rate_limited", + "unavailable", + "timeout", + "provider" + ] + }, + "Refusal": { + "type": "string", + "enum": [ + "Expired", + "IdentitySubstitution", + "ProviderDigestSubstitution", + "ModelNotAllowed", + "OutputBound", + "PayloadBound", + "Capacity", + "InFlight", + "Indeterminate", + "ReplayConflict", + "Malformed" + ] + }, + "StopReason": { + "description": "Why a provider stopped producing output.", + "type": "string", + "enum": [ + "end_turn", + "tool_use", + "max_output_tokens", + "stop_sequence", + "refusal", + "context_window" + ] + }, + "TransportDisposition": { + "oneOf": [ + { + "type": "object", + "properties": { + "Refused": { + "$ref": "#/$defs/Refusal" + } + }, + "additionalProperties": false, + "required": [ + "Refused" + ] + }, + { + "type": "object", + "properties": { + "Transported": { + "type": "object", + "properties": { + "events": { + "type": "array", + "items": { + "$ref": "#/$defs/TransportEvent" + } + }, + "receipt": { + "$ref": "#/$defs/TransportReceipt" + } + }, + "additionalProperties": false, + "required": [ + "events", + "receipt" + ] + } + }, + "additionalProperties": false, + "required": [ + "Transported" + ] + } + ] + }, + "TransportEvent": { + "type": "object", + "properties": { + "payload": { + "$ref": "#/$defs/TransportEventPayload" + }, + "sequence": { + "type": "integer", + "format": "uint64", + "minimum": 0 + } + }, + "additionalProperties": false, + "required": [ + "sequence", + "payload" + ] + }, + "TransportEventPayload": { + "oneOf": [ + { + "type": "object", + "properties": { + "TextDelta": { + "type": "object", + "properties": { + "text": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "text" + ] + } + }, + "additionalProperties": false, + "required": [ + "TextDelta" + ] + }, + { + "type": "object", + "properties": { + "ToolCall": { + "type": "object", + "properties": { + "arguments": { + "type": "string" + }, + "call_id": { + "type": "string" + }, + "name": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "call_id", + "name", + "arguments" + ] + } + }, + "additionalProperties": false, + "required": [ + "ToolCall" + ] + } + ] + }, + "TransportOutcome": { + "oneOf": [ + { + "type": "object", + "properties": { + "Completed": { + "type": "object", + "properties": { + "provider_response_id": { + "type": [ + "string", + "null" + ] + }, + "stop_reason": { + "$ref": "#/$defs/StopReason" + }, + "usage": { + "$ref": "#/$defs/Usage" + } + }, + "additionalProperties": false, + "required": [ + "stop_reason", + "usage" + ] + } + }, + "additionalProperties": false, + "required": [ + "Completed" + ] + }, + { + "type": "object", + "properties": { + "Failed": { + "type": "object", + "properties": { + "kind": { + "$ref": "#/$defs/FailureKind" + }, + "message": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "kind", + "message" + ] + } + }, + "additionalProperties": false, + "required": [ + "Failed" + ] + } + ] + }, + "TransportReceipt": { + "type": "object", + "properties": { + "caller_runtime_id": { + "type": "string" + }, + "model": { + "type": "string" + }, + "native_request_sha256": { + "type": "array", + "items": { + "type": "integer", + "format": "uint8", + "maximum": 255, + "minimum": 0 + }, + "maxItems": 32, + "minItems": 32 + }, + "outcome": { + "$ref": "#/$defs/TransportOutcome" + }, + "provider": { + "type": "string" + }, + "provider_request_sha256": { + "type": "array", + "items": { + "type": "integer", + "format": "uint8", + "maximum": 255, + "minimum": 0 + }, + "maxItems": 32, + "minItems": 32 + }, + "request_id": { + "type": "string" + }, + "schema_id": { + "type": "string" + } + }, + "additionalProperties": false, + "required": [ + "schema_id", + "request_id", + "caller_runtime_id", + "native_request_sha256", + "provider_request_sha256", + "model", + "provider", + "outcome" + ] + }, + "Usage": { + "description": "Token accounting for one completed request. Cache reads and writes are\ncounted apart from plain input; `reasoning_tokens` is absent when the\nprovider does not report it.", + "type": "object", + "properties": { + "cache_read_tokens": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "cache_write_tokens": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "input_tokens": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "output_tokens": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "reasoning_tokens": { + "type": [ + "integer", + "null" + ], + "format": "uint64", + "minimum": 0 + } + }, + "additionalProperties": false, + "required": [ + "input_tokens", + "cache_read_tokens", + "cache_write_tokens", + "output_tokens" + ] + } + } +}