Skip to content

Commit f5e5664

Browse files
modackermoc
andauthored
fix: deterministic wait polling and checkpoint lineage in repair reuse (#44)
- workspace-router check advances the workflow_wait event cursor so the polling loop blocks on new events instead of racing run completion - repair reuse accepts checkpoint dependencies while the recomputed value hash matches the source run, and still invalidates on value changes - drop the unreachable second requestId lookup in start() - dist/main.mjs rebuilt from the changed engine (byte-reproducible build) Co-authored-by: moc <moc@sgmov.com>
1 parent 00878c7 commit f5e5664

4 files changed

Lines changed: 22 additions & 9 deletions

File tree

‎plugins/hetaoBackend/mcode-dynamic-workflows/checks/repair.check.mjs‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -133,3 +133,16 @@ test('the false JSON Schema cannot silently accept a node output',async()=>{
133133
assert.equal(run.steps[0].errorDetails.code,'OUTPUT_SCHEMA_INVALID');assert.deepEqual(run.steps[0].rawOutput,{unexpected:true});
134134
}finally{await f.cleanup();}
135135
});
136+
137+
test('checkpoint dependencies with unchanged values keep candidates reusable; changed values invalidate downstream',async()=>{
138+
const calls=[],f=await fixture(async s=>{calls.push(s.id);return {output:s.id};});try{
139+
const sourceScript=`await ctx.checkpoint('seed','v1');const a=await ctx.agent({id:'a',prompt:'a',dependsOn:'checkpoint:seed'});throw Error('bad synthesis');`;
140+
const source=await start(f.engine,sourceScript);
141+
const draft=await f.engine.repair(source.id,request(source,{script:sourceScript.replace("throw Error('bad synthesis');",'return a.output;'),reuseStepIds:['a']}));
142+
await f.engine.approve(draft.id,{revision:1});const end=await finish(f.engine,draft.id);
143+
assert.equal(end.status,'succeeded');assert.equal(end.attempts,0);assert.equal(end.steps.find(s=>s.id==='a').reusedFrom.runId,source.id);
144+
const changed=await f.engine.repair(source.id,request(source,{script:sourceScript.replace("'v1'","'v2'").replace("throw Error('bad synthesis');",'return a.output;'),reuseStepIds:['a']}));
145+
await f.engine.approve(changed.id,{revision:1});const done=await finish(f.engine,changed.id);
146+
assert.equal(done.status,'succeeded');assert.equal(done.attempts,1);assert.ok(!done.steps.find(s=>s.id==='a').reusedFrom);assert.deepEqual(calls,['a','a']);
147+
}finally{await f.cleanup();}
148+
});

