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

fix(usage): refine stream terminal capture gating for OpenAI responses

Browse Source
This commit is contained in:
zhefox
2026-05-23 03:31:15 +08:00
parent bd1e155332
commit 91cb2bbbcc
2 changed files with 131 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/aether-usage-runtime/src/report.rs
+97 -4
View File
@@ -402,13 +402,79 @@ fn is_openai_responses_family_format_alias(value: &str) -> bool {
fn stream_report_captured_terminal_state(
payload: &GatewayStreamReportRequest,
) -> Option<StreamCapturedTerminalState> {
let provider_state =
stream_capture_terminal_state_from_base64(payload.provider_body_base64.as_deref());
let client_state =
stream_capture_terminal_state_from_base64(payload.client_body_base64.as_deref());
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> {
@@ -837,6 +903,33 @@ mod tests {
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
+34 -8
View File
@@ -929,8 +929,13 @@ 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(provider_response_full.as_ref(), client_response.as_ref());
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(),
@@ -1074,12 +1079,29 @@ 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_response.and_then(stream_capture_terminal_state),
client_response.and_then(stream_capture_terminal_state),
(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(),
)
}
@@ -1104,15 +1126,19 @@ fn combine_stream_capture_terminal_states(
}
}
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 {
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_")
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)
}