Files
Genarrative/server-rs/crates/api-server/src/external_generation.rs
T
lhk229 56fe1cb85d
Project CI / Frontend tests (pull_request) Successful in 2m58s
Project CI / Native shell tests (pull_request) Successful in 12m45s
Project CI / Repository checks (pull_request) Failing after 9s
Project CI / Backend tests (pull_request) Failing after 9s
收紧编辑器内部处理元数据边界
普通用户与公开响应统一移除生成 provider、内部处理模型和抠图审计字段。
普通生成请求移除 segModel,由后端固定抠图策略并清理客户端伪造的保留字段。
手动去背景继承可信源模型,角色动作与去背景失败统一使用稳定用户文案。
后台原始审计继续保留真实 provider 与抠图模型,并补齐 Owner、Public、Admin 和紧凑结果测试。
同步更新前端响应契约、外部 OpenAPI、后端架构文档和项目决策记录。
2026-07-31 04:58:07 +00:00

483 lines
19 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
use axum::{
Json,
extract::{Extension, Path, Query, State},
http::StatusCode,
response::Response,
};
use serde::Deserialize;
use serde_json::json;
use shared_contracts::external_generation::{
ExternalGenerationJobStatus, ExternalGenerationJobStatusDetailRecord,
ExternalGenerationJobStatusRecord, ExternalGenerationJobStatusResponse,
ExternalGenerationQueueOverview, ExternalGenerationQueueOverviewResponse,
ExternalGenerationTaskAcknowledgeRequest, ExternalGenerationTaskAcknowledgeResponse,
ExternalGenerationTaskListResponse, ExternalGenerationTaskRecord,
};
use spacetime_client::{
ExternalGenerationJobAcknowledgeRecordInput, ExternalGenerationJobGetRecordInput,
ExternalGenerationJobListRecordInput, ExternalGenerationJobSummaryListRecord,
ExternalGenerationJobSummaryRecord, SpacetimeClientError,
};
use crate::editor_generation_queue::{
EDITOR_BACKGROUND_REMOVAL_JOB_KIND, EDITOR_CHARACTER_ANIMATION_GENERATION_JOB_KIND,
};
use crate::{
api_response::json_success_body, auth::AuthenticatedAccessToken, http_error::AppError,
request_context::RequestContext, state::AppState,
};
const EXTERNAL_GENERATION_PROVIDER: &str = "external_generation";
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ExternalGenerationTaskListQuery {
limit: Option<u32>,
include_acknowledged_terminal: Option<bool>,
statuses: Option<String>,
}
pub async fn get_external_generation_queue_overview(
State(state): State<AppState>,
Extension(request_context): Extension<RequestContext>,
Extension(authenticated): Extension<AuthenticatedAccessToken>,
) -> Result<Json<serde_json::Value>, Response> {
let owner_user_id = authenticated.claims().user_id().to_string();
let list = state
.spacetime_client()
.list_external_generation_job_summaries(ExternalGenerationJobListRecordInput {
owner_user_id,
limit: 1,
include_acknowledged_terminal: false,
statuses: Vec::new(),
})
.await
.map_err(|error| external_generation_error_response(&request_context, error))?;
Ok(json_success_body(
Some(&request_context),
ExternalGenerationQueueOverviewResponse {
overview: map_external_generation_queue_overview(&list),
},
))
}
pub async fn list_external_generation_tasks(
State(state): State<AppState>,
Extension(request_context): Extension<RequestContext>,
Extension(authenticated): Extension<AuthenticatedAccessToken>,
Query(query): Query<ExternalGenerationTaskListQuery>,
) -> Result<Json<serde_json::Value>, Response> {
let owner_user_id = authenticated.claims().user_id().to_string();
let requested_limit = query.limit.unwrap_or(50).clamp(1, 100);
let status_filter = parse_external_generation_status_filter(query.statuses.as_deref());
let statuses = external_generation_status_filter_input(&status_filter);
let list = state
.spacetime_client()
.list_external_generation_job_summaries(ExternalGenerationJobListRecordInput {
owner_user_id,
limit: requested_limit,
include_acknowledged_terminal: query.include_acknowledged_terminal.unwrap_or(false),
statuses,
})
.await
.map_err(|error| external_generation_error_response(&request_context, error))?;
let overview = map_external_generation_queue_overview(&list);
let mut tasks: Vec<_> = list
.jobs
.into_iter()
.map(map_external_generation_task_record)
.filter(|task| {
status_filter.is_empty() || status_filter.iter().any(|status| status == &task.status)
})
.collect();
tasks.truncate(requested_limit as usize);
Ok(json_success_body(
Some(&request_context),
ExternalGenerationTaskListResponse { overview, tasks },
))
}
pub async fn acknowledge_external_generation_tasks(
State(state): State<AppState>,
Extension(request_context): Extension<RequestContext>,
Extension(authenticated): Extension<AuthenticatedAccessToken>,
Json(payload): Json<ExternalGenerationTaskAcknowledgeRequest>,
) -> Result<Json<serde_json::Value>, Response> {
let owner_user_id = authenticated.claims().user_id().to_string();
let acknowledged = state
.spacetime_client()
.acknowledge_external_generation_job_summaries(
ExternalGenerationJobAcknowledgeRecordInput {
owner_user_id,
job_ids: payload.job_ids,
acknowledged_at_micros: current_utc_micros(),
},
)
.await
.map_err(|error| external_generation_error_response(&request_context, error))?;
Ok(json_success_body(
Some(&request_context),
ExternalGenerationTaskAcknowledgeResponse {
acknowledged_tasks: acknowledged
.jobs
.into_iter()
.map(map_external_generation_task_record)
.collect(),
},
))
}
pub async fn get_external_generation_job_status(
State(state): State<AppState>,
Extension(request_context): Extension<RequestContext>,
Extension(authenticated): Extension<AuthenticatedAccessToken>,
Path(job_id): Path<String>,
) -> Result<Json<serde_json::Value>, Response> {
let owner_user_id = authenticated.claims().user_id().to_string();
let job = state
.spacetime_client()
.get_external_generation_job_summary(ExternalGenerationJobGetRecordInput {
job_id,
owner_user_id,
})
.await
.map_err(|error| external_generation_error_response(&request_context, error))?;
Ok(json_success_body(
Some(&request_context),
ExternalGenerationJobStatusResponse {
job: map_external_generation_job_status_detail(job),
},
))
}
fn map_external_generation_queue_overview(
list: &ExternalGenerationJobSummaryListRecord,
) -> ExternalGenerationQueueOverview {
ExternalGenerationQueueOverview {
pending_count: list.pending_count,
running_count: list.running_count,
unacknowledged_terminal_count: list.unacknowledged_terminal_count,
updated_at_micros: list.now_micros,
}
}
fn map_external_generation_job_status(
job: ExternalGenerationJobSummaryRecord,
) -> ExternalGenerationJobStatusRecord {
let (status, phase_detail, progress, error) = match job.status.as_str() {
"completed" => (
ExternalGenerationJobStatus::Completed,
"生成已完成。",
100,
None,
),
"running" if job.phase.as_deref() == Some("processing") => {
(ExternalGenerationJobStatus::Running, "正在处理。", 70, None)
}
"running" => (ExternalGenerationJobStatus::Running, "正在生成。", 35, None),
"failed" => (
ExternalGenerationJobStatus::Failed,
"生成失败。",
0,
user_visible_external_generation_error(
job.job_kind.as_str(),
job.last_error_message.clone(),
),
),
_ => (ExternalGenerationJobStatus::Queued, "排队中。", 8, None),
};
ExternalGenerationJobStatusRecord {
operation_id: job.job_id.clone(),
status,
phase_label: job.request_label.clone(),
phase_detail: phase_detail.to_string(),
progress,
error,
updated_at_micros: job.updated_at_micros,
}
}
/// 外部任务表保存的失败文本同时服务于 worker 诊断和用户通知。抠图及角色动作透明化任务的
/// 原始失败文本可能包含内部 provider、分割模型或实现细节,因此只能在普通用户读取边界
/// 替换成稳定文案。
/// 原始值仍留在任务记录、tracing 和后台审计路径中。
fn user_visible_external_generation_error(job_kind: &str, error: Option<String>) -> Option<String> {
if job_kind == EDITOR_BACKGROUND_REMOVAL_JOB_KIND && error.is_some() {
return Some("去除背景失败,请稍后重试。".to_string());
}
if job_kind == EDITOR_CHARACTER_ANIMATION_GENERATION_JOB_KIND && error.is_some() {
return Some("角色动作生成失败,请稍后重试。".to_string());
}
error
}
fn map_external_generation_job_status_detail(
job: ExternalGenerationJobSummaryRecord,
) -> ExternalGenerationJobStatusDetailRecord {
let warning = job.warning_message.clone();
ExternalGenerationJobStatusDetailRecord {
status: map_external_generation_job_status(job),
warning,
}
}
fn map_external_generation_task_record(
job: ExternalGenerationJobSummaryRecord,
) -> ExternalGenerationTaskRecord {
let status_record = map_external_generation_job_status(job.clone());
ExternalGenerationTaskRecord {
job_id: job.job_id,
job_kind: job.job_kind,
source_module: job.source_module,
source_entity_id: job.source_entity_id,
request_label: job.request_label,
request_prompt: job.request_prompt,
status: status_record.status,
phase_label: status_record.phase_label,
phase_detail: status_record.phase_detail,
progress: status_record.progress,
error: status_record.error,
warning: job.warning_message,
price_mud_points: job.price_mud_points,
refund_ledger_id: job.refund_ledger_id,
notification_acknowledged_at: job.notification_acknowledged_at,
created_at: job.created_at,
started_at: job.started_at,
completed_at: job.completed_at,
updated_at: job.updated_at,
updated_at_micros: job.updated_at_micros,
}
}
fn parse_external_generation_status_filter(
statuses: Option<&str>,
) -> Vec<ExternalGenerationJobStatus> {
statuses
.unwrap_or_default()
.split(',')
.filter_map(|status| match status.trim() {
"queued" => Some(ExternalGenerationJobStatus::Queued),
"running" => Some(ExternalGenerationJobStatus::Running),
"completed" => Some(ExternalGenerationJobStatus::Completed),
"failed" => Some(ExternalGenerationJobStatus::Failed),
_ => None,
})
.collect()
}
fn external_generation_status_filter_input(
statuses: &[ExternalGenerationJobStatus],
) -> Vec<String> {
statuses
.iter()
.map(|status| match status {
ExternalGenerationJobStatus::Queued => "queued",
ExternalGenerationJobStatus::Running => "running",
ExternalGenerationJobStatus::Completed => "completed",
ExternalGenerationJobStatus::Failed => "failed",
})
.map(str::to_string)
.collect()
}
fn current_utc_micros() -> i64 {
shared_kernel::offset_datetime_to_unix_micros(time::OffsetDateTime::now_utc())
}
fn external_generation_error_response(
request_context: &RequestContext,
error: SpacetimeClientError,
) -> Response {
AppError::from_status(StatusCode::BAD_GATEWAY)
.with_details(json!({
"provider": EXTERNAL_GENERATION_PROVIDER,
"message": error.to_string(),
}))
.into_response_with_context(Some(request_context))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parses_external_generation_status_filter() {
assert_eq!(
parse_external_generation_status_filter(Some("running,queued,unknown")),
vec![
ExternalGenerationJobStatus::Running,
ExternalGenerationJobStatus::Queued
]
);
}
#[test]
fn maps_task_from_payload_free_summary_projection() {
let summary = ExternalGenerationJobSummaryRecord {
job_id: "task-1".to_string(),
job_kind: "editor_image_generation".to_string(),
owner_user_id: "user-1".to_string(),
source_module: "editor-canvas".to_string(),
source_entity_id: "project-1".to_string(),
request_label: "图片生成".to_string(),
request_prompt: Some("发光主视觉".to_string()),
status: "completed".to_string(),
last_error_message: Some("上一次尝试失败".to_string()),
created_at: "2026-07-10T08:00:00Z".to_string(),
started_at: Some("2026-07-10T08:00:01Z".to_string()),
completed_at: Some("2026-07-10T08:00:10Z".to_string()),
updated_at: "2026-07-10T08:00:10Z".to_string(),
updated_at_micros: 1_000,
price_mud_points: 4,
refund_ledger_id: None,
notification_acknowledged_at: None,
notification_acknowledged_at_micros: None,
phase: None,
warning_message: Some("连通域数量不足".to_string()),
};
let status = map_external_generation_job_status_detail(summary.clone());
let task = map_external_generation_task_record(summary);
assert_eq!(status.warning.as_deref(), Some("连通域数量不足"));
assert!(status.status.error.is_none());
assert_eq!(task.request_prompt.as_deref(), Some("发光主视觉"));
assert_eq!(task.status, ExternalGenerationJobStatus::Completed);
assert_eq!(task.warning.as_deref(), Some("连通域数量不足"));
assert!(task.error.is_none());
}
#[test]
fn maps_running_processing_phase_from_backend_projection() {
let status = map_external_generation_job_status(ExternalGenerationJobSummaryRecord {
job_id: "task-processing".to_string(),
job_kind: "editor_character_animation_generation".to_string(),
owner_user_id: "user-1".to_string(),
source_module: "editor-canvas".to_string(),
source_entity_id: "project-1".to_string(),
request_label: "角色动作生成".to_string(),
request_prompt: None,
status: "running".to_string(),
last_error_message: None,
created_at: "2026-07-13T08:00:00Z".to_string(),
started_at: Some("2026-07-13T08:00:01Z".to_string()),
completed_at: None,
updated_at: "2026-07-13T08:00:10Z".to_string(),
updated_at_micros: 1_000,
price_mud_points: 4,
refund_ledger_id: None,
notification_acknowledged_at: None,
notification_acknowledged_at_micros: None,
phase: Some("processing".to_string()),
warning_message: None,
});
assert_eq!(status.status, ExternalGenerationJobStatus::Running);
assert_eq!(status.phase_detail, "正在处理。");
assert_eq!(status.progress, 70);
}
#[test]
fn maps_legacy_running_job_without_phase_as_generating() {
let mut job = ExternalGenerationJobSummaryRecord {
job_id: "task-legacy".to_string(),
job_kind: "editor_image_generation".to_string(),
owner_user_id: "user-1".to_string(),
source_module: "editor-canvas".to_string(),
source_entity_id: "project-1".to_string(),
request_label: "图片生成".to_string(),
request_prompt: None,
status: "running".to_string(),
last_error_message: None,
created_at: "2026-07-13T08:00:00Z".to_string(),
started_at: Some("2026-07-13T08:00:01Z".to_string()),
completed_at: None,
updated_at: "2026-07-13T08:00:10Z".to_string(),
updated_at_micros: 1_000,
price_mud_points: 4,
refund_ledger_id: None,
notification_acknowledged_at: None,
notification_acknowledged_at_micros: None,
phase: None,
warning_message: None,
};
let legacy = map_external_generation_job_status(job.clone());
assert_eq!(legacy.phase_detail, "正在生成。");
job.phase = Some("generating".to_string());
let generating = map_external_generation_job_status(job);
assert_eq!(generating.phase_detail, "正在生成。");
}
#[test]
fn background_removal_failure_hides_internal_processing_details_from_owner() {
let job = ExternalGenerationJobSummaryRecord {
job_id: "task-matting-failed".to_string(),
job_kind: EDITOR_BACKGROUND_REMOVAL_JOB_KIND.to_string(),
owner_user_id: "user-1".to_string(),
source_module: "editor-canvas".to_string(),
source_entity_id: "project-1".to_string(),
request_label: "去除背景".to_string(),
request_prompt: None,
status: "failed".to_string(),
last_error_message: Some(
"BgFilter birefnet provider 返回失败,model=segment-common-image".to_string(),
),
created_at: "2026-07-31T08:00:00Z".to_string(),
started_at: Some("2026-07-31T08:00:01Z".to_string()),
completed_at: Some("2026-07-31T08:00:10Z".to_string()),
updated_at: "2026-07-31T08:00:10Z".to_string(),
updated_at_micros: 1_000,
price_mud_points: 0,
refund_ledger_id: None,
notification_acknowledged_at: None,
notification_acknowledged_at_micros: None,
phase: None,
warning_message: None,
};
let status = map_external_generation_job_status(job.clone());
let task = map_external_generation_task_record(job);
assert_eq!(status.error.as_deref(), Some("去除背景失败,请稍后重试。"));
assert_eq!(task.error.as_deref(), Some("去除背景失败,请稍后重试。"));
for value in [status.error, task.error].into_iter().flatten() {
let lower = value.to_ascii_lowercase();
assert!(!lower.contains("bgfilter"));
assert!(!lower.contains("birefnet"));
assert!(!lower.contains("segment-common-image"));
}
}
#[test]
fn character_animation_failure_hides_internal_matting_details_from_owner() {
let message = user_visible_external_generation_error(
EDITOR_CHARACTER_ANIMATION_GENERATION_JOB_KIND,
Some("解析 BgFilter 输入动作帧失败,seg_model=birefnetprovider=BgFilter".to_string()),
);
assert_eq!(message.as_deref(), Some("角色动作生成失败,请稍后重试。"));
let lower = message
.expect("失败任务应返回稳定文案")
.to_ascii_lowercase();
assert!(!lower.contains("bgfilter"));
assert!(!lower.contains("birefnet"));
assert!(!lower.contains("provider"));
}
#[test]
fn unrelated_generation_failure_keeps_its_user_visible_message() {
let message = user_visible_external_generation_error(
"editor_image_generation",
Some("生成服务暂时不可用。".to_string()),
);
assert_eq!(message.as_deref(), Some("生成服务暂时不可用。"));
}
}