GenarrativeAI/Genarrative
10
0
Fork 0
Code Issues 21 Pull Requests 6 Actions Packages Projects 1 Releases Wiki Activity

refactor: improve result reconciliation by distinguishing retryable errors and fatal failures

Browse Source
This commit is contained in:
王德宇
2026-07-17 19:05:44 +08:00
parent 15c8a2f8b1
commit 0f3c63a6c6
3 changed files with 101 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/project-memory/shared-memory/decision-log.md
+1 -1
View File
@@ -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`。
docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md
+1 -1
View File
@@ -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` 的业务真相。
### 创作 / 游玩统一流程主干
server-rs/crates/api-server/src/editor_agent/reconcile.rs
+99 -20
View File
@@ -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<String> 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::<Value>(payload).map_err(|e| e.to_string()))?
.ok_or_else(|| ResultReconcileFailure::RetryLater("生成任务缺少结果数据".to_string()))
.and_then(|payload| {
serde_json::from_str::<Value>(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"));
}
}