i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-10 05:00:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/pr/547'

Browse Source
This commit is contained in:
fawney19
2026-05-23 22:03:17 +08:00
parent 3eb9614e68 0f2a96554f
commit 6b1c1e4f50
7 changed files with 936 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/aether-gateway/src/usage/reporting/mod.rs
+73 -3
View File
@@ -18,7 +18,10 @@ use context::{report_context_is_locally_actionable, resolve_locally_actionable_r
use aether_usage_runtime::{
is_local_ai_stream_report_kind, is_local_ai_sync_report_kind, report_request_id,
should_handle_local_stream_report, should_handle_local_sync_report,
stream_report_represents_failure, sync_report_represents_failure,
stream_report_missing_terminal_event, stream_report_represents_failure,
sync_report_represents_failure, STREAM_MISSING_TERMINAL_EVENT_CATEGORY,
STREAM_MISSING_TERMINAL_EVENT_MESSAGE, STREAM_TERMINAL_ERROR_CATEGORY,
STREAM_TERMINAL_ERROR_MESSAGE,
};
pub(crate) use aether_usage_runtime::{GatewayStreamReportRequest, GatewaySyncReportRequest};
@@ -257,6 +260,7 @@ async fn handle_local_stream_report(state: &AppState, payload: &GatewayStreamRep
.as_ref()
.and_then(|telemetry| telemetry.elapsed_ms);
let failed = stream_report_represents_failure(payload);
let missing_terminal_event = stream_report_missing_terminal_event(payload);
record_report_request_candidate_status(
state,
payload.report_context.as_ref(),
@@ -270,8 +274,10 @@ async fn handle_local_stream_report(state: &AppState, payload: &GatewayStreamRep
error_type: failed.then(|| {
if payload.status_code >= 400 {
"stream_http_error".to_string()
} else if missing_terminal_event {
STREAM_MISSING_TERMINAL_EVENT_CATEGORY.to_string()
} else {
"stream_terminal_error".to_string()
STREAM_TERMINAL_ERROR_CATEGORY.to_string()
}
}),
error_message: failed.then(|| {
@@ -280,7 +286,11 @@ async fn handle_local_stream_report(state: &AppState, payload: &GatewayStreamRep
.as_ref()
.and_then(|summary| summary.parser_error.clone())
.unwrap_or_else(|| {
"execution runtime stream ended with a terminal error".to_string()
if missing_terminal_event {
STREAM_MISSING_TERMINAL_EVENT_MESSAGE.to_string()
} else {
STREAM_TERMINAL_ERROR_MESSAGE.to_string()
}
})
}),
latency_ms,
@@ -310,9 +320,11 @@ mod tests {
use aether_data_contracts::repository::provider_catalog::{
ProviderCatalogReadRepository, StoredProviderCatalogKey, StoredProviderCatalogProvider,
};
use aether_data_contracts::repository::usage::UsageBodyCaptureState;
use aether_data_contracts::repository::video_tasks::{
UpsertVideoTask, VideoTaskStatus, VideoTaskWriteRepository,
};
use base64::Engine as _;
use serde_json::json;
use super::{
@@ -757,6 +769,64 @@ mod tests {
);
}
#[tokio::test]
async fn submit_openai_responses_stream_report_marks_missing_terminal_event_as_failed() {
let repository = Arc::new(InMemoryRequestCandidateRepository::seed(vec![
sample_request_candidate(
"cand-reporting-stream-missing-terminal-1",
"req-reporting-stream-missing-terminal-1",
),
]));
let state = build_test_state(Arc::clone(&repository));
let provider_sse = concat!(
"event: response.created\n",
"data: {\"type\":\"response.created\"}\n\n",
"event: response.output_text.delta\n",
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"hello\"}\n\n"
);
submit_stream_report(
&state,
GatewayStreamReportRequest {
trace_id: "trace-reporting-stream-missing-terminal-1".to_string(),
report_kind: "openai_responses_stream_success".to_string(),
report_context: Some(json!({
"request_id": "req-reporting-stream-missing-terminal-1",
"client_api_format": "openai:responses",
"provider_api_format": "openai:responses"
})),
status_code: 200,
headers: BTreeMap::new(),
provider_body_base64: Some(
base64::engine::general_purpose::STANDARD.encode(provider_sse.as_bytes()),
),
provider_body_state: Some(UsageBodyCaptureState::Inline),
client_body_base64: None,
client_body_state: None,
terminal_summary: None,
telemetry: None,
},
)
.await
.expect("stream report should stay local");
let stored = repository
.list_by_request_id("req-reporting-stream-missing-terminal-1")
.await
.expect("request candidates should list");
assert_eq!(stored.len(), 1);
assert_eq!(stored[0].status, RequestCandidateStatus::Failed);
assert_eq!(stored[0].status_code, Some(200));
assert_eq!(
stored[0].error_type.as_deref(),
Some("stream_missing_terminal_event")
);
assert_eq!(
stored[0].error_message.as_deref(),
Some("execution runtime stream ended before provider terminal event")
);
}
#[tokio::test]
async fn submit_sync_report_updates_codex_quota_from_response_headers() {
crate::orchestration::clear_local_report_effect_caches_for_tests();
crates/aether-ai-formats/src/formats/shared/standard_matrix.rs
+95
View File
@@ -551,6 +551,101 @@ mod tests {
assert_eq!(converted["stream"], true);
}
#[test]
fn standard_openai_chat_to_claude_normalizes_multiturn_tool_history() {
let request = json!({
"model": "deepseek-v4-flash",
"messages": [
{"role": "system", "content": "Be precise."},
{"role": "user", "content": "check two things"},
{
"role": "assistant",
"content": null,
"tool_calls": [
{
"id": "weather-1",
"type": "function",
"function": {"name": "get_weather", "arguments": "{\"city\":\"NYC\"}"}
},
{
"id": "call_2",
"type": "function",
"function": {"name": "lookup", "arguments": "{\"q\":\"db\"}"}
}
]
},
{"role": "tool", "tool_call_id": "weather-1", "content": ""},
{"role": "tool", "tool_call_id": "call_2", "content": [{"type": "text", "text": "rows=1"}]},
{"role": "user", "content": "now answer"}
],
"tools": [
{
"type": "function",
"function": {"name": "get_weather", "description": "Get weather", "parameters": null}
},
{
"type": "function",
"function": {
"name": "lookup",
"description": "Lookup data",
"parameters": {"properties": {"q": {"type": "string"}}}
}
}
],
"parallel_tool_calls": true,
"max_tokens": 128,
"stream": true
});
let converted = build_standard_request_body(
&request,
"openai:chat",
"claude-sonnet-4-5",
"custom",
"claude:messages",
"/v1/chat/completions",
true,
None,
None,
)
.expect("openai chat tool history should build as claude messages");
assert_eq!(converted["model"], "claude-sonnet-4-5");
assert_eq!(converted["system"], "Be precise.");
assert_eq!(converted["messages"][0]["role"], "user");
assert_eq!(converted["messages"][1]["role"], "assistant");
assert_eq!(converted["messages"][1]["content"][0]["type"], "tool_use");
assert_eq!(
converted["messages"][1]["content"][0]["id"],
"toolu_weather-1"
);
assert_eq!(converted["messages"][1]["content"][1]["id"], "call_2");
assert_eq!(converted["messages"][2]["role"], "user");
assert_eq!(
converted["messages"][2]["content"][0]["type"],
"tool_result"
);
assert_eq!(
converted["messages"][2]["content"][0]["tool_use_id"],
"toolu_weather-1"
);
assert_eq!(converted["messages"][2]["content"][0]["content"], "(empty)");
assert_eq!(
converted["messages"][2]["content"][1]["tool_use_id"],
"call_2"
);
assert_eq!(converted["messages"][2]["content"][1]["content"], "rows=1");
assert_eq!(converted["messages"][2]["content"][2]["type"], "text");
assert_eq!(converted["messages"][2]["content"][2]["text"], "now answer");
assert_eq!(converted["tools"][0]["input_schema"]["type"], "object");
assert_eq!(
converted["tools"][0]["input_schema"]["properties"],
json!({})
);
assert_eq!(converted["tools"][1]["input_schema"]["type"], "object");
assert_eq!(converted["stream"], true);
}
fn codex_default_body_rules() -> Value {
json!([
{"action":"drop","path":"max_output_tokens"},
crates/aether-ai-formats/src/protocol/canonical.rs
+99 -9
View File
@@ -11,6 +11,7 @@ pub(crate) const OPENAI_RESPONSES_EXTENSION_NAMESPACE: &str = "openai_responses"
pub(crate) const OPENAI_RESPONSES_LEGACY_EXTENSION_NAMESPACE: &str = "openai_cli";
const AETHER_EXTENSION_NAMESPACE: &str = "aether";
const CLAUDE_TOOL_RESULT_SOURCE_MARKER: &str = "claude_tool_result";
const OPENAI_CHAT_TOOL_RESULT_SOURCE_MARKER: &str = "openai_chat_tool_result";
const OPENAI_CHAT_TOOL_ERROR_PREFIX: &str = "[tool error]";
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
@@ -725,7 +726,7 @@ pub(crate) fn openai_role_to_canonical(role: &str) -> CanonicalRole {
"assistant" => CanonicalRole::Assistant,
"system" => CanonicalRole::System,
"developer" => CanonicalRole::Developer,
"tool" => CanonicalRole::Tool,
"tool" | "function" => CanonicalRole::Tool,
_ => CanonicalRole::Unknown,
}
}
@@ -1329,10 +1330,12 @@ pub(crate) fn openai_message_content_blocks(
blocks.splice(0..0, reasoning_blocks);
}
}
let mut saw_tool_calls = false;
if let Some(tool_calls) = message.get("tool_calls").and_then(Value::as_array) {
for tool_call in tool_calls {
let tool_call = tool_call.as_object()?;
let function = tool_call.get("function").and_then(Value::as_object)?;
saw_tool_calls = true;
blocks.push(CanonicalContentBlock::ToolUse {
id: tool_call
.get("id")
@@ -1349,12 +1352,38 @@ pub(crate) fn openai_message_content_blocks(
});
}
}
if role == CanonicalRole::Assistant && !saw_tool_calls {
if let Some(function_call) = message.get("function_call").and_then(Value::as_object) {
let name = function_call
.get("name")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
blocks.push(CanonicalContentBlock::ToolUse {
id: message
.get("id")
.and_then(Value::as_str)
.filter(|value| !value.trim().is_empty())
.unwrap_or(name.as_str())
.to_string(),
name,
input: parse_jsonish_value(function_call.get("arguments")),
extensions: openai_extensions(message, &["role", "content", "function_call"]),
});
}
}
if role == CanonicalRole::Tool {
let text = openai_content_text(message.get("content"));
let mut extensions = openai_extensions(message, &["role", "content", "tool_call_id"]);
extensions.insert(
AETHER_EXTENSION_NAMESPACE.to_string(),
json!({ "source": OPENAI_CHAT_TOOL_RESULT_SOURCE_MARKER }),
);
blocks.push(CanonicalContentBlock::ToolResult {
tool_use_id: message
.get("tool_call_id")
.and_then(Value::as_str)
.or_else(|| message.get("name").and_then(Value::as_str))
.unwrap_or_default()
.to_string(),
name: None,
@@ -1374,7 +1403,7 @@ pub(crate) fn openai_message_content_blocks(
text
}),
is_error: false,
extensions: openai_extensions(message, &["role", "content", "tool_call_id"]),
extensions,
});
}
Some(blocks)
@@ -3798,7 +3827,10 @@ pub(crate) fn canonical_block_to_claude(
} => {
let mut out = Map::new();
out.insert("type".to_string(), Value::String("tool_use".to_string()));
out.insert("id".to_string(), Value::String(id.clone()));
out.insert(
"id".to_string(),
Value::String(claude_compatible_tool_use_id(id)),
);
out.insert("name".to_string(), Value::String(name.clone()));
out.insert("input".to_string(), input.clone());
out.extend(namespace_extension_object(extensions, "claude", &out));
@@ -3816,7 +3848,7 @@ pub(crate) fn canonical_block_to_claude(
out.insert("type".to_string(), Value::String("tool_result".to_string()));
out.insert(
"tool_use_id".to_string(),
Value::String(tool_use_id.clone()),
Value::String(claude_compatible_tool_use_id(tool_use_id)),
);
out.insert(
"content".to_string(),
@@ -3824,6 +3856,7 @@ pub(crate) fn canonical_block_to_claude(
output.as_ref(),
content_text.as_deref(),
role,
extensions,
),
);
out.insert("is_error".to_string(), Value::Bool(*is_error));
@@ -3838,6 +3871,7 @@ fn canonical_tool_result_content_to_claude(
output: Option<&Value>,
content_text: Option<&str>,
role: &CanonicalRole,
extensions: &BTreeMap<String, Value>,
) -> Value {
if matches!(role, CanonicalRole::Assistant) {
return output
@@ -3845,15 +3879,54 @@ fn canonical_tool_result_content_to_claude(
.unwrap_or_else(|| Value::String(content_text.unwrap_or_default().to_string()));
}
if is_openai_chat_tool_result(extensions) {
let text = content_text
.map(ToOwned::to_owned)
.or_else(|| output.map(openai_responses_tool_output_text))
.unwrap_or_default();
return Value::String(non_empty_tool_result_text(&text));
}
match output {
Some(Value::String(text)) => Value::String(text.clone()),
Some(Value::String(text)) => Value::String(non_empty_tool_result_text(text)),
Some(Value::Array(parts)) if claude_tool_result_content_blocks_are_wire_safe(parts) => {
Value::Array(parts.clone())
}
Some(Value::Null) => Value::String(non_empty_tool_result_text("")),
Some(value) => serde_json::to_string(value)
.map(Value::String)
.unwrap_or_else(|_| Value::String(content_text.unwrap_or_default().to_string())),
None => Value::String(content_text.unwrap_or_default().to_string()),
.map(|text| Value::String(non_empty_tool_result_text(&text)))
.unwrap_or_else(|_| {
Value::String(non_empty_tool_result_text(content_text.unwrap_or_default()))
}),
None => Value::String(non_empty_tool_result_text(content_text.unwrap_or_default())),
}
}
fn is_openai_chat_tool_result(extensions: &BTreeMap<String, Value>) -> bool {
extensions
.get(AETHER_EXTENSION_NAMESPACE)
.and_then(|value| value.get("source"))
.and_then(Value::as_str)
== Some(OPENAI_CHAT_TOOL_RESULT_SOURCE_MARKER)
}
fn non_empty_tool_result_text(text: &str) -> String {
if text.trim().is_empty() {
"(empty)".to_string()
} else {
text.to_string()
}
}
fn claude_compatible_tool_use_id(id: &str) -> String {
let trimmed = id.trim();
if trimmed.is_empty() {
return "toolu_".to_string();
}
if trimmed.starts_with("toolu_") || trimmed.starts_with("call_") {
trimmed.to_string()
} else {
format!("toolu_{trimmed}")
}
}
@@ -3986,7 +4059,7 @@ pub(crate) fn canonical_tools_to_claude(canonical: &CanonicalRequest) -> Vec<Val
}
out.insert(
"input_schema".to_string(),
tool.parameters.clone().unwrap_or_else(|| json!({})),
claude_input_schema_from_tool_parameters(tool.parameters.as_ref()),
);
out.extend(namespace_extension_object(&tool.extensions, "claude", &out));
Value::Object(out)
@@ -4004,6 +4077,23 @@ pub(crate) fn canonical_tools_to_claude(canonical: &CanonicalRequest) -> Vec<Val
tools
}
fn claude_input_schema_from_tool_parameters(parameters: Option<&Value>) -> Value {
match parameters {
Some(Value::Object(schema)) => {
let mut schema = schema.clone();
schema
.entry("type".to_string())
.or_insert_with(|| Value::String("object".to_string()));
schema
.entry("properties".to_string())
.or_insert_with(|| json!({}));
Value::Object(schema)
}
Some(Value::Null) | None => json!({"type": "object", "properties": {}}),
Some(_) => json!({"type": "object", "properties": {}}),
}
}
pub(crate) fn canonical_tool_choice_to_claude(
choice: Option<&CanonicalToolChoice>,
parallel_tool_calls: Option<bool>,
crates/aether-data-contracts/src/repository/candidates/types.rs
+79 -4
View File
@@ -261,10 +261,9 @@ impl RequestCandidateTrace {
pub fn derive_request_candidate_final_status(
candidates: &[StoredRequestCandidate],
) -> RequestCandidateFinalStatus {
let has_success = candidates.iter().any(|candidate| {
candidate.status == RequestCandidateStatus::Success
|| matches!(candidate.status_code, Some(status_code) if (200..300).contains(&status_code))
});
let has_success = candidates
.iter()
.any(|candidate| candidate.status == RequestCandidateStatus::Success);
if has_success {
return RequestCandidateFinalStatus::Success;
}
@@ -297,6 +296,13 @@ pub fn derive_request_candidate_final_status(
return RequestCandidateFinalStatus::Pending;
}
let has_legacy_success_status_code = candidates
.iter()
.any(|candidate| matches!(candidate.status_code, Some(status_code) if (200..300).contains(&status_code)));
if has_legacy_success_status_code {
return RequestCandidateFinalStatus::Success;
}
RequestCandidateFinalStatus::Failed
}
@@ -536,3 +542,72 @@ impl<T> RequestCandidateRepository for T where
T: RequestCandidateReadRepository + RequestCandidateWriteRepository + Send + Sync
{
}
#[cfg(test)]
mod tests {
use super::{
derive_request_candidate_final_status, RequestCandidateFinalStatus, RequestCandidateStatus,
StoredRequestCandidate,
};
fn candidate(
id: &str,
status: RequestCandidateStatus,
status_code: Option<i32>,
) -> StoredRequestCandidate {
StoredRequestCandidate::new(
id.to_string(),
"req-1".to_string(),
None,
None,
None,
None,
0,
0,
None,
None,
None,
status,
None,
false,
status_code,
None,
None,
Some(100),
None,
None,
None,
1_700_000_000_000,
Some(1_700_000_000_000),
Some(1_700_000_000_100),
)
.expect("candidate should build")
}
#[test]
fn failed_candidate_with_http_200_stays_final_failed() {
let candidates = vec![candidate(
"cand-1",
RequestCandidateStatus::Failed,
Some(200),
)];
assert_eq!(
derive_request_candidate_final_status(&candidates),
RequestCandidateFinalStatus::Failed
);
}
#[test]
fn explicit_success_candidate_still_wins_after_failed_attempt() {
let candidates = vec![
candidate("cand-1", RequestCandidateStatus::Failed, Some(503)),
candidate("cand-2", RequestCandidateStatus::Success, Some(200)),
];
assert_eq!(
derive_request_candidate_final_status(&candidates),
RequestCandidateFinalStatus::Success
);
}
}
crates/aether-usage-runtime/src/lib.rs
+7 -3
View File
@@ -27,9 +27,13 @@ pub use report::{
infer_internal_finalize_signature, is_local_ai_stream_report_kind,
is_local_ai_sync_report_kind, normalize_gemini_file_name, report_request_id,
resolve_internal_finalize_route, should_handle_local_stream_report,
should_handle_local_sync_report, stream_report_represents_failure,
sync_report_represents_failure, GatewayStreamReportRequest, GatewaySyncReportRequest,
GeminiFileMappingEntry, InternalFinalizeRoute, GEMINI_FILE_MAPPING_TTL_SECONDS,
should_handle_local_sync_report, stream_capture_terminal_state,
stream_report_missing_terminal_event, stream_report_represents_failure,
stream_report_requires_observed_terminal_event, sync_report_represents_failure,
GatewayStreamReportRequest, GatewaySyncReportRequest, GeminiFileMappingEntry,
InternalFinalizeRoute, StreamCapturedTerminalState, GEMINI_FILE_MAPPING_TTL_SECONDS,
STREAM_MISSING_TERMINAL_EVENT_CATEGORY, STREAM_MISSING_TERMINAL_EVENT_MESSAGE,
STREAM_TERMINAL_ERROR_CATEGORY, STREAM_TERMINAL_ERROR_MESSAGE,
};
pub use report_context::{
build_locally_actionable_report_context_from_request_candidate,
crates/aether-usage-runtime/src/report.rs
+409 -8
View File
@@ -8,6 +8,19 @@ use serde_json::Value;
pub const GEMINI_FILE_MAPPING_TTL_SECONDS: u64 = 60 * 60 * 48;
const GEMINI_FILE_MAPPING_CACHE_PREFIX: &str = "gemini_files:key";
pub const STREAM_MISSING_TERMINAL_EVENT_CATEGORY: &str = "stream_missing_terminal_event";
pub const STREAM_TERMINAL_ERROR_CATEGORY: &str = "stream_terminal_error";
pub const STREAM_MISSING_TERMINAL_EVENT_MESSAGE: &str =
"execution runtime stream ended before provider terminal event";
pub const STREAM_TERMINAL_ERROR_MESSAGE: &str =
"execution runtime stream ended with a terminal error";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StreamCapturedTerminalState {
Completed,
Failed,
Missing,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct GatewaySyncReportRequest {
@@ -292,22 +305,342 @@ pub fn sync_report_represents_failure(
.is_some_and(|value| !value.is_null())
}
fn stream_terminal_summary_represents_failure(summary: &ExecutionStreamTerminalSummary) -> bool {
summary.parser_error.is_some()
|| (!summary.observed_finish
&& !summary
fn stream_terminal_summary_missing_observed_finish(
summary: &ExecutionStreamTerminalSummary,
requires_observed_terminal_event: bool,
) -> bool {
!summary.observed_finish
&& (requires_observed_terminal_event
|| !summary
.standardized_usage
.as_ref()
.is_some_and(aether_contracts::StandardizedUsage::has_token_signal))
}
fn stream_terminal_summary_represents_failure(
summary: &ExecutionStreamTerminalSummary,
requires_observed_terminal_event: bool,
) -> bool {
summary.parser_error.is_some()
|| stream_terminal_summary_missing_observed_finish(
summary,
requires_observed_terminal_event,
)
}
pub fn stream_report_represents_failure(payload: &GatewayStreamReportRequest) -> bool {
let requires_observed_terminal_event = stream_report_requires_observed_terminal_event(
payload.report_kind.as_str(),
payload.report_context.as_ref(),
);
payload.status_code >= 400
|| payload.report_kind.contains("error")
|| payload
.terminal_summary
.as_ref()
.is_some_and(stream_terminal_summary_represents_failure)
|| payload.terminal_summary.as_ref().is_some_and(|summary| {
stream_terminal_summary_represents_failure(summary, requires_observed_terminal_event)
})
|| stream_report_captured_terminal_failure(payload)
|| stream_report_missing_terminal_event(payload)
}
pub fn stream_report_missing_terminal_event(payload: &GatewayStreamReportRequest) -> bool {
let requires_observed_terminal_event = stream_report_requires_observed_terminal_event(
payload.report_kind.as_str(),
payload.report_context.as_ref(),
);
if !requires_observed_terminal_event {
return false;
}
payload
.terminal_summary
.as_ref()
.is_some_and(|summary| stream_terminal_summary_missing_observed_finish(summary, true))
|| matches!(
stream_report_captured_terminal_state(payload),
Some(StreamCapturedTerminalState::Missing)
)
}
pub fn stream_report_captured_terminal_failure(payload: &GatewayStreamReportRequest) -> bool {
matches!(
stream_report_captured_terminal_state(payload),
Some(StreamCapturedTerminalState::Failed)
)
}
pub fn stream_report_requires_observed_terminal_event(
report_kind: &str,
report_context: Option<&Value>,
) -> bool {
let report_kind = report_kind.trim().to_ascii_lowercase();
if report_kind.starts_with("openai_responses_")
|| report_kind.starts_with("openai_compact_")
|| report_kind.starts_with("openai_cli_")
{
return true;
}
let Some(context) = report_context.and_then(Value::as_object) else {
return false;
};
[
"provider_stream_event_api_format",
"provider_stream_api_format",
"provider_api_format",
"client_api_format",
]
.into_iter()
.filter_map(|field| context.get(field).and_then(Value::as_str))
.any(is_openai_responses_family_format_alias)
}
fn is_openai_responses_family_format_alias(value: &str) -> bool {
let normalized = value.trim().to_ascii_lowercase().replace('_', ":");
aether_ai_formats::is_openai_responses_family_format(normalized.as_str())
}
fn stream_report_captured_terminal_state(
payload: &GatewayStreamReportRequest,
) -> Option<StreamCapturedTerminalState> {
let provider_state = stream_report_provider_capture_requires_terminal_event(payload)
.then(|| stream_capture_terminal_state_from_base64(payload.provider_body_base64.as_deref()))
.flatten();
let client_state = stream_report_client_capture_requires_terminal_event(payload)
.then(|| stream_capture_terminal_state_from_base64(payload.client_body_base64.as_deref()))
.flatten();
combine_stream_terminal_states(provider_state, client_state)
}
fn stream_report_provider_capture_requires_terminal_event(
payload: &GatewayStreamReportRequest,
) -> bool {
let context = payload.report_context.as_ref();
stream_report_context_has_openai_responses_format(
context,
&[
"provider_stream_event_api_format",
"provider_stream_api_format",
"provider_api_format",
],
) || (stream_report_kind_requires_observed_terminal_event(payload.report_kind.as_str())
&& !stream_report_context_has_any_format(context))
}
fn stream_report_client_capture_requires_terminal_event(
payload: &GatewayStreamReportRequest,
) -> bool {
let context = payload.report_context.as_ref();
stream_report_context_has_openai_responses_format(context, &["client_api_format"])
|| (stream_report_kind_requires_observed_terminal_event(payload.report_kind.as_str())
&& !stream_report_context_has_any_format(context))
}
fn stream_report_context_has_openai_responses_format(
report_context: Option<&Value>,
fields: &[&str],
) -> bool {
fields
.iter()
.filter_map(|field| {
report_context
.and_then(Value::as_object)
.and_then(|context| context.get(*field))
.and_then(Value::as_str)
})
.any(is_openai_responses_family_format_alias)
}
fn stream_report_context_has_any_format(report_context: Option<&Value>) -> bool {
[
"provider_stream_event_api_format",
"provider_stream_api_format",
"provider_api_format",
"client_api_format",
]
.into_iter()
.any(|field| {
report_context
.and_then(Value::as_object)
.and_then(|context| context.get(field))
.and_then(Value::as_str)
.map(str::trim)
.is_some_and(|value| !value.is_empty())
})
}
fn stream_report_kind_requires_observed_terminal_event(report_kind: &str) -> bool {
let report_kind = report_kind.trim().to_ascii_lowercase();
report_kind.starts_with("openai_responses_")
|| report_kind.starts_with("openai_compact_")
|| report_kind.starts_with("openai_cli_")
}
fn stream_capture_terminal_state_from_base64(
body_base64: Option<&str>,
) -> Option<StreamCapturedTerminalState> {
let body_base64 = body_base64?;
let bytes = base64::engine::general_purpose::STANDARD
.decode(body_base64)
.ok()?;
if let Ok(value) = serde_json::from_slice::<Value>(&bytes) {
return stream_capture_terminal_state(&value);
}
let text = String::from_utf8(bytes).ok()?;
stream_capture_terminal_state_from_sse_text(text.as_str())
}
pub fn stream_capture_terminal_state(value: &Value) -> Option<StreamCapturedTerminalState> {
if let Some(chunks) = value.get("chunks").and_then(Value::as_array) {
let mut state = None;
for chunk in chunks {
state = combine_stream_terminal_states(state, openai_response_terminal_state(chunk));
}
return state.or_else(|| {
stream_capture_looks_like_stream(value).then_some(StreamCapturedTerminalState::Missing)
});
}
openai_response_terminal_state(value)
}
fn stream_capture_looks_like_stream(value: &Value) -> bool {
value
.get("metadata")
.and_then(|metadata| metadata.get("stream"))
.and_then(Value::as_bool)
.unwrap_or(false)
|| value.get("chunks").and_then(Value::as_array).is_some()
}
fn openai_response_terminal_state(value: &Value) -> Option<StreamCapturedTerminalState> {
if value.get("error").is_some_and(|error| !error.is_null()) {
return Some(StreamCapturedTerminalState::Failed);
}
let event_type = value
.get("type")
.or_else(|| value.get("event"))
.and_then(Value::as_str)
.map(str::trim)
.unwrap_or_default();
match event_type {
"response.completed" => return Some(StreamCapturedTerminalState::Completed),
"response.failed" | "response.incomplete" | "error" => {
return Some(StreamCapturedTerminalState::Failed);
}
_ => {}
}
let response = value.get("response");
if response
.and_then(|response| response.get("error"))
.is_some_and(|error| !error.is_null())
|| response
.and_then(|response| response.get("incomplete_details"))
.is_some_and(|details| !details.is_null())
{
return Some(StreamCapturedTerminalState::Failed);
}
match response
.and_then(|response| response.get("status"))
.and_then(Value::as_str)
.map(str::trim)
{
Some("completed") => Some(StreamCapturedTerminalState::Completed),
Some("failed" | "incomplete" | "cancelled") => Some(StreamCapturedTerminalState::Failed),
_ => None,
}
}
fn combine_stream_terminal_states(
current: Option<StreamCapturedTerminalState>,
next: Option<StreamCapturedTerminalState>,
) -> Option<StreamCapturedTerminalState> {
match (current, next) {
(Some(StreamCapturedTerminalState::Failed), _)
| (_, Some(StreamCapturedTerminalState::Failed)) => {
Some(StreamCapturedTerminalState::Failed)
}
(Some(StreamCapturedTerminalState::Missing), _)
| (_, Some(StreamCapturedTerminalState::Missing)) => {
Some(StreamCapturedTerminalState::Missing)
}
(Some(StreamCapturedTerminalState::Completed), _)
| (_, Some(StreamCapturedTerminalState::Completed)) => {
Some(StreamCapturedTerminalState::Completed)
}
(None, None) => None,
}
}
fn stream_capture_terminal_state_from_sse_text(text: &str) -> Option<StreamCapturedTerminalState> {
let mut state = None;
let mut saw_stream_payload = false;
for_each_sse_payload(text, |payload| {
saw_stream_payload = true;
if payload == "[DONE]" {
return;
}
if let Ok(value) = serde_json::from_str::<Value>(payload) {
state = combine_stream_terminal_states(state, openai_response_terminal_state(&value));
}
});
state.or_else(|| saw_stream_payload.then_some(StreamCapturedTerminalState::Missing))
}
fn flush_sse_payload<F>(payload: &mut String, has_payload: &mut bool, on_payload: &mut F)
where
F: FnMut(&str),
{
if !*has_payload {
return;
}
on_payload(payload);
payload.clear();
*has_payload = false;
}
fn for_each_sse_payload<F>(text: &str, mut on_payload: F)
where
F: FnMut(&str),
{
let mut payload = String::new();
let mut has_payload = false;
let bytes = text.as_bytes();
let mut line_start = 0usize;
let mut cursor = 0usize;
while cursor <= bytes.len() {
if cursor < bytes.len() && bytes[cursor] != b'\n' && bytes[cursor] != b'\r' {
cursor += 1;
continue;
}
let line = text[line_start..cursor].trim();
if line.is_empty() {
flush_sse_payload(&mut payload, &mut has_payload, &mut on_payload);
} else if let Some(data) = line.strip_prefix("data:").map(str::trim) {
if !data.is_empty() {
if has_payload {
payload.push('\n');
}
payload.push_str(data);
has_payload = true;
}
}
if cursor == bytes.len() {
break;
}
if bytes[cursor] == b'\r' && bytes.get(cursor + 1) == Some(&b'\n') {
cursor += 2;
} else {
cursor += 1;
}
line_start = cursor;
}
flush_sse_payload(&mut payload, &mut has_payload, &mut on_payload);
}
pub fn should_handle_local_sync_report(
@@ -392,6 +725,7 @@ mod tests {
use std::collections::BTreeMap;
use aether_contracts::ExecutionStreamTerminalSummary;
use aether_data_contracts::repository::usage::UsageBodyCaptureState;
use base64::Engine as _;
use serde_json::json;
@@ -529,6 +863,73 @@ mod tests {
assert!(stream_report_represents_failure(&payload));
}
#[test]
fn detects_openai_responses_stream_missing_terminal_from_captured_sse() {
let provider_sse = concat!(
"event: response.created\n",
"data: {\"type\":\"response.created\"}\n\n",
"event: response.output_text.delta\n",
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"hello\"}\n\n"
);
let mut payload = sample_stream_report("openai_responses_stream_success", 200);
payload.report_context = Some(json!({
"client_api_format": "openai:responses",
"provider_api_format": "openai:responses"
}));
payload.provider_body_base64 =
Some(base64::engine::general_purpose::STANDARD.encode(provider_sse.as_bytes()));
payload.provider_body_state = Some(UsageBodyCaptureState::Inline);
assert!(stream_report_represents_failure(&payload));
assert!(super::stream_report_missing_terminal_event(&payload));
}
#[test]
fn accepts_openai_responses_stream_completed_from_captured_sse() {
let provider_sse = concat!(
"event: response.completed\n",
"data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\"}}\n\n"
);
let mut payload = sample_stream_report("openai_responses_stream_success", 200);
payload.report_context = Some(json!({
"client_api_format": "openai:responses",
"provider_api_format": "openai:responses"
}));
payload.provider_body_base64 =
Some(base64::engine::general_purpose::STANDARD.encode(provider_sse.as_bytes()));
payload.provider_body_state = Some(UsageBodyCaptureState::Inline);
assert!(!stream_report_represents_failure(&payload));
assert!(!super::stream_report_missing_terminal_event(&payload));
}
#[test]
fn accepts_completed_openai_responses_provider_with_rewritten_openai_chat_client_stream() {
let provider_sse = concat!(
"event: response.completed\n",
"data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\"}}\n\n"
);
let client_sse = concat!(
"data: {\"object\":\"chat.completion.chunk\",\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n",
"data: [DONE]\n\n"
);
let mut payload = sample_stream_report("openai_chat_stream_success", 200);
payload.report_context = Some(json!({
"client_api_format": "openai:chat",
"provider_api_format": "openai:responses",
"provider_stream_event_api_format": "openai:responses"
}));
payload.provider_body_base64 =
Some(base64::engine::general_purpose::STANDARD.encode(provider_sse.as_bytes()));
payload.provider_body_state = Some(UsageBodyCaptureState::Inline);
payload.client_body_base64 =
Some(base64::engine::general_purpose::STANDARD.encode(client_sse.as_bytes()));
payload.client_body_state = Some(UsageBodyCaptureState::Inline);
assert!(!stream_report_represents_failure(&payload));
assert!(!super::stream_report_missing_terminal_event(&payload));
}
#[test]
fn infers_internal_finalize_signature_from_context_or_report_kind() {
let from_context = sample_sync_report_with_context(
crates/aether-usage-runtime/src/write.rs
+174 -10
View File
@@ -18,8 +18,11 @@ use crate::request_metadata::{
sanitize_usage_request_metadata_ref,
};
use crate::{
map_usage_from_response, GatewayStreamReportRequest, GatewaySyncReportRequest,
StandardizedUsage, UsageEvent, UsageEventData, UsageEventType,
map_usage_from_response, stream_capture_terminal_state, GatewayStreamReportRequest,
GatewaySyncReportRequest, StandardizedUsage, StreamCapturedTerminalState, UsageEvent,
UsageEventData, UsageEventType, STREAM_MISSING_TERMINAL_EVENT_CATEGORY,
STREAM_MISSING_TERMINAL_EVENT_MESSAGE, STREAM_TERMINAL_ERROR_CATEGORY,
STREAM_TERMINAL_ERROR_MESSAGE,
};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -926,10 +929,30 @@ pub fn build_stream_terminal_usage_seed(
map_usage_from_response(response, context_seed.provider_contract.as_str())
})
});
let captured_terminal_state = captured_stream_terminal_state(
report_kind.as_str(),
context_seed.client_contract.as_str(),
context_seed.provider_contract.as_str(),
provider_response_full.as_ref(),
client_response.as_ref(),
);
let requires_observed_terminal_event = stream_usage_requires_observed_terminal_event(
report_kind.as_str(),
context_seed.client_contract.as_str(),
context_seed.provider_contract.as_str(),
);
let observed_stream_finish = observed_stream_finish.or_else(|| {
captured_terminal_state.map(|state| state != StreamCapturedTerminalState::Missing)
});
let missing_observed_finish = matches!(observed_stream_finish, Some(false))
&& !standardized_usage
.as_ref()
.is_some_and(StandardizedUsage::has_token_signal);
&& (requires_observed_terminal_event
|| !standardized_usage
.as_ref()
.is_some_and(StandardizedUsage::has_token_signal));
let captured_terminal_failure = matches!(
captured_terminal_state,
Some(StreamCapturedTerminalState::Failed)
);
let terminal_error_message = terminal_error_message
.or_else(|| {
provider_response_full
@@ -940,17 +963,17 @@ pub fn build_stream_terminal_usage_seed(
client_response
.as_ref()
.and_then(extract_explicit_error_message_from_json)
});
})
.or_else(|| captured_terminal_failure.then(|| STREAM_TERMINAL_ERROR_MESSAGE.to_string()));
let terminal_failure_category = if terminal_error_message.is_some() {
Some("stream_terminal_error".to_string())
Some(STREAM_TERMINAL_ERROR_CATEGORY.to_string())
} else if missing_observed_finish {
Some("stream_missing_terminal_event".to_string())
Some(STREAM_MISSING_TERMINAL_EVENT_CATEGORY.to_string())
} else {
None
};
let terminal_error_message = terminal_error_message.or_else(|| {
missing_observed_finish
.then(|| "execution runtime stream ended before provider terminal event".to_string())
missing_observed_finish.then(|| STREAM_MISSING_TERMINAL_EVENT_MESSAGE.to_string())
});
if client_response.is_none() {
if let (Some(message), Some(category)) = (
@@ -1055,6 +1078,76 @@ fn infer_stream_terminal_state(
}
}
fn captured_stream_terminal_state(
report_kind: &str,
client_contract: &str,
provider_contract: &str,
provider_response: Option<&Value>,
client_response: Option<&Value>,
) -> Option<StreamCapturedTerminalState> {
let report_kind_requires_terminal_event =
stream_report_kind_requires_observed_terminal_event(report_kind);
let provider_contract_requires_terminal_event =
is_openai_responses_family_format_alias(provider_contract);
let client_contract_requires_terminal_event =
is_openai_responses_family_format_alias(client_contract);
let fallback_requires_terminal_event = report_kind_requires_terminal_event
&& !provider_contract_requires_terminal_event
&& !client_contract_requires_terminal_event;
combine_stream_capture_terminal_states(
(provider_contract_requires_terminal_event || fallback_requires_terminal_event)
.then(|| provider_response.and_then(stream_capture_terminal_state))
.flatten(),
(client_contract_requires_terminal_event || fallback_requires_terminal_event)
.then(|| client_response.and_then(stream_capture_terminal_state))
.flatten(),
)
}
fn combine_stream_capture_terminal_states(
current: Option<StreamCapturedTerminalState>,
next: Option<StreamCapturedTerminalState>,
) -> Option<StreamCapturedTerminalState> {
match (current, next) {
(Some(StreamCapturedTerminalState::Failed), _)
| (_, Some(StreamCapturedTerminalState::Failed)) => {
Some(StreamCapturedTerminalState::Failed)
}
(Some(StreamCapturedTerminalState::Missing), _)
| (_, Some(StreamCapturedTerminalState::Missing)) => {
Some(StreamCapturedTerminalState::Missing)
}
(Some(StreamCapturedTerminalState::Completed), _)
| (_, Some(StreamCapturedTerminalState::Completed)) => {
Some(StreamCapturedTerminalState::Completed)
}
(None, None) => None,
}
}
fn stream_report_kind_requires_observed_terminal_event(report_kind: &str) -> bool {
let report_kind = report_kind.trim().to_ascii_lowercase();
report_kind.starts_with("openai_responses_")
|| report_kind.starts_with("openai_compact_")
|| report_kind.starts_with("openai_cli_")
}
fn stream_usage_requires_observed_terminal_event(
report_kind: &str,
client_contract: &str,
provider_contract: &str,
) -> bool {
stream_report_kind_requires_observed_terminal_event(report_kind)
|| is_openai_responses_family_format_alias(client_contract)
|| is_openai_responses_family_format_alias(provider_contract)
}
fn is_openai_responses_family_format_alias(value: &str) -> bool {
let normalized = value.trim().to_ascii_lowercase().replace('_', ":");
aether_ai_formats::is_openai_responses_family_format(normalized.as_str())
}
fn resolve_has_format_conversion(
context: Option<&Map<String, Value>>,
client_contract: &str,
@@ -3945,6 +4038,77 @@ mod tests {
assert_eq!(event.data.output_tokens, None);
}
#[test]
fn stream_terminal_usage_marks_missing_captured_openai_responses_terminal_as_failed() {
let plan = ExecutionPlan {
request_id: "req-stream-missing-captured-finish-1".to_string(),
candidate_id: Some("cand-stream-missing-captured-finish-1".to_string()),
provider_name: Some("OpenAI".to_string()),
provider_id: "provider-1".to_string(),
endpoint_id: "endpoint-1".to_string(),
key_id: "key-1".to_string(),
method: "POST".to_string(),
url: "https://example.com/v1/responses".to_string(),
headers: BTreeMap::new(),
content_type: None,
content_encoding: None,
body: RequestBody {
json_body: None,
body_bytes_b64: None,
body_ref: None,
},
stream: true,
client_api_format: "openai:responses".to_string(),
provider_api_format: "openai:responses".to_string(),
model_name: Some("gpt-5.5".to_string()),
proxy: None,
transport_profile: None,
timeouts: None,
};
let provider_sse = concat!(
"event: response.created\n",
"data: {\"type\":\"response.created\"}\n\n",
"event: response.output_text.delta\n",
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"hello\"}\n\n"
);
let payload = GatewayStreamReportRequest {
trace_id: "trace-stream-missing-captured-finish-1".to_string(),
report_kind: "openai_responses_stream_success".to_string(),
report_context: Some(json!({
"client_api_format": "openai:responses",
"provider_api_format": "openai:responses"
})),
status_code: 200,
headers: BTreeMap::from([(
"content-type".to_string(),
"text/event-stream".to_string(),
)]),
provider_body_base64: Some(
base64::engine::general_purpose::STANDARD.encode(provider_sse.as_bytes()),
),
provider_body_state: Some(UsageBodyCaptureState::Inline),
client_body_base64: None,
client_body_state: Some(UsageBodyCaptureState::None),
terminal_summary: None,
telemetry: None,
};
let event =
build_stream_terminal_usage_event(&plan, payload.report_context.as_ref(), &payload)
.expect("usage event should build");
assert_eq!(event.event_type, UsageEventType::Failed);
assert_eq!(event.data.status_code, Some(200));
assert_eq!(
event.data.error_category.as_deref(),
Some("stream_missing_terminal_event")
);
assert_eq!(
event.data.error_message.as_deref(),
Some("execution runtime stream ended before provider terminal event")
);
}
#[test]
fn stream_terminal_usage_marks_http_200_response_failed_as_stream_terminal_error() {
let plan = ExecutionPlan {