From 032d25d5f30d7b7eb0c92a018a081acf12581181 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Mon, 24 Aug 2026 14:43:02 +0200 Subject: [PATCH] Add isolated multi-caller transport core --- Cargo.lock | 136 +++++++++ Cargo.toml | 2 + README.md | 17 +- src/lib.rs | 798 +++++++++++++++++++++++++++++++++++++++++++++++++---- 4 files changed, 899 insertions(+), 54 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1d82556..6add5f5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,41 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "aead" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d122413f284cf2d62fb1b7db97e02edb8cda96d769b16e443a4f6195e35662b0" +dependencies = [ + "crypto-common", + "generic-array", +] + +[[package]] +name = "aes" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" +dependencies = [ + "cfg-if", + "cipher", + "cpufeatures", +] + +[[package]] +name = "aes-gcm" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "831010a0f742e1209b3bcea8fab6a8e149051ba6099432c8cb2cc117dec3ead1" +dependencies = [ + "aead", + "aes", + "cipher", + "ctr", + "ghash", + "subtle", +] + [[package]] name = "autocfg" version = "1.5.1" @@ -23,10 +58,22 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cipher" +version = "0.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" +dependencies = [ + "crypto-common", + "inout", +] + [[package]] name = "codex-connector" version = "0.1.0" dependencies = [ + "aes-gcm", + "rand_core", "rmp-serde", "serde", "sha2", @@ -49,9 +96,19 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" dependencies = [ "generic-array", + "rand_core", "typenum", ] +[[package]] +name = "ctr" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0369ee1ad671834580515889b80f2ea915f23b8be8d0daa4bbaf2ac5c7590835" +dependencies = [ + "cipher", +] + [[package]] name = "digest" version = "0.10.7" @@ -72,6 +129,36 @@ dependencies = [ "version_check", ] +[[package]] +name = "getrandom" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" +dependencies = [ + "cfg-if", + "libc", + "wasi", +] + +[[package]] +name = "ghash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0d8a4362ccb29cb0b265253fb0a2728f592895ee6854fd9bc13f2ffda266ff1" +dependencies = [ + "opaque-debug", + "polyval", +] + +[[package]] +name = "inout" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01" +dependencies = [ + "generic-array", +] + [[package]] name = "libc" version = "0.2.189" @@ -87,6 +174,24 @@ dependencies = [ "autocfg", ] +[[package]] +name = "opaque-debug" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" + +[[package]] +name = "polyval" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1fe60d06143b2430aa532c94cfe9e29783047f06c0d7fd359a9a51b729fa25" +dependencies = [ + "cfg-if", + "cpufeatures", + "opaque-debug", + "universal-hash", +] + [[package]] name = "proc-macro2" version = "1.0.107" @@ -105,6 +210,15 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "rand_core" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" +dependencies = [ + "getrandom", +] + [[package]] name = "rmp" version = "0.8.15" @@ -165,6 +279,12 @@ dependencies = [ "digest", ] +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + [[package]] name = "syn" version = "3.0.4" @@ -208,8 +328,24 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "universal-hash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc1de2c688dc15305988b563c3854064043356019f97a4b46276fe734c4f07ea" +dependencies = [ + "crypto-common", + "subtle", +] + [[package]] name = "version_check" version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" diff --git a/Cargo.toml b/Cargo.toml index bb1a230..5adce1e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,6 +12,8 @@ repository = "https://github.com/GameCult/CodexConnector" path = "src/lib.rs" [dependencies] +aes-gcm = "0.10" +rand_core = { version = "0.6", features = ["getrandom"] } rmp-serde = "1" serde = { version = "1", features = ["derive"] } sha2 = "0.10" diff --git a/README.md b/README.md index b3b1812..6840e6a 100644 --- a/README.md +++ b/README.md @@ -9,10 +9,13 @@ The connector owns one writable credential/refresh store, authenticated caller admission, bounded transport physiology, exact provider-request receipts, and redacted CultMesh health. It owns no prompt policy or decision authority. -The current source establishes the v2 typed contract before adding the daemon. -This is deliberate: the existing Yggdrasil connector already proves the -process seam, but its single-caller, fixed-model, stateless, no-tools wire law -cannot become the shared contract by inertia. +The current source establishes the v2 typed contract and its transport-neutral +service core before adding the daemon shell. The core admits distinct caller +keys and policies, returns an exact cached response for byte-identical retries, +refuses conflicting or concurrent reuse of one caller/request identity, and +leaves provider execution outside its lock. The existing Yggdrasil connector +already proves the process seam, but its single-caller, fixed-model, stateless, +no-tools wire law cannot become the shared contract by inertia. ## Contract invariant @@ -23,3 +26,9 @@ audit correlation; the connector does not interpret or rederive consumer-native cognition. Tool calls are typed output. The connector never executes them. + +The encrypted envelope contains only MessagePack contracts and an AES-256-GCM +payload. Prompt text and results are not visible to the framing layer. The +caller runtime identity remains visible so the daemon can select that caller's +distinct key; the authenticated inner contract repeats the identity and must +match it exactly. diff --git a/src/lib.rs b/src/lib.rs index 2367a99..715f0c9 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,5 +1,8 @@ -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; +use aes_gcm::aead::{Aead, Payload}; +use aes_gcm::{Aes256Gcm, KeyInit, Nonce}; +use rand_core::{OsRng, RngCore}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use thiserror::Error; @@ -8,6 +11,19 @@ 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"; + +#[derive(Clone, PartialEq, Eq)] +pub struct CodexTransportKey([u8; 32]); + +impl CodexTransportKey { + pub fn from_connection_secret(secret: &str) -> Result { + if secret.trim().is_empty() || secret.trim() != secret { + return Err(ServiceError::InvalidAdmission); + } + Ok(Self(Sha256::digest(secret.as_bytes()).into())) + } +} #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct CodexProviderRequest { @@ -200,61 +216,62 @@ impl CodexTransportInvocation { } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub enum CodexTransportResult { - Refused { - schema_id: String, - request_id: String, - caller_runtime_id: String, - reason: CodexRefusal, - }, +pub struct CodexTransportResult { + pub schema_id: String, + pub request_id: String, + pub caller_runtime_id: String, + pub disposition: CodexTransportDisposition, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub enum CodexTransportDisposition { + Refused(CodexRefusal), Transported { - schema_id: String, - request_id: String, - caller_runtime_id: String, - provider_request_sha256: [u8; 32], events: Vec, - receipt: CodexTransportReceipt, + receipt: Box, }, } impl CodexTransportResult { + pub fn refused(invocation: &CodexTransportInvocation, reason: CodexRefusal) -> Self { + Self { + schema_id: RESULT_SCHEMA_ID.to_string(), + request_id: invocation.request_id().to_string(), + caller_runtime_id: invocation.caller_runtime_id.clone(), + disposition: CodexTransportDisposition::Refused(reason), + } + } + + pub fn transported( + invocation: &CodexTransportInvocation, + events: Vec, + receipt: CodexTransportReceipt, + ) -> Self { + Self { + schema_id: RESULT_SCHEMA_ID.to_string(), + request_id: invocation.request_id().to_string(), + caller_runtime_id: invocation.caller_runtime_id.clone(), + disposition: CodexTransportDisposition::Transported { + events, + receipt: Box::new(receipt), + }, + } + } + pub fn validate_against( &self, invocation: &CodexTransportInvocation, ) -> Result<(), ContractError> { - let (schema_id, request_id, caller_runtime_id) = match self { - Self::Refused { - schema_id, - request_id, - caller_runtime_id, - .. - } - | Self::Transported { - schema_id, - request_id, - caller_runtime_id, - .. - } => (schema_id, request_id, caller_runtime_id), - }; - if schema_id != RESULT_SCHEMA_ID { + if self.schema_id != RESULT_SCHEMA_ID { return Err(ContractError::Schema); } - if request_id != invocation.request_id() - || caller_runtime_id != &invocation.caller_runtime_id + if self.request_id != invocation.request_id() + || self.caller_runtime_id != invocation.caller_runtime_id { return Err(ContractError::Identity); } - if let Self::Transported { - provider_request_sha256, - events, - receipt, - .. - } = self - { - if provider_request_sha256 != &invocation.provider_request_sha256 { - return Err(ContractError::ProviderDigest); - } + if let CodexTransportDisposition::Transported { events, receipt } = &self.disposition { for (expected, event) in events.iter().enumerate() { if event.sequence != expected as u64 { return Err(ContractError::EventSequence); @@ -373,11 +390,12 @@ pub enum CodexTransportOutcome { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub enum CodexRefusal { - CallerNotAdmitted, Expired, IdentitySubstitution, ProviderDigestSubstitution, Policy, + Capacity, + InFlight, ReplayConflict, Malformed, } @@ -393,6 +411,481 @@ pub fn provider_request_sha256(request: &CodexProviderRequest) -> Result<[u8; 32 Ok(Sha256::digest(canonical_provider_request_bytes(request)?).into()) } +pub fn canonical_invocation_bytes( + invocation: &CodexTransportInvocation, +) -> Result, ContractError> { + invocation.request.validate()?; + if invocation.provider_request_sha256 != provider_request_sha256(&invocation.request)? { + return Err(ContractError::ProviderDigest); + } + rmp_serde::to_vec(invocation).map_err(|_| ContractError::Encoding) +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum CodexEnvelopeKind { + Invocation, + Result, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct CodexTransportEnvelope { + pub schema_id: String, + pub caller_runtime_id: String, + pub request_id: String, + pub message_kind: CodexEnvelopeKind, + pub nonce: [u8; 12], + pub ciphertext: Vec, +} + +pub fn encrypt_invocation( + invocation: &CodexTransportInvocation, + security: &CodexTransportKey, +) -> Result { + let plaintext = canonical_invocation_bytes(invocation)?; + encrypt_message( + &invocation.caller_runtime_id, + invocation.request_id(), + CodexEnvelopeKind::Invocation, + &plaintext, + security, + ) +} + +pub fn decrypt_invocation( + envelope: &CodexTransportEnvelope, + security: &CodexTransportKey, +) -> Result { + validate_envelope(envelope, CodexEnvelopeKind::Invocation)?; + let plaintext = decrypt_message(envelope, security)?; + let invocation: CodexTransportInvocation = + rmp_serde::from_slice(&plaintext).map_err(|_| ServiceError::Encoding)?; + if invocation.caller_runtime_id != envelope.caller_runtime_id + || invocation.request_id() != envelope.request_id + { + return Err(ServiceError::OuterIdentity); + } + Ok(invocation) +} + +pub fn encrypt_result( + result: &CodexTransportResult, + security: &CodexTransportKey, +) -> Result { + let plaintext = rmp_serde::to_vec(result).map_err(|_| ServiceError::Encoding)?; + encrypt_message( + &result.caller_runtime_id, + &result.request_id, + CodexEnvelopeKind::Result, + &plaintext, + security, + ) +} + +pub fn decrypt_result( + envelope: &CodexTransportEnvelope, + security: &CodexTransportKey, + invocation: &CodexTransportInvocation, +) -> Result { + validate_envelope(envelope, CodexEnvelopeKind::Result)?; + let plaintext = decrypt_message(envelope, security)?; + let result: CodexTransportResult = + rmp_serde::from_slice(&plaintext).map_err(|_| ServiceError::Encoding)?; + if result.caller_runtime_id != envelope.caller_runtime_id + || result.request_id != envelope.request_id + { + return Err(ServiceError::OuterIdentity); + } + result.validate_against(invocation)?; + Ok(result) +} + +fn encrypt_message( + caller_runtime_id: &str, + request_id: &str, + message_kind: CodexEnvelopeKind, + plaintext: &[u8], + security: &CodexTransportKey, +) -> Result { + require_id(caller_runtime_id, "caller_runtime_id")?; + require_id(request_id, "request_id")?; + let mut nonce = [0; 12]; + OsRng.fill_bytes(&mut nonce); + let cipher = Aes256Gcm::new_from_slice(&security.0).map_err(|_| ServiceError::Encryption)?; + let aad = envelope_aad( + ENVELOPE_SCHEMA_ID, + caller_runtime_id, + request_id, + message_kind, + )?; + let ciphertext = cipher + .encrypt( + Nonce::from_slice(&nonce), + Payload { + msg: plaintext, + aad: &aad, + }, + ) + .map_err(|_| ServiceError::Encryption)?; + Ok(CodexTransportEnvelope { + schema_id: ENVELOPE_SCHEMA_ID.to_string(), + caller_runtime_id: caller_runtime_id.to_string(), + request_id: request_id.to_string(), + message_kind, + nonce, + ciphertext, + }) +} + +fn decrypt_message( + envelope: &CodexTransportEnvelope, + security: &CodexTransportKey, +) -> Result, ServiceError> { + let cipher = Aes256Gcm::new_from_slice(&security.0).map_err(|_| ServiceError::Encryption)?; + let aad = envelope_aad( + &envelope.schema_id, + &envelope.caller_runtime_id, + &envelope.request_id, + envelope.message_kind, + )?; + cipher + .decrypt( + Nonce::from_slice(&envelope.nonce), + Payload { + msg: &envelope.ciphertext, + aad: &aad, + }, + ) + .map_err(|_| ServiceError::Encryption) +} + +fn envelope_aad( + schema_id: &str, + caller_runtime_id: &str, + request_id: &str, + message_kind: CodexEnvelopeKind, +) -> Result, ServiceError> { + rmp_serde::to_vec(&(schema_id, caller_runtime_id, request_id, message_kind)) + .map_err(|_| ServiceError::Encoding) +} + +fn validate_envelope( + envelope: &CodexTransportEnvelope, + expected_kind: CodexEnvelopeKind, +) -> Result<(), ServiceError> { + if envelope.schema_id != ENVELOPE_SCHEMA_ID || envelope.message_kind != expected_kind { + return Err(ServiceError::Envelope); + } + require_id(&envelope.caller_runtime_id, "caller_runtime_id")?; + require_id(&envelope.request_id, "request_id")?; + if envelope.ciphertext.is_empty() { + return Err(ServiceError::Envelope); + } + Ok(()) +} + +pub struct CodexCallerAdmission { + caller_runtime_id: String, + security: CodexTransportKey, + allowed_models: HashSet, + max_concurrent_requests: usize, + max_payload_bytes: usize, + max_output_tokens: u32, +} + +impl CodexCallerAdmission { + pub fn new( + caller_runtime_id: impl Into, + connection_key: impl Into, + allowed_models: impl IntoIterator, + max_concurrent_requests: usize, + max_payload_bytes: usize, + max_output_tokens: u32, + ) -> Result { + let caller_runtime_id = caller_runtime_id.into(); + require_id(&caller_runtime_id, "caller_runtime_id")?; + let allowed_models = allowed_models.into_iter().collect::>(); + if allowed_models.is_empty() + || allowed_models + .iter() + .any(|model| require_id(model, "allowed_model").is_err()) + || max_concurrent_requests == 0 + || max_payload_bytes == 0 + || max_output_tokens == 0 + { + return Err(ServiceError::InvalidAdmission); + } + let connection_key = connection_key.into(); + let security = CodexTransportKey::from_connection_secret(&connection_key)?; + Ok(Self { + caller_runtime_id, + security, + allowed_models, + max_concurrent_requests, + max_payload_bytes, + max_output_tokens, + }) + } + + pub fn caller_runtime_id(&self) -> &str { + &self.caller_runtime_id + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct RequestIdentity { + caller_runtime_id: String, + request_id: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct ActiveRequest { + invocation_sha256: [u8; 32], + expires_at_unix_ms: u64, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct CompletedRequest { + invocation_sha256: [u8; 32], + expires_at_unix_ms: u64, + response: CodexTransportEnvelope, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CodexExecutionClaim { + invocation: CodexTransportInvocation, + invocation_sha256: [u8; 32], +} + +impl CodexExecutionClaim { + pub fn invocation(&self) -> &CodexTransportInvocation { + &self.invocation + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum CodexTransportAdmission { + Execute(Box), + Reply(CodexTransportEnvelope), +} + +pub struct CodexTransportService { + callers: HashMap, + active: HashMap, + completed: HashMap, + max_expiry_skew_ms: u64, +} + +impl CodexTransportService { + pub fn new( + admissions: impl IntoIterator, + max_expiry_skew_ms: u64, + ) -> Result { + if max_expiry_skew_ms == 0 { + return Err(ServiceError::InvalidAdmission); + } + let mut callers = HashMap::new(); + let mut keys = HashSet::new(); + for admission in admissions { + if !keys.insert(admission.security.0) { + return Err(ServiceError::SharedCallerKey); + } + let caller_runtime_id = admission.caller_runtime_id.clone(); + if callers.insert(caller_runtime_id, admission).is_some() { + return Err(ServiceError::DuplicateCaller); + } + } + if callers.is_empty() { + return Err(ServiceError::InvalidAdmission); + } + Ok(Self { + callers, + active: HashMap::new(), + completed: HashMap::new(), + max_expiry_skew_ms, + }) + } + + pub fn begin( + &mut self, + envelope: &CodexTransportEnvelope, + now_unix_ms: u64, + ) -> Result { + validate_envelope(envelope, CodexEnvelopeKind::Invocation)?; + let caller = self + .callers + .get(&envelope.caller_runtime_id) + .ok_or(ServiceError::CallerNotAdmitted)?; + if envelope.ciphertext.len() > caller.max_payload_bytes { + return Err(ServiceError::PayloadTooLarge); + } + let security = caller.security.clone(); + let max_concurrent_requests = caller.max_concurrent_requests; + let invocation = decrypt_invocation(envelope, &security)?; + if let Err(error) = invocation.validate(now_unix_ms, self.max_expiry_skew_ms) { + return self.reply_refusal(&invocation, contract_refusal(&error), &security); + } + if !caller.allowed_models.contains(&invocation.request.model) + || invocation + .request + .max_output_tokens + .is_some_and(|limit| limit > caller.max_output_tokens) + { + return self.reply_refusal(&invocation, CodexRefusal::Policy, &security); + } + + self.active + .retain(|_, request| request.expires_at_unix_ms >= now_unix_ms); + self.completed + .retain(|_, request| request.expires_at_unix_ms >= now_unix_ms); + + let identity = RequestIdentity { + caller_runtime_id: invocation.caller_runtime_id.clone(), + request_id: invocation.request_id().to_string(), + }; + let invocation_sha256: [u8; 32] = + Sha256::digest(canonical_invocation_bytes(&invocation)?).into(); + if let Some(completed) = self.completed.get(&identity) { + return if completed.invocation_sha256 == invocation_sha256 { + Ok(CodexTransportAdmission::Reply(completed.response.clone())) + } else { + self.reply_refusal(&invocation, CodexRefusal::ReplayConflict, &security) + }; + } + if let Some(active) = self.active.get(&identity) { + let refusal = if active.invocation_sha256 == invocation_sha256 { + CodexRefusal::InFlight + } else { + CodexRefusal::ReplayConflict + }; + return self.reply_refusal(&invocation, refusal, &security); + } + let active_for_caller = self + .active + .keys() + .filter(|key| key.caller_runtime_id == invocation.caller_runtime_id) + .count(); + if active_for_caller >= max_concurrent_requests { + return self.reply_refusal(&invocation, CodexRefusal::Capacity, &security); + } + + self.active.insert( + identity, + ActiveRequest { + invocation_sha256, + expires_at_unix_ms: invocation.expires_at_unix_ms, + }, + ); + Ok(CodexTransportAdmission::Execute(Box::new( + CodexExecutionClaim { + invocation, + invocation_sha256, + }, + ))) + } + + pub fn complete( + &mut self, + claim: CodexExecutionClaim, + result: CodexTransportResult, + ) -> Result { + if !matches!( + &result.disposition, + CodexTransportDisposition::Transported { .. } + ) { + return Err(ServiceError::InvalidCompletion); + } + result.validate_against(&claim.invocation)?; + let identity = RequestIdentity { + caller_runtime_id: claim.invocation.caller_runtime_id.clone(), + request_id: claim.invocation.request_id().to_string(), + }; + let active = self + .active + .get(&identity) + .ok_or(ServiceError::NoActiveRequest)?; + if active.invocation_sha256 != claim.invocation_sha256 { + return Err(ServiceError::ActiveRequestMismatch); + } + let caller = self + .callers + .get(&identity.caller_runtime_id) + .ok_or(ServiceError::CallerNotAdmitted)?; + let response = encrypt_result(&result, &caller.security)?; + self.active.remove(&identity); + self.completed.insert( + identity, + CompletedRequest { + invocation_sha256: claim.invocation_sha256, + expires_at_unix_ms: claim.invocation.expires_at_unix_ms, + response: response.clone(), + }, + ); + Ok(response) + } + + pub fn cancel(&mut self, claim: &CodexExecutionClaim) -> bool { + let identity = RequestIdentity { + caller_runtime_id: claim.invocation.caller_runtime_id.clone(), + request_id: claim.invocation.request_id().to_string(), + }; + self.active + .get(&identity) + .is_some_and(|active| active.invocation_sha256 == claim.invocation_sha256) + && self.active.remove(&identity).is_some() + } + + fn reply_refusal( + &self, + invocation: &CodexTransportInvocation, + refusal: CodexRefusal, + security: &CodexTransportKey, + ) -> Result { + Ok(CodexTransportAdmission::Reply(encrypt_result( + &CodexTransportResult::refused(invocation, refusal), + security, + )?)) + } +} + +fn contract_refusal(error: &ContractError) -> CodexRefusal { + match error { + ContractError::Expiry => CodexRefusal::Expired, + ContractError::Identity => CodexRefusal::IdentitySubstitution, + ContractError::ProviderDigest => CodexRefusal::ProviderDigestSubstitution, + _ => CodexRefusal::Malformed, + } +} + +#[derive(Debug, Error, PartialEq, Eq)] +pub enum ServiceError { + #[error(transparent)] + Contract(#[from] ContractError), + #[error("invalid caller admission")] + InvalidAdmission, + #[error("caller runtime was admitted twice")] + DuplicateCaller, + #[error("callers must not share transport keys")] + SharedCallerKey, + #[error("caller runtime is not admitted")] + CallerNotAdmitted, + #[error("transport payload exceeds the caller bound")] + PayloadTooLarge, + #[error("unexpected transport envelope")] + Envelope, + #[error("transport envelope identity substitution")] + OuterIdentity, + #[error("transport encryption or authentication failed")] + Encryption, + #[error("transport MessagePack encoding failed")] + Encoding, + #[error("completion did not carry a transported provider outcome")] + InvalidCompletion, + #[error("completion has no active request")] + NoActiveRequest, + #[error("completion does not match the active request")] + ActiveRequestMismatch, +} + fn require_id(value: &str, field: &'static str) -> Result<(), ContractError> { if value.trim().is_empty() || value.trim() != value { return Err(ContractError::Invalid(field)); @@ -530,12 +1023,9 @@ mod tests { #[test] fn transported_result_binds_identity_digest_and_event_order() { let invocation = invocation(); - let result = CodexTransportResult::Transported { - schema_id: RESULT_SCHEMA_ID.to_string(), - request_id: invocation.request_id().to_string(), - caller_runtime_id: invocation.caller_runtime_id.clone(), - provider_request_sha256: invocation.provider_request_sha256, - events: vec![CodexTransportEvent { + let result = CodexTransportResult::transported( + &invocation, + vec![CodexTransportEvent { sequence: 0, payload: CodexTransportEventPayload::ToolCall { call_id: "call-1".to_string(), @@ -543,12 +1033,13 @@ mod tests { arguments: "{}".to_string(), }, }], - receipt: receipt(&invocation), - }; + receipt(&invocation), + ); assert_eq!(result.validate_against(&invocation), Ok(())); let mut reordered = result; - let CodexTransportResult::Transported { events, .. } = &mut reordered else { + let CodexTransportDisposition::Transported { events, .. } = &mut reordered.disposition + else { unreachable!() }; events[0].sequence = 1; @@ -575,4 +1066,211 @@ mod tests { Err(ContractError::NativeDigest) ); } + + fn caller( + caller_runtime_id: &str, + key: &str, + max_concurrent_requests: usize, + ) -> CodexCallerAdmission { + CodexCallerAdmission::new( + caller_runtime_id, + key, + ["gpt-5.4".to_string()], + max_concurrent_requests, + 64 * 1024, + 32_768, + ) + .unwrap() + } + + fn service(max_concurrent_requests: usize) -> CodexTransportService { + CodexTransportService::new( + [ + caller( + "epiphany-yggdrasil", + "epiphany-distinct-test-key", + max_concurrent_requests, + ), + caller( + "ghostlight-yggdrasil", + "ghostlight-distinct-test-key", + max_concurrent_requests, + ), + ], + 5_000, + ) + .unwrap() + } + + fn security(key: &str) -> CodexTransportKey { + CodexTransportKey::from_connection_secret(key).unwrap() + } + + fn invocation_for(caller_runtime_id: &str, request_id: &str) -> CodexTransportInvocation { + let mut request = request(); + request.request_id = request_id.to_string(); + CodexTransportInvocation::new(caller_runtime_id, 2_000, [7; 32], request).unwrap() + } + + #[test] + fn encrypted_invocation_hides_content_and_binds_outer_identity() { + let invocation = invocation(); + let epiphany_key = security("epiphany-distinct-test-key"); + let envelope = encrypt_invocation(&invocation, &epiphany_key).unwrap(); + assert!( + !envelope + .ciphertext + .windows("Projected state".len()) + .any(|window| window == b"Projected state") + ); + assert_eq!( + decrypt_invocation(&envelope, &epiphany_key).unwrap(), + invocation + ); + assert_eq!( + decrypt_invocation(&envelope, &security("wrong-caller-key")), + Err(ServiceError::Encryption) + ); + + let mut substituted = envelope; + substituted.request_id = "request-2".to_string(); + assert_eq!( + decrypt_invocation(&substituted, &epiphany_key), + Err(ServiceError::Encryption) + ); + } + + #[test] + fn service_refuses_shared_keys_and_isolates_caller_request_identities() { + assert!(matches!( + CodexTransportService::new( + [ + caller("epiphany-yggdrasil", "shared-key", 1), + caller("ghostlight-yggdrasil", "shared-key", 1), + ], + 5_000, + ), + Err(ServiceError::SharedCallerKey) + )); + + let epiphany = invocation_for("epiphany-yggdrasil", "shared-request-id"); + let ghostlight = invocation_for("ghostlight-yggdrasil", "shared-request-id"); + let mut service = service(1); + assert!(matches!( + service + .begin( + &encrypt_invocation(&epiphany, &security("epiphany-distinct-test-key")) + .unwrap(), + 1_000, + ) + .unwrap(), + CodexTransportAdmission::Execute(_) + )); + assert!(matches!( + service + .begin( + &encrypt_invocation(&ghostlight, &security("ghostlight-distinct-test-key")) + .unwrap(), + 1_000, + ) + .unwrap(), + CodexTransportAdmission::Execute(_) + )); + } + + #[test] + fn service_returns_exact_completed_replay_and_refuses_conflicting_replay() { + let invocation = invocation(); + let security = security("epiphany-distinct-test-key"); + let mut service = service(1); + let claim = match service + .begin(&encrypt_invocation(&invocation, &security).unwrap(), 1_000) + .unwrap() + { + CodexTransportAdmission::Execute(claim) => claim, + CodexTransportAdmission::Reply(_) => panic!("first invocation must execute"), + }; + let response = service + .complete( + *claim, + CodexTransportResult::transported( + &invocation, + vec![CodexTransportEvent { + sequence: 0, + payload: CodexTransportEventPayload::ToolCall { + call_id: "call-1".to_string(), + name: "read_source".to_string(), + arguments: "{}".to_string(), + }, + }], + receipt(&invocation), + ), + ) + .unwrap(); + + let replay = match service + .begin(&encrypt_invocation(&invocation, &security).unwrap(), 1_000) + .unwrap() + { + CodexTransportAdmission::Reply(response) => response, + CodexTransportAdmission::Execute(_) => panic!("completed invocation re-executed"), + }; + assert_eq!(replay, response); + let decoded = decrypt_result(&replay, &security, &invocation).unwrap(); + assert!(matches!( + decoded.disposition, + CodexTransportDisposition::Transported { .. } + )); + + let mut changed_request = request(); + changed_request.instructions.push_str(" Different cargo."); + let conflicting = + CodexTransportInvocation::new("epiphany-yggdrasil", 2_000, [7; 32], changed_request) + .unwrap(); + let refusal = match service + .begin(&encrypt_invocation(&conflicting, &security).unwrap(), 1_000) + .unwrap() + { + CodexTransportAdmission::Reply(response) => response, + CodexTransportAdmission::Execute(_) => panic!("conflicting replay executed"), + }; + assert!(matches!( + decrypt_result(&refusal, &security, &conflicting) + .unwrap() + .disposition, + CodexTransportDisposition::Refused(CodexRefusal::ReplayConflict) + )); + } + + #[test] + fn service_refuses_duplicate_inflight_and_per_caller_capacity() { + let first = invocation_for("epiphany-yggdrasil", "request-1"); + let second = invocation_for("epiphany-yggdrasil", "request-2"); + let security = security("epiphany-distinct-test-key"); + let mut service = service(1); + let first_envelope = encrypt_invocation(&first, &security).unwrap(); + assert!(matches!( + service.begin(&first_envelope, 1_000).unwrap(), + CodexTransportAdmission::Execute(_) + )); + + for (invocation, expected) in [ + (&first, CodexRefusal::InFlight), + (&second, CodexRefusal::Capacity), + ] { + let response = match service + .begin(&encrypt_invocation(invocation, &security).unwrap(), 1_000) + .unwrap() + { + CodexTransportAdmission::Reply(response) => response, + CodexTransportAdmission::Execute(_) => panic!("refused invocation executed"), + }; + assert_eq!( + decrypt_result(&response, &security, invocation) + .unwrap() + .disposition, + CodexTransportDisposition::Refused(expected) + ); + } + } }