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

feat: 实现 executor 流式执行路径及完善 gateway 多格式 plan 构建

Browse Source 
- Rust executor 新增流式 plan 支持，覆盖 openai/claude/gemini 的 chat/cli/video 格式
- Gateway 新增 finalize-sync、report-stream 端点及 stream report 模型
- 新增 video sync 操作(create/cancel/remix/delete)的 plan 构建
- 修正 OpenAI Responses(openai:cli) SSE 格式: 添加 event: 行、移除 [DONE] 哨兵
- 统一 OpenAI CLI normalizer 的 ID 生成方法
This commit is contained in:
fawney19
2026-03-21 15:51:37 +08:00
parent 53bb23b510
commit 455234e797
14 changed files with 7692 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

365
crates/aether-gateway/src/executor.rs
View File

@@ -11,6 +11,7 @@ use base64::Engine as _;
use futures_util::{StreamExt, TryStreamExt};
use serde::{Deserialize, Serialize};
use serde_json::json;
use tokio::sync::mpsc;
use tokio_util::codec::{FramedRead, LinesCodec};
use tokio_util::io::StreamReader;
use tracing::warn;
@@ -28,6 +29,18 @@ const GEMINI_FILES_UPLOAD_PLAN_KIND: &str = "gemini_files_upload";
const GEMINI_FILES_DELETE_PLAN_KIND: &str = "gemini_files_delete";
const GEMINI_FILES_DOWNLOAD_PLAN_KIND: &str = "gemini_files_download";
const OPENAI_VIDEO_CONTENT_PLAN_KIND: &str = "openai_video_content";
const OPENAI_VIDEO_CANCEL_SYNC_PLAN_KIND: &str = "openai_video_cancel_sync";
const OPENAI_VIDEO_REMIX_SYNC_PLAN_KIND: &str = "openai_video_remix_sync";
const OPENAI_VIDEO_DELETE_SYNC_PLAN_KIND: &str = "openai_video_delete_sync";
const GEMINI_VIDEO_CREATE_SYNC_PLAN_KIND: &str = "gemini_video_create_sync";
const GEMINI_VIDEO_CANCEL_SYNC_PLAN_KIND: &str = "gemini_video_cancel_sync";
const OPENAI_CHAT_STREAM_PLAN_KIND: &str = "openai_chat_stream";
const CLAUDE_CHAT_STREAM_PLAN_KIND: &str = "claude_chat_stream";
const GEMINI_CHAT_STREAM_PLAN_KIND: &str = "gemini_chat_stream";
const OPENAI_CLI_STREAM_PLAN_KIND: &str = "openai_cli_stream";
const CLAUDE_CLI_STREAM_PLAN_KIND: &str = "claude_cli_stream";
const GEMINI_CLI_STREAM_PLAN_KIND: &str = "gemini_cli_stream";
const OPENAI_VIDEO_CREATE_SYNC_PLAN_KIND: &str = "openai_video_create_sync";
const OPENAI_CHAT_SYNC_PLAN_KIND: &str = "openai_chat_sync";
const OPENAI_CLI_SYNC_PLAN_KIND: &str = "openai_cli_sync";
const OPENAI_COMPACT_SYNC_PLAN_KIND: &str = "openai_compact_sync";
@@ -81,6 +94,20 @@ struct GatewaySyncReportRequest {
telemetry: Option<ExecutionTelemetry>,
}
#[derive(Debug, Serialize)]
struct GatewayStreamReportRequest {
trace_id: String,
report_kind: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
report_context: Option<serde_json::Value>,
status_code: u16,
headers: BTreeMap<String, String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
body_base64: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
telemetry: Option<ExecutionTelemetry>,
}
pub(crate) async fn maybe_execute_via_executor_sync(
state: &AppState,
parts: &http::request::Parts,
@@ -205,6 +232,7 @@ pub(crate) async fn maybe_execute_via_executor_sync(
pub(crate) async fn maybe_execute_via_executor_stream(
state: &AppState,
parts: &http::request::Parts,
body_bytes: &Bytes,
trace_id: &str,
decision: Option<&GatewayControlDecision>,
) -> Result<Option<Response<Body>>, GatewayError> {
@@ -222,14 +250,35 @@ pub(crate) async fn maybe_execute_via_executor_stream(
return Ok(None);
};
let (body_json, body_base64) = if is_json_request(&parts.headers) {
if body_bytes.is_empty() {
(json!({}), None)
} else {
match serde_json::from_slice::<serde_json::Value>(body_bytes) {
Ok(value) => (value, None),
Err(_) => return Ok(None),
}
}
} else {
(
json!({}),
(!body_bytes.is_empty())
.then(|| base64::engine::general_purpose::STANDARD.encode(body_bytes)),
)
};
if !is_matching_stream_request(plan_kind, parts, &body_json) {
return Ok(None);
}
let request_payload = GatewayControlPlanRequest {
trace_id: trace_id.to_string(),
method: parts.method.to_string(),
path: parts.uri.path().to_string(),
query_string: parts.uri.query().map(ToOwned::to_owned),
headers: collect_control_headers(&parts.headers),
body_json: json!({}),
body_base64: None,
body_json,
body_base64,
auth_context: decision.auth_context.clone(),
};
@@ -295,11 +344,14 @@ pub(crate) async fn maybe_execute_via_executor_stream(
execute_executor_stream(
state,
control_base_url,
executor_base_url,
plan,
trace_id,
decision,
plan_kind,
payload.report_kind,
payload.report_context,
)
.await
}
@@ -308,19 +360,69 @@ fn resolve_direct_executor_stream_plan_kind(
parts: &http::request::Parts,
decision: &GatewayControlDecision,
) -> Option<&'static str> {
if parts.method != http::Method::GET || decision.route_class.as_deref() != Some("ai_public") {
if decision.route_class.as_deref() != Some("ai_public") {
return None;
}
if decision.route_family.as_deref() == Some("gemini")
&& decision.route_kind.as_deref() == Some("files")
&& parts.method == http::Method::GET
&& parts.uri.path().ends_with(":download")
{
return Some(GEMINI_FILES_DOWNLOAD_PLAN_KIND);
}
if decision.route_family.as_deref() == Some("openai")
&& decision.route_kind.as_deref() == Some("chat")
&& parts.method == http::Method::POST
&& parts.uri.path() == "/v1/chat/completions"
{
return Some(OPENAI_CHAT_STREAM_PLAN_KIND);
}
if decision.route_family.as_deref() == Some("claude")
&& decision.route_kind.as_deref() == Some("chat")
&& parts.method == http::Method::POST
&& parts.uri.path() == "/v1/messages"
{
return Some(CLAUDE_CHAT_STREAM_PLAN_KIND);
}
if decision.route_family.as_deref() == Some("claude")
&& decision.route_kind.as_deref() == Some("cli")
&& parts.method == http::Method::POST
&& parts.uri.path() == "/v1/messages"
{
return Some(CLAUDE_CLI_STREAM_PLAN_KIND);
}
if decision.route_family.as_deref() == Some("gemini")
&& decision.route_kind.as_deref() == Some("chat")
&& parts.method == http::Method::POST
&& parts.uri.path().ends_with(":streamGenerateContent")
{
return Some(GEMINI_CHAT_STREAM_PLAN_KIND);
}
if decision.route_family.as_deref() == Some("gemini")
&& decision.route_kind.as_deref() == Some("cli")
&& parts.method == http::Method::POST
&& parts.uri.path().ends_with(":streamGenerateContent")
{
return Some(GEMINI_CLI_STREAM_PLAN_KIND);
}
if decision.route_family.as_deref() == Some("openai")
&& decision.route_kind.as_deref() == Some("cli")
&& parts.method == http::Method::POST
&& parts.uri.path() == "/v1/responses"
{
return Some(OPENAI_CLI_STREAM_PLAN_KIND);
}
if decision.route_family.as_deref() == Some("openai")
&& decision.route_kind.as_deref() == Some("video")
&& parts.method == http::Method::GET
&& parts.uri.path().ends_with("/content")
{
return Some(OPENAI_VIDEO_CONTENT_PLAN_KIND);
@@ -337,6 +439,56 @@ fn resolve_direct_executor_sync_plan_kind(
return None;
}
if decision.route_family.as_deref() == Some("openai")
&& decision.route_kind.as_deref() == Some("video")
&& parts.method == http::Method::POST
&& parts.uri.path().starts_with("/v1/videos/")
&& parts.uri.path().ends_with("/cancel")
{
return Some(OPENAI_VIDEO_CANCEL_SYNC_PLAN_KIND);
}
if decision.route_family.as_deref() == Some("openai")
&& decision.route_kind.as_deref() == Some("video")
&& parts.method == http::Method::POST
&& parts.uri.path().starts_with("/v1/videos/")
&& parts.uri.path().ends_with("/remix")
{
return Some(OPENAI_VIDEO_REMIX_SYNC_PLAN_KIND);
}
if decision.route_family.as_deref() == Some("openai")
&& decision.route_kind.as_deref() == Some("video")
&& parts.method == http::Method::POST
&& parts.uri.path() == "/v1/videos"
{
return Some(OPENAI_VIDEO_CREATE_SYNC_PLAN_KIND);
}
if decision.route_family.as_deref() == Some("openai")
&& decision.route_kind.as_deref() == Some("video")
&& parts.method == http::Method::DELETE
&& parts.uri.path().starts_with("/v1/videos/")
{
return Some(OPENAI_VIDEO_DELETE_SYNC_PLAN_KIND);
}
if decision.route_family.as_deref() == Some("gemini")
&& decision.route_kind.as_deref() == Some("video")
&& parts.method == http::Method::POST
&& parts.uri.path().ends_with(":cancel")
{
return Some(GEMINI_VIDEO_CANCEL_SYNC_PLAN_KIND);
}
if decision.route_family.as_deref() == Some("gemini")
&& decision.route_kind.as_deref() == Some("video")
&& parts.method == http::Method::POST
&& parts.uri.path().ends_with(":predictLongRunning")
{
return Some(GEMINI_VIDEO_CREATE_SYNC_PLAN_KIND);
}
if decision.route_family.as_deref() == Some("openai")
&& decision.route_kind.as_deref() == Some("chat")
&& parts.method == http::Method::POST
@@ -419,13 +571,17 @@ fn resolve_direct_executor_sync_plan_kind(
None
}
#[allow(clippy::too_many_arguments)] // internal function, grouping would add unnecessary indirection
async fn execute_executor_stream(
state: &AppState,
control_base_url: &str,
executor_base_url: &str,
plan: ExecutionPlan,
trace_id: &str,
decision: &GatewayControlDecision,
plan_kind: &str,
report_kind: Option<String>,
report_context: Option<serde_json::Value>,
) -> Result<Option<Response<Body>>, GatewayError> {
let response = match state
.client
@@ -469,6 +625,10 @@ async fn execute_executor_stream(
));
};
if should_fallback_to_control_stream(plan_kind, status_code) {
return Ok(None);
}
if status_code >= 400 {
let error_body = collect_error_body(&mut lines).await?;
return Ok(Some(build_executor_error_response(
@@ -481,8 +641,18 @@ async fn execute_executor_stream(
)?));
}
let (tx, mut rx) = mpsc::channel::<Result<Bytes, IoError>>(16);
let state_for_report = state.clone();
let trace_id_owned = trace_id.to_string();
let body_stream = stream! {
let control_base_url_owned = control_base_url.to_string();
let headers_for_report = headers.clone();
let report_kind_owned = report_kind.clone();
let report_context_owned = report_context.clone();
tokio::spawn(async move {
let mut buffered_body = Vec::new();
let mut telemetry: Option<ExecutionTelemetry> = None;
let mut reached_eof = false;
loop {
let next_frame = match read_next_frame(&mut lines).await {
Ok(frame) => frame,
@@ -496,20 +666,35 @@ async fn execute_executor_stream(
};
match frame.payload {
StreamFramePayload::Data { chunk_b64, text } => {
if let Some(chunk_b64) = chunk_b64 {
let chunk = if let Some(chunk_b64) = chunk_b64 {
match base64::engine::general_purpose::STANDARD.decode(chunk_b64) {
Ok(decoded) => yield Ok::<Bytes, IoError>(Bytes::from(decoded)),
Ok(decoded) => decoded,
Err(err) => {
warn!(trace_id = %trace_id_owned, error = %err, "gateway failed to decode executor chunk");
break;
}
}
} else if let Some(text) = text {
yield Ok::<Bytes, IoError>(Bytes::from(text.into_bytes()));
text.into_bytes()
} else {
Vec::new()
};
if chunk.is_empty() {
continue;
}
buffered_body.extend_from_slice(&chunk);
let _ = tx.send(Ok(Bytes::from(chunk))).await;
}
StreamFramePayload::Telemetry {
telemetry: frame_telemetry,
} => {
telemetry = Some(frame_telemetry);
}
StreamFramePayload::Eof { .. } => {
reached_eof = true;
break;
}
StreamFramePayload::Telemetry { .. } => {}
StreamFramePayload::Eof { .. } => break,
StreamFramePayload::Error { error } => {
warn!(trace_id = %trace_id_owned, error = %error.message, "executor stream emitted error frame");
break;
@@ -517,6 +702,39 @@ async fn execute_executor_stream(
StreamFramePayload::Headers { .. } => {}
}
}
drop(tx);
if reached_eof {
if let Some(report_kind) = report_kind_owned {
let report = GatewayStreamReportRequest {
trace_id: trace_id_owned.clone(),
report_kind,
report_context: report_context_owned,
status_code,
headers: headers_for_report,
body_base64: (!buffered_body.is_empty())
.then(|| base64::engine::general_purpose::STANDARD.encode(&buffered_body)),
telemetry,
};
if let Err(err) = submit_stream_report(
&state_for_report,
&control_base_url_owned,
&trace_id_owned,
report,
)
.await
{
warn!(trace_id = %trace_id_owned, error = ?err, "gateway failed to submit stream execution report");
}
}
}
});
let body_stream = stream! {
while let Some(item) = rx.recv().await {
yield item;
}
};
Ok(Some(build_client_response_from_parts(
@@ -574,6 +792,25 @@ async fn execute_executor_sync(
return Ok(None);
}
if should_finalize_sync_response(plan_kind) {
let payload = GatewaySyncReportRequest {
trace_id: trace_id.to_string(),
report_kind: report_kind.unwrap_or_default(),
report_context,
status_code: result.status_code,
headers: headers.clone(),
body_json: body_json.clone(),
body_base64: body_base64.clone(),
telemetry: result.telemetry.clone(),
};
let response = submit_sync_finalize(state, control_base_url, trace_id, payload).await?;
return Ok(Some(build_client_response(
response,
trace_id,
Some(decision),
)?));
}
if let Some(report_kind) = report_kind {
let report = GatewaySyncReportRequest {
trace_id: trace_id.to_string(),
@@ -604,9 +841,23 @@ fn should_fallback_to_control_sync(
result: &ExecutionResult,
body_json: Option<&serde_json::Value>,
) -> bool {
if matches!(
plan_kind,
OPENAI_VIDEO_CANCEL_SYNC_PLAN_KIND | GEMINI_VIDEO_CANCEL_SYNC_PLAN_KIND
) {
return result.status_code >= 400;
}
if plan_kind == OPENAI_VIDEO_DELETE_SYNC_PLAN_KIND {
return result.status_code >= 400 && result.status_code != 404;
}
if !matches!(
plan_kind,
OPENAI_CHAT_SYNC_PLAN_KIND
OPENAI_VIDEO_CREATE_SYNC_PLAN_KIND
| OPENAI_VIDEO_REMIX_SYNC_PLAN_KIND
| GEMINI_VIDEO_CREATE_SYNC_PLAN_KIND
| OPENAI_CHAT_SYNC_PLAN_KIND
| OPENAI_CLI_SYNC_PLAN_KIND
| OPENAI_COMPACT_SYNC_PLAN_KIND
| CLAUDE_CHAT_SYNC_PLAN_KIND
@@ -628,6 +879,50 @@ fn should_fallback_to_control_sync(
body_json.get("error").is_some()
}
fn should_finalize_sync_response(plan_kind: &str) -> bool {
matches!(
plan_kind,
OPENAI_VIDEO_CANCEL_SYNC_PLAN_KIND
| OPENAI_VIDEO_CREATE_SYNC_PLAN_KIND
| OPENAI_VIDEO_REMIX_SYNC_PLAN_KIND
| OPENAI_VIDEO_DELETE_SYNC_PLAN_KIND
| GEMINI_VIDEO_CREATE_SYNC_PLAN_KIND
| GEMINI_VIDEO_CANCEL_SYNC_PLAN_KIND
)
}
fn should_fallback_to_control_stream(plan_kind: &str, status_code: u16) -> bool {
matches!(
plan_kind,
OPENAI_CHAT_STREAM_PLAN_KIND
| CLAUDE_CHAT_STREAM_PLAN_KIND
| GEMINI_CHAT_STREAM_PLAN_KIND
| OPENAI_CLI_STREAM_PLAN_KIND
| CLAUDE_CLI_STREAM_PLAN_KIND
| GEMINI_CLI_STREAM_PLAN_KIND
) && status_code >= 400
}
fn is_matching_stream_request(
plan_kind: &str,
parts: &http::request::Parts,
body_json: &serde_json::Value,
) -> bool {
match plan_kind {
OPENAI_CHAT_STREAM_PLAN_KIND
| CLAUDE_CHAT_STREAM_PLAN_KIND
| OPENAI_CLI_STREAM_PLAN_KIND
| CLAUDE_CLI_STREAM_PLAN_KIND => body_json
.get("stream")
.and_then(|value| value.as_bool())
.unwrap_or(false),
GEMINI_CHAT_STREAM_PLAN_KIND | GEMINI_CLI_STREAM_PLAN_KIND => {
parts.uri.path().ends_with(":streamGenerateContent")
}
_ => true,
}
}
type DecodedBody = (Vec<u8>, Option<serde_json::Value>, Option<String>);
fn decode_execution_result_body(
@@ -687,6 +982,56 @@ async fn submit_sync_report(
Ok(())
}
async fn submit_sync_finalize(
state: &AppState,
control_base_url: &str,
trace_id: &str,
payload: GatewaySyncReportRequest,
) -> Result<reqwest::Response, GatewayError> {
state
.client
.post(format!(
"{control_base_url}/api/internal/gateway/finalize-sync"
))
.header(TRACE_ID_HEADER, trace_id)
.json(&payload)
.send()
.await
.map_err(|err| GatewayError::ControlUnavailable {
trace_id: trace_id.to_string(),
message: err.to_string(),
})
}
async fn submit_stream_report(
state: &AppState,
control_base_url: &str,
trace_id: &str,
payload: GatewayStreamReportRequest,
) -> Result<(), GatewayError> {
let response = state
.client
.post(format!(
"{control_base_url}/api/internal/gateway/report-stream"
))
.header(TRACE_ID_HEADER, trace_id)
.json(&payload)
.send()
.await
.map_err(|err| GatewayError::ControlUnavailable {
trace_id: trace_id.to_string(),
message: err.to_string(),
})?;
response
.error_for_status()
.map_err(|err| GatewayError::ControlUnavailable {
trace_id: trace_id.to_string(),
message: err.to_string(),
})?;
Ok(())
}
async fn collect_error_body<R>(
lines: &mut FramedRead<R, LinesCodec>,
) -> Result<Vec<u8>, GatewayError>

1111
crates/aether-gateway/src/gateway/tests/ai_execute.rs
View File

File diff suppressed because it is too large Load Diff

1863
crates/aether-gateway/src/gateway/tests/video.rs
View File

File diff suppressed because it is too large Load Diff

46
crates/aether-gateway/src/handlers.rs
View File

@@ -11,7 +11,7 @@ use tracing::info;
use crate::gateway::constants::*;
use crate::gateway::headers::{
extract_or_generate_trace_id, header_value_str, should_skip_request_header,
extract_or_generate_trace_id, header_value_str, is_json_request, should_skip_request_header,
};
use crate::gateway::{
build_client_response, maybe_execute_via_control, maybe_execute_via_executor_stream,
@@ -132,6 +132,20 @@ pub(crate) async fn proxy_request(
let buffered_body = to_bytes(body, usize::MAX)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))?;
let stream_request = request_wants_stream(&parts, &buffered_body);
if stream_request {
if let Some(executor_response) = maybe_execute_via_executor_stream(
&state,
&parts,
&buffered_body,
&trace_id,
control_decision.as_ref(),
)
.await?
{
return Ok(executor_response);
}
}
if let Some(executor_response) = maybe_execute_via_executor_sync(
&state,
&parts,
@@ -143,11 +157,18 @@ pub(crate) async fn proxy_request(
{
return Ok(executor_response);
}
if let Some(executor_response) =
maybe_execute_via_executor_stream(&state, &parts, &trace_id, control_decision.as_ref())
.await?
{
return Ok(executor_response);
if parts.method != http::Method::POST {
if let Some(executor_response) = maybe_execute_via_executor_stream(
&state,
&parts,
&buffered_body,
&trace_id,
control_decision.as_ref(),
)
.await?
{
return Ok(executor_response);
}
}
if let Some(control_response) = maybe_execute_via_control(
&state,
@@ -202,3 +223,16 @@ pub(crate) async fn proxy_request(
Ok(response)
}
fn request_wants_stream(parts: &http::request::Parts, body: &axum::body::Bytes) -> bool {
if parts.uri.path().contains(":streamGenerateContent") {
return true;
}
if !is_json_request(&parts.headers) || body.is_empty() {
return false;
}
serde_json::from_slice::<serde_json::Value>(body)
.ok()
.and_then(|value| value.get("stream").and_then(|stream| stream.as_bool()))
.unwrap_or(false)
}