fix(dynamic-workflow): deterministic wait polling and checkpoint lineage in repair reuse

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

@@ -133,3 +133,16 @@ test('the false JSON Schema cannot silently accept a node output',async()=>{
  assert.equal(run.steps[0].errorDetails.code,'OUTPUT_SCHEMA_INVALID');assert.deepEqual(run.steps[0].rawOutput,{unexpected:true});
  }finally{await f.cleanup();}
 });
+
+test('checkpoint dependencies with unchanged values keep candidates reusable; changed values invalidate downstream',async()=>{
+ const calls=[],f=await fixture(async s=>{calls.push(s.id);return {output:s.id};});try{
+ const sourceScript=`await ctx.checkpoint('seed','v1');const a=await ctx.agent({id:'a',prompt:'a',dependsOn:'checkpoint:seed'});throw Error('bad synthesis');`;
+ const source=await start(f.engine,sourceScript);
+ const draft=await f.engine.repair(source.id,request(source,{script:sourceScript.replace("throw Error('bad synthesis');",'return a.output;'),reuseStepIds:['a']}));
+ await f.engine.approve(draft.id,{revision:1});const end=await finish(f.engine,draft.id);
+ assert.equal(end.status,'succeeded');assert.equal(end.attempts,0);assert.equal(end.steps.find(s=>s.id==='a').reusedFrom.runId,source.id);
+ 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']}));
+ await f.engine.approve(changed.id,{revision:1});const done=await finish(f.engine,changed.id);
+ 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']);
+ }finally{await f.cleanup();}
+});

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

@@ -46,7 +46,7 @@ test('packaged MCP launched in plugin root routes concurrent projects and execut
   for(let i=0;i<2;i++){
    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);
    let status;
-   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;}
+   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;}
    assert.equal(status.status,'succeeded');
    const results=await value(a,'workflow_results',{workspace:projects[i],runId:drafts[i].id});
    assert.deepEqual(results.steps.find(s=>s.id==='cwd').output,{cwd:projects[i],arg:projects[i]});

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

@@ -14248,11 +14248,6 @@ var Engine = class extends EventEmitter {
     }
     const fingerprints = {};
     const topology = assertValidDependencies(previewTopology(request.script, request.input ?? {}));
-    const duplicate = this.store.byRequest(request.requestId);
-    if (duplicate) {
-      check(duplicate.requestHash === requestHash, "requestId \u53C2\u6570\u51B2\u7A81");
-      return this.snapshot(duplicate.id);
-    }
     check(!this.closing, "\u670D\u52A1\u6B63\u5728\u5173\u95ED");
     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 };
     this.store.transaction(() => {
@@ -14581,7 +14576,10 @@ var Engine = class extends EventEmitter {
       return cached2.promise;
     }
     const repair = ctx.run.repair, candidate = !previous && repair?.reuseStepIds.includes(spec.id) ? this.store.repairCandidate(ctx.run.id, spec.id) : null;
-    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)) {
+    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) => {
+      const dep = this.store.step(ctx.run.id, id2);
+      return dep?.reusedFrom?.runId === repair.sourceRunId || dep?.kind === "checkpoint" && dep.requestHash === this.store.step(repair.sourceRunId, id2)?.requestHash;
+    })) {
       let valid = true;
       try {
         if (validateOutput) valid = validateOutput(candidate.output);

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

@@ -47,7 +47,6 @@ export class Engine extends EventEmitter {
    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);
    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);}
    const fingerprints={};const topology=assertValidDependencies(previewTopology(request.script,request.input??{}));
-   const duplicate=this.store.byRequest(request.requestId);if(duplicate){check(duplicate.requestHash===requestHash,'requestId 参数冲突');return this.snapshot(duplicate.id);}
    check(!this.closing,'服务正在关闭');
    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};
    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 {
    const repair=ctx.run.repair,candidate=!previous&&repair?.reuseStepIds.includes(spec.id)?this.store.repairCandidate(ctx.run.id,spec.id):null;
    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(id=>this.store.step(ctx.run.id,id)?.reusedFrom?.runId===repair.sourceRunId)){
+      &&deps.every(id=>{const dep=this.store.step(ctx.run.id,id);return dep?.reusedFrom?.runId===repair.sourceRunId
+       // Checkpoints recompute every run by design; reuse stays valid while the
+       // recomputed value matches the source run, breaking lineage if it changed.
+       ||(dep?.kind==='checkpoint'&&dep.requestHash===this.store.step(repair.sourceRunId,id)?.requestHash);})){
      // Recheck the output against today's validator, including legacy candidates.
      let valid=true;try{if(validateOutput)valid=validateOutput(candidate.output);}catch{valid=false;}
      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,