GameCult/Epiphany
Archived
Watch
9
0
Fork
You've already forked Epiphany
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Delete mirrored provider event family

Browse source
This commit is contained in:
Meta Vens 2026-08-25 01:33:40 +02:00
commit b83c041670
4 changed files with 166 additions and 313 deletions
Download patch file Download diff file Expand all files Collapse all files

190
epiphany-openai-adapter/src/lib.rs
View file

@ -1,7 +1,10 @@
use anyhow::{Result, anyhow};
use codex_connector::{CodexInputItem, CodexProviderRequest, CodexToolChoice, CodexToolDefinition};
use cultcache_rs::DatabaseEntry;
use epiphany_model_adapter::{EpiphanyModelInputItem, EpiphanyModelRequest};
use epiphany_model_adapter::{
EpiphanyModelInputItem, EpiphanyModelReceipt, EpiphanyModelRequest, EpiphanyModelStreamEvent,
EpiphanyModelStreamPayload, EpiphanyModelToolDefinition, MODEL_ADAPTER_EVENT_SCHEMA_ID,
};
use serde::Deserialize;
use serde::Serialize;
use sha2::{Digest, Sha256};
@ -19,7 +22,7 @@ pub struct EpiphanyProviderRequest {
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum EpiphanyProviderRequestPayload {
Codex(CodexProviderRequest),
OpenRouter(EpiphanyOpenAiModelRequest),
OpenRouter(EpiphanyOpenRouterRequest),
}
impl EpiphanyProviderRequest {
@ -46,22 +49,22 @@ impl EpiphanyProviderRequest {
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct EpiphanyOpenAiModelRequest {
pub struct EpiphanyOpenRouterRequest {
pub request_id: String,
pub conversation_id: String,
pub model: String,
pub instructions: String,
pub input: Vec<EpiphanyOpenAiInputItem>,
pub input: Vec<EpiphanyModelInputItem>,
pub reasoning_effort: Option<String>,
pub reasoning_summary: Option<String>,
pub service_tier: Option<String>,
pub output_contract_id: Option<String>,
pub previous_response_id: Option<String>,
pub tools: Vec<EpiphanyOpenAiToolDefinition>,
pub tools: Vec<EpiphanyModelToolDefinition>,
pub output_schema_json: Option<String>,
}
impl EpiphanyOpenAiModelRequest {
impl EpiphanyOpenRouterRequest {
pub fn new(
request_id: impl Into<String>,
conversation_id: impl Into<String>,
@ -99,27 +102,19 @@ pub fn request_from_native(request: &EpiphanyModelRequest) -> Result<EpiphanyPro
Ok(EpiphanyProviderRequest { payload })
}
fn openrouter_request_from_native(request: &EpiphanyModelRequest) -> EpiphanyOpenAiModelRequest {
EpiphanyOpenAiModelRequest {
fn openrouter_request_from_native(request: &EpiphanyModelRequest) -> EpiphanyOpenRouterRequest {
EpiphanyOpenRouterRequest {
request_id: request.request_id.clone(),
conversation_id: request.conversation_id.clone(),
model: request.model.clone(),
instructions: request.instructions.clone(),
input: request.input.iter().map(input_from_native).collect(),
input: request.input.clone(),
reasoning_effort: request.reasoning_effort.clone(),
reasoning_summary: request.reasoning_summary.clone(),
service_tier: request.service_tier.clone(),
output_contract_id: request.output_contract_id.clone(),
previous_response_id: request.previous_response_id.clone(),
tools: request
.tools
.iter()
.map(|tool| EpiphanyOpenAiToolDefinition {
name: tool.name.clone(),
description: tool.description.clone(),
parameters_json: tool.parameters_json.clone(),
})
.collect(),
tools: request.tools.clone(),
output_schema_json: request.output_schema_json.clone(),
}
}
@ -204,119 +199,6 @@ fn codex_input_from_native(input: &EpiphanyModelInputItem) -> CodexInputItem {
}
}
fn input_from_native(input: &EpiphanyModelInputItem) -> EpiphanyOpenAiInputItem {
match input {
EpiphanyModelInputItem::UserText { text } => {
EpiphanyOpenAiInputItem::UserText { text: text.clone() }
}
EpiphanyModelInputItem::AssistantText { text } => {
EpiphanyOpenAiInputItem::AssistantText { text: text.clone() }
}
EpiphanyModelInputItem::ToolCall {
call_id,
name,
arguments,
} => EpiphanyOpenAiInputItem::ToolCall {
call_id: call_id.clone(),
name: name.clone(),
arguments: arguments.clone(),
},
EpiphanyModelInputItem::ToolResult { call_id, output } => {
EpiphanyOpenAiInputItem::ToolResult {
call_id: call_id.clone(),
output: output.clone(),
}
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct EpiphanyOpenAiToolDefinition {
pub name: String,
pub description: String,
pub parameters_json: String,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum EpiphanyOpenAiInputItem {
UserText {
text: String,
},
AssistantText {
text: String,
},
ToolCall {
call_id: String,
name: String,
arguments: String,
},
ToolResult {
call_id: String,
output: String,
},
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct EpiphanyOpenAiStreamEvent {
pub request_id: String,
pub sequence: u64,
pub payload: EpiphanyOpenAiStreamPayload,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum EpiphanyOpenAiStreamPayload {
TextDelta {
text: String,
},
ReasoningDelta {
text: String,
},
ToolCall {
call_id: String,
name: String,
arguments: String,
},
Completed {
receipt: Box<EpiphanyOpenAiModelReceipt>,
},
Failed {
message: String,
},
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct EpiphanyOpenAiModelReceipt {
pub request_id: String,
pub model: String,
pub response_id: Option<String>,
pub input_tokens: Option<u64>,
pub output_tokens: Option<u64>,
pub reasoning_output_tokens: Option<u64>,
pub transport: Option<String>,
pub caller_runtime_id: Option<String>,
pub native_request_sha256: Option<String>,
pub provider_request_sha256: Option<String>,
pub cached_input_tokens: Option<u64>,
}
impl EpiphanyOpenAiModelReceipt {
pub fn new(request_id: impl Into<String>, model: impl Into<String>) -> Self {
Self {
request_id: request_id.into(),
model: model.into(),
response_id: None,
input_tokens: None,
output_tokens: None,
reasoning_output_tokens: None,
transport: None,
caller_runtime_id: None,
native_request_sha256: None,
provider_request_sha256: None,
cached_input_tokens: None,
}
}
}
/// Project Epiphany's canonical JSON Schema into the strict subset accepted by
/// model providers. Native decoding remains the final validation authority.
pub fn strict_provider_schema(schema_json: &str) -> anyhow::Result<serde_json::Value> {
@ -375,12 +257,12 @@ pub const OPENROUTER_TERMINAL_TOOL_NAME: &str = "epiphany_submit_typed_result";
/// Lower one typed Epiphany request to OpenRouter's terminal-tool dialect.
pub fn openrouter_request_body(
request: &EpiphanyOpenAiModelRequest,
request: &EpiphanyOpenRouterRequest,
) -> anyhow::Result<serde_json::Value> {
let has_tool_result = request
.input
.iter()
.any(|item| matches!(item, EpiphanyOpenAiInputItem::ToolResult { .. }));
.any(|item| matches!(item, EpiphanyModelInputItem::ToolResult { .. }));
let output_schema = request
.output_schema_json
.as_deref()
@ -404,13 +286,13 @@ pub fn openrouter_request_body(
})];
for item in &request.input {
messages.push(match item {
EpiphanyOpenAiInputItem::UserText { text } => {
EpiphanyModelInputItem::UserText { text } => {
serde_json::json!({"role": "user", "content": text})
}
EpiphanyOpenAiInputItem::AssistantText { text } => {
EpiphanyModelInputItem::AssistantText { text } => {
serde_json::json!({"role": "assistant", "content": text})
}
EpiphanyOpenAiInputItem::ToolCall {
EpiphanyModelInputItem::ToolCall {
call_id,
name,
arguments,
@ -422,7 +304,7 @@ pub fn openrouter_request_body(
"function": {"name": name, "arguments": arguments},
}],
}),
EpiphanyOpenAiInputItem::ToolResult { call_id, output } => serde_json::json!({
EpiphanyModelInputItem::ToolResult { call_id, output } => serde_json::json!({
"role": "tool",
"tool_call_id": call_id,
"content": output,
@ -484,9 +366,9 @@ pub fn openrouter_request_body(
/// event family. The model transcript remains optional; the terminal receipt
/// is authoritative transport evidence.
pub fn openrouter_events_from_response(
request: &EpiphanyOpenAiModelRequest,
request: &EpiphanyOpenRouterRequest,
response_bytes: &[u8],
) -> anyhow::Result<Vec<EpiphanyOpenAiStreamEvent>> {
) -> anyhow::Result<Vec<EpiphanyModelStreamEvent>> {
let response: OpenRouterResponse = serde_json::from_slice(response_bytes)
.map_err(|error| anyhow::anyhow!("OpenRouter returned an invalid response: {error}"))?;
let choice = response
@ -499,7 +381,7 @@ pub fn openrouter_events_from_response(
push_openrouter_event(
&mut events,
&request.request_id,
EpiphanyOpenAiStreamPayload::ReasoningDelta { text: reasoning },
EpiphanyModelStreamPayload::ReasoningDelta { text: reasoning },
);
}
let terminal_calls = choice
@ -526,7 +408,7 @@ pub fn openrouter_events_from_response(
push_openrouter_event(
&mut events,
&request.request_id,
EpiphanyOpenAiStreamPayload::TextDelta {
EpiphanyModelStreamPayload::TextDelta {
text: arguments.clone(),
},
);
@ -535,14 +417,14 @@ pub fn openrouter_events_from_response(
push_openrouter_event(
&mut events,
&request.request_id,
EpiphanyOpenAiStreamPayload::TextDelta { text: content },
EpiphanyModelStreamPayload::TextDelta { text: content },
);
}
for call in choice.message.tool_calls {
push_openrouter_event(
&mut events,
&request.request_id,
EpiphanyOpenAiStreamPayload::ToolCall {
EpiphanyModelStreamPayload::ToolCall {
call_id: call.id,
name: call.function.name,
arguments: call.function.arguments,
@ -556,11 +438,12 @@ pub fn openrouter_events_from_response(
));
}
let usage = response.usage;
let mut receipt = EpiphanyOpenAiModelReceipt::new(
let mut receipt = EpiphanyModelReceipt::new(
&request.request_id,
"openrouter",
response.model.unwrap_or_else(|| request.model.clone()),
);
receipt.response_id = response.id;
receipt.provider_response_id = response.id;
receipt.input_tokens = usage.as_ref().and_then(|usage| usage.prompt_tokens);
receipt.output_tokens = usage.as_ref().and_then(|usage| usage.completion_tokens);
receipt.reasoning_output_tokens = usage
@ -571,7 +454,7 @@ pub fn openrouter_events_from_response(
push_openrouter_event(
&mut events,
&request.request_id,
EpiphanyOpenAiStreamPayload::Completed {
EpiphanyModelStreamPayload::Completed {
receipt: Box::new(receipt),
},
);
@ -625,12 +508,14 @@ struct OpenRouterCompletionTokenDetails {
}
fn push_openrouter_event(
events: &mut Vec<EpiphanyOpenAiStreamEvent>,
events: &mut Vec<EpiphanyModelStreamEvent>,
request_id: &str,
payload: EpiphanyOpenAiStreamPayload,
payload: EpiphanyModelStreamPayload,
) {
events.push(EpiphanyOpenAiStreamEvent {
events.push(EpiphanyModelStreamEvent {
schema_id: MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(),
request_id: request_id.into(),
provider: "openrouter".into(),
sequence: events.len() as u64,
payload,
});
@ -1002,8 +887,7 @@ mod tests {
#[test]
fn openrouter_terminal_tool_preserves_structured_decision() {
let mut request =
EpiphanyOpenAiModelRequest::new("request", "conversation", "ox", "decide");
let mut request = EpiphanyOpenRouterRequest::new("request", "conversation", "ox", "decide");
request.output_schema_json = Some(
r#"{"type":"object","properties":{"verdict":{"type":"string"}},"required":["verdict"]}"#.into(),
);
@ -1027,11 +911,11 @@ mod tests {
.unwrap();
assert!(matches!(
&events[0].payload,
EpiphanyOpenAiStreamPayload::TextDelta { text } if text == "{\"verdict\":\"pass\"}"
EpiphanyModelStreamPayload::TextDelta { text } if text == "{\"verdict\":\"pass\"}"
));
assert!(matches!(
&events[1].payload,
EpiphanyOpenAiStreamPayload::Completed { receipt }
EpiphanyModelStreamPayload::Completed { receipt }
if receipt.transport.as_deref() == Some("openrouter-chat-completions")
));
}

32
epiphany-openai-runtime/src/bin/epiphany-openai-runtime.rs
View file

@ -17,10 +17,12 @@ use anyhow::Result;
use anyhow::anyhow;
use epiphany_model_adapter::EpiphanyModelRequest;
#[cfg(test)]
use epiphany_openai_adapter::{
EpiphanyOpenAiModelReceipt, EpiphanyOpenAiModelRequest, EpiphanyOpenAiStreamEvent,
EpiphanyOpenAiStreamPayload, EpiphanyProviderRequestPayload,
use epiphany_model_adapter::{
EpiphanyModelReceipt, EpiphanyModelStreamEvent, EpiphanyModelStreamPayload,
MODEL_ADAPTER_EVENT_SCHEMA_ID,
};
#[cfg(test)]
use epiphany_openai_adapter::{EpiphanyOpenRouterRequest, EpiphanyProviderRequestPayload};
use epiphany_openai_runtime::DEFAULT_CODEX_CONNECTOR_ENDPOINT;
use epiphany_openai_runtime::DEFAULT_PROVIDER_REQUEST_TIMEOUT;
use epiphany_openai_runtime::EpiphanyOpenAiRuntimeOptions;
@ -993,9 +995,9 @@ mod tests {
},
};
let events = provider_transport::events_from_connector_result(result)?;
let events = provider_transport::events_from_connector_result("openai-codex", result)?;
assert_eq!(events.len(), 1);
let EpiphanyOpenAiStreamPayload::Completed { receipt } = &events[0].payload else {
let EpiphanyModelStreamPayload::Completed { receipt } = &events[0].payload else {
panic!("connector result must produce one terminal receipt")
};
let native_digest = "11".repeat(32);
@ -1038,7 +1040,7 @@ mod tests {
];
for (index, schema) in schemas.into_iter().enumerate() {
let mut request = EpiphanyOpenAiModelRequest::new(
let mut request = EpiphanyOpenRouterRequest::new(
format!("strict-worker-schema-{index}"),
"strict-worker-schema",
"gpt-5.4",
@ -1056,7 +1058,7 @@ mod tests {
#[test]
fn research_contract_projects_typed_decision_to_provider_strict_shape() -> Result<()> {
let mut request = EpiphanyOpenAiModelRequest::new(
let mut request = EpiphanyOpenRouterRequest::new(
"strict-research-schema",
"strict-worker-schema",
"gpt-5.4",
@ -1127,8 +1129,8 @@ mod tests {
},
&model_request,
)?;
let mut receipt = EpiphanyOpenAiModelReceipt::new(request_id, "gpt-test");
receipt.response_id = Some(format!("response-{request_id}"));
let mut receipt = EpiphanyModelReceipt::new(request_id, DEFAULT_PROVIDER, "gpt-test");
receipt.provider_response_id = Some(format!("response-{request_id}"));
receipt.transport = Some("test".to_string());
record_model_turn_events(
store,
@ -1146,19 +1148,23 @@ mod tests {
},
&model_request,
&[
EpiphanyOpenAiStreamEvent {
EpiphanyModelStreamEvent {
schema_id: MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(),
request_id: request_id.to_string(),
provider: DEFAULT_PROVIDER.to_string(),
sequence: 0,
payload: EpiphanyOpenAiStreamPayload::ToolCall {
payload: EpiphanyModelStreamPayload::ToolCall {
call_id: format!("call-{request_id}"),
name: tool_name.to_string(),
arguments: arguments.to_string(),
},
},
EpiphanyOpenAiStreamEvent {
EpiphanyModelStreamEvent {
schema_id: MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(),
request_id: request_id.to_string(),
provider: DEFAULT_PROVIDER.to_string(),
sequence: 1,
payload: EpiphanyOpenAiStreamPayload::Completed {
payload: EpiphanyModelStreamPayload::Completed {
receipt: Box::new(receipt),
},
},

195
epiphany-openai-runtime/src/lib.rs
View file

@ -24,13 +24,10 @@ use epiphany_core::runtime_identity;
use epiphany_core::runtime_spine_cache;
use epiphany_core::terminalize_runtime_job;
use epiphany_model_adapter::EpiphanyModelInputItem;
use epiphany_model_adapter::EpiphanyModelReceipt;
use epiphany_model_adapter::EpiphanyModelRequest;
use epiphany_model_adapter::EpiphanyModelStreamEvent;
use epiphany_model_adapter::EpiphanyModelStreamPayload;
use epiphany_model_adapter::EpiphanyModelToolDefinition;
use epiphany_openai_adapter::EpiphanyOpenAiStreamEvent;
use epiphany_openai_adapter::EpiphanyOpenAiStreamPayload;
use epiphany_openai_adapter::EpiphanyProviderRequest;
use epiphany_tool_adapter::EPIPHANY_TOOL_RUNTIME_ADAPTER_ID;
use epiphany_tool_adapter::EpiphanyToolInvocationIntent;
@ -120,12 +117,12 @@ pub fn record_model_turn_events(
store_path: impl AsRef<Path>,
options: &EpiphanyOpenAiRuntimeOptions,
request: &EpiphanyModelRequest,
events: &[EpiphanyOpenAiStreamEvent],
events: &[EpiphanyModelStreamEvent],
) -> Result<EpiphanyOpenAiRuntimeRunSummary> {
let store_path = store_path.as_ref();
let provider_request = provider_request_from_model_request(request)?;
let events = compact_openai_events_for_storage(events);
let tool_intents = tool_invocation_intents_from_openai_events(&request.provider, &events);
let events = compact_model_events_for_storage(events);
let tool_intents = tool_invocation_intents_from_model_events(&events);
for intent in &tool_intents {
put_runtime_tool_execution_intent(
store_path,
@ -140,19 +137,15 @@ pub fn record_model_turn_events(
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
for event in &events {
let model_event = model_event_from_openai_event(&request.provider, event);
cache.put(
model_event_key(&model_event.request_id, model_event.sequence),
&model_event,
)?;
if let EpiphanyModelStreamPayload::Completed { receipt } = &model_event.payload {
cache.put(model_event_key(&event.request_id, event.sequence), event)?;
if let EpiphanyModelStreamPayload::Completed { receipt } = &event.payload {
cache.put(model_receipt_key(&receipt.request_id), receipt.as_ref())?;
}
match &event.payload {
EpiphanyOpenAiStreamPayload::Completed { receipt: completed } => {
EpiphanyModelStreamPayload::Completed { receipt: completed } => {
receipt = Some(completed.as_ref().clone());
}
EpiphanyOpenAiStreamPayload::Failed { message } => {
EpiphanyModelStreamPayload::Failed { message } => {
failure = Some(message.clone());
}
_ => {}
@ -207,19 +200,19 @@ pub fn record_model_turn_events(
})
}
fn compact_openai_events_for_storage(
events: &[EpiphanyOpenAiStreamEvent],
) -> Vec<EpiphanyOpenAiStreamEvent> {
fn compact_model_events_for_storage(
events: &[EpiphanyModelStreamEvent],
) -> Vec<EpiphanyModelStreamEvent> {
let mut compacted = Vec::new();
let mut text_buffer = String::new();
let mut reasoning_buffer = String::new();
for event in events {
match &event.payload {
EpiphanyOpenAiStreamPayload::TextDelta { text } => {
EpiphanyModelStreamPayload::TextDelta { text } => {
flush_reasoning_buffer(&mut compacted, event, &mut reasoning_buffer);
text_buffer.push_str(text);
}
EpiphanyOpenAiStreamPayload::ReasoningDelta { text } => {
EpiphanyModelStreamPayload::ReasoningDelta { text } => {
flush_text_buffer(&mut compacted, event, &mut text_buffer);
reasoning_buffer.push_str(text);
}
@ -238,8 +231,8 @@ fn compact_openai_events_for_storage(
}
fn flush_text_buffer(
compacted: &mut Vec<EpiphanyOpenAiStreamEvent>,
source: &EpiphanyOpenAiStreamEvent,
compacted: &mut Vec<EpiphanyModelStreamEvent>,
source: &EpiphanyModelStreamEvent,
buffer: &mut String,
) {
if !buffer.is_empty() {
@ -247,14 +240,14 @@ fn flush_text_buffer(
push_compacted_event(
compacted,
source,
EpiphanyOpenAiStreamPayload::TextDelta { text },
EpiphanyModelStreamPayload::TextDelta { text },
);
}
}
fn flush_reasoning_buffer(
compacted: &mut Vec<EpiphanyOpenAiStreamEvent>,
source: &EpiphanyOpenAiStreamEvent,
compacted: &mut Vec<EpiphanyModelStreamEvent>,
source: &EpiphanyModelStreamEvent,
buffer: &mut String,
) {
if !buffer.is_empty() {
@ -262,18 +255,20 @@ fn flush_reasoning_buffer(
push_compacted_event(
compacted,
source,
EpiphanyOpenAiStreamPayload::ReasoningDelta { text },
EpiphanyModelStreamPayload::ReasoningDelta { text },
);
}
}
fn push_compacted_event(
compacted: &mut Vec<EpiphanyOpenAiStreamEvent>,
source: &EpiphanyOpenAiStreamEvent,
payload: EpiphanyOpenAiStreamPayload,
compacted: &mut Vec<EpiphanyModelStreamEvent>,
source: &EpiphanyModelStreamEvent,
payload: EpiphanyModelStreamPayload,
) {
compacted.push(EpiphanyOpenAiStreamEvent {
compacted.push(EpiphanyModelStreamEvent {
schema_id: source.schema_id.clone(),
request_id: source.request_id.clone(),
provider: source.provider.clone(),
sequence: compacted.len() as u64,
payload,
});
@ -837,74 +832,12 @@ fn repository_source_tools() -> Vec<EpiphanyModelToolDefinition> {
]
}
pub fn model_event_from_openai_event(
provider: &str,
event: &EpiphanyOpenAiStreamEvent,
) -> EpiphanyModelStreamEvent {
EpiphanyModelStreamEvent {
schema_id: epiphany_model_adapter::MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(),
request_id: event.request_id.clone(),
provider: provider.to_string(),
sequence: event.sequence,
payload: match &event.payload {
EpiphanyOpenAiStreamPayload::TextDelta { text } => {
EpiphanyModelStreamPayload::TextDelta { text: text.clone() }
}
EpiphanyOpenAiStreamPayload::ReasoningDelta { text } => {
EpiphanyModelStreamPayload::ReasoningDelta { text: text.clone() }
}
EpiphanyOpenAiStreamPayload::ToolCall {
call_id,
name,
arguments,
} => EpiphanyModelStreamPayload::ToolCall {
call_id: call_id.clone(),
name: name.clone(),
arguments: arguments.clone(),
},
EpiphanyOpenAiStreamPayload::Completed { receipt } => {
EpiphanyModelStreamPayload::Completed {
receipt: Box::new(model_receipt_from_openai_receipt(provider, receipt)),
}
}
EpiphanyOpenAiStreamPayload::Failed { message } => EpiphanyModelStreamPayload::Failed {
message: message.clone(),
},
},
}
}
pub fn model_receipt_from_openai_receipt(
provider: &str,
receipt: &epiphany_openai_adapter::EpiphanyOpenAiModelReceipt,
) -> EpiphanyModelReceipt {
EpiphanyModelReceipt {
schema_id: epiphany_model_adapter::MODEL_ADAPTER_RECEIPT_SCHEMA_ID.to_string(),
request_id: receipt.request_id.clone(),
provider: provider.to_string(),
model: receipt.model.clone(),
provider_response_id: receipt.response_id.clone(),
input_tokens: receipt.input_tokens,
output_tokens: receipt.output_tokens,
reasoning_output_tokens: receipt.reasoning_output_tokens,
transport: receipt.transport.clone(),
caller_runtime_id: receipt.caller_runtime_id.clone(),
native_request_sha256: receipt.native_request_sha256.clone(),
provider_request_sha256: receipt.provider_request_sha256.clone(),
cached_input_tokens: receipt.cached_input_tokens,
}
}
pub fn tool_invocation_intents_from_openai_events(
provider: &str,
events: &[EpiphanyOpenAiStreamEvent],
pub fn tool_invocation_intents_from_model_events(
events: &[EpiphanyModelStreamEvent],
) -> Vec<EpiphanyToolInvocationIntent> {
events
.iter()
.filter_map(|event| {
let model_event = model_event_from_openai_event(provider, event);
tool_invocation_intent_from_model_event(&model_event)
})
.filter_map(tool_invocation_intent_from_model_event)
.collect()
}
@ -1929,7 +1862,7 @@ mod tests {
use epiphany_core::EpiphanyWorkerLaunchDocument;
use epiphany_core::RuntimeSpineHeartbeatJobOptions;
use epiphany_core::runtime_job;
use epiphany_openai_adapter::EpiphanyOpenAiModelReceipt;
use epiphany_model_adapter::EpiphanyModelReceipt;
use sha2::{Digest, Sha256};
use tempfile::tempdir;
@ -2752,73 +2685,79 @@ mod tests {
Ok(())
}
fn test_openai_event(
fn test_model_event(
request_id: &str,
sequence: u64,
payload: EpiphanyOpenAiStreamPayload,
) -> EpiphanyOpenAiStreamEvent {
EpiphanyOpenAiStreamEvent {
payload: EpiphanyModelStreamPayload,
) -> EpiphanyModelStreamEvent {
EpiphanyModelStreamEvent {
schema_id: epiphany_model_adapter::MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(),
request_id: request_id.to_string(),
provider: DEFAULT_MODEL_PROVIDER.to_string(),
sequence,
payload,
}
}
#[test]
fn compacts_openai_text_and_reasoning_deltas_before_storage() {
fn compacts_native_model_text_and_reasoning_deltas_before_storage() {
let events = vec![
test_openai_event(
test_model_event(
"req-1",
0,
EpiphanyOpenAiStreamPayload::ReasoningDelta {
EpiphanyModelStreamPayload::ReasoningDelta {
text: "think".to_string(),
},
),
test_openai_event(
test_model_event(
"req-1",
1,
EpiphanyOpenAiStreamPayload::ReasoningDelta {
EpiphanyModelStreamPayload::ReasoningDelta {
text: " small".to_string(),
},
),
test_openai_event(
test_model_event(
"req-1",
2,
EpiphanyOpenAiStreamPayload::TextDelta {
EpiphanyModelStreamPayload::TextDelta {
text: "{\"role".to_string(),
},
),
test_openai_event(
test_model_event(
"req-1",
3,
EpiphanyOpenAiStreamPayload::TextDelta {
EpiphanyModelStreamPayload::TextDelta {
text: "Id\":\"modeling\"}".to_string(),
},
),
test_openai_event(
test_model_event(
"req-1",
4,
EpiphanyOpenAiStreamPayload::Completed {
receipt: Box::new(EpiphanyOpenAiModelReceipt::new("req-1", "gpt-5.4")),
EpiphanyModelStreamPayload::Completed {
receipt: Box::new(EpiphanyModelReceipt::new(
"req-1",
DEFAULT_MODEL_PROVIDER,
"gpt-5.4",
)),
},
),
];
let compacted = compact_openai_events_for_storage(&events);
let compacted = compact_model_events_for_storage(&events);
assert_eq!(compacted.len(), 3);
assert_eq!(compacted[0].sequence, 0);
assert!(matches!(
&compacted[0].payload,
EpiphanyOpenAiStreamPayload::ReasoningDelta { text } if text == "think small"
EpiphanyModelStreamPayload::ReasoningDelta { text } if text == "think small"
));
assert!(matches!(
&compacted[1].payload,
EpiphanyOpenAiStreamPayload::TextDelta { text } if text == "{\"roleId\":\"modeling\"}"
EpiphanyModelStreamPayload::TextDelta { text } if text == "{\"roleId\":\"modeling\"}"
));
assert!(matches!(
compacted[2].payload,
EpiphanyOpenAiStreamPayload::Completed { .. }
EpiphanyModelStreamPayload::Completed { .. }
));
}
@ -2844,13 +2783,15 @@ mod tests {
},
&native_request,
)?;
let mut receipt = EpiphanyOpenAiModelReceipt::new("req-1", "gpt-5.4");
receipt.response_id = Some("resp-1".to_string());
let mut receipt = EpiphanyModelReceipt::new("req-1", DEFAULT_MODEL_PROVIDER, "gpt-5.4");
receipt.provider_response_id = Some("resp-1".to_string());
receipt.transport = Some("test".to_string());
let events = vec![EpiphanyOpenAiStreamEvent {
let events = vec![EpiphanyModelStreamEvent {
schema_id: epiphany_model_adapter::MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(),
request_id: "req-1".to_string(),
provider: DEFAULT_MODEL_PROVIDER.to_string(),
sequence: 0,
payload: EpiphanyOpenAiStreamPayload::Completed {
payload: EpiphanyModelStreamPayload::Completed {
receipt: Box::new(receipt),
},
}];
@ -3290,23 +3231,27 @@ mod tests {
},
&native_request,
)?;
let mut receipt = EpiphanyOpenAiModelReceipt::new("req-tools", "gpt-5.4");
receipt.response_id = Some("resp-tools".to_string());
let mut receipt = EpiphanyModelReceipt::new("req-tools", DEFAULT_MODEL_PROVIDER, "gpt-5.4");
receipt.provider_response_id = Some("resp-tools".to_string());
receipt.transport = Some("test".to_string());
let events = vec![
EpiphanyOpenAiStreamEvent {
EpiphanyModelStreamEvent {
schema_id: epiphany_model_adapter::MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(),
request_id: "req-tools".to_string(),
provider: DEFAULT_MODEL_PROVIDER.to_string(),
sequence: 0,
payload: EpiphanyOpenAiStreamPayload::ToolCall {
payload: EpiphanyModelStreamPayload::ToolCall {
call_id: "call-original".to_string(),
name: "mcp__smoke_server__smoke_tool".to_string(),
arguments: "{}".to_string(),
},
},
EpiphanyOpenAiStreamEvent {
EpiphanyModelStreamEvent {
schema_id: epiphany_model_adapter::MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(),
request_id: "req-tools".to_string(),
provider: DEFAULT_MODEL_PROVIDER.to_string(),
sequence: 1,
payload: EpiphanyOpenAiStreamPayload::Completed {
payload: EpiphanyModelStreamPayload::Completed {
receipt: Box::new(receipt),
},
},

62
epiphany-openai-runtime/src/provider_transport.rs
View file

@ -7,10 +7,12 @@ use codex_connector::{
CodexConnectorClient, CodexProviderRequest, CodexTransportDisposition,
CodexTransportEventPayload, CodexTransportInvocation, CodexTransportOutcome,
};
use epiphany_model_adapter::EpiphanyModelRequest;
use epiphany_model_adapter::{
EpiphanyModelReceipt, EpiphanyModelRequest, EpiphanyModelStreamEvent,
EpiphanyModelStreamPayload, MODEL_ADAPTER_EVENT_SCHEMA_ID,
};
use epiphany_openai_adapter::{
EpiphanyOpenAiModelReceipt, EpiphanyOpenAiModelRequest, EpiphanyOpenAiStreamEvent,
EpiphanyOpenAiStreamPayload, EpiphanyProviderRequestPayload, openrouter_events_from_response,
EpiphanyOpenRouterRequest, EpiphanyProviderRequestPayload, openrouter_events_from_response,
openrouter_request_body,
};
use epiphany_openai_runtime::{
@ -37,11 +39,13 @@ pub async fn run_model_turn(
collect_transport_events(
execute_codex_connector(&options, &request, codex_request),
&request_id,
provider,
)
.await
}
(OPENROUTER_MODEL_PROVIDER, EpiphanyProviderRequestPayload::OpenRouter(request)) => {
collect_transport_events(execute_openrouter(&options, request), &request_id).await
collect_transport_events(execute_openrouter(&options, request), &request_id, provider)
.await
}
_ => {
return Err(anyhow!(
@ -56,7 +60,7 @@ async fn execute_codex_connector(
options: &EpiphanyOpenAiRuntimeOptions,
native_request: &EpiphanyModelRequest,
provider_request: CodexProviderRequest,
) -> Result<Vec<EpiphanyOpenAiStreamEvent>> {
) -> Result<Vec<EpiphanyModelStreamEvent>> {
let endpoint = options
.connector_endpoint
.ok_or_else(|| anyhow!("Codex provider requires an explicit connector endpoint"))?;
@ -86,17 +90,19 @@ async fn execute_codex_connector(
MAX_CONNECTOR_FRAME_BYTES,
options.request_timeout,
)?;
let provider = native_request.provider.clone();
tokio::task::spawn_blocking(move || {
let result = client.execute(&invocation)?;
events_from_connector_result(result)
events_from_connector_result(&provider, result)
})
.await
.context("Codex connector client task failed")?
}
pub(super) fn events_from_connector_result(
provider: &str,
result: codex_connector::CodexTransportResult,
) -> Result<Vec<EpiphanyOpenAiStreamEvent>> {
) -> Result<Vec<EpiphanyModelStreamEvent>> {
let request_id = result.request_id.clone();
match result.disposition {
CodexTransportDisposition::Refused(reason) => Err(anyhow!(
@ -105,18 +111,20 @@ pub(super) fn events_from_connector_result(
CodexTransportDisposition::Transported { events, receipt } => {
let mut projected = events
.into_iter()
.map(|event| EpiphanyOpenAiStreamEvent {
.map(|event| EpiphanyModelStreamEvent {
schema_id: MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(),
request_id: request_id.clone(),
provider: provider.to_string(),
sequence: event.sequence,
payload: match event.payload {
CodexTransportEventPayload::TextDelta { text } => {
EpiphanyOpenAiStreamPayload::TextDelta { text }
EpiphanyModelStreamPayload::TextDelta { text }
}
CodexTransportEventPayload::ToolCall {
call_id,
name,
arguments,
} => EpiphanyOpenAiStreamPayload::ToolCall {
} => EpiphanyModelStreamPayload::ToolCall {
call_id,
name,
arguments,
@ -124,7 +132,7 @@ pub(super) fn events_from_connector_result(
},
})
.collect::<Vec<_>>();
let mut terminal = EpiphanyOpenAiModelReceipt::new(&request_id, &receipt.model);
let mut terminal = EpiphanyModelReceipt::new(&request_id, provider, &receipt.model);
terminal.transport = Some(receipt.transport.clone());
terminal.caller_runtime_id = Some(receipt.caller_runtime_id.clone());
terminal.native_request_sha256 = Some(hex_digest(receipt.native_request_sha256));
@ -137,15 +145,17 @@ pub(super) fn events_from_connector_result(
reasoning_output_tokens,
cached_input_tokens,
} => {
terminal.response_id = provider_response_id;
terminal.provider_response_id = provider_response_id;
terminal.input_tokens = input_tokens;
terminal.output_tokens = output_tokens;
terminal.reasoning_output_tokens = reasoning_output_tokens;
terminal.cached_input_tokens = cached_input_tokens;
projected.push(EpiphanyOpenAiStreamEvent {
projected.push(EpiphanyModelStreamEvent {
schema_id: MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(),
request_id,
provider: provider.to_string(),
sequence: projected.len() as u64,
payload: EpiphanyOpenAiStreamPayload::Completed {
payload: EpiphanyModelStreamPayload::Completed {
receipt: Box::new(terminal),
},
});
@ -154,10 +164,12 @@ pub(super) fn events_from_connector_result(
failure_kind,
message,
} => {
projected.push(EpiphanyOpenAiStreamEvent {
projected.push(EpiphanyModelStreamEvent {
schema_id: MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(),
request_id,
provider: provider.to_string(),
sequence: projected.len() as u64,
payload: EpiphanyOpenAiStreamPayload::Failed {
payload: EpiphanyModelStreamPayload::Failed {
message: format!("Codex connector {failure_kind}: {message}"),
},
});
@ -170,8 +182,8 @@ pub(super) fn events_from_connector_result(
async fn execute_openrouter(
options: &EpiphanyOpenAiRuntimeOptions,
request: EpiphanyOpenAiModelRequest,
) -> Result<Vec<EpiphanyOpenAiStreamEvent>> {
request: EpiphanyOpenRouterRequest,
) -> Result<Vec<EpiphanyModelStreamEvent>> {
let api_key = Zeroizing::new(read_static_provider_credential(
options.provider_credential_path.as_deref(),
OPENROUTER_MODEL_PROVIDER,
@ -210,16 +222,22 @@ async fn execute_openrouter(
.context("OpenRouter client task failed")?
}
async fn collect_transport_events<F>(future: F, request_id: &str) -> Vec<EpiphanyOpenAiStreamEvent>
async fn collect_transport_events<F>(
future: F,
request_id: &str,
provider: &str,
) -> Vec<EpiphanyModelStreamEvent>
where
F: std::future::Future<Output = Result<Vec<EpiphanyOpenAiStreamEvent>>>,
F: std::future::Future<Output = Result<Vec<EpiphanyModelStreamEvent>>>,
{
match future.await {
Ok(events) => events,
Err(error) => vec![EpiphanyOpenAiStreamEvent {
Err(error) => vec![EpiphanyModelStreamEvent {
schema_id: MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(),
request_id: request_id.to_string(),
provider: provider.to_string(),
sequence: 0,
payload: EpiphanyOpenAiStreamPayload::Failed {
payload: EpiphanyModelStreamPayload::Failed {
message: error.to_string(),
},
}],