fix(client): wait for project saves and classify undispatched operations
This commit is contained in:
parent
f893869f69
commit
fbf9ff20c8
@ -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 });
|
||||
|
||||
68
apps/detail-agent-pi/test/workflow-execute.test.mjs
Normal file
68
apps/detail-agent-pi/test/workflow-execute.test.mjs
Normal file
@ -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);
|
||||
});
|
||||
96
apps/detail-agent-pi/test/workflow-not-dispatched.test.mjs
Normal file
96
apps/detail-agent-pi/test/workflow-not-dispatched.test.mjs
Normal file
@ -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);
|
||||
});
|
||||
63
apps/detail-agent-pi/test/workflow-save-wait.test.mjs
Normal file
63
apps/detail-agent-pi/test/workflow-save-wait.test.mjs
Normal file
@ -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);
|
||||
});
|
||||
@ -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();}
|
||||
});
|
||||
|
||||
@ -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<Value, String> {
|
||||
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<Value, String> {
|
||||
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());
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user