fix(client): recover delayed image results without resubmitting

This commit is contained in:
WuYongJie 2026-11-13 09:32:46 +08:00
parent 64358a7870
commit dcd68ff494
6 changed files with 99 additions and 15 deletions

View File

@ -2,8 +2,13 @@ import { randomUUID } from 'node:crypto';
import { setTimeout as delay } from 'node:timers/promises';
import { workflowResult } from './workflow-context.mjs';
// The native image worker can wait 10 minutes for the POST and 3 more for
// read-only recovery. Leave one minute for local preparation and final receipts.
const IMAGE_OPERATION_WAIT_MS = 14 * 60_000;
const LOCAL_OPERATION_WAIT_MS = 120_000;
// One new operation, then only read its status. No transport or model retry.
export async function executeWorkflow(host, args, signal, { timeoutMs = 120000, pause = delay } = {}) {
export async function executeWorkflow(host, args, signal, { timeoutMs, pause = delay } = {}) {
if (host.canvasNotificationsDetached || host.canvasWritesStopped) throw Error('原项目已切换或原操作需核查,本轮不再派发');
if (host.nativeUnresolved) throw Error('原操作结果未知,先核查原请求;可使用 native_reconcile 只读核查,不会重发');
if (host.nativeToolActive) throw Error('已有原生操作正在执行,请依次等待');
@ -30,13 +35,13 @@ export async function executeWorkflow(host, args, signal, { timeoutMs = 120000,
if (host.canvasNotificationsDetached || host.canvasWritesStopped) throw Error('原项目已切换或用户已接手,未派发采纳');
}
dispatched = true;
const deadline = Date.now() + timeoutMs;
const saveDeadline = Date.now() + (timeoutMs ?? LOCAL_OPERATION_WAIT_MS);
let result = await host.canvas.workflowExecute({ ...args, operationId });
while (result?.operationId === operationId && result.phase === 'waiting_save' && result.dispatched === false) {
dispatched = false;
checkTurn();
signal?.throwIfAborted();
if (Date.now() >= deadline) throw Error('项目仍在保存,本次尚未开始处理');
if (Date.now() >= saveDeadline) throw Error('项目仍在保存,本次尚未开始处理');
await pause(100, undefined, { signal });
checkTurn();
signal?.throwIfAborted();
@ -67,9 +72,10 @@ export async function executeWorkflow(host, args, signal, { timeoutMs = 120000,
}
if (result?.operationId !== operationId || result.phase !== 'submitted') throw Error('工作流操作提交状态无效,原操作需核查;不会重发');
await host.store.appendEvent({ type:'workflow_operation_submitted', run:host.canvas.run, operationId, intent:args.intent, slotId:args.slotId, productVariantId:args.productVariantId });
const deadline = Date.now() + (timeoutMs ?? (args.intent === 'generate_candidate' ? IMAGE_OPERATION_WAIT_MS : LOCAL_OPERATION_WAIT_MS));
while (result.phase === 'submitted' && Date.now() < deadline) {
signal?.throwIfAborted();
await pause(200, undefined, { signal });
await pause(args.intent === 'generate_candidate' ? 500 : 200, undefined, { signal });
result = await host.canvas.workflowPoll(operationId);
checkTurn();
if (result?.operationId !== operationId || !['submitted','complete','failed','unknown'].includes(result.phase)) throw Error('工作流操作回执无效,原操作需核查;不会重发');

View File

@ -66,3 +66,15 @@ test('generation never requests a write lease',async()=>{
const x=setup(['complete']);x.host.canvas.read=async()=>{throw Error('unexpected canvas read')};x.host.canvas.requestWrite=async()=>{throw Error('unexpected permission')};
await executeWorkflow(x.host,args,undefined,{pause:async()=>{}});assert.equal(x.host.canvasWriteRequested,undefined);
});
test('a long image generation keeps polling the original operation after the former two-minute cutoff', async t => {
const x = setup(['submitted', 'submitted', 'complete']);
let now = 0;
t.mock.method(Date, 'now', () => now);
const result = await executeWorkflow(x.host, args, undefined, {
pause: async () => { now += 65_000; },
});
assert.equal(result.operation.phase, 'complete');
assert.deepEqual(x.calls.map(([type]) => type), ['execute', 'poll', 'poll', 'poll', 'read']);
assert.ok(x.calls.filter(([type]) => type === 'poll').every(([, id]) => id === x.calls[0][1].operationId));
});

View File

@ -823,8 +823,14 @@ impl Backend {
let brand = json!({"name":creation["brief"]["brandName"],"tone":profile["tone"],"visualRules":visual.chars().take(4000).collect::<String>(),"fontFamily":"Noto Sans SC"});
let call = self.image_job(Some(Arc::new(references)), &json!({"doc":doc,"revision":input["baseRevision"],"model":model,"mode":if parent.is_some(){"refine"}else{"generate"},"text":prompt,"screen":output["name"],"references":layers,"slotBinding":binding,"slotFrame":field["frame"],"brand":if knowledge.is_null(){brand}else{json!({"name":creation["brief"]["brandName"],"knowledge":knowledge})},"facts":creation["facts"]}), false, events, ctx);
let file = dir.join("image-jobs").join(&candidate).join("task.json");
let journal =
let mut journal =
if file.is_file() { read_json(&file)? } else { json!({"id":candidate,"slotBinding":binding,"phase":"prepared","notDispatched":true}) };
let mut call = call;
// Keep the same worker and original candidate until read-only recovery settles.
if matches!(journal["phase"].as_str(), Some("dispatched" | "running" | "unknown")) {
call = self.recover_original_image(doc, &candidate, events, ctx);
journal = read_json(&file)?;
}
candidate_result = Some(candidate_receipt(&reconcile_candidate(&dir, &journal)?));
operation_error = call.err();
} else if method == "slot-recover" {

View File

@ -343,6 +343,8 @@ pub(super) fn retry_original_queries(
_ => return result,
};
let transient = message.contains("processing")
|| message.contains("原任务仍在生成")
|| message.contains("HTTP 202")
|| message.contains("连接中断")
|| message.contains("HTTP 502")
|| message.contains("HTTP 503")
@ -449,6 +451,15 @@ impl Backend {
egui::ColorImage::from_rgba_unmultiplied([w, h], &rgba),
))
}
pub(super) fn recover_original_image(&mut self, doc: u64, task: &str, tx: &mpsc::Sender<Event>, ctx: &egui::Context) -> R<Value> {
let started = Instant::now();
retry_original_queries(
|budget| self.image_job(None, &json!({"doc":doc,"taskId":task,"recoveryBudgetMs":budget.as_millis() as u64}), true, tx, ctx),
std::thread::sleep,
|| started.elapsed(),
)
}
pub(super) fn image_job(&mut self, doc: Option<Arc<Document>>, input: &Value, recover: bool, tx: &mpsc::Sender<Event>, ctx: &egui::Context) -> R<Value> {
let deadline = if recover { input["recoveryBudgetMs"].as_u64().map(|ms| Instant::now() + Duration::from_millis(ms.min(180_000))) } else { None };
let id = doc.as_ref().map(|d| d.id.0).or_else(|| input["doc"].as_u64()).ok_or("请先打开项目")?;

View File

@ -2727,8 +2727,13 @@ impl Workspace {
self.note("本地 Agent 连接已断开,草稿已保留");
}
if self.pi_run.is_some() {
let pointer = ui.input(|i| i.pointer.any_pressed() && i.pointer.interact_pos().is_some_and(|p| app.last_canvas_rect.contains(p)));
if pointer || current != self.owner_doc.map(|id| (id, self.expected_revision)) {
// Browsing is allowed during conversation and candidate generation.
// A primary press can still interrupt an explicitly authorized canvas write.
let editing_pointer = self.allow_apply && ui.input(|input| {
input.pointer.primary_pressed()
&& input.pointer.interact_pos().is_some_and(|position| app.last_canvas_rect.contains(position))
});
if editing_pointer || current != self.owner_doc.map(|id| (id, self.expected_revision)) {
self.handover();
}
}

View File

@ -540,15 +540,26 @@ fn execute(w: &mut Workspace, app: &mut PhotocraftApp, input: &Value, version: u
if w.workflow.operations.iter().filter(|op| matches!(op["phase"].as_str(), Some("complete" | "failed"))).count() < excess {
return Err("本次会话操作记录已达上限,未执行新操作".into());
}
if !available || w.pi_write_blocked || w.agent_effects_paused || w.unknown || w.pi_unknown || w.unknown_image
|| w.agent_native.cancelled(&run) || w.pending_project_open.is_some() || !w.queue.is_empty()
|| w.pending_image.is_some() || w.pending_slot_apply.is_some()
if !available || w.pending_project_open.is_some() {
return Err("当前项目资料尚未就绪或已切换,请确认项目后继续".into());
}
if w.unknown || w.pi_unknown || w.unknown_image {
return Err("有一项原请求需要核查,请先查看该请求的结果".into());
}
if app.session.active().is_none_or(|active| active.revision != w.expected_revision) {
return Err("画布版本已变化,请确认当前图位后继续".into());
}
if w.pi_write_blocked || w.agent_effects_paused || w.agent_native.cancelled(&run) {
return Err("当前回合的画布操作已取消,可在对话中确认下一步".into());
}
if w.config || w.model_settings || w.key_settings || !w.api_key.is_empty() || w.delete_confirm.is_some()
|| app.file_dialog_open() || app.discard.is_some() {
return Err("请先完成当前设置或确认窗口中的操作".into());
}
if !w.queue.is_empty() || w.pending_image.is_some() || w.pending_slot_apply.is_some()
|| w.canvas_workspace.scope_switch_in_flight || w.canvas_workspace.pending_adoption.is_some()
|| w.workflow.operation_in_flight(&run) || w.busy.as_deref() != Some("pi-agent")
|| w.config || w.model_settings || w.key_settings || !w.api_key.is_empty() || w.delete_confirm.is_some()
|| app.file_dialog_open() || app.discard.is_some()
|| app.session.active().is_none_or(|active| active.revision != w.expected_revision) {
return Err("用户已接手或原任务尚未落定,未执行操作".into());
|| w.workflow.operation_in_flight(&run) || w.busy.as_deref() != Some("pi-agent") {
return Err("工作台正在处理其他操作,请等它结束后继续".into());
}
if w.template_workflow["canvasFirst"] != true || input["version"].as_u64() != Some(version) || w.workflow.last_read_version != Some(version) {
return Err("工作流已变化或不支持直接执行,请重新读取".into());
@ -1171,6 +1182,31 @@ mod tests {
assert!(requests.try_recv().is_err());
}
#[test]
fn canvas_click_or_pan_without_edits_does_not_stop_candidate_dispatch() {
for button in [egui::PointerButton::Primary, egui::PointerButton::Middle] {
let (mut w, mut app, requests, ctx) = direct_fixture();
w.observed_doc = w.owner_doc;
app.last_canvas_rect = egui::Rect::from_min_size(egui::Pos2::ZERO, egui::vec2(800.0, 600.0));
let position = egui::pos2(100.0, 100.0);
let input = egui::RawInput {
screen_rect: Some(app.last_canvas_rect),
events: vec![
egui::Event::PointerMoved(position),
egui::Event::PointerButton { pos: position, button, pressed: true, modifiers: Default::default() },
],
..Default::default()
};
let mut frame = ctx.run_ui(input, |ui| assert!(w.tick_context(&mut app, ui)));
frame.textures_delta.clear();
let read = private_call(&mut w, &mut app, &ctx, "pi-workflow-read", json!({})).unwrap();
let result = private_call(&mut w, &mut app, &ctx, "pi-workflow-execute", direct_input(read["version"].clone())).unwrap();
assert_eq!(result["phase"], "submitted", "{button:?}: {result}");
assert!(result["__workflowDelegation"].is_object());
assert!(requests.try_recv().is_err(), "only the delegation can dispatch the candidate");
}
}
#[test]
fn private_dispatch_waits_for_save_ack_before_one_delegation() {
let (mut w, mut app, requests, ctx) = direct_fixture();
@ -1211,6 +1247,14 @@ mod tests {
assert_eq!(rejected["phase"], "not_dispatched", "{scenario}");
assert_eq!(rejected["accepted"], false);
assert_eq!(rejected["dispatched"], false);
assert_eq!(rejected["error"]["message"], match scenario {
"stale" => "工作流已变化或不支持直接执行,请重新读取",
"handover" => "当前回合的画布操作已取消,可在对话中确认下一步",
"unknown" => "有一项原请求需要核查,请先查看该请求的结果",
"drift" => "画布版本已变化,请确认当前图位后继续",
"adoption" => "本轮原生操作尚未取得当前修订许可",
_ => unreachable!(),
});
assert_eq!(rejected["run"], w.pi_run.as_deref().unwrap());
assert_eq!(rejected["documentId"], w.owner_doc.unwrap());
assert_eq!(rejected["baseRevision"], app.session.active().unwrap().revision);