diff --git a/apps/detail-agent-pi/src/workflow-execute.mjs b/apps/detail-agent-pi/src/workflow-execute.mjs index f785f451..26fcd45d 100644 --- a/apps/detail-agent-pi/src/workflow-execute.mjs +++ b/apps/detail-agent-pi/src/workflow-execute.mjs @@ -30,11 +30,43 @@ export async function executeWorkflow(host, args, signal, { timeoutMs = 120000, if (host.canvasNotificationsDetached || host.canvasWritesStopped) throw Error('原项目已切换或用户已接手,未派发采纳'); } dispatched = true; + const deadline = Date.now() + timeoutMs; 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('项目仍在保存,本次尚未开始处理'); + await pause(100, undefined, { signal }); + checkTurn(); + signal?.throwIfAborted(); + const state = await host.canvas.read(); + checkTurn(); + if (state?.conversationActive === false || state?.writeBlocked || state?.revision !== result.revision) throw Error('画布已变化或用户已接手,尚未派发操作'); + if (state?.error) throw Error(state.error); + dispatched = true; + result = await host.canvas.workflowExecute({ ...args, operationId }); + } + const notDispatched = result?.phase === 'not_dispatched' && result.accepted === false && result.dispatched === false + && result.operationId === operationId && typeof run === 'string' && run && result.run === run + && Number.isSafeInteger(documentId) && documentId > 0 && result.documentId === documentId + && Number.isSafeInteger(result.baseRevision) && result.baseRevision >= 0 + && ['accountId','origin','project'].every(key => typeof identity[key] === 'string' && identity[key]) + && result.accountId === identity.accountId && result.origin === identity.origin + && result.projectLocalId === `${identity.origin}|${identity.accountId}|${identity.project}` + && result.error?.code === 'local_precondition_failed' && typeof result.error.message === 'string' && result.error.message.trim(); + if (notDispatched) { dispatched = false; terminal = true; } checkTurn(); + if (notDispatched) { + const message = `本次操作未派发:${workflowResult({message:result.error.message}).message}`; + host.canvasWritesStopped = true; + host.workflowStopReason = message; + await host.store.appendEvent({type:'workflow_operation_not_dispatched',run,documentId,operationId,baseRevision:result.baseRevision, + intent:args.intent,slotId:args.slotId,productVariantId:args.productVariantId,code:result.error.code,message}); + throw Error(message); + } 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; while (result.phase === 'submitted' && Date.now() < deadline) { signal?.throwIfAborted(); await pause(200, undefined, { signal }); diff --git a/apps/detail-agent-pi/test/workflow-execute.test.mjs b/apps/detail-agent-pi/test/workflow-execute.test.mjs new file mode 100644 index 00000000..f1dab472 --- /dev/null +++ b/apps/detail-agent-pi/test/workflow-execute.test.mjs @@ -0,0 +1,68 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { executeWorkflow } from '../src/workflow-execute.mjs'; +import { StdioCanvas } from '../src/stdio-canvas.mjs'; +const args={version:1,intent:'generate_candidate',outputId:'detail',slotId:'hero',productVariantId:'walnut'}; +function setup(phases) { + const calls=[], events=[]; let id; + const host={canvas:{run:'r',async workflowExecute(input){ calls.push(['execute',input]);id=input.operationId;return {operationId:id,phase:'submitted'}; },async workflowPoll(op){calls.push(['poll',op]);return {operationId:id,phase:phases.shift()||'submitted'};}},store:{async appendEvent(v){events.push(v)}}, async readWorkflow(input){calls.push(['read',input]);return {version:2,section:'slots'};}}; + return {host,calls,events}; +} +test('late prior-turn workflow receipt cannot stop the new turn and retains unknown operation identity',async()=>{ + const x=setup([]);x.host.turnEpoch=1; + let release; + x.host.canvas.workflowExecute=input=>{x.calls.push(['execute',input]);return new Promise(resolve=>{release=()=>resolve({operationId:input.operationId,phase:'submitted'});});}; + const task=executeWorkflow(x.host,args,undefined,{pause:async()=>{}}); + await new Promise(resolve=>setImmediate(resolve)); + const operationId=x.calls[0][1].operationId; + x.host.turnEpoch=2;x.host.canvas.run='new-run';x.host.canvasWritesStopped=false; + release();await assert.rejects(task,/原回合已结束/); + assert.equal(x.host.canvasWritesStopped,false); + assert.deepEqual(x.host.nativeUnresolved,{kind:'workflow',operationId,run:'r',documentId:undefined,identity:{scope:undefined,project:undefined,accountId:undefined,origin:undefined},epoch:1}); + await assert.rejects(executeWorkflow(x.host,args),/先核查原请求/); + assert.equal(x.calls.length,1); +}); +test('one dispatch polls matching operation and refreshes after complete',async()=>{ + const x=setup(['submitted','complete']);const result=await executeWorkflow(x.host,args,undefined,{pause:async()=>{}}); + assert.deepEqual(x.calls.map(c=>c[0]),['execute','poll','poll','read']);assert.equal(result.operation.phase,'complete');assert.equal(result.workflow.version,2); + assert.equal(x.calls[0][1].operationId,x.calls[1][1]);assert.equal(x.host.canvasWritesStopped,undefined); +}); +test('unknown, failed and invalid receipt never redispatch and stop writes',async()=>{ + for (const phase of ['unknown','failed','bad']) {const x=setup([phase]);await assert.rejects(executeWorkflow(x.host,args,undefined,{pause:async()=>{}}));assert.equal(x.calls.filter(c=>c[0]==='execute').length,1);assert.equal(x.host.canvasWritesStopped,true);assert.equal(x.calls.some(c=>c[0]==='read'),false);} +}); + +test('an upstream failure keeps its real reason instead of describing a user handover',async()=>{ + const x=setup([]); + x.host.canvas.workflowPoll=async operationId=>({operationId,phase:'unknown',candidateId:'original',message:'图片服务暂时不可用(503),原请求待核查;不会重发'}); + await assert.rejects(executeWorkflow(x.host,args,undefined,{pause:async()=>{}}),/503/); + assert.match(x.host.workflowStopReason,/503/); + assert.equal(x.host.canvasWritesStopped,true); + assert.equal(x.calls.filter(c=>c[0]==='execute').length,1); + await assert.rejects(executeWorkflow(x.host,args),/原操作需核查/); +}); +test('cancellation and broken poll preserve operation and block further dispatch',async()=>{ + const x=setup([]); x.host.canvas.workflowPoll=async()=>{throw Error('pipe closed')}; + await assert.rejects(executeWorkflow(x.host,args,undefined,{pause:async()=>{}}));await assert.rejects(executeWorkflow(x.host,args));assert.equal(x.calls.filter(c=>c[0]==='execute').length,1); + const y=setup([]), controller=new AbortController();controller.abort();await assert.rejects(executeWorkflow(y.host,args,controller.signal));assert.equal(y.calls.length,0); +}); +test('stdio transport carries session identity and operation id',async()=>{ + const sent=[];const c=new StdioCanvas(async (action,input)=>{sent.push({action,input});},7);c.run='run';await c.workflowExecute({...args,operationId:'op'});await c.workflowPoll('op');assert.deepEqual(sent.map(v=>v.action),['pi-workflow-execute','pi-workflow-poll']);assert.equal(sent[1].input.run,'run');assert.equal(sent[1].input.documentId,7);assert.equal(sent[1].input.operationId,'op'); +}); +test('adoption obtains current revision write permission before dispatch',async()=>{ + const x=setup(['complete']); x.host.canvas.read=async()=>{x.calls.push(['canvas-read']);return {revision:17,conversationActive:true,writeBlocked:false};}; + x.host.canvas.requestWrite=async revision=>{x.calls.push(['permission',revision]);return {allowed:true,revision};}; + await executeWorkflow(x.host,{...args,intent:'adopt_candidate',candidateId:'chosen'},undefined,{pause:async()=>{}}); + assert.deepEqual(x.calls.map(c=>c[0]),['canvas-read','permission','execute','poll','read']); assert.equal(x.calls[1][1],17); assert.equal(x.host.canvasWriteRequested,true); +}); +test('failed or stale adoption permission never dispatches',async()=>{ + for (const permission of [{allowed:false,revision:17},{allowed:true,revision:18},null]) { + const x=setup(['complete']);x.host.canvas.read=async()=>({revision:17});x.host.canvas.requestWrite=async()=>permission; + await assert.rejects(executeWorkflow(x.host,{...args,intent:'adopt_candidate',candidateId:'chosen'})); + assert.equal(x.calls.length,0); assert.equal(x.host.canvasWritesStopped,true); + } + const x=setup([]);await assert.rejects(executeWorkflow(x.host,{...args,intent:'adopt_candidate',candidateId:'chosen'}));assert.equal(x.calls.length,0); +}); +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); +}); diff --git a/apps/detail-agent-pi/test/workflow-not-dispatched.test.mjs b/apps/detail-agent-pi/test/workflow-not-dispatched.test.mjs new file mode 100644 index 00000000..d5f7d673 --- /dev/null +++ b/apps/detail-agent-pi/test/workflow-not-dispatched.test.mjs @@ -0,0 +1,96 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { executeWorkflow } from '../src/workflow-execute.mjs'; + +const args = {version:1,intent:'generate_candidate',outputId:'detail',slotId:'hero',productVariantId:'walnut'}; +function fixture(change = value => value) { + const calls = [], events = []; + const identity = {scope:'scope',project:'project'}; + const host = {turnEpoch:1,currentWorkflow:{revisions:{canvas:17}}, + canvas:{run:'run',documentId:7,identity:{accountId:'account',origin:'https://platform.test'}, + async workflowExecute({operationId}) { + calls.push(operationId); + return change({operationId,phase:'not_dispatched',accepted:false,dispatched:false, + run:'run',documentId:7,baseRevision:17,accountId:'account',origin:'https://platform.test', + projectLocalId:'https://platform.test|account|project', + error:{code:'local_precondition_failed',message:'local precondition failed'}}); + }, + async workflowPoll() { assert.fail('a rejected operation must not be polled'); }}, + store:{identity,async appendEvent(event) { events.push(event); }}, + async readWorkflow() { assert.fail('a rejected operation must not refresh as completed'); }}; + return {host,calls,events}; +} + +test('an identity-bound local rejection stops the turn without an unknown image request',async()=>{ + const x = fixture(); + await assert.rejects(executeWorkflow(x.host,args),/未派发.*local precondition failed/); + assert.equal(x.calls.length,1); + assert.equal(x.host.nativeUnresolved,undefined); + assert.equal(x.host.nativeToolActive,false); + assert.equal(x.host.canvasWritesStopped,true); + assert.match(x.host.workflowStopReason,/未派发.*local precondition failed/); + assert.deepEqual(x.events.map(event=>event.type),['workflow_operation_not_dispatched']); + assert.equal(x.events[0].operationId,x.calls[0]); + await assert.rejects(executeWorkflow(x.host,args),/本轮不再派发/); + assert.equal(x.calls.length,1); +}); + +test('a malformed or foreign rejection cannot clear a possibly dispatched operation',async()=>{ + for (const [key,value] of Object.entries({operationId:'other',run:'other',documentId:8,baseRevision:-1, + accountId:'other',origin:'https://other.test',projectLocalId:'other',accepted:true,dispatched:true, + error:{code:'other',message:'not sent'}})) { + const x = fixture(receipt=>({...receipt,[key]:value})); + await assert.rejects(executeWorkflow(x.host,args)); + assert.equal(x.host.nativeUnresolved.operationId,x.calls[0],key); + assert.equal(x.host.canvasWritesStopped,true,key); + assert.equal(x.events.length,0,key); + } +}); + +test('a user edit after the workflow read leaves an authoritative rejection known not dispatched',async()=>{ + const x = fixture(receipt=>({...receipt,baseRevision:18})); + await assert.rejects(executeWorkflow(x.host,args),/未派发/); + assert.equal(x.host.nativeUnresolved,undefined); + assert.equal(x.calls.length,1); + assert.equal(x.host.canvasWritesStopped,true); + assert.equal(x.events[0].baseRevision,18); +}); + +test('a transport error keeps the original unknown identity even when its words say not sent',async()=>{ + const x = fixture(); + x.host.canvas.workflowExecute=async input=>{x.calls.push(input.operationId);throw Error('not_dispatched: 用户已接手,未执行操作');}; + await assert.rejects(executeWorkflow(x.host,args)); + assert.equal(x.host.nativeUnresolved.operationId,x.calls[0]); + assert.equal(x.host.nativeUnresolved.run,'run'); + assert.equal(x.events.length,0); +}); + +test('a rejection after waiting for a save is still known not dispatched',async()=>{ + const x = fixture(), reject = x.host.canvas.workflowExecute; + x.host.canvas.read=async()=>({revision:17,conversationActive:true,writeBlocked:false}); + x.host.canvas.workflowExecute=async input=>x.calls.length===0 + ? (x.calls.push(input.operationId),{operationId:input.operationId,phase:'waiting_save',dispatched:false,revision:17}) + : reject(input); + await assert.rejects(executeWorkflow(x.host,args,undefined,{pause:async()=>{}}),/未派发/); + assert.equal(new Set(x.calls).size,1); + assert.equal(x.calls.length,2); + assert.equal(x.host.nativeUnresolved,undefined); +}); + +test('a late authoritative rejection cannot stop or contaminate a new user turn',async()=>{ + const x = fixture(); + let release; + const rejected = x.host.canvas.workflowExecute; + x.host.canvas.workflowExecute=async input=>{ + const receipt=await rejected(input); + return new Promise(resolve=>{release=()=>resolve(receipt);}); + }; + const task=executeWorkflow(x.host,args); + await new Promise(resolve=>setImmediate(resolve)); + x.host.turnEpoch=2;x.host.canvas.run='new-run'; + release(); + await assert.rejects(task,/原回合已结束/); + assert.equal(x.host.nativeUnresolved,undefined); + assert.equal(x.host.canvasWritesStopped,undefined); + assert.equal(x.events.length,0); +}); diff --git a/apps/detail-agent-pi/test/workflow-save-wait.test.mjs b/apps/detail-agent-pi/test/workflow-save-wait.test.mjs new file mode 100644 index 00000000..715ff2ea --- /dev/null +++ b/apps/detail-agent-pi/test/workflow-save-wait.test.mjs @@ -0,0 +1,63 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { executeWorkflow } from '../src/workflow-execute.mjs'; + +const args = {version:1,intent:'generate_candidate',outputId:'detail',slotId:'hero',productVariantId:'walnut'}; +function setup() { + const calls = []; + const state = {revision:17,conversationActive:true,writeBlocked:false}; + let accepted = 0; + let attempts = 0; + const host = {turnEpoch:1,canvas:{run:'run',documentId:7, + async read() { return {...state}; }, + async workflowExecute(input) { + calls.push(input.operationId); + if (attempts++ < 2) return {operationId:input.operationId,phase:'waiting_save',dispatched:false,revision:17}; + accepted++; + return {operationId:input.operationId,phase:'submitted'}; + }, + async workflowPoll(operationId) { return {operationId,phase:'complete'}; } + },store:{async appendEvent() {}},async readWorkflow() { return {version:1}; }}; + return {host,calls,state,accepted:()=>accepted}; +} + +test('normal saves defer one candidate intent until its original save acknowledgement',async()=>{ + const x = setup(); + const result = await executeWorkflow(x.host,args,undefined,{pause:async()=>{}}); + assert.equal(result.operation.phase,'complete'); + assert.equal(new Set(x.calls).size,1); + assert.equal(x.calls.length,3); + assert.equal(x.accepted(),1); + assert.equal(x.host.nativeUnresolved,undefined); +}); + +test('handover, revision drift and save failure stop a waiting intent before dispatch',async()=>{ + for (const change of [state=>state.writeBlocked=true,state=>state.revision++,state=>state.error='本机保存失败']) { + const x = setup(); + await assert.rejects(executeWorkflow(x.host,args,undefined,{pause:async()=>change(x.state)})); + assert.equal(x.calls.length,1); + assert.equal(x.accepted(),0); + assert.equal(x.host.nativeUnresolved,undefined); + } +}); + +test('cancelling or replacing a turn while saving never creates a candidate',async()=>{ + for (const action of ['cancel','replace']) { + const x = setup(); + const controller = new AbortController(); + await assert.rejects(executeWorkflow(x.host,args,controller.signal,{pause:async()=>{ + if (action==='cancel') controller.abort(); else x.host.turnEpoch++; + }})); + assert.equal(x.calls.length,1); + assert.equal(x.accepted(),0); + assert.equal(x.host.nativeUnresolved,undefined); + } +}); + +test('a save wait timeout is known not dispatched and cannot become an unknown image request',async()=>{ + const x = setup(); + await assert.rejects(executeWorkflow(x.host,args,undefined,{timeoutMs:0}),/仍在保存/); + assert.equal(x.calls.length,1); + assert.equal(x.accepted(),0); + assert.equal(x.host.nativeUnresolved,undefined); +}); diff --git a/apps/detail-agent-pi/test/workflow.test.mjs b/apps/detail-agent-pi/test/workflow.test.mjs index e6d0b8ee..a4519b84 100644 --- a/apps/detail-agent-pi/test/workflow.test.mjs +++ b/apps/detail-agent-pi/test/workflow.test.mjs @@ -39,7 +39,7 @@ function baseCanvas() { return {activities:[],writes:0,grants:0,async begin(){},async read(){return {conversationActive:true,allowed:false,writeBlocked:false,revision:1,context:{id:7,name:'本机画布',width:1440,height:1800,layers:[]}};}, async requestWrite(){this.grants++;throw Error('not used');},async edit(){this.writes++;throw Error('not used');},async activity(text){this.activities.push(text);},async delta(){},async end(){}}; } -async function fixture(reply,{adapter='full',broker,history=false,nativeEvidence=true,nativeBrandId='native-brand',nativeCategoryId='native-category'}={}) { +async function fixture(reply,{adapter='full',execute=false,broker,history=false,nativeEvidence=true,nativeBrandId='native-brand',nativeCategoryId='native-category'}={}) { const root=await fs.mkdtemp(path.join(os.tmpdir(),'jiansun-pi-')),project=randomUUID(),identity={scope,project}; const store=await ProjectStore.open(root,identity,binding,{continuous:true}),canvas=baseCanvas(); if(history){ @@ -60,6 +60,7 @@ async function fixture(reply,{adapter='full',broker,history=false,nativeEvidence canvas.workflowNavigate=async args=>{native.navigations.push(args);native.stage=args.stage;return {available:true,version:native.version,stage:native.stage,outputId:args.outputId};}; canvas.workflowPropose=async args=>{native.proposals.push(args);if(args.version!==native.version)throw Error('工作流版本已变化,请重新读取;未添加建议');if(args.instruction?.trim()&&args.intent!=='selling_points')throw Error('补充要求仅用于卖点表达方向,其他提案使用当前配置');return {proposed:true,awaitingReview:true,executed:false,version:native.version,proposal:{...args,title:'拟定页面方案',awaitingReview:true,executed:false}};}; } + if(execute){canvas.workflowExecute=async args=>({operationId:args.operationId,phase:'submitted'});canvas.workflowPoll=async operationId=>({operationId,phase:'complete'});} const calls=[]; const transport={info,stream(_model,context){calls.push({context,injected:injection(context)});return stream(reply(context,calls.length,native,canvas));}}; const host=await DetailPiAgent.open({store,canvas,transport,broker,projectEvidence:nativeEvidence?evidence(project,'胡桃色',native.brandId,native.categoryId):undefined,documentId:7}); @@ -281,3 +282,28 @@ test('recovery can restore workflow_read but never replays navigation or review }finally{await host.dispose();await cleanup(x.root);} } }); + +test('SDK stops at a known local rejection without another model step or a reconciliation loop',async()=>{ + const x=await fixture(()=>tool('workflow_execute',{version:1,intent:'generate_candidate',outputId:'detail',slotId:'hero',productVariantId:'walnut'}),{execute:true}); + x.canvas.documentId=7; + x.canvas.begin=async()=>{x.canvas.run='rejected-run';}; + const read=x.canvas.workflowRead; + x.canvas.workflowRead=async input=>({...await read(input),revisions:{canvas:1}}); + let executions=0; + x.canvas.workflowExecute=async({operationId})=>{ + executions++; + return {operationId,phase:'not_dispatched',accepted:false,dispatched:false,run:'rejected-run',documentId:7, + baseRevision:1,accountId:info.accountId,origin:info.origin,projectLocalId:`${info.origin}|${info.accountId}|${x.project}`, + error:{code:'local_precondition_failed',message:'当前画布修订已变化'}}; + }; + x.canvas.workflowPoll=async()=>assert.fail('no registered operation to poll'); + try{ + const result=await x.host.prompt('生成第一张候选,不采纳'); + assert.match(result.text,/未派发.*当前画布修订已变化/); + assert.equal(executions,1); + assert.equal(x.calls.length,1); + assert.equal(x.host.nativeUnresolved,undefined); + assert.equal(x.canvas.writes,0); + assert.ok(x.host.session.messages.some(message=>message.role==='toolResult'&&message.toolName==='workflow_execute'&&message.isError)); + }finally{await x.close();} +}); diff --git a/apps/detail-photocraft/crates/ui-egui/src/jiansun/workflow_bridge.rs b/apps/detail-photocraft/crates/ui-egui/src/jiansun/workflow_bridge.rs index 58323858..702e5d4f 100644 --- a/apps/detail-photocraft/crates/ui-egui/src/jiansun/workflow_bridge.rs +++ b/apps/detail-photocraft/crates/ui-egui/src/jiansun/workflow_bridge.rs @@ -518,6 +518,76 @@ pub(super) fn proposal_current(w: &Workspace, app: &PhotocraftApp) -> bool { w.workflow.pending.as_ref().is_some_and(|pending| owned(w, app) && !w.pi_write_blocked && guard(w, app) == pending.guard) } +fn operation_id(input: &Value) -> Result<&str, String> { + input["operationId"].as_str().filter(|s| s.len() == 36 && s.bytes().enumerate().all(|(i,b)| if [8,13,18,23].contains(&i) { b == b'-' } else { b.is_ascii_hexdigit() })).ok_or("操作身份无效".into()) +} + +fn execute(w: &mut Workspace, app: &mut PhotocraftApp, input: &Value, version: u64, available: bool) -> Result { + arguments(input, &["version", "intent", "instruction", "outputId", "slotId", "productVariantId", "candidateId", "operationId"])?; + let id = operation_id(input)?; + let run = w.pi_run.clone().ok_or("本地Agent会话已结束")?; + if let Some(op) = w.workflow.operations.iter().find(|op| op["operationId"] == id) { + if op["run"] != run { return Err("操作身份不属于当前回合".into()); } + return Ok(json!({"operationId":id,"phase":op["phase"],"candidateId":op["candidateId"]})); + } + if w.workflow.operations.iter().any(|op| !matches!(op["phase"].as_str(), Some("complete" | "failed"))) { + return Err("原操作尚未落定,未执行新操作".into()); + } + let excess = w.workflow.operations.len().saturating_add(1).saturating_sub(64); + 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() + || 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()); + } + if w.template_workflow["canvasFirst"] != true || input["version"].as_u64() != Some(version) || w.workflow.last_read_version != Some(version) { + return Err("工作流已变化或不支持直接执行,请重新读取".into()); + } + let intent = input["intent"].as_str().filter(|s| matches!(*s, "generate_candidate" | "adopt_candidate")).ok_or("此操作不能直接执行")?; + if intent == "adopt_candidate" && !w.allow_apply { return Err("本轮原生操作尚未取得当前修订许可".into()); } + let instruction = input.get("instruction").map_or(Ok(""), |v| v.as_str().ok_or("调整要求须为文本"))?; + if instruction.chars().count() > 500 || instruction.chars().any(|ch| ch.is_control() && !matches!(ch, '\n' | '\r' | '\t')) || (intent != "generate_candidate" && !instruction.trim().is_empty()) { + return Err("调整要求无效".into()); + } + let operation_input = template_input(w, intent, input)?; + if w.saving.is_some() || app.saving() { + return Ok(json!({"operationId":id,"phase":"waiting_save","dispatched":false,"revision":w.expected_revision})); + } + if w.workflow.operations.iter().any(|op| op["run"] == run && op["intent"] == intent + && op["slotId"] == operation_input["slotId"] && op["productVariantId"] == operation_input["variantId"] + && op["candidateIdInput"] == input["candidateId"] && op["instruction"] == instruction) { + return Err("本轮相同操作已有原记录,不会重复执行".into()); + } + let method = if intent == "generate_candidate" { "slot-append" } else { "slot-adopt-prep" }; + let op = json!({"operationId":id,"run":run,"intent":intent,"phase":"submitted","accountId":w.account["id"],"origin":w.origin, + "documentId":app.session.active().map(|s|s.doc.id.0),"projectLocalId":w.pi_request_scope["project"]["localId"],"baseRevision":w.expected_revision, + "slotId":operation_input["slotId"],"productVariantId":operation_input["variantId"], + "candidateId":operation_input["candidateId"],"candidateIdInput":input["candidateId"],"instruction":instruction}); + for _ in 0..excess { + if let Some(index) = w.workflow.operations.iter().position(|old| matches!(old["phase"].as_str(), Some("complete" | "failed"))) { + w.workflow.operations.remove(index); + } + } + w.workflow.operations.push(op); + let submitted = match w.slot_action_inner(app, method, operation_input, true) { + Ok(result) => result, + Err(error) => { + if let Some(op) = w.workflow.operations.last_mut() { op["phase"] = json!("failed"); } + return Err(error); + } + }; + let delegated = submitted.get("delegatedInput").filter(|value| value.is_object()).ok_or("本机委托未创建")?; + Ok(json!({"operationId":id,"phase":"submitted","__workflowDelegation":{"method":method,"input":delegated, + "run":run,"operationId":id,"accountId":w.account["id"],"origin":w.origin,"projectLocalId":w.pi_request_scope["project"]["localId"]}})) +} + pub(super) fn action(w: &mut Workspace, app: &mut PhotocraftApp, method: &str, input: &Value) -> Result { active_run(w, app, input)?; let current_guard = guard(w, app); @@ -558,66 +628,15 @@ pub(super) fn action(w: &mut Workspace, app: &mut PhotocraftApp, method: &str, i Ok(json!({"operationId":id,"phase":op["phase"],"candidateId":op["candidateId"],"message":op["message"]})) } "pi-workflow-execute" => { - arguments(input, &["version", "intent", "instruction", "outputId", "slotId", "productVariantId", "candidateId", "operationId"])?; - let id = input["operationId"].as_str().filter(|s| s.len() == 36 && s.bytes().enumerate().all(|(i,b)| if [8,13,18,23].contains(&i) { b == b'-' } else { b.is_ascii_hexdigit() })).ok_or("操作身份无效")?; - let run = w.pi_run.clone().ok_or("本地Agent会话已结束")?; - if let Some(op) = w.workflow.operations.iter().find(|op| op["operationId"] == id) { - if op["run"] != run { return Err("操作身份不属于当前回合".into()); } - return Ok(json!({"operationId":id,"phase":op["phase"],"candidateId":op["candidateId"]})); + let result = execute(w, app, input, version, available); + if let (Err(message), Ok(id)) = (&result, operation_id(input)) + && available && !w.workflow.operations.iter().any(|op| op["operationId"] == id) { + return Ok(json!({"operationId":id,"phase":"not_dispatched","accepted":false,"dispatched":false, + "run":w.pi_run,"documentId":w.owner_doc,"baseRevision":current_guard["canvas"]["revision"],"accountId":w.account["id"], + "origin":w.origin,"projectLocalId":w.pi_request_scope["project"]["localId"], + "error":{"code":"local_precondition_failed","message":message}})); } - if w.workflow.operations.iter().any(|op| !matches!(op["phase"].as_str(), Some("complete" | "failed"))) { - return Err("原操作尚未落定,未执行新操作".into()); - } - let excess = w.workflow.operations.len().saturating_add(1).saturating_sub(64); - 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() || w.saving.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.saving() - || app.session.active().is_none_or(|active| active.revision != w.expected_revision) { - 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()); - } - let intent = input["intent"].as_str().filter(|s| matches!(*s, "generate_candidate" | "adopt_candidate")).ok_or("此操作不能直接执行")?; - if intent == "adopt_candidate" && !w.allow_apply { return Err("本轮原生操作尚未取得当前修订许可".into()); } - let instruction = input.get("instruction").map_or(Ok(""), |v| v.as_str().ok_or("调整要求须为文本"))?; - if instruction.chars().count() > 500 || instruction.chars().any(|ch| ch.is_control() && !matches!(ch, '\n' | '\r' | '\t')) || (intent != "generate_candidate" && !instruction.trim().is_empty()) { - return Err("调整要求无效".into()); - } - let operation_input = template_input(w, intent, input)?; - if w.workflow.operations.iter().any(|op| op["run"] == run && op["intent"] == intent - && op["slotId"] == operation_input["slotId"] && op["productVariantId"] == operation_input["variantId"] - && op["candidateIdInput"] == input["candidateId"] && op["instruction"] == instruction) { - return Err("本轮相同操作已有原记录,不会重复执行".into()); - } - let method = if intent == "generate_candidate" { "slot-append" } else { "slot-adopt-prep" }; - let op = json!({"operationId":id,"run":run,"intent":intent,"phase":"submitted","accountId":w.account["id"],"origin":w.origin, - "documentId":app.session.active().map(|s|s.doc.id.0),"projectLocalId":w.pi_request_scope["project"]["localId"],"baseRevision":w.expected_revision, - "slotId":operation_input["slotId"],"productVariantId":operation_input["variantId"], - "candidateId":operation_input["candidateId"],"candidateIdInput":input["candidateId"],"instruction":instruction}); - for _ in 0..excess { - if let Some(index) = w.workflow.operations.iter().position(|old| matches!(old["phase"].as_str(), Some("complete" | "failed"))) { - w.workflow.operations.remove(index); - } - } - w.workflow.operations.push(op); - let submitted = match w.slot_action_inner(app, method, operation_input, true) { - Ok(result) => result, - Err(error) => { - if let Some(op) = w.workflow.operations.last_mut() { op["phase"] = json!("failed"); } - return Err(error); - } - }; - let delegated = submitted.get("delegatedInput").filter(|value| value.is_object()).ok_or("本机委托未创建")?; - Ok(json!({"operationId":id,"phase":"submitted","__workflowDelegation":{"method":method,"input":delegated, - "run":run,"operationId":id,"accountId":w.account["id"],"origin":w.origin,"projectLocalId":w.pi_request_scope["project"]["localId"]}})) + result } "pi-workflow-propose" => { arguments(input, &["version", "intent", "instruction", "outputId", "slotId", "variant", "productVariantId", "candidateId"])?; @@ -1149,6 +1168,28 @@ mod tests { assert!(requests.try_recv().is_err()); } + #[test] + fn private_dispatch_waits_for_save_ack_before_one_delegation() { + let (mut w, mut app, requests, ctx) = direct_fixture(); + let read = private_call(&mut w, &mut app, &ctx, "pi-workflow-read", json!({})).unwrap(); + let input = direct_input(read["version"].clone()); + w.saving = Some((w.owner_doc.unwrap(), w.expected_revision)); + let waiting = private_call(&mut w, &mut app, &ctx, "pi-workflow-execute", input.clone()).unwrap(); + assert_eq!(waiting["phase"], "waiting_save"); + assert_eq!(waiting["dispatched"], false); + assert!(waiting["__workflowDelegation"].is_null()); + assert!(w.workflow.operations.is_empty()); + assert!(requests.try_recv().is_err()); + w.saving = None; + let submitted = private_call(&mut w, &mut app, &ctx, "pi-workflow-execute", input.clone()).unwrap(); + assert_eq!(submitted["phase"], "submitted"); + assert!(submitted["__workflowDelegation"].is_object()); + let repeated = private_call(&mut w, &mut app, &ctx, "pi-workflow-execute", input).unwrap(); + assert!(repeated["__workflowDelegation"].is_null()); + assert_eq!(w.workflow.operations.len(), 1); + assert!(requests.try_recv().is_err()); + } + #[test] fn private_dispatch_rejects_stale_handover_unknown_drift_and_ungranted_adoption() { for scenario in ["stale", "handover", "unknown", "drift", "adoption"] { @@ -1163,7 +1204,16 @@ mod tests { "adoption" => { input["intent"] = json!("adopt_candidate"); input["instruction"] = json!(""); w.allow_apply = false; }, _ => {}, } - assert!(private_call(&mut w, &mut app, &ctx, "pi-workflow-execute", input).is_err(), "{scenario}"); + let rejected = private_call(&mut w, &mut app, &ctx, "pi-workflow-execute", input).unwrap(); + assert_eq!(rejected["phase"], "not_dispatched", "{scenario}"); + assert_eq!(rejected["accepted"], false); + assert_eq!(rejected["dispatched"], false); + 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); + assert_eq!(rejected["accountId"], w.account["id"]); + assert_eq!(rejected["origin"], w.origin); + assert_eq!(rejected["projectLocalId"], w.pi_request_scope["project"]["localId"]); assert!(requests.try_recv().is_err(), "{scenario}: no request must be sent"); assert!(w.workflow.operations.is_empty(), "{scenario}: no operation must be recorded"); } @@ -1190,7 +1240,8 @@ mod tests { } let original = w.workflow.operations.clone(); let read = private_call(&mut w, &mut app, &ctx, "pi-workflow-read", json!({})).unwrap(); - assert!(private_call(&mut w, &mut app, &ctx, "pi-workflow-execute", direct_input(read["version"].clone())).is_err()); + let rejected = private_call(&mut w, &mut app, &ctx, "pi-workflow-execute", direct_input(read["version"].clone())).unwrap(); + assert_eq!(rejected["phase"], "not_dispatched"); assert_eq!(w.workflow.operations, original); assert!(requests.try_recv().is_err()); }