From 0f3c63a6c68e1c88983980d203b3e31c87f8f9de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E5=BE=B7=E5=AE=87?= Date: Fri, 17 Jul 2026 19:05:44 +0800 Subject: [PATCH] refactor: improve result reconciliation by distinguishing retryable errors and fatal failures --- .../shared-memory/decision-log.md | 2 +- ...】server-rs与SpacetimeDB数据契约-2026-05-15.md | 2 +- .../api-server/src/editor_agent/reconcile.rs | 119 +++++++++++++++--- 3 files changed, 101 insertions(+), 22 deletions(-) diff --git a/docs/project-memory/shared-memory/decision-log.md b/docs/project-memory/shared-memory/decision-log.md index 14b871a71..7af62ba89 100644 --- a/docs/project-memory/shared-memory/decision-log.md +++ b/docs/project-memory/shared-memory/decision-log.md @@ -3251,7 +3251,7 @@ ## 2026-07-13 画布 Agent 工具执行状态复用外部生成任务 - 决策：画布 Agent 的 OSS 工具消息使用 `status=not_completed|completed|failed|cancelled` 和可选 `externalJobId`；不使用 `cancelledAt`，不新增关联表。`external_generation_job` 是排队、执行、lease 与计费结算的唯一真相；OSS status 只表达该消息回填结果，不复制 queued / running。确认接口按 `conversationId + messageId + toolName` 稳定去重并复用既有编辑器 worker job kind。 -- 懒回填：`GET /conversation` 会在持有 conversation lock 后扫描 `status=not_completed` 且有 `externalJobId` 的工具消息，按 job id 定向读取主任务；完成时复用原工具 formatter 更新 system text、写入轻量媒体引用并标记 `completed`，任务本身失败时写入 `error` 并标记 `failed`。任务结果读取或 completed payload 解析 / formatter 首次失败后，在同一次 GET 内最多重试 3 次，每次等待 100ms 并重新读取主任务；仍失败才把工具消息写为 `failed`，不依赖前端再次刷新。排队 / 执行保持 `not_completed`。worker 的 `result_payload_json` 只保留 formatter 与媒体引用所需的轻量生成回包，不向通用 summary 状态接口投影。 +- 懒回填：`GET /conversation` 会在持有 conversation lock 后扫描 `status=not_completed` 且有 `externalJobId` 的工具消息，按 job id 定向读取主任务；完成时复用原工具 formatter 更新 system text、写入轻量媒体引用并标记 `completed`，任务本身失败时写入 `error` 并标记 `failed`。任务结果读取或 completed payload 解析 / formatter 首次失败后，在同一次 GET 内最多重试 3 次，每次等待 100ms 并重新读取主任务；读取失败或 completed 任务暂缺 `result_payload_json` 时，本次重试耗尽后保留 `not_completed + externalJobId` 供下次 GET 继续 reconcile，确定性的 payload 损坏、结构不兼容或 formatter 错误才在重试耗尽后写为 `failed`，避免致命错误永久循环。排队 / 执行保持 `not_completed`。worker 的 `result_payload_json` 只保留 formatter 与媒体引用所需的轻量生成回包，不向通用 summary 状态接口投影。 - 影响范围：画布 Agent 共享契约、确认/取消接口、编辑器生成入队 helper、对话状态展示与恢复。 - 验证方式：`cargo check -p api-server -p shared-contracts --manifest-path server-rs/Cargo.toml`、画布 Agent 定向前端测试、`npm run typecheck`、`npm run check:encoding`、`git diff --check`。 diff --git a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md index 4c3a08304..a5bc0abfc 100644 --- a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md +++ b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md @@ -85,7 +85,7 @@ npm run check:server-rs-ddd - 对话附件只允许引用当前工程 `editor_project_resource` 或当前账号 `editor_asset` 的图片；前端可提交展示用 `imageSrc` / `thumbnailSrc`，后端必须按 `resourceId` / `assetId` 重新归一、校验 owner / project 和 `objectKey`，再给 LLM 或生成工具使用。 - 画布 Agent 工具复用既有编辑器图片生成 / 修改 / 图标 spritesheet BFF，并继续使用后端模型定价和 `execute_billable_asset_operation_with_cost`；前端不提交 `priceMudPoints`。 - `/messages/{messageId}/confirm` 与 `/messages/{messageId}/cancel` 只返回成功确认；前端成功后立即重新读取整个会话，以会话详情中的权威消息状态和 `externalJobId` 驱动气泡展示与任务轮询。 -- 会话详情的终态懒回填必须在单次 GET 和同一 conversation lock 内完成有界重试：任务结果读取、completed payload 解析或工具 formatter 首次失败后最多重试 3 次，每次等待 100ms 并重新读取主任务；仍失败后把 OSS 工具消息原子写为 `failed`，保存“重试 3 次后仍失败”的最后错误。不得依赖浏览器再次轮询才能耗尽重试，也不得让消息永久停在 `not_completed`。 +- 会话详情的终态懒回填必须在单次 GET 和同一 conversation lock 内完成有界重试：任务结果读取、completed payload 解析或工具 formatter 首次失败后最多重试 3 次，每次等待 100ms 并重新读取主任务。任务读取失败或 completed 任务暂缺 `result_payload_json` 时，本次重试耗尽后仍保留 OSS 工具消息的 `not_completed + externalJobId`，由下次会话读取继续 reconcile；JSON 损坏、结果结构不兼容或 formatter 失败等确定性致命错误在重试耗尽后原子写为 `failed`，保存“重试 3 次后仍失败”的最后错误，避免永久循环。 - 画布 Agent 是“正式任务 payload 不进入通用用户 read model”规则的窄例外消费者：`GET /api/editor/agent-conversations/{conversationId}` 只按会话中已有的 `externalJobId` 定向读取主任务，完成后由对应工具 formatter 从 `result_payload_json` 提取并归一有界的图片 / 视频 / 音频引用，写入 OSS 工具消息后返回。前端仍不得通过通用任务列表 / 状态接口读取或解析 `request_payload_json` / `result_payload_json`；OSS 轻量媒体引用只是会话展示与后续 Agent 上下文，不替代 `editor_project_resource`、`editor_asset`、`editor_canvas.layers_json` 或 `external_generation_job` 的业务真相。 ### 创作 / 游玩统一流程主干 diff --git a/server-rs/crates/api-server/src/editor_agent/reconcile.rs b/server-rs/crates/api-server/src/editor_agent/reconcile.rs index 3e7606d6e..20d21a02f 100644 --- a/server-rs/crates/api-server/src/editor_agent/reconcile.rs +++ b/server-rs/crates/api-server/src/editor_agent/reconcile.rs @@ -34,6 +34,18 @@ use tokio::time::{Duration, sleep}; pub(crate) const EDITOR_AGENT_RESULT_RECONCILE_MAX_RETRIES: u8 = 3; const EDITOR_AGENT_RESULT_RECONCILE_RETRY_DELAY: Duration = Duration::from_millis(100); +#[derive(Debug, PartialEq)] +enum ResultReconcileFailure { + RetryLater(String), + Fatal(String), +} + +impl From for ResultReconcileFailure { + fn from(error: String) -> Self { + Self::Fatal(error) + } +} + pub async fn reconcile_editor_agent_tool_calls( state: &AppState, conversation: &EditorAgentConversationRecord, @@ -59,15 +71,19 @@ pub async fn reconcile_editor_agent_tool_calls( for retry_count in 0..=EDITOR_AGENT_RESULT_RECONCILE_MAX_RETRIES { let job = match state .spacetime_client() - .get_external_generation_job_generated_artifacts(ExternalGenerationJobGetRecordInput { - job_id: job_id.clone(), - owner_user_id: conversation.owner_user_id.clone(), - }) + .get_external_generation_job_generated_artifacts( + ExternalGenerationJobGetRecordInput { + job_id: job_id.clone(), + owner_user_id: conversation.owner_user_id.clone(), + }, + ) .await { Ok(job) => job, Err(error) => { - last_reconcile_error = Some(format!("读取生成任务结果失败：{error}")); + last_reconcile_error = Some(ResultReconcileFailure::RetryLater(format!( + "读取生成任务结果失败：{error}" + ))); if should_retry_result_reconcile(retry_count) { sleep(EDITOR_AGENT_RESULT_RECONCILE_RETRY_DELAY).await; } @@ -114,12 +130,12 @@ pub async fn reconcile_editor_agent_tool_calls( } } - if let Some(error) = last_reconcile_error { - // unlikely. - // if we don't mark this as failed, - // the client will poll external job id -> found completed -> get conv and reconcile again, - // if here is fatal, it will be stuck in a loop. - mark_result_reconcile_failed(&mut document.messages[message_index], error); + if let Some(error) = last_reconcile_error + && apply_exhausted_result_reconcile_failure( + &mut document.messages[message_index], + error, + ) + { reconciled.push(document.messages[message_index].clone()); } } @@ -141,7 +157,7 @@ fn mark_job_failed(message: &mut EditorAgentMessage, error: String) { message.text = format!("[tool_call:{}] output: {error}", tool_call.tool_name); } -fn mark_result_reconcile_failed(message: &mut EditorAgentMessage, error: String) { +fn mark_fatal_result_reconcile_failed(message: &mut EditorAgentMessage, error: String) { mark_job_failed( message, format!( @@ -150,21 +166,39 @@ fn mark_result_reconcile_failed(message: &mut EditorAgentMessage, error: String) ); } +fn apply_exhausted_result_reconcile_failure( + message: &mut EditorAgentMessage, + failure: ResultReconcileFailure, +) -> bool { + match failure { + ResultReconcileFailure::RetryLater(_) => false, + ResultReconcileFailure::Fatal(error) => { + mark_fatal_result_reconcile_failed(message, error); + true + } + } +} + fn reconcile_completed_editor_agent_tool_call( message: &mut EditorAgentMessage, result_payload_json: Option<&str>, -) -> Result<(), String> { +) -> Result<(), ResultReconcileFailure> { let response = result_payload_json - .ok_or_else(|| "生成任务缺少结果数据".to_string()) - .and_then(|payload| serde_json::from_str::(payload).map_err(|e| e.to_string()))? + .ok_or_else(|| ResultReconcileFailure::RetryLater("生成任务缺少结果数据".to_string())) + .and_then(|payload| { + serde_json::from_str::(payload) + .map_err(|error| ResultReconcileFailure::Fatal(error.to_string())) + })? // TODO extract const .get("editor-agent-tool-call-result") .cloned() - .ok_or_else(|| "生成任务缺少 Agent 工具调用结果".to_string())?; + .ok_or_else(|| { + ResultReconcileFailure::Fatal("生成任务缺少 Agent 工具调用结果".to_string()) + })?; let tool_call = message .tool_call .as_mut() - .ok_or_else(|| "消息缺少工具调用".to_string())?; + .ok_or_else(|| ResultReconcileFailure::Fatal("消息缺少工具调用".to_string()))?; let tool_name = tool_call.tool_name.clone(); match tool_name.as_str() { @@ -226,7 +260,11 @@ fn reconcile_completed_editor_agent_tool_call( }; tool_call.audios = vec![resp_to_asset::editor_agent_generated_audio(&result)]; } - _ => return Err(format!("不支持回填工具 {tool_name}")), + _ => { + return Err(ResultReconcileFailure::Fatal(format!( + "不支持回填工具 {tool_name}" + ))); + } } tool_call.error = None; tool_call.status = EditorAgentToolCallStatus::Completed; @@ -274,11 +312,15 @@ mod tests { } #[test] - fn exhausted_terminal_result_reconcile_marks_the_tool_call_failed() { + fn exhausted_fatal_result_reconcile_marks_the_tool_call_failed() { let mut message = pending_tool_message(); - mark_result_reconcile_failed(&mut message, "invalid result payload".to_string()); + let changed = apply_exhausted_result_reconcile_failure( + &mut message, + ResultReconcileFailure::Fatal("invalid result payload".to_string()), + ); + assert!(changed); let tool_call = message.tool_call.expect("tool call should remain present"); assert_eq!(tool_call.status, EditorAgentToolCallStatus::Failed); assert_eq!( @@ -287,4 +329,41 @@ mod tests { ); assert!(message.text.contains("重试 3 次后仍失败")); } + + #[test] + fn missing_result_payload_is_retryable_and_preserves_pending_job_reference() { + let mut message = pending_tool_message(); + + let error = reconcile_completed_editor_agent_tool_call(&mut message, None) + .expect_err("missing result payload should be retried later"); + + assert_eq!( + error, + ResultReconcileFailure::RetryLater("生成任务缺少结果数据".to_string()) + ); + assert!(!apply_exhausted_result_reconcile_failure( + &mut message, + error, + )); + let tool_call = message.tool_call.expect("tool call should remain present"); + assert_eq!(tool_call.status, EditorAgentToolCallStatus::NotCompleted); + assert_eq!(tool_call.external_job_id.as_deref(), Some("job-1")); + assert_eq!(tool_call.error, None); + } + + #[test] + fn read_failure_is_retryable_without_terminal_message_mutation() { + let mut message = pending_tool_message(); + let error = ResultReconcileFailure::RetryLater( + "读取生成任务结果失败：temporarily unavailable".to_string(), + ); + + assert!(!apply_exhausted_result_reconcile_failure( + &mut message, + error, + )); + let tool_call = message.tool_call.expect("tool call should remain present"); + assert_eq!(tool_call.status, EditorAgentToolCallStatus::NotCompleted); + assert_eq!(tool_call.external_job_id.as_deref(), Some("job-1")); + } }