‎plugins/hetaoBackend/mcode-dynamic-workflows/checks/workspace-router.check.mjs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ test('packaged MCP launched in plugin root routes concurrent projects and execut
4646
for(let i=0;i<2;i++){
4747
assert.equal((await fetch(new URL(`/api/runs/${drafts[i].id}/approve`,dashboards[i].url),{method:'POST',headers,body:JSON.stringify({revision:1})})).status,200);
4848
let status;
49-
for(let n=0;n<30;n++){status=await value(a,'workflow_wait',{workspace:projects[i],runId:drafts[i].id,timeoutMs:500});if(status.status==='succeeded')break;}
49+
for(let n=0;n<30;n++){status=await value(a,'workflow_wait',{workspace:projects[i],runId:drafts[i].id,timeoutMs:500,afterSequence:status?.nextSequence??0});if(status.status==='succeeded')break;}
5050
assert.equal(status.status,'succeeded');
5151
const results=await value(a,'workflow_results',{workspace:projects[i],runId:drafts[i].id});
5252
assert.deepEqual(results.steps.find(s=>s.id==='cwd').output,{cwd:projects[i],arg:projects[i]});

‎plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs‎

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -14248,11 +14248,6 @@ var Engine = class extends EventEmitter {
1424814248
}
1424914249
const fingerprints = {};
1425014250
const topology = assertValidDependencies(previewTopology(request.script, request.input ?? {}));
14251-
const duplicate = this.store.byRequest(request.requestId);
14252-
if (duplicate) {
14253-
check(duplicate.requestHash === requestHash, "requestId \u53C2\u6570\u51B2\u7A81");
14254-
return this.snapshot(duplicate.id);
14255-
}
1425614251
check(!this.closing, "\u670D\u52A1\u6B63\u5728\u5173\u95ED");
1425714252
const run = { ...repair ? { repair } : {}, id: randomUUID2(), requestId: request.requestId, requestHash, ...definition, scriptHash: hash(request.script), fingerprints, workspace: this.options.workspace, revision: 1, topology, status: "pending_review", createdAt: Date.now(), updatedAt: Date.now(), attempts: 0, phases: [], result: null, error: null };
1425814253
this.store.transaction(() => {
@@ -14581,7 +14576,10 @@ var Engine = class extends EventEmitter {
1458114576
return cached2.promise;
1458214577
}
1458314578
const repair = ctx.run.repair, candidate = !previous && repair?.reuseStepIds.includes(spec.id) ? this.store.repairCandidate(ctx.run.id, spec.id) : null;
14584-
if (candidate && candidate.requestHash === requestHash && repair.contextHash === hash({ workspace: ctx.run.workspace, input: ctx.run.input, executor: ctx.run.executor, fingerprints: ctx.run.fingerprints }) && deps.every((id2) => this.store.step(ctx.run.id, id2)?.reusedFrom?.runId === repair.sourceRunId)) {
14579+
if (candidate && candidate.requestHash === requestHash && repair.contextHash === hash({ workspace: ctx.run.workspace, input: ctx.run.input, executor: ctx.run.executor, fingerprints: ctx.run.fingerprints }) && deps.every((id2) => {
14580+
const dep = this.store.step(ctx.run.id, id2);
14581+
return dep?.reusedFrom?.runId === repair.sourceRunId || dep?.kind === "checkpoint" && dep.requestHash === this.store.step(repair.sourceRunId, id2)?.requestHash;
14582+
})) {
1458514583
let valid = true;
1458614584
try {
1458714585
if (validateOutput) valid = validateOutput(candidate.output);

‎plugins/hetaoBackend/mcode-dynamic-workflows/src/engine.mjs‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,6 @@ export class Engine extends EventEmitter {
4747
const definition={...limits,...metadata,name:request.name,script:request.script,input:request.input??{},executor:request.executor,concurrency,maxCalls};const requestHash=hash(repair?{...definition,repair}:definition);
4848
const existing=this.store.byRequest(request.requestId);if(existing){const legacyDefinition={...definition};for(const key of Object.keys(DEFAULT_LIMITS))delete legacyDefinition[key];check(existing.requestHash===requestHash||(existing.maxSteps===undefined&&Object.keys(DEFAULT_LIMITS).every(k=>request[k]===undefined)&&existing.requestHash===hash(legacyDefinition)),'requestId 已用于不同参数');return this.snapshot(existing.id);}
4949
const fingerprints={};const topology=assertValidDependencies(previewTopology(request.script,request.input??{}));
50-
const duplicate=this.store.byRequest(request.requestId);if(duplicate){check(duplicate.requestHash===requestHash,'requestId 参数冲突');return this.snapshot(duplicate.id);}
5150
check(!this.closing,'服务正在关闭');
5251
const run={...(repair?{repair}:{}),id:randomUUID(),requestId:request.requestId,requestHash,...definition,scriptHash:hash(request.script),fingerprints,workspace:this.options.workspace,revision:1,topology,status:'pending_review',createdAt:Date.now(),updatedAt:Date.now(),attempts:0,phases:[],result:null,error:null};
5352
this.store.transaction(()=>{this.store.save(run);for(const step of candidates)this.store.saveRepairCandidate(run.id,step);this.store.event(run.id,'run.created',{name:run.name});});return this.snapshot(run.id);
@@ -186,7 +185,10 @@ export class Engine extends EventEmitter {
186185
const repair=ctx.run.repair,candidate=!previous&&repair?.reuseStepIds.includes(spec.id)?this.store.repairCandidate(ctx.run.id,spec.id):null;
187186
if(candidate&&candidate.requestHash===requestHash
188187
&&repair.contextHash===hash({workspace:ctx.run.workspace,input:ctx.run.input,executor:ctx.run.executor,fingerprints:ctx.run.fingerprints})
189-
&&deps.every(id=>this.store.step(ctx.run.id,id)?.reusedFrom?.runId===repair.sourceRunId)){
188+
&&deps.every(id=>{const dep=this.store.step(ctx.run.id,id);return dep?.reusedFrom?.runId===repair.sourceRunId
189+
// Checkpoints recompute every run by design; reuse stays valid while the
190+
// recomputed value matches the source run, breaking lineage if it changed.
191+
||(dep?.kind==='checkpoint'&&dep.requestHash===this.store.step(repair.sourceRunId,id)?.requestHash);})){
190192
// Recheck the output against today's validator, including legacy candidates.
191193
let valid=true;try{if(validateOutput)valid=validateOutput(candidate.output);}catch{valid=false;}
192194
if(valid){const step={...candidate,...(typeof planId==='string'?{planId}:{}),attempt:0,createdAt:Date.now(),startedAt:null,endedAt:Date.now(),usage:null,usageHistory:[],sessionId:undefined,turnId:undefined,

0 commit comments

Comments
 (0)