diff --git a/Cargo.lock b/Cargo.lock index ecf1894d..7d71f4c5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2050,7 +2050,10 @@ version = "0.1.0" dependencies = [ "anyhow", "chrono", + "codex-login", "cultcache-rs", + "cultmesh-rs", + "cultnet-rs", "epiphany-core", "epiphany-model-adapter", "epiphany-openai-adapter", diff --git a/Cargo.toml b/Cargo.toml index 4f7c6291..038914fa 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -102,6 +102,11 @@ name = "epiphany-model-runtime" path = "epiphany-openai-runtime/src/bin/epiphany-model-runtime.rs" required-features = ["openai-runtime"] +[[bin]] +name = "epiphany-model-connector" +path = "epiphany-openai-runtime/src/bin/epiphany-model-connector.rs" +required-features = ["openai-runtime"] + [[bin]] name = "epiphany-tool-mcp-runtime" path = "epiphany-tool-mcp-runtime/src/main.rs" diff --git a/README.md b/README.md index 5c05bc1e..c640dc83 100644 --- a/README.md +++ b/README.md @@ -129,6 +129,10 @@ Her body is made of: authority surfaces - Codex retained only for honest OpenAI subscription auth and model transport; Epiphany does not publish a Codex-owned project-state surface +- an independently supervised, loopback-only + [model provider connector](docs/model-provider-connector.md) can lend that + transport to an admitted GameCult service without lending Epiphany's Mind or + credentials That last clause matters. Codex and other frontier agents can be excellent workers. They do not decide which organizational state is authoritative, who diff --git a/docs/README.md b/docs/README.md index 02786fdd..954cd487 100644 --- a/docs/README.md +++ b/docs/README.md @@ -25,6 +25,9 @@ outside the working notes. - [Fork implementation plan](../notes/epiphany-fork-implementation-plan.md) records the current architecture and cut lines. - [Project map](../state/map.yaml) is the canonical slow machine map. +- [Model provider connector](./model-provider-connector.md) records the typed + loopback boundary that lends Epiphany's Codex transport without lending her + Mind, credentials, or swarm runtime. For historical GameCult-level investor context, the sibling site dossier lives at `F:\Projects\gamecult-site\docs\gamecult_integrated_dossier.tex`. It is a diff --git a/docs/model-provider-connector.md b/docs/model-provider-connector.md new file mode 100644 index 00000000..82a1b383 --- /dev/null +++ b/docs/model-provider-connector.md @@ -0,0 +1,117 @@ +# Epiphany Model Provider Connector + +## Objective + +Let another GameCult service use Epiphany's admitted Codex subscription +transport without acquiring Epiphany's Mind, credentials, scheduler, swarm +runtime, or authority. The first consumer is Ghostlight Dungeon. + +The connector is an independently supervised loopback service. Its daemon is +earned by credential isolation, request-concurrency ownership, and failure +isolation from both the consumer and the Epiphany swarm. + +## Authority map + +- **Owner:** `epiphany-model-connector` owns Codex authentication custody, + provider transport, physical request concurrency, transport replay handling, + and a redacted capability/status projection. +- **Inputs:** one encrypted, expiring + `epiphany.model_connector_invocation.v1` carried in a bounded CultNet + `DocumentPutRaw` frame. The admitted caller and model are configured by the + operator. +- **Outputs:** one encrypted `epiphany.model_connector_result.v1` containing + ordered provider events and the terminal model receipt, or a standard + CultNet refusal. Odin receives only + `epiphany.model_connector_status.v1`. +- **Derived state:** connector readiness and capacity are discovery state. They + are not provider credentials, inference permission, campaign state, or + Epiphany Mind state. +- **Forbidden writers:** the caller cannot select an unadmitted model, attach + tools, request reasoning summaries, chain provider response state, inspect + credentials, or mutate Epiphany. The connector cannot interpret model output + as application state or commit application mutations. +- **Shared path:** every accepted consumer request uses the same bounded frame, + authenticated envelope, caller/model checks, semaphore, Codex transport, and + terminal receipt validation. +- **Cut line:** the connector calls the existing Codex transport directly. It + does not start Epiphany's runtime, open her CultCache Mind, emit heartbeat + state, or participate in swarm scheduling. + +## Wire flow + +```text +consumer-owned typed stage + -> provider-neutral model request + -> MessagePack connector invocation + -> AES-GCM authenticated envelope + -> bounded TCP-framed CultNet on loopback + -> Epiphany Codex transport + -> ordered text/provider receipt events + -> encrypted typed connector result + -> consumer validation and local interpretation + -> consumer-owned commit authority +``` + +The outer CultNet message ID, document key, encrypted envelope request ID, +invocation request ID, model request ID, every stream-event request ID, and +terminal receipt request ID must agree. Requests carry a short expiry and an +exact payload digest. A repeated request ID with identical bytes receives the +cached terminal response for that connector lifetime; a conflicting or +concurrently active duplicate is rejected. Application idempotency and commit +authority remain with the consumer. + +The frame decoder rejects a declared payload larger than its advertised limit +before allocating the payload. This is a CultNet transport invariant rather +than a connector-specific workaround. + +## Security and privacy + +- The listener must bind to loopback. It is never routed through nginx or a + public firewall. +- A service-private shared key authenticates and encrypts every invocation and + result. It is stored outside source and release directories. +- The Codex credential stays in the connector's service-readable credential + directory. It is never copied into Ghostlight, arguments, logs, CultMesh, or + Odin. +- Tools, provider-side state chaining, reasoning summaries, reasoning deltas, + and tool-call events are refused. +- Prompts and results are not persisted by the connector. The consumer owns + its own permitted stage receipts. + +## Cache and token efficiency + +The request contract carries `prompt_cache_key` and `max_output_tokens` through +the existing model spine. Consumers construct the cache key from stable stage +and output-contract identity, not volatile world state. Provider receipts +report prompt, completion, reasoning, and cached-input token counts when the +provider supplies them. Epiphany neither fabricates cache hits nor interprets +their application value. + +## Deployment boundary + +Idunn manages this connector as a separate immutable artifact and service. An +Epiphany deployment brake may prevent changing Epiphany's swarm release; it +does not make the connector part of that release and must not wake the swarm. +The connector's own deployment and continuity remain explicit. + +The default endpoint is `127.0.0.1:4103`, maximum payload is one MiB, and +physical concurrency is eight. On Yggdrasil, the Odin publication endpoint is +`10.77.0.1:17871`; local `127.0.0.1:17871` belongs to Idunn. + +## Verification + +Focused tests prove: + +- encrypted round trips and exact request correlation; +- wrong caller, model, expiry, replay, tool, and reasoning refusals; +- no reasoning or tool material crosses the accepted result boundary; +- connector status contains no credential or application state; +- oversized TCP-framed payloads are refused before allocation; +- provider cache-token telemetry survives the Codex spine; +- a consumer transport can use the real framed/encrypted boundary without + learning connector credentials or Epiphany state. + +The build surface is one new production binary in the existing OpenAI runtime +crate. It reuses the existing CultNet, CultMesh, Codex login, and model-spine +dependencies. It does not add a second model client or a second orchestration +runtime. diff --git a/epiphany-core/src/reasoning_context.rs b/epiphany-core/src/reasoning_context.rs index 70f70241..d656cd35 100644 --- a/epiphany-core/src/reasoning_context.rs +++ b/epiphany-core/src/reasoning_context.rs @@ -2052,6 +2052,8 @@ mod tests { output_schema_json: None, provider_id: native.provider.clone(), wire_dialect: epiphany_openai_adapter::EpiphanyOpenAiWireDialect::Responses, + max_output_tokens: native.max_output_tokens, + prompt_cache_key: native.prompt_cache_key.clone(), }; (native, provider) } diff --git a/epiphany-model-adapter/src/lib.rs b/epiphany-model-adapter/src/lib.rs index 2cb5e668..8abea7e0 100644 --- a/epiphany-model-adapter/src/lib.rs +++ b/epiphany-model-adapter/src/lib.rs @@ -6,6 +6,10 @@ pub const MODEL_ADAPTER_REQUEST_SCHEMA_ID: &str = "epiphany.model_request.v0"; pub const MODEL_ADAPTER_EVENT_SCHEMA_ID: &str = "epiphany.model_stream_event.v0"; pub const MODEL_ADAPTER_RECEIPT_SCHEMA_ID: &str = "epiphany.model_receipt.v0"; pub const MODEL_ADAPTER_STATUS_SCHEMA_ID: &str = "epiphany.model_adapter_status.v0"; +pub const MODEL_CONNECTOR_ENVELOPE_SCHEMA_ID: &str = "epiphany.model_connector_envelope.v1"; +pub const MODEL_CONNECTOR_INVOCATION_SCHEMA_ID: &str = "epiphany.model_connector_invocation.v1"; +pub const MODEL_CONNECTOR_RESULT_SCHEMA_ID: &str = "epiphany.model_connector_result.v1"; +pub const MODEL_CONNECTOR_STATUS_SCHEMA_ID: &str = "epiphany.model_connector_status.v1"; #[derive(Debug, Clone, PartialEq, Eq, DatabaseEntry)] #[cultcache( @@ -62,6 +66,10 @@ pub struct EpiphanyModelRequest { pub source_worker_job_id: Option, #[cultcache(key = 15, default)] pub reasoning_basis_id: Option, + #[cultcache(key = 16, default)] + pub max_output_tokens: Option, + #[cultcache(key = 17, default)] + pub prompt_cache_key: Option, } impl EpiphanyModelRequest { @@ -89,10 +97,95 @@ impl EpiphanyModelRequest { output_schema_json: None, source_worker_job_id: None, reasoning_basis_id: None, + max_output_tokens: None, + prompt_cache_key: None, } } } +/// Encrypted CultNet cargo. The correlation fields are deliberately repeated +/// inside the authenticated plaintext and must match after decryption. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct EpiphanyModelConnectorEnvelope { + pub schema_id: String, + pub request_id: String, + pub message_kind: String, + pub nonce: Vec, + pub ciphertext: Vec, +} + +#[derive(Debug, Clone, PartialEq, DatabaseEntry)] +#[cultcache( + type = "epiphany.model_connector_invocation.v1", + schema = "EpiphanyModelConnectorInvocation" +)] +pub struct EpiphanyModelConnectorInvocation { + #[cultcache(key = 0)] + pub schema_id: String, + #[cultcache(key = 1)] + pub request_id: String, + #[cultcache(key = 2)] + pub caller_runtime_id: String, + #[cultcache(key = 3)] + pub expires_at_unix_ms: u64, + #[cultcache(key = 4)] + pub request: EpiphanyModelRequest, +} + +#[derive(Debug, Clone, PartialEq, DatabaseEntry)] +#[cultcache( + type = "epiphany.model_connector_result.v1", + schema = "EpiphanyModelConnectorResult" +)] +pub struct EpiphanyModelConnectorResult { + #[cultcache(key = 0)] + pub schema_id: String, + #[cultcache(key = 1)] + pub request_id: String, + #[cultcache(key = 2)] + pub accepted: bool, + #[cultcache(key = 3, default)] + pub events: Vec, + #[cultcache(key = 4, default)] + pub error: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, DatabaseEntry)] +#[cultcache( + type = "epiphany.model_connector_status.v1", + schema = "EpiphanyModelConnectorStatus" +)] +pub struct EpiphanyModelConnectorStatus { + #[cultcache(key = 0)] + pub schema_id: String, + #[cultcache(key = 1)] + pub provider_id: String, + #[cultcache(key = 2)] + pub capability_id: String, + #[cultcache(key = 3)] + pub request_schema_id: String, + #[cultcache(key = 4)] + pub result_schema_id: String, + #[cultcache(key = 5)] + pub envelope_schema_id: String, + #[cultcache(key = 6)] + pub transport_protocol: String, + #[cultcache(key = 7)] + pub host: String, + #[cultcache(key = 8)] + pub port: u16, + #[cultcache(key = 9)] + pub max_payload_bytes: u32, + #[cultcache(key = 10)] + pub max_parallel_requests: u32, + #[cultcache(key = 11)] + pub model: String, + #[cultcache(key = 12)] + pub ready: bool, + #[cultcache(key = 13)] + pub updated_at_utc: String, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct EpiphanyModelToolDefinition { pub name: String, @@ -179,6 +272,8 @@ pub struct EpiphanyModelReceipt { pub reasoning_output_tokens: Option, #[cultcache(key = 8, default)] pub transport: Option, + #[cultcache(key = 9, default)] + pub cached_input_tokens: Option, } impl EpiphanyModelReceipt { @@ -197,6 +292,7 @@ impl EpiphanyModelReceipt { output_tokens: None, reasoning_output_tokens: None, transport: None, + cached_input_tokens: None, } } } diff --git a/epiphany-openai-adapter/src/lib.rs b/epiphany-openai-adapter/src/lib.rs index 9b02de1d..f286ecf9 100644 --- a/epiphany-openai-adapter/src/lib.rs +++ b/epiphany-openai-adapter/src/lib.rs @@ -82,6 +82,10 @@ pub struct EpiphanyOpenAiModelRequest { pub provider_id: String, #[cultcache(key = 14)] pub wire_dialect: EpiphanyOpenAiWireDialect, + #[cultcache(key = 15, default)] + pub max_output_tokens: Option, + #[cultcache(key = 16, default)] + pub prompt_cache_key: Option, } impl EpiphanyOpenAiModelRequest { @@ -107,6 +111,8 @@ impl EpiphanyOpenAiModelRequest { output_schema_json: None, provider_id: "openai-codex".to_string(), wire_dialect: EpiphanyOpenAiWireDialect::Responses, + max_output_tokens: None, + prompt_cache_key: None, } } } @@ -143,6 +149,8 @@ pub fn request_from_native(request: &EpiphanyModelRequest) -> EpiphanyOpenAiMode } else { EpiphanyOpenAiWireDialect::Responses }, + max_output_tokens: request.max_output_tokens, + prompt_cache_key: request.prompt_cache_key.clone(), } } @@ -257,6 +265,8 @@ pub struct EpiphanyOpenAiModelReceipt { pub reasoning_output_tokens: Option, #[cultcache(key = 7, default)] pub transport: Option, + #[cultcache(key = 8, default)] + pub cached_input_tokens: Option, } impl EpiphanyOpenAiModelReceipt { @@ -270,6 +280,7 @@ impl EpiphanyOpenAiModelReceipt { output_tokens: None, reasoning_output_tokens: None, transport: None, + cached_input_tokens: None, } } } @@ -280,13 +291,14 @@ mod tests { #[test] fn native_provider_identity_selects_one_exact_wire_dialect() { - let openai = EpiphanyModelRequest::new( + let mut openai = EpiphanyModelRequest::new( "openai-request", "conversation", "openai-codex", "gpt-test", "decide", ); + openai.max_output_tokens = Some(512); let openrouter = EpiphanyModelRequest::new( "openrouter-request", "conversation", @@ -298,6 +310,7 @@ mod tests { let openai = request_from_native(&openai); let openrouter = request_from_native(&openrouter); assert_eq!(openai.provider_id, "openai-codex"); + assert_eq!(openai.max_output_tokens, Some(512)); assert_eq!(openai.wire_dialect, EpiphanyOpenAiWireDialect::Responses); assert_eq!(openrouter.provider_id, "openrouter"); assert_eq!( diff --git a/epiphany-openai-codex-spine/src/lib.rs b/epiphany-openai-codex-spine/src/lib.rs index 435be7ba..028e9f53 100644 --- a/epiphany-openai-codex-spine/src/lib.rs +++ b/epiphany-openai-codex-spine/src/lib.rs @@ -9,6 +9,11 @@ use codex_client::Request; use codex_client::ReqwestTransport; use codex_client::TransportError; use codex_client::sse_stream; +use codex_login::AuthCredentialsStoreMode; +use codex_login::AuthManager; +use codex_login::AuthMode; +use codex_login::CodexAuth; +use codex_login::default_client::build_reqwest_client; use epiphany_openai_adapter::EpiphanyOpenAiAdapterStatus; use epiphany_openai_adapter::EpiphanyOpenAiAuthMode; use epiphany_openai_adapter::EpiphanyOpenAiInputItem; @@ -18,11 +23,6 @@ use epiphany_openai_adapter::EpiphanyOpenAiStreamEvent; use epiphany_openai_adapter::EpiphanyOpenAiStreamPayload; use epiphany_openai_adapter::EpiphanyOpenAiWireDialect; use epiphany_openai_adapter::OPENAI_ADAPTER_STATUS_SCHEMA_ID; -use codex_login::AuthCredentialsStoreMode; -use codex_login::AuthManager; -use codex_login::AuthMode; -use codex_login::CodexAuth; -use codex_login::default_client::build_reqwest_client; use serde::Deserialize; use serde::Serialize; use serde_json::Value; @@ -391,6 +391,9 @@ pub fn chat_completions_body_from_epiphany( if let Some(effort) = request.reasoning_effort.as_deref() { body["reasoning"] = serde_json::json!({"effort": effort}); } + if let Some(max_output_tokens) = request.max_output_tokens { + body["max_tokens"] = serde_json::json!(max_output_tokens); + } Ok(body) } @@ -440,6 +443,12 @@ struct OpenRouterChatUsage { prompt_tokens: Option, completion_tokens: Option, completion_tokens_details: Option, + prompt_tokens_details: Option, +} + +#[derive(Debug, Deserialize)] +struct OpenRouterPromptTokenDetails { + cached_tokens: Option, } #[derive(Debug, Deserialize)] @@ -529,6 +538,10 @@ fn openrouter_events_from_chat_completion( .as_ref() .and_then(|item| item.completion_tokens_details.as_ref()) .and_then(|item| item.reasoning_tokens); + receipt.cached_input_tokens = usage + .as_ref() + .and_then(|item| item.prompt_tokens_details.as_ref()) + .and_then(|item| item.cached_tokens); receipt.transport = Some("openrouter-chat-completions".to_string()); push_openrouter_event( &mut events, @@ -600,7 +613,8 @@ pub fn responses_body_from_epiphany( include: Vec::new(), service_tier: parse_service_tier(request.service_tier.as_deref())?, previous_response_id: request.previous_response_id, - prompt_cache_key: None, + prompt_cache_key: request.prompt_cache_key, + max_output_tokens: request.max_output_tokens, text, client_metadata: None, }; @@ -1011,6 +1025,8 @@ struct EpiphanyResponsesBody { #[serde(skip_serializing_if = "Option::is_none")] prompt_cache_key: Option, #[serde(skip_serializing_if = "Option::is_none")] + max_output_tokens: Option, + #[serde(skip_serializing_if = "Option::is_none")] text: Option, #[serde(skip_serializing_if = "Option::is_none")] client_metadata: Option>, @@ -1113,9 +1129,15 @@ struct EpiphanyResponseCompleted { struct EpiphanyResponseCompletedUsage { input_tokens: i64, output_tokens: i64, + input_tokens_details: Option, output_tokens_details: Option, } +#[derive(Debug, Deserialize)] +struct EpiphanyResponseCompletedInputTokensDetails { + cached_tokens: i64, +} + #[derive(Debug, Deserialize)] struct EpiphanyResponseCompletedOutputTokensDetails { reasoning_tokens: i64, @@ -1257,6 +1279,9 @@ impl EpiphanyResponsesStreamState { receipt.reasoning_output_tokens = usage .output_tokens_details .and_then(|details| nonnegative_i64_to_u64(details.reasoning_tokens)); + receipt.cached_input_tokens = usage + .input_tokens_details + .and_then(|details| nonnegative_i64_to_u64(details.cached_tokens)); } self.push_payload(EpiphanyOpenAiStreamPayload::Completed { receipt }); } @@ -1547,11 +1572,13 @@ mod tests { }) .to_string(), ); + request.max_output_tokens = Some(768); let body = chat_completions_body_from_epiphany(&request).expect("request should map"); assert!(body.get("response_format").is_none()); assert_eq!(body["tool_choice"], "required"); + assert_eq!(body["max_tokens"], 768); assert_eq!(body["tools"].as_array().expect("tools").len(), 1); assert_eq!( body["tools"][0]["function"]["name"], @@ -1650,6 +1677,9 @@ mod tests { completion_tokens_details: Some(OpenRouterCompletionTokenDetails { reasoning_tokens: Some(1), }), + prompt_tokens_details: Some(OpenRouterPromptTokenDetails { + cached_tokens: Some(8), + }), }), }; @@ -1665,6 +1695,7 @@ mod tests { EpiphanyOpenAiStreamPayload::Completed { receipt } if receipt.transport.as_deref() == Some("openrouter-chat-completions") && receipt.input_tokens == Some(12) + && receipt.cached_input_tokens == Some(8) && receipt.reasoning_output_tokens == Some(1) )); } @@ -1693,6 +1724,8 @@ mod tests { request.reasoning_summary = Some("concise".to_string()); request.service_tier = Some("flex".to_string()); request.previous_response_id = Some("resp-1".to_string()); + request.max_output_tokens = Some(1_024); + request.prompt_cache_key = Some("ghostlight:narrator:v1".to_string()); request .tools .push(epiphany_openai_adapter::EpiphanyOpenAiToolDefinition { @@ -1721,6 +1754,8 @@ mod tests { assert_eq!(responses["stream"], true); assert_eq!(responses["store"], false); assert_eq!(responses["service_tier"], "flex"); + assert_eq!(responses["max_output_tokens"], 1_024); + assert_eq!(responses["prompt_cache_key"], "ghostlight:narrator:v1"); assert_eq!(responses["tools"].as_array().expect("tools").len(), 1); assert_eq!( responses["tools"][0]["name"], diff --git a/epiphany-openai-runtime/Cargo.toml b/epiphany-openai-runtime/Cargo.toml index 21a86974..24911dbc 100644 --- a/epiphany-openai-runtime/Cargo.toml +++ b/epiphany-openai-runtime/Cargo.toml @@ -14,6 +14,9 @@ path = "src/lib.rs" anyhow = "1" chrono = "0.4.44" cultcache-rs = { path = "../vendor/cultcache-rs" } +cultmesh-rs = { path = "../vendor/cultmesh-rs" } +cultnet-rs = { path = "../vendor/cultnet-rs" } +codex-login = { path = "../vendor/codex/codex-rs/login" } epiphany-core = { path = "../epiphany-core" } epiphany-model-adapter = { path = "../epiphany-model-adapter" } epiphany-openai-adapter = { path = "../epiphany-openai-adapter" } @@ -23,7 +26,7 @@ rmp-serde = "1" serde = { version = "1", features = ["derive"] } serde_json = "1" sha2 = "0.10" -tokio = { version = "1", features = ["macros", "rt-multi-thread"] } +tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "signal", "sync"] } uuid = { version = "1", features = ["v4"] } [dev-dependencies] diff --git a/epiphany-openai-runtime/src/bin/epiphany-model-connector.rs b/epiphany-openai-runtime/src/bin/epiphany-model-connector.rs new file mode 100644 index 00000000..8012974e --- /dev/null +++ b/epiphany-openai-runtime/src/bin/epiphany-model-connector.rs @@ -0,0 +1,7 @@ +use anyhow::Result; +use epiphany_openai_runtime::{EpiphanyModelConnectorOptions, serve_model_connector}; + +#[tokio::main] +async fn main() -> Result<()> { + serve_model_connector(EpiphanyModelConnectorOptions::from_env_args()?).await +} diff --git a/epiphany-openai-runtime/src/lib.rs b/epiphany-openai-runtime/src/lib.rs index 43f99f41..cc93faaa 100644 --- a/epiphany-openai-runtime/src/lib.rs +++ b/epiphany-openai-runtime/src/lib.rs @@ -56,6 +56,8 @@ use sha2::Digest; mod persona_executor; pub use persona_executor::*; +mod model_connector; +pub use model_connector::*; pub const OPENAI_RUNTIME_ROLE: &str = "openai-model-adapter"; pub const DEFAULT_PROVIDER_REQUEST_TIMEOUT: Duration = Duration::from_secs(90); @@ -1321,6 +1323,7 @@ pub fn model_receipt_from_openai_receipt( output_tokens: receipt.output_tokens, reasoning_output_tokens: receipt.reasoning_output_tokens, transport: receipt.transport.clone(), + cached_input_tokens: receipt.cached_input_tokens, } } diff --git a/epiphany-openai-runtime/src/model_connector.rs b/epiphany-openai-runtime/src/model_connector.rs new file mode 100644 index 00000000..af4984c6 --- /dev/null +++ b/epiphany-openai-runtime/src/model_connector.rs @@ -0,0 +1,916 @@ +use std::collections::{HashMap, HashSet}; +use std::net::{SocketAddr, TcpStream}; +use std::path::PathBuf; +use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; + +use anyhow::{Context, Result, bail}; +use cultmesh_rs::{CultMesh, CultMeshNodeOptions, CultMeshRudpDocumentPublishOptions}; +use cultnet_rs::{ + CultNetClientSecurityOptions, CultNetMessage, CultNetRawDocumentRecord, + CultNetRawPayloadEncoding, CultNetSecret, CultNetWireContract, TcpFramedTransportConnection, + TcpFramedTransportProfileOptions, create_tcp_framed_transport_profile, + decode_cultnet_message_from_slice, encode_cultnet_message_to_vec, +}; +use epiphany_model_adapter::{ + EpiphanyModelConnectorEnvelope, EpiphanyModelConnectorInvocation, EpiphanyModelConnectorResult, + EpiphanyModelConnectorStatus, EpiphanyModelRequest, EpiphanyModelStreamEvent, + EpiphanyModelStreamPayload, MODEL_ADAPTER_EVENT_SCHEMA_ID, MODEL_ADAPTER_REQUEST_SCHEMA_ID, + MODEL_CONNECTOR_ENVELOPE_SCHEMA_ID, MODEL_CONNECTOR_INVOCATION_SCHEMA_ID, + MODEL_CONNECTOR_RESULT_SCHEMA_ID, MODEL_CONNECTOR_STATUS_SCHEMA_ID, +}; +use epiphany_openai_codex_spine::{EpiphanyCodexOpenAiTransport, auth_manager}; +use sha2::{Digest, Sha256}; +use tokio::net::TcpListener; +use tokio::sync::{Mutex, Semaphore}; + +const CONNECTOR_PROVIDER_ID: &str = "epiphany.codex-model"; +const CONNECTOR_CAPABILITY_ID: &str = "model.generate.structured"; +const CONNECTOR_STATUS_KEY: &str = "epiphany.codex-model"; +const CONNECTOR_RUNTIME_ID: &str = "epiphany-model-connector"; +const CONNECTOR_REQUEST_KIND: &str = "model_request"; +const CONNECTOR_RESULT_KIND: &str = "model_result"; +const CONNECTOR_MAX_EXPIRY_SKEW_MS: u64 = 5 * 60 * 1_000; +const CONNECTOR_MAX_OUTPUT_TOKENS: u32 = 32_768; + +cultmesh_rs::cultmesh_documents!(ModelConnectorDocuments { + EpiphanyModelConnectorStatus => MODEL_CONNECTOR_STATUS_SCHEMA_ID, +}); + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct EpiphanyModelConnectorOptions { + pub bind: SocketAddr, + pub state_path: PathBuf, + pub codex_home: PathBuf, + pub connection_key_path: PathBuf, + pub allowed_caller_runtime_id: String, + pub model: String, + pub max_parallel_requests: u32, + pub max_payload_bytes: u32, + pub odin_endpoint: Option, +} + +impl EpiphanyModelConnectorOptions { + pub fn from_env_args() -> Result { + let mut options = Self { + bind: "127.0.0.1:4103".parse()?, + state_path: PathBuf::from("state/epiphany-model-connector.cc"), + codex_home: crate::default_codex_home()?, + connection_key_path: PathBuf::new(), + allowed_caller_runtime_id: "ghostlight-dungeon-yggdrasil".to_string(), + model: "gpt-5.4".to_string(), + max_parallel_requests: 8, + max_payload_bytes: 1_048_576, + odin_endpoint: Some("127.0.0.1:17871".parse()?), + }; + let mut args = std::env::args().skip(1); + while let Some(arg) = args.next() { + match arg.as_str() { + "--bind" => options.bind = next_arg(&mut args, "--bind")?.parse()?, + "--state" => options.state_path = PathBuf::from(next_arg(&mut args, "--state")?), + "--codex-home" => { + options.codex_home = PathBuf::from(next_arg(&mut args, "--codex-home")?) + } + "--connection-key-file" => { + options.connection_key_path = + PathBuf::from(next_arg(&mut args, "--connection-key-file")?) + } + "--allowed-caller" => { + options.allowed_caller_runtime_id = next_arg(&mut args, "--allowed-caller")? + } + "--model" => options.model = next_arg(&mut args, "--model")?, + "--max-parallel" => { + options.max_parallel_requests = + next_arg(&mut args, "--max-parallel")?.parse()? + } + "--max-payload-bytes" => { + options.max_payload_bytes = + next_arg(&mut args, "--max-payload-bytes")?.parse()? + } + "--odin" => { + let endpoint = next_arg(&mut args, "--odin")?; + options.odin_endpoint = if endpoint == "none" { + None + } else { + Some(endpoint.parse()?) + }; + } + other => bail!("unknown epiphany-model-connector argument {other:?}"), + } + } + options.validate()?; + Ok(options) + } + + fn validate(&self) -> Result<()> { + if !self.bind.ip().is_loopback() { + bail!("model connector must bind to a loopback address") + } + if self.connection_key_path.as_os_str().is_empty() { + bail!("--connection-key-file is required") + } + if self.allowed_caller_runtime_id.trim().is_empty() { + bail!("--allowed-caller must be non-empty") + } + if self.model.trim().is_empty() { + bail!("--model must be non-empty") + } + if self.max_parallel_requests == 0 { + bail!("--max-parallel must be greater than zero") + } + if self.max_payload_bytes < 4_096 { + bail!("--max-payload-bytes must be at least 4096") + } + Ok(()) + } +} + +fn next_arg(args: &mut impl Iterator, name: &str) -> Result { + args.next().with_context(|| format!("missing {name} value")) +} + +#[derive(Default)] +struct ReplayCache { + active: HashSet, + completed: HashMap, +} + +#[derive(Clone)] +struct CachedResponse { + request_digest: [u8; 32], + expires_at_unix_ms: u64, + document: CultNetRawDocumentRecord, +} + +pub async fn serve_model_connector(options: EpiphanyModelConnectorOptions) -> Result<()> { + options.validate()?; + let connection_key = read_secret_file(&options.connection_key_path)?; + let security = CultNetClientSecurityOptions::new(connection_key)?; + let listener = TcpListener::bind(options.bind) + .await + .with_context(|| format!("failed to bind model connector at {}", options.bind))?; + let local_addr = listener.local_addr()?; + let auth = auth_manager(options.codex_home.clone()); + if auth.auth().await.is_none() { + bail!( + "Codex subscription authentication is unavailable at {}", + options.codex_home.display() + ) + } + publish_connector_status(&options, local_addr)?; + + let profile = create_tcp_framed_transport_profile( + CONNECTOR_RUNTIME_ID, + TcpFramedTransportProfileOptions { + host: Some(local_addr.ip().to_string()), + port: Some(local_addr.port()), + max_payload_bytes: Some(options.max_payload_bytes), + ..TcpFramedTransportProfileOptions::default() + }, + ); + let permits = Arc::new(Semaphore::new(options.max_parallel_requests as usize)); + let replay = Arc::new(Mutex::new(ReplayCache::default())); + eprintln!( + "{CONNECTOR_RUNTIME_ID} ready at {local_addr} for {} with {} concurrent calls", + options.allowed_caller_runtime_id, options.max_parallel_requests + ); + + loop { + let (stream, peer) = listener.accept().await?; + if !peer.ip().is_loopback() { + continue; + } + let stream = stream.into_std()?; + stream.set_nonblocking(false)?; + let profile = profile.clone(); + let security = security.clone(); + let permits = permits.clone(); + let replay = replay.clone(); + let auth = auth.clone(); + let allowed_caller = options.allowed_caller_runtime_id.clone(); + let model = options.model.clone(); + tokio::spawn(async move { + let _permit = match permits.acquire_owned().await { + Ok(permit) => permit, + Err(_) => return, + }; + if let Err(error) = serve_one_connection( + stream, + profile, + security, + replay, + auth, + &allowed_caller, + &model, + ) + .await + { + eprintln!("model connector request refused: {error:#}"); + } + }); + } +} + +async fn serve_one_connection( + stream: TcpStream, + profile: cultnet_rs::CultNetTransportProfile, + security: CultNetClientSecurityOptions, + replay: Arc>, + auth: Arc, + allowed_caller: &str, + model: &str, +) -> Result<()> { + let (connection, request_document) = tokio::task::spawn_blocking(move || { + let mut connection = TcpFramedTransportConnection::new(stream, profile); + let frame = connection.receive()?; + let message = decode_cultnet_message_from_slice( + &frame.payload, + CultNetWireContract::CultNetSchemaV0, + )?; + let document = request_document_from_message(message)?; + Ok::<_, anyhow::Error>((connection, document)) + }) + .await??; + + let outer_request_id = request_document.record_key.clone(); + let request_digest: [u8; 32] = Sha256::digest(&request_document.payload).into(); + match claim_request(&replay, &outer_request_id, request_digest).await { + Ok(Some(cached)) => return send_document(connection, cached).await, + Ok(None) => {} + Err(error) => { + eprintln!("model connector replay refusal for {outer_request_id}: {error:#}"); + return send_error(connection, "model connector rejected the replay").await; + } + } + + let outcome = + process_request_document(&request_document, &security, auth, allowed_caller, model).await; + let (response, expires_at) = match outcome { + Ok(value) => value, + Err(error) => { + release_request(&replay, &outer_request_id).await; + return send_error(connection, "model connector rejected the invocation") + .await + .with_context(|| error.to_string()); + } + }; + complete_request( + &replay, + outer_request_id, + request_digest, + expires_at, + response.clone(), + ) + .await; + send_document(connection, response).await +} + +fn request_document_from_message(message: CultNetMessage) -> Result { + let CultNetMessage::DocumentPutRaw { + message_id, + document, + } = message + else { + bail!("model connector accepts only CultNet DocumentPutRaw invocations") + }; + if message_id != document.record_key { + bail!("model connector message id does not match the document key") + } + Ok(document) +} + +async fn claim_request( + cache: &Mutex, + request_id: &str, + request_digest: [u8; 32], +) -> Result> { + let now = unix_ms()?; + let mut cache = cache.lock().await; + cache + .completed + .retain(|_, response| response.expires_at_unix_ms >= now); + if let Some(response) = cache.completed.get(request_id) { + if response.request_digest != request_digest { + bail!("request id was reused with different encrypted cargo") + } + return Ok(Some(response.document.clone())); + } + if !cache.active.insert(request_id.to_string()) { + bail!("request is already in flight") + } + Ok(None) +} + +async fn release_request(cache: &Mutex, request_id: &str) { + cache.lock().await.active.remove(request_id); +} + +async fn complete_request( + cache: &Mutex, + request_id: String, + request_digest: [u8; 32], + expires_at_unix_ms: u64, + document: CultNetRawDocumentRecord, +) { + let mut cache = cache.lock().await; + cache.active.remove(&request_id); + cache.completed.insert( + request_id, + CachedResponse { + request_digest, + expires_at_unix_ms, + document, + }, + ); +} + +async fn process_request_document( + document: &CultNetRawDocumentRecord, + security: &CultNetClientSecurityOptions, + auth: Arc, + allowed_caller: &str, + allowed_model: &str, +) -> Result<(CultNetRawDocumentRecord, u64)> { + if document.schema_id != MODEL_CONNECTOR_ENVELOPE_SCHEMA_ID { + bail!("unexpected request envelope schema") + } + let envelope: EpiphanyModelConnectorEnvelope = + rmp_serde::from_slice(&document.payload).context("request envelope is malformed")?; + if envelope.schema_id != MODEL_CONNECTOR_ENVELOPE_SCHEMA_ID + || envelope.message_kind != CONNECTOR_REQUEST_KIND + || envelope.request_id != document.record_key + { + bail!("request envelope identity is inconsistent") + } + let plaintext = CultNetSecret::decrypt_bytes(&envelope.ciphertext, &envelope.nonce, security) + .context("request envelope authentication failed")?; + let invocation: EpiphanyModelConnectorInvocation = + rmp_serde::from_slice(&plaintext).context("request invocation is malformed")?; + let validation = validate_invocation( + &invocation, + &envelope.request_id, + allowed_caller, + allowed_model, + unix_ms()?, + ); + let result = match validation { + Ok(()) => execute_invocation(invocation.clone(), auth).await, + Err(error) => EpiphanyModelConnectorResult { + schema_id: MODEL_CONNECTOR_RESULT_SCHEMA_ID.to_string(), + request_id: invocation.request_id.clone(), + accepted: false, + events: Vec::new(), + error: Some(error.to_string()), + }, + }; + let response_envelope = encrypt_result(&result, security)?; + let response_payload = rmp_serde::to_vec_named(&response_envelope)?; + Ok(( + CultNetRawDocumentRecord { + schema_id: MODEL_CONNECTOR_ENVELOPE_SCHEMA_ID.to_string(), + record_key: invocation.request_id, + stored_at: now_utc(), + payload_encoding: CultNetRawPayloadEncoding::Messagepack, + payload: response_payload, + source_runtime_id: Some(CONNECTOR_RUNTIME_ID.to_string()), + source_agent_id: None, + source_role: Some("model-provider-connector".to_string()), + tags: Some(vec![CONNECTOR_CAPABILITY_ID.to_string()]), + }, + invocation.expires_at_unix_ms, + )) +} + +fn validate_invocation( + invocation: &EpiphanyModelConnectorInvocation, + outer_request_id: &str, + allowed_caller: &str, + allowed_model: &str, + now_unix_ms: u64, +) -> Result<()> { + if invocation.schema_id != MODEL_CONNECTOR_INVOCATION_SCHEMA_ID + || invocation.request.schema_id != MODEL_ADAPTER_REQUEST_SCHEMA_ID + { + bail!("unexpected model invocation schema") + } + if invocation.request_id != outer_request_id + || invocation.request.request_id != invocation.request_id + { + bail!("model invocation substituted request identity") + } + if invocation.caller_runtime_id != allowed_caller { + bail!("caller runtime is not admitted") + } + if invocation.expires_at_unix_ms < now_unix_ms + || invocation.expires_at_unix_ms > now_unix_ms + CONNECTOR_MAX_EXPIRY_SKEW_MS + { + bail!("model invocation is expired or exceeds the expiry horizon") + } + if invocation.request.provider != "openai-codex" { + bail!("connector admits only the openai-codex provider") + } + if invocation.request.model != allowed_model { + bail!("requested model is not admitted") + } + if invocation.request.instructions.trim().is_empty() { + bail!("model instructions must be non-empty") + } + if !invocation.request.tools.is_empty() { + bail!("shared model connector does not admit tool execution") + } + if invocation.request.previous_response_id.is_some() { + bail!("shared model connector admits stateless requests only") + } + if invocation.request.reasoning_summary.is_some() { + bail!("shared model connector does not expose reasoning summaries") + } + if matches!(invocation.request.max_output_tokens, Some(0)) + || invocation + .request + .max_output_tokens + .is_some_and(|value| value > CONNECTOR_MAX_OUTPUT_TOKENS) + { + bail!("max_output_tokens exceeds the connector bound") + } + if invocation + .request + .prompt_cache_key + .as_deref() + .is_some_and(|value| value.trim().is_empty() || value.len() > 128) + { + bail!("prompt_cache_key must contain at most 128 non-blank bytes") + } + Ok(()) +} + +async fn execute_invocation( + invocation: EpiphanyModelConnectorInvocation, + auth: Arc, +) -> EpiphanyModelConnectorResult { + let provider_request = epiphany_openai_adapter::request_from_native(&invocation.request); + let transport = EpiphanyCodexOpenAiTransport::openai(auth); + let events = match transport.collect_model_events(provider_request).await { + Ok(events) => sanitize_model_events( + &invocation.request, + events + .iter() + .map(|event| crate::model_event_from_openai_event("openai-codex", event)) + .collect(), + ), + Err(error) => { + eprintln!( + "model connector provider failure for {}: {error:#}", + invocation.request_id + ); + vec![EpiphanyModelStreamEvent { + schema_id: MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(), + request_id: invocation.request_id.clone(), + provider: "openai-codex".to_string(), + sequence: 0, + payload: EpiphanyModelStreamPayload::Failed { + message: "model provider transport failed".to_string(), + }, + }] + } + }; + EpiphanyModelConnectorResult { + schema_id: MODEL_CONNECTOR_RESULT_SCHEMA_ID.to_string(), + request_id: invocation.request_id, + accepted: true, + events, + error: None, + } +} + +fn sanitize_model_events( + request: &EpiphanyModelRequest, + events: Vec, +) -> Vec { + let mut sanitized = Vec::new(); + for event in events { + if event.request_id != request.request_id || event.provider != request.provider { + return vec![connector_failed_event( + &request.request_id, + "model provider substituted event identity", + )]; + } + match event.payload { + EpiphanyModelStreamPayload::ReasoningDelta { .. } => continue, + EpiphanyModelStreamPayload::ToolCall { .. } => { + return vec![connector_failed_event( + &request.request_id, + "model provider returned an inadmissible tool call", + )]; + } + payload => sanitized.push(EpiphanyModelStreamEvent { + schema_id: MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(), + request_id: request.request_id.clone(), + provider: request.provider.clone(), + sequence: sanitized.len() as u64, + payload, + }), + } + } + if sanitized.is_empty() { + sanitized.push(connector_failed_event( + &request.request_id, + "model provider returned no public events", + )); + } + sanitized +} + +fn connector_failed_event(request_id: &str, message: &str) -> EpiphanyModelStreamEvent { + EpiphanyModelStreamEvent { + schema_id: MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(), + request_id: request_id.to_string(), + provider: "openai-codex".to_string(), + sequence: 0, + payload: EpiphanyModelStreamPayload::Failed { + message: message.to_string(), + }, + } +} + +pub fn encrypt_invocation( + invocation: &EpiphanyModelConnectorInvocation, + security: &CultNetClientSecurityOptions, +) -> Result { + let nonce = CultNetSecret::new_nonce(); + let plaintext = rmp_serde::to_vec_named(invocation)?; + Ok(EpiphanyModelConnectorEnvelope { + schema_id: MODEL_CONNECTOR_ENVELOPE_SCHEMA_ID.to_string(), + request_id: invocation.request_id.clone(), + message_kind: CONNECTOR_REQUEST_KIND.to_string(), + ciphertext: CultNetSecret::encrypt_bytes(&plaintext, &nonce, security)?, + nonce: nonce.to_vec(), + }) +} + +fn encrypt_result( + result: &EpiphanyModelConnectorResult, + security: &CultNetClientSecurityOptions, +) -> Result { + let nonce = CultNetSecret::new_nonce(); + let plaintext = rmp_serde::to_vec_named(result)?; + Ok(EpiphanyModelConnectorEnvelope { + schema_id: MODEL_CONNECTOR_ENVELOPE_SCHEMA_ID.to_string(), + request_id: result.request_id.clone(), + message_kind: CONNECTOR_RESULT_KIND.to_string(), + ciphertext: CultNetSecret::encrypt_bytes(&plaintext, &nonce, security)?, + nonce: nonce.to_vec(), + }) +} + +pub fn decrypt_result( + envelope: &EpiphanyModelConnectorEnvelope, + security: &CultNetClientSecurityOptions, +) -> Result { + if envelope.schema_id != MODEL_CONNECTOR_ENVELOPE_SCHEMA_ID + || envelope.message_kind != CONNECTOR_RESULT_KIND + { + bail!("unexpected connector result envelope") + } + let plaintext = CultNetSecret::decrypt_bytes(&envelope.ciphertext, &envelope.nonce, security)?; + let result: EpiphanyModelConnectorResult = rmp_serde::from_slice(&plaintext)?; + if result.schema_id != MODEL_CONNECTOR_RESULT_SCHEMA_ID + || result.request_id != envelope.request_id + { + bail!("connector result substituted request identity") + } + Ok(result) +} + +async fn send_document( + connection: TcpFramedTransportConnection, + document: CultNetRawDocumentRecord, +) -> Result<()> { + let message = CultNetMessage::SnapshotResponseRaw { + message_id: document.record_key.clone(), + documents: vec![document], + }; + let payload = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0)?; + tokio::task::spawn_blocking(move || { + let mut connection = connection; + connection.send("schema", &payload) + }) + .await??; + Ok(()) +} + +async fn send_error( + connection: TcpFramedTransportConnection, + message: &str, +) -> Result<()> { + let payload = encode_cultnet_message_to_vec( + &CultNetMessage::Error { + error: message.to_string(), + }, + CultNetWireContract::CultNetSchemaV0, + )?; + tokio::task::spawn_blocking(move || { + let mut connection = connection; + connection.send("schema", &payload) + }) + .await??; + Ok(()) +} + +fn publish_connector_status( + options: &EpiphanyModelConnectorOptions, + bind: SocketAddr, +) -> Result<()> { + if let Some(parent) = options.state_path.parent() { + std::fs::create_dir_all(parent)?; + } + let status = EpiphanyModelConnectorStatus { + schema_id: MODEL_CONNECTOR_STATUS_SCHEMA_ID.to_string(), + provider_id: CONNECTOR_PROVIDER_ID.to_string(), + capability_id: CONNECTOR_CAPABILITY_ID.to_string(), + request_schema_id: MODEL_CONNECTOR_INVOCATION_SCHEMA_ID.to_string(), + result_schema_id: MODEL_CONNECTOR_RESULT_SCHEMA_ID.to_string(), + envelope_schema_id: MODEL_CONNECTOR_ENVELOPE_SCHEMA_ID.to_string(), + transport_protocol: "cultnet.tcp_framed.v0".to_string(), + host: bind.ip().to_string(), + port: bind.port(), + max_payload_bytes: options.max_payload_bytes, + max_parallel_requests: options.max_parallel_requests, + model: options.model.clone(), + ready: true, + updated_at_utc: now_utc(), + }; + let mut node = CultMesh::create_node( + &options.state_path, + ModelConnectorDocuments, + CultMeshNodeOptions { + runtime_id: CONNECTOR_RUNTIME_ID.to_string(), + pull_on_start: true, + }, + )?; + node.put(CONNECTOR_STATUS_KEY, &status)?; + node.flush()?; + if let Some(target) = options.odin_endpoint { + if let Err(error) = node.publish_document_to_rudp_catalog( + CONNECTOR_STATUS_KEY, + &status, + CultMeshRudpDocumentPublishOptions::odin(target, CONNECTOR_RUNTIME_ID), + ) { + eprintln!("model connector Odin publication failed: {error:#}"); + } + } + Ok(()) +} + +fn read_secret_file(path: &PathBuf) -> Result { + let raw = std::fs::read_to_string(path) + .with_context(|| format!("failed to read connector key {}", path.display()))?; + let key = raw.trim(); + if key.is_empty() || raw.trim_matches(['\r', '\n']) != key { + bail!("connector key file is empty or contains surrounding whitespace") + } + Ok(key.to_string()) +} + +fn unix_ms() -> Result { + Ok(SystemTime::now() + .duration_since(UNIX_EPOCH) + .context("system clock predates Unix epoch")? + .as_millis() + .try_into() + .context("system clock does not fit u64 milliseconds")?) +} + +fn now_utc() -> String { + chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true) +} + +#[cfg(test)] +mod tests { + use super::*; + use epiphany_model_adapter::EpiphanyModelReceipt; + + fn invocation(now: u64) -> EpiphanyModelConnectorInvocation { + let mut request = EpiphanyModelRequest::new( + "request-1", + "conversation-1", + "openai-codex", + "gpt-5.4", + "Return one bounded answer.", + ); + request.max_output_tokens = Some(512); + EpiphanyModelConnectorInvocation { + schema_id: MODEL_CONNECTOR_INVOCATION_SCHEMA_ID.to_string(), + request_id: request.request_id.clone(), + caller_runtime_id: "ghostlight-dungeon-yggdrasil".to_string(), + expires_at_unix_ms: now + 60_000, + request, + } + } + + #[test] + fn encrypted_connector_cargo_round_trips_without_exposing_plaintext() -> Result<()> { + let security = CultNetClientSecurityOptions::new("not-a-production-secret")?; + let invocation = invocation(1_000); + let envelope = encrypt_invocation(&invocation, &security)?; + assert!( + !envelope + .ciphertext + .windows(10) + .any(|part| part == b"conversation") + ); + let plaintext = + CultNetSecret::decrypt_bytes(&envelope.ciphertext, &envelope.nonce, &security)?; + let decoded: EpiphanyModelConnectorInvocation = rmp_serde::from_slice(&plaintext)?; + assert_eq!(decoded, invocation); + Ok(()) + } + + #[test] + fn connector_rejects_wrong_caller_expiry_tools_and_output_overrun() { + let now = 1_000; + let mut value = invocation(now); + assert!( + validate_invocation( + &value, + "request-1", + "ghostlight-dungeon-yggdrasil", + "gpt-5.4", + now, + ) + .is_ok() + ); + value.caller_runtime_id = "foreign-runtime".to_string(); + assert!( + validate_invocation( + &value, + "request-1", + "ghostlight-dungeon-yggdrasil", + "gpt-5.4", + now, + ) + .is_err() + ); + value = invocation(now); + value.expires_at_unix_ms = now - 1; + assert!( + validate_invocation( + &value, + "request-1", + "ghostlight-dungeon-yggdrasil", + "gpt-5.4", + now, + ) + .is_err() + ); + value = invocation(now); + value.request.max_output_tokens = Some(CONNECTOR_MAX_OUTPUT_TOKENS + 1); + assert!( + validate_invocation( + &value, + "request-1", + "ghostlight-dungeon-yggdrasil", + "gpt-5.4", + now, + ) + .is_err() + ); + value = invocation(now); + value.request.model = "another-model".to_string(); + assert!( + validate_invocation( + &value, + "request-1", + "ghostlight-dungeon-yggdrasil", + "gpt-5.4", + now, + ) + .is_err() + ); + value = invocation(now); + value + .request + .tools + .push(epiphany_model_adapter::EpiphanyModelToolDefinition { + name: "forbidden".to_string(), + description: "must not execute".to_string(), + parameters_json: "{}".to_string(), + }); + assert!( + validate_invocation( + &value, + "request-1", + "ghostlight-dungeon-yggdrasil", + "gpt-5.4", + now, + ) + .is_err() + ); + value = invocation(now); + value.request.previous_response_id = Some("provider-state".to_string()); + assert!( + validate_invocation( + &value, + "request-1", + "ghostlight-dungeon-yggdrasil", + "gpt-5.4", + now, + ) + .is_err() + ); + value = invocation(now); + value.request.reasoning_summary = Some("brief".to_string()); + assert!( + validate_invocation( + &value, + "request-1", + "ghostlight-dungeon-yggdrasil", + "gpt-5.4", + now, + ) + .is_err() + ); + } + + #[test] + fn connector_refuses_outer_message_and_document_identity_split() { + let document = CultNetRawDocumentRecord { + schema_id: MODEL_CONNECTOR_ENVELOPE_SCHEMA_ID.to_string(), + record_key: "request-1".to_string(), + stored_at: now_utc(), + payload_encoding: CultNetRawPayloadEncoding::Messagepack, + payload: Vec::new(), + source_runtime_id: None, + source_agent_id: None, + source_role: None, + tags: None, + }; + assert!( + request_document_from_message(CultNetMessage::DocumentPutRaw { + message_id: "request-2".to_string(), + document, + }) + .is_err() + ); + } + + #[test] + fn reasoning_is_removed_and_public_events_are_renumbered() { + let invocation = invocation(1_000); + let receipt = EpiphanyModelReceipt::new("request-1", "openai-codex", "gpt-5.4"); + let events = vec![ + EpiphanyModelStreamEvent { + schema_id: MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(), + request_id: "request-1".to_string(), + provider: "openai-codex".to_string(), + sequence: 0, + payload: EpiphanyModelStreamPayload::ReasoningDelta { + text: "private".to_string(), + }, + }, + EpiphanyModelStreamEvent { + schema_id: MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(), + request_id: "request-1".to_string(), + provider: "openai-codex".to_string(), + sequence: 1, + payload: EpiphanyModelStreamPayload::TextDelta { + text: "public".to_string(), + }, + }, + EpiphanyModelStreamEvent { + schema_id: MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(), + request_id: "request-1".to_string(), + provider: "openai-codex".to_string(), + sequence: 2, + payload: EpiphanyModelStreamPayload::Completed { receipt }, + }, + ]; + let events = sanitize_model_events(&invocation.request, events); + assert_eq!(events.len(), 2); + assert_eq!(events[0].sequence, 0); + assert_eq!(events[1].sequence, 1); + assert!(matches!( + events[0].payload, + EpiphanyModelStreamPayload::TextDelta { .. } + )); + assert!(matches!( + events[1].payload, + EpiphanyModelStreamPayload::Completed { .. } + )); + } + + #[test] + fn encrypted_result_refuses_outer_identity_substitution() -> Result<()> { + let security = CultNetClientSecurityOptions::new("not-a-production-secret")?; + let result = EpiphanyModelConnectorResult { + schema_id: MODEL_CONNECTOR_RESULT_SCHEMA_ID.to_string(), + request_id: "request-1".to_string(), + accepted: true, + events: Vec::new(), + error: None, + }; + let mut envelope = encrypt_result(&result, &security)?; + envelope.request_id = "request-2".to_string(); + assert!(decrypt_result(&envelope, &security).is_err()); + Ok(()) + } +} diff --git a/schemas/cultnet/epiphany.model-connector-envelope.schema.json b/schemas/cultnet/epiphany.model-connector-envelope.schema.json new file mode 100644 index 00000000..f20df35d --- /dev/null +++ b/schemas/cultnet/epiphany.model-connector-envelope.schema.json @@ -0,0 +1,24 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://gamecult.dev/epiphany/cultnet/epiphany.model-connector-envelope.schema.json", + "title": "Epiphany Model Connector Encrypted Envelope", + "type": "object", + "required": ["schema_id", "request_id", "message_kind", "nonce", "ciphertext"], + "properties": { + "schema_id": { "const": "epiphany.model_connector_envelope.v1" }, + "request_id": { "type": "string", "minLength": 1 }, + "message_kind": { "enum": ["model_request", "model_result"] }, + "nonce": { + "type": "array", + "items": { "type": "integer", "minimum": 0, "maximum": 255 }, + "minItems": 12, + "maxItems": 12 + }, + "ciphertext": { + "type": "array", + "items": { "type": "integer", "minimum": 0, "maximum": 255 }, + "minItems": 1 + } + }, + "additionalProperties": false +} diff --git a/schemas/cultnet/epiphany.model-connector-invocation.schema.json b/schemas/cultnet/epiphany.model-connector-invocation.schema.json new file mode 100644 index 00000000..e46bf586 --- /dev/null +++ b/schemas/cultnet/epiphany.model-connector-invocation.schema.json @@ -0,0 +1,15 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://gamecult.dev/epiphany/cultnet/epiphany.model-connector-invocation.schema.json", + "title": "Epiphany Model Connector Invocation", + "type": "object", + "required": ["schema_id", "request_id", "caller_runtime_id", "expires_at_unix_ms", "request"], + "properties": { + "schema_id": { "const": "epiphany.model_connector_invocation.v1" }, + "request_id": { "type": "string", "minLength": 1 }, + "caller_runtime_id": { "type": "string", "minLength": 1 }, + "expires_at_unix_ms": { "type": "integer", "minimum": 1 }, + "request": { "$ref": "epiphany.model-request.schema.json" } + }, + "additionalProperties": false +} diff --git a/schemas/cultnet/epiphany.model-connector-result.schema.json b/schemas/cultnet/epiphany.model-connector-result.schema.json new file mode 100644 index 00000000..2b04bd77 --- /dev/null +++ b/schemas/cultnet/epiphany.model-connector-result.schema.json @@ -0,0 +1,18 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://gamecult.dev/epiphany/cultnet/epiphany.model-connector-result.schema.json", + "title": "Epiphany Model Connector Result", + "type": "object", + "required": ["schema_id", "request_id", "accepted", "events"], + "properties": { + "schema_id": { "const": "epiphany.model_connector_result.v1" }, + "request_id": { "type": "string", "minLength": 1 }, + "accepted": { "type": "boolean" }, + "events": { + "type": "array", + "items": { "$ref": "epiphany.model-stream-event.schema.json" } + }, + "error": { "type": ["string", "null"] } + }, + "additionalProperties": false +} diff --git a/schemas/cultnet/epiphany.model-connector-status.schema.json b/schemas/cultnet/epiphany.model-connector-status.schema.json new file mode 100644 index 00000000..0ef2a8ac --- /dev/null +++ b/schemas/cultnet/epiphany.model-connector-status.schema.json @@ -0,0 +1,39 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://gamecult.dev/epiphany/cultnet/epiphany.model-connector-status.schema.json", + "title": "Epiphany Model Connector Status", + "type": "object", + "required": [ + "schema_id", + "provider_id", + "capability_id", + "request_schema_id", + "result_schema_id", + "envelope_schema_id", + "transport_protocol", + "host", + "port", + "max_payload_bytes", + "max_parallel_requests", + "model", + "ready", + "updated_at_utc" + ], + "properties": { + "schema_id": { "const": "epiphany.model_connector_status.v1" }, + "provider_id": { "const": "epiphany.codex-model" }, + "capability_id": { "const": "model.generate.structured" }, + "request_schema_id": { "const": "epiphany.model_connector_invocation.v1" }, + "result_schema_id": { "const": "epiphany.model_connector_result.v1" }, + "envelope_schema_id": { "const": "epiphany.model_connector_envelope.v1" }, + "transport_protocol": { "const": "cultnet.tcp_framed.v0" }, + "host": { "type": "string", "minLength": 1 }, + "port": { "type": "integer", "minimum": 1, "maximum": 65535 }, + "max_payload_bytes": { "type": "integer", "minimum": 4096 }, + "max_parallel_requests": { "type": "integer", "minimum": 1 }, + "model": { "type": "string", "minLength": 1 }, + "ready": { "type": "boolean" }, + "updated_at_utc": { "type": "string", "format": "date-time" } + }, + "additionalProperties": false +} diff --git a/schemas/cultnet/epiphany.model-receipt.schema.json b/schemas/cultnet/epiphany.model-receipt.schema.json index 314f1094..8f7fd3e6 100644 --- a/schemas/cultnet/epiphany.model-receipt.schema.json +++ b/schemas/cultnet/epiphany.model-receipt.schema.json @@ -13,7 +13,8 @@ "input_tokens": { "type": ["integer", "null"], "minimum": 0 }, "output_tokens": { "type": ["integer", "null"], "minimum": 0 }, "reasoning_output_tokens": { "type": ["integer", "null"], "minimum": 0 }, - "transport": { "type": ["string", "null"] } + "transport": { "type": ["string", "null"] }, + "cached_input_tokens": { "type": ["integer", "null"], "minimum": 0 } }, "additionalProperties": false } diff --git a/schemas/cultnet/epiphany.model-request.schema.json b/schemas/cultnet/epiphany.model-request.schema.json index 8474b826..2c4f67eb 100644 --- a/schemas/cultnet/epiphany.model-request.schema.json +++ b/schemas/cultnet/epiphany.model-request.schema.json @@ -34,7 +34,10 @@ } }, "output_schema_json": { "type": ["string", "null"] }, - "source_worker_job_id": { "type": ["string", "null"] } + "source_worker_job_id": { "type": ["string", "null"] }, + "reasoning_basis_id": { "type": ["string", "null"] }, + "max_output_tokens": { "type": ["integer", "null"], "minimum": 1 }, + "prompt_cache_key": { "type": ["string", "null"], "minLength": 1 } }, "additionalProperties": false, "$defs": { diff --git a/schemas/cultnet/epiphany.openai-model-receipt.schema.json b/schemas/cultnet/epiphany.openai-model-receipt.schema.json index 861c4a8d..a5049c65 100644 --- a/schemas/cultnet/epiphany.openai-model-receipt.schema.json +++ b/schemas/cultnet/epiphany.openai-model-receipt.schema.json @@ -12,7 +12,8 @@ "input_tokens": { "type": ["integer", "null"], "minimum": 0 }, "output_tokens": { "type": ["integer", "null"], "minimum": 0 }, "reasoning_output_tokens": { "type": ["integer", "null"], "minimum": 0 }, - "transport": { "type": ["string", "null"] } + "transport": { "type": ["string", "null"] }, + "cached_input_tokens": { "type": ["integer", "null"], "minimum": 0 } }, "additionalProperties": false } diff --git a/schemas/cultnet/epiphany.openai-model-request.schema.json b/schemas/cultnet/epiphany.openai-model-request.schema.json index eaaa522d..de678d80 100644 --- a/schemas/cultnet/epiphany.openai-model-request.schema.json +++ b/schemas/cultnet/epiphany.openai-model-request.schema.json @@ -25,7 +25,9 @@ }, "output_schema_json": { "type": ["string", "null"] }, "provider_id": { "enum": ["openai-codex", "openai", "openrouter"] }, - "wire_dialect": { "enum": ["Responses", "ChatCompletionsTerminalTool"] } + "wire_dialect": { "enum": ["Responses", "ChatCompletionsTerminalTool"] }, + "max_output_tokens": { "type": ["integer", "null"], "minimum": 1 }, + "prompt_cache_key": { "type": ["string", "null"], "minLength": 1 } }, "additionalProperties": false, "$defs": { diff --git a/vendor/cultnet-rs b/vendor/cultnet-rs index 69f2498d..6bf659b0 160000 --- a/vendor/cultnet-rs +++ b/vendor/cultnet-rs @@ -1 +1 @@ -Subproject commit 69f2498daa5a2e3fe9bb9c532c95b75a5f0f61bb +Subproject commit 6bf659b02fea0f5896856b5c13de7cf18e4c40d8