diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/README.md b/plugins/hetaoBackend/mcode-dynamic-workflows/README.md index b49ae56..495e8f5 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/README.md +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/README.md @@ -45,6 +45,12 @@ Tracked files are regular workspace files (up to 1 MB each), fingerprinted from The repaired script runs from its beginning; checkpoints are recomputed and unreached branches are not premarked complete. Declare every data/control dependency in `dependsOn`. Untracked files, external evidence and side effects cannot be checked automatically, so stale or incorrect results must not be selected for reuse. Schema-constrained outputs accept native values, complete JSON text, or one complete JSON fence; validation errors preserve the raw output. Each node has an independent schema namespace, including local references, so repeated schema identifiers cannot conflict across nodes or runs. +## Cross-run reuse + +Stored results are normally reused only within a run (resume) or through explicit repair selection. Passing `reuseAcrossRuns: true` to `workflow_start`, or setting it while a draft is pending review, additionally lets a node adopt a stored result from an earlier run of the same project. Adoption requires the node's full spec to hash identically (prompt, model, effort, input, schema, dependencies) and the run context to match on all four keys — workspace, input, executor and tracked-file fingerprints. The node's upstream lineage must match too: every step carries a lineage hash over its own spec and the lineage hashes of its declared dependencies in order, so a changed upstream (for example an edited upstream prompt) invalidates downstream candidates even when their own specs are unchanged. mcode nodes must declare an explicit model to be cross-run candidates — the default model comes from the CLI environment and is not part of the match key. Only runs executed with this feature stamp the context and lineage hashes on their steps, so legacy runs are not candidates. Adopted outputs are re-validated against the node's current schema; the step records provenance in `reusedFrom` — the immediate source run and step, `crossRun: true` — and in `originalProducer`, the run and step that originally produced the output, so chained adoptions stay traceable to their origin; it emits a `step.reused` event, and does not consume the workflow's agent-call budget. + +Reuse proves only that the context was identical and that the stored result was carried over faithfully — it does not prove the original run's output was semantically correct. For critical nodes, prefer schemas with evidence fields or place an independent verification node downstream, and keep `reuseAcrossRuns` off when in doubt. + ## Data, permissions and network - Every project-scoped tool requires an absolute `workspace`; plugin process cwd is never treated as your project. Canonical project paths isolate runs, templates, history, limits and ports. Never use an unrelated project's path. diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/cross-reuse-mcp.check.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/cross-reuse-mcp.check.mjs new file mode 100644 index 0000000..a036360 --- /dev/null +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/cross-reuse-mcp.check.mjs @@ -0,0 +1,14 @@ +import test from 'node:test';import assert from 'node:assert/strict';import {Client} from '@modelcontextprotocol/sdk/client/index.js';import {StdioClientTransport} from '@modelcontextprotocol/sdk/client/stdio.js';import {mkdtemp,rm,writeFile} from 'node:fs/promises';import {tmpdir} from 'node:os';import {join,resolve} from 'node:path'; +test('packaged MCP advertises reuseAcrossRuns and accepts it through the public tool surface',async()=>{ + const dir=await mkdtemp(join(tmpdir(),'wf-cross-mcp-'));await writeFile(join(dir,'settings.json'),JSON.stringify({workspace:dir,dataDir:dir})); + const client=new Client({name:'cross-reuse-mcp-test',version:'1'});const transport=new StdioClientTransport({command:process.execPath,args:[resolve('dist/main.mjs'),'--stdio','--settings',join(dir,'settings.json')],stderr:'pipe'}); + try{ + await client.connect(transport);const {tools}=await client.listTools(); + for(const name of ['workflow_start','workflow_update']){const tool=tools.find(t=>t.name===name);assert.ok(tool,`${name} listed`);assert.equal(tool.inputSchema.properties.reuseAcrossRuns?.type,'boolean',`${name} schema must advertise reuseAcrossRuns (additionalProperties:false)`);} + const started=await client.callTool({name:'workflow_start',arguments:{requestId:'cross-mcp',name:'Cross-run via public surface',executor:'demo',reuseAcrossRuns:true,script:'return await ctx.agent({id:"a",prompt:"p"});'}}); + assert.ok(!started.isError,started.content?.[0]?.text);const run=JSON.parse(started.content[0].text); + assert.equal(run.reuseAcrossRuns,true,'the flag must survive the public tool surface'); + const rejected=await client.callTool({name:'workflow_start',arguments:{requestId:'cross-mcp-bad',name:'Bad flag type',executor:'demo',reuseAcrossRuns:'yes',script:'return 1;'}}); + assert.ok(rejected.isError,'a non-boolean flag must be rejected by the public surface'); + }finally{await client.close();await transport.close();await rm(dir,{recursive:true,force:true});} +}); diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/cross-reuse.check.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/cross-reuse.check.mjs new file mode 100644 index 0000000..f5c7bec --- /dev/null +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/cross-reuse.check.mjs @@ -0,0 +1,218 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {mkdtemp,rm,writeFile} from 'node:fs/promises'; +import {tmpdir} from 'node:os'; +import {join} from 'node:path'; +import {setTimeout as delay} from 'node:timers/promises'; +import {Store} from '../src/store.mjs'; +import {Engine} from '../src/engine.mjs'; +import {TOOLS} from '../src/tools.mjs'; +// SDD contract suite for run-level reuseAcrossRuns (cross-run reuse of succeeded +// agent nodes). The engine/store behavior specified here may not exist yet; this +// file is the contract the implementation must satisfy. +async function fixture(execute){const dir=await mkdtemp(join(tmpdir(),'wf-xreuse-'));const store=new Store(dir),engine=new Engine(store,{workspace:dir,execute});return {dir,store,engine,cleanup:async()=>{await engine.close();store.close();await rm(dir,{recursive:true,force:true});}};} +async function finish(engine,id){for(let i=0;i<300;i++){if(!engine.active.has(id))return engine.snapshot(id);await delay(20);}throw Error('timeout');} +async function run(engine,script,input={},opts={}){const r=await engine.start({requestId:crypto.randomUUID(),name:'Cross reuse',executor:'demo',script,input,...opts});await engine.approve(r.id,{revision:1});return finish(engine,r.id);} +const probe=`return await ctx.agent({id:'a',prompt:'a'});`; +const chain=`const a=await ctx.agent({id:'a',prompt:'a'});const b=await ctx.agent({id:'b',prompt:'b',dependsOn:['a']});`; +const broken=chain+`throw Error('bad synthesis');`; +const repaired=chain+`return {a:a.output,b:b.output};`; +test('cross-run reuse is opt-in; without the flag a second identical run makes fresh calls and reuses nothing',async()=>{ + const calls=[],f=await fixture(async s=>{calls.push(s.id);return {output:s.id};});try{ + const first=await run(f.engine,probe),second=await run(f.engine,probe); + assert.equal(first.status,'succeeded');assert.equal(second.status,'succeeded'); + assert.deepEqual(calls,['a','a']);assert.ok(second.steps.every(s=>!s.reusedFrom)); + }finally{await f.cleanup();} +}); +test('an opted-in run adopts the newest succeeded node from an identical context without a new model call',async()=>{ + const calls=[],f=await fixture(async s=>{calls.push(s.id);return {output:s.id};});try{ + const source=await run(f.engine,probe,{},{reuseAcrossRuns:true}); + const end=await run(f.engine,probe,{},{reuseAcrossRuns:true}); + const step=end.steps.find(s=>s.id==='a'); + assert.equal(end.status,'succeeded');assert.equal(end.attempts,0);assert.deepEqual(calls,['a']); + assert.equal(step.attempt,0);assert.equal(step.usage,null);assert.deepEqual(step.usageHistory,[]); + assert.equal(step.reusedFrom.runId,source.id);assert.equal(step.reusedFrom.stepId,'a');assert.equal(step.reusedFrom.crossRun,true);assert.equal(typeof step.reusedFrom.endedAt,'number'); + assert.ok(f.store.events(end.id).some(e=>e.type==='step.reused'&&e.stepId==='a')); + }finally{await f.cleanup();} +}); +test('a changed input hashes to a different context so the opted-in run calls again',async()=>{ + const calls=[],f=await fixture(async s=>{calls.push(s.id);return {output:s.id};});try{ + await run(f.engine,probe,{},{reuseAcrossRuns:true}); + const end=await run(f.engine,probe,{tenant:'other'},{reuseAcrossRuns:true}); + assert.equal(end.status,'succeeded');assert.deepEqual(calls,['a','a']);assert.ok(end.steps.every(s=>!s.reusedFrom)); + }finally{await f.cleanup();} +}); +test('a changed executor hashes to a different context so the opted-in run cannot reuse; the fresh call is observable',async()=>{ + const calls=[],f=await fixture(async s=>{calls.push(s.id);return {output:s.id};});try{ + await run(f.engine,probe,{},{reuseAcrossRuns:true,executor:'demo'}); + const end=await run(f.engine,probe,{},{reuseAcrossRuns:true,executor:'mcode'}); + assert.equal(end.status,'succeeded');assert.deepEqual(calls,['a','a']);assert.ok(end.steps.every(s=>!s.reusedFrom)); + }finally{await f.cleanup();} +}); +test('a changed prompt misses cross-run reuse even in an otherwise identical context',async()=>{ + const calls=[],f=await fixture(async s=>{calls.push(s.id);return {output:s.id};});try{ + await run(f.engine,probe,{},{reuseAcrossRuns:true}); + const end=await run(f.engine,probe.replace("prompt:'a'","prompt:'changed'"),{},{reuseAcrossRuns:true}); + assert.equal(end.status,'succeeded');assert.deepEqual(calls,['a','a']);assert.ok(end.steps.every(s=>!s.reusedFrom)); + }finally{await f.cleanup();} +}); +test('adding an output schema changes the node request hash so cross-run reuse misses',async()=>{ + const calls=[],f=await fixture(async s=>{calls.push(s.id);return {output:s.id};});try{ + await run(f.engine,probe,{},{reuseAcrossRuns:true}); + const end=await run(f.engine,`return await ctx.agent({id:'a',prompt:'a',schema:{type:'string'}});`,{},{reuseAcrossRuns:true}); + assert.equal(end.status,'succeeded');assert.deepEqual(calls,['a','a']);assert.ok(end.steps.every(s=>!s.reusedFrom)); + }finally{await f.cleanup();} +}); +test('only succeeded nodes are cross-run candidates; a failed source node is called again',async()=>{ + let fail=true;const calls=[],f=await fixture(async s=>{calls.push(s.id);if(s.id==='b'&&fail)throw Error('provider failure');return {output:s.id};});try{ + const source=await run(f.engine,broken,{},{reuseAcrossRuns:true}); + assert.equal(source.status,'failed');fail=false; + const end=await run(f.engine,repaired,{},{reuseAcrossRuns:true}); + assert.equal(end.status,'succeeded');assert.deepEqual(end.result,{a:'a',b:'b'});assert.deepEqual(calls,['a','b','b']); + assert.equal(end.steps.find(s=>s.id==='a').reusedFrom.runId,source.id);assert.equal(end.steps.find(s=>s.id==='b').reusedFrom,undefined);assert.equal(end.attempts,1); + }finally{await f.cleanup();} +}); +test('with several qualifying runs the newest succeeded run wins',async()=>{ + const calls=[],f=await fixture(async s=>{calls.push(s.id);return {output:s.id};});try{ + const ids=[];for(let i=0;i<3;i++){const r=await run(f.engine,probe);assert.ok(r.steps.every(s=>!s.reusedFrom));ids.push(r.id);} + assert.deepEqual(calls,['a','a','a']); + const end=await run(f.engine,probe,{},{reuseAcrossRuns:true}); + const step=end.steps.find(s=>s.id==='a'); + assert.equal(end.status,'succeeded');assert.equal(end.attempts,0);assert.deepEqual(calls,['a','a','a']); + assert.equal(step.reusedFrom.runId,ids[2]);assert.equal(step.reusedFrom.crossRun,true); + }finally{await f.cleanup();} +}); +test('pure cross-run reuse succeeds on a one-call budget without consuming it',async()=>{ + const calls=[],f=await fixture(async s=>{calls.push(s.id);return {output:s.id};});try{ + await run(f.engine,probe,{},{reuseAcrossRuns:true}); + const end=await run(f.engine,probe,{},{reuseAcrossRuns:true,maxCalls:1}); + assert.equal(end.status,'succeeded');assert.equal(end.attempts,0);assert.deepEqual(calls,['a']); + }finally{await f.cleanup();} +}); +test('tracked-file changes between approvals change fingerprints and the context hash so reuse misses',async()=>{ + const calls=[],f=await fixture(async s=>{calls.push(s.id);return {output:s.id};});try{ + await writeFile(join(f.dir,'evidence.txt'),'X'); + await run(f.engine,probe,{files:['evidence.txt']},{reuseAcrossRuns:true}); + await writeFile(join(f.dir,'evidence.txt'),'Y'); + const end=await run(f.engine,probe,{files:['evidence.txt']},{reuseAcrossRuns:true}); + assert.equal(end.status,'succeeded');assert.deepEqual(calls,['a','a']);assert.ok(end.steps.every(s=>!s.reusedFrom)); + }finally{await f.cleanup();} +}); +test('reusing a requestId with a flipped reuseAcrossRuns value is rejected as a parameter conflict',async()=>{ + const f=await fixture(async s=>({output:s.id}));try{ + const requestId=crypto.randomUUID(); + await f.engine.start({requestId,name:'Cross reuse',executor:'demo',script:probe,input:{}}); + await assert.rejects(f.engine.start({requestId,name:'Cross reuse',executor:'demo',script:probe,input:{},reuseAcrossRuns:true}),/requestId 已用于不同参数/); + }finally{await f.cleanup();} +}); +test('a changed upstream prompt invalidates the downstream candidate even though its own spec is unchanged',async()=>{ + const calls=[],f=await fixture(async s=>{calls.push(s.id);return {output:s.prompt};});try{ + const first=await run(f.engine,repaired,{},{reuseAcrossRuns:true}); + assert.equal(first.status,'succeeded');assert.deepEqual(first.result,{a:'a',b:'b'});assert.deepEqual(calls,['a','b']); + const second=await run(f.engine,repaired.replace("prompt:'a'","prompt:'new'"),{},{reuseAcrossRuns:true}); + assert.equal(second.status,'succeeded'); + assert.deepEqual(calls,['a','b','a','b']); + assert.deepEqual(second.result,{a:'new',b:'b'}); + assert.ok(second.steps.every(s=>!s.reusedFrom)); + }finally{await f.cleanup();} +}); +test('mcode nodes without an explicit model are never cross-run candidates; with an explicit model they are',async()=>{ + const calls=[],f=await fixture(async s=>{calls.push(s.id);return {output:s.model??'default'};});try{ + const opts={reuseAcrossRuns:true,executor:'mcode'}; + const withModel=`return await ctx.agent({id:'a',prompt:'a',model:'m2'});`; + const r1=await run(f.engine,withModel,{},opts),r2=await run(f.engine,withModel,{},opts); + assert.equal(r1.status,'succeeded');assert.equal(r2.status,'succeeded'); + assert.deepEqual(calls,['a']);assert.equal(r2.attempts,0); + assert.equal(r2.steps.find(s=>s.id==='a').reusedFrom.crossRun,true); + const r3=await run(f.engine,probe,{},opts),r4=await run(f.engine,probe,{},opts); + assert.equal(r3.status,'succeeded');assert.equal(r4.status,'succeeded'); + assert.equal(r3.attempts,1);assert.equal(r4.attempts,1);assert.deepEqual(calls,['a','a','a']); + assert.ok([...r3.steps,...r4.steps].every(s=>!s.reusedFrom)); + }finally{await f.cleanup();} +}); +test('workflow_start and workflow_update expose reuseAcrossRuns as a boolean parameter',()=>{ + for(const name of ['workflow_start','workflow_update']){ + const tool=TOOLS.find(t=>t.name===name); + assert.ok(tool,`${name} missing from TOOLS`); + assert.equal(tool.inputSchema.additionalProperties,false); + assert.equal(tool.inputSchema.properties.reuseAcrossRuns.type,'boolean'); + assert.equal(typeof tool.inputSchema.properties.reuseAcrossRuns.description,'string'); + } +}); +test('chained adoption keeps reusedFrom on the immediate source and originalProducer on the first producer',async()=>{ + const calls=[],f=await fixture(async s=>{calls.push(s.id);return {output:s.id};});try{ + const run1=await run(f.engine,repaired,{},{reuseAcrossRuns:true}); + const run2=await run(f.engine,repaired,{},{reuseAcrossRuns:true}); + const run3=await run(f.engine,repaired,{},{reuseAcrossRuns:true}); + assert.deepEqual(calls,['a','b']);assert.equal(run3.attempts,0); + const b1=run1.steps.find(s=>s.id==='b'),b2=run2.steps.find(s=>s.id==='b'),b3=run3.steps.find(s=>s.id==='b'); + assert.equal(b2.reusedFrom.runId,run1.id);assert.equal(b2.reusedFrom.crossRun,true); + assert.equal(b2.originalProducer.runId,run1.id);assert.equal(b2.originalProducer.stepId,'b'); + assert.equal(b3.reusedFrom.runId,run2.id);assert.equal(b3.reusedFrom.stepId,'b');assert.equal(b3.reusedFrom.crossRun,true); + assert.equal(b3.originalProducer.runId,run1.id);assert.equal(b3.originalProducer.stepId,'b'); + assert.equal(b3.usage,null);assert.deepEqual(b3.usageHistory,[]); + }finally{await f.cleanup();} +}); +test('an upstream that re-executes with a different output invalidates downstream adoption',async()=>{ + // Maintainer's divergence shape: the upstream is not a cross-run candidate itself + // (mcode node without an explicit model re-executes every run) and its output + // differs between runs; the eligible downstream must not adopt the stale result. + let aOutput='old';const calls=[],f=await fixture(async s=>{calls.push(s.id);return {output:s.id==='a'?aOutput:s.id};});try{ + const script=`const a=await ctx.agent({id:'a',prompt:'write'});const b=await ctx.agent({id:'b',prompt:'read',model:'m2',dependsOn:['a'],input:{content:a.output}});return {a:a.output,b:b.output};`; + const run1=await run(f.engine,script,{},{executor:'mcode'}); + aOutput='new'; + const run2=await run(f.engine,script,{},{executor:'mcode',reuseAcrossRuns:true}); + assert.deepEqual(calls,['a','b','a','b'],'a re-executes (no model => never a candidate); b must not adopt the stale run-1 result'); + assert.deepEqual(run2.result,{a:'new',b:'b'}); + assert.ok(!run2.steps.find(s=>s.id==='b').reusedFrom,'divergent upstream output must break lineage'); + }finally{await f.cleanup();} +}); +test('cross-run adoption restamps planId to the current topology node',async()=>{ + const calls=[],f=await fixture(async s=>{calls.push(s.id);return {output:s.id};});try{ + const body='return await ctx.agent({id:"a",prompt:"a"});'; + const run1=await run(f.engine,body); + // Same agent spec, script text differs by a leading comment -> different topology planId. + const run2=await run(f.engine,'// shifted\n'+body,{},{reuseAcrossRuns:true}); + assert.equal(calls.length,1,'reuse hits'); + const a2=run2.steps.find(s=>s.id==='a'); + assert.ok(a2.reusedFrom?.crossRun); + assert.notEqual(a2.planId,run1.steps.find(s=>s.id==='a').planId,'adopted step must not carry the source run planId'); + const node=run2.topology.nodes.find(n=>n.stepId==='a'); + assert.equal(a2.planId,node?.planId??'a','adopted step maps onto the current topology node'); + assert.equal(run2.topology.nodes.filter(n=>n.stepId==='a'||n.id==='a').length,1,'current topology exposes exactly one node for the agent'); + }finally{await f.cleanup();} +}); +test('a fresh upstream execution with identical output blocks downstream adoption (same-output, changed-file shape)',async()=>{ + // Maintainer's repro shape: A (no model, never a candidate) re-executes and returns + // the SAME value while its filesystem effect changes; B (eligible, unchanged spec) + // must re-execute rather than adopt the old result. + const {writeFile,readFile}=await import('node:fs/promises');const {join}=await import('node:path'); + let aWrites='old';const calls=[],f=await fixture(async s=>{calls.push(s.id); + if(s.id==='a'){await writeFile(join(f.dir,'produced.txt'),aWrites);return {output:'done'};} + return {output:(await readFile(join(f.dir,'produced.txt'),'utf8')).trim()};}); + try{ + const script=`const a=await ctx.agent({id:'a',prompt:'write'});const b=await ctx.agent({id:'b',prompt:'read',model:'m2',dependsOn:['a']});return {a:a.output,b:b.output};`; + await writeFile(join(f.dir,'produced.txt'),'same baseline'); + const run1=await run(f.engine,script,{files:['produced.txt']},{executor:'mcode'}); + await writeFile(join(f.dir,'produced.txt'),'same baseline');aWrites='new'; + const run2=await run(f.engine,script,{files:['produced.txt']},{executor:'mcode',reuseAcrossRuns:true}); + assert.deepEqual(calls,['a','b','a','b'],'a re-executes (no model); identical output must NOT let b adopt'); + assert.deepEqual(run2.result,{a:'done',b:'new'}); + assert.ok(!run2.steps.find(s=>s.id==='b').reusedFrom); + }finally{await f.cleanup();} +}); + +test('originalProducer survives multi-repair chains into cross-run adoption',async()=>{ + const f=await fixture(async s=>({output:s.id}));try{ + const script=`return (await ctx.agent({id:'a',prompt:'a'})).output;`; + const broken=`const a=await ctx.agent({id:'a',prompt:'a'});throw Error('x');`; + const R1=await run(f.engine,broken); + const repair=async src=>{const d=await f.engine.repair(src.id,{requestId:crypto.randomUUID(),sourceUpdatedAt:f.store.get(src.id).updatedAt,script,reason:'fix',reuseStepIds:['a']});await f.engine.approve(d.id,{revision:1});for(let i=0;i<300;i++){if(!f.engine.active.has(d.id))return f.engine.snapshot(d.id);await new Promise(r=>setTimeout(r,20));}}; + const R2=await repair(R1),R3=await repair(R2); + const R4=await run(f.engine,script,{},{reuseAcrossRuns:true}); + const a4=R4.steps.find(s=>s.id==='a'); + assert.equal(a4.reusedFrom.runId,R3.id,'immediate source is the newest run'); + assert.equal(a4.reusedFrom.crossRun,true); + assert.equal(a4.originalProducer.runId,R1.id,'first producer survives two repairs into cross-run adoption'); + }finally{await f.cleanup();} +}); diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs index fd6a073..c0a37ee 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs @@ -7817,6 +7817,16 @@ var Store = class { steps(runId) { return this.db.prepare("SELECT body FROM steps WHERE runId=? ORDER BY rowid").all(runId).map((r) => JSON.parse(r.body)); } + // All match keys (contextHash, lineageHash) are stamped on the step body at + // creation, so filtering happens in SQL and LIMIT applies after the full match. + // Rows without the stamped hashes (legacy runs) never match: cross-run reuse is + // an opt-in feature and older steps are not candidates. + findCrossRunReuse({ contextHash, requestHash, lineageHash, excludeRunId, limit = 20 }) { + return this.db.prepare("SELECT runId,body AS stepBody FROM steps WHERE runId<>? AND json_extract(body,'$.kind')='agent' AND json_extract(body,'$.status')='succeeded' AND json_extract(body,'$.requestHash')=? AND json_extract(body,'$.contextHash')=? AND json_extract(body,'$.lineageHash')=? ORDER BY rowid DESC LIMIT ?").all(excludeRunId, requestHash, contextHash, lineageHash, limit).map((r) => { + const step = JSON.parse(r.stepBody); + return { runId: r.runId, stepId: step.id, step }; + }); + } saveStep(runId, step) { this.db.prepare("INSERT INTO steps VALUES(?,?,?) ON CONFLICT(runId,id) DO UPDATE SET body=excluded.body").run(runId, step.id, JSON.stringify(step)); } @@ -14237,7 +14247,8 @@ var Engine = class extends EventEmitter { check(Number.isInteger(maxCalls) && maxCalls >= 1 && maxCalls <= 100, "\u8C03\u7528\u6570\u8303\u56F4 1\u2013100"); const limits = resolveLimits(request, this.defaults); const metadata = request.metadata === void 0 ? {} : { metadata: normalizeMetadata(request.metadata) }; - const definition = { ...limits, ...metadata, name: request.name, script: request.script, input: request.input ?? {}, executor: request.executor, concurrency, maxCalls }; + const reuseAcrossRuns = request.reuseAcrossRuns === void 0 ? false : (check(typeof request.reuseAcrossRuns === "boolean", "reuseAcrossRuns \u5FC5\u987B\u4E3A\u5E03\u5C14"), request.reuseAcrossRuns); + const definition = { ...limits, ...metadata, name: request.name, script: request.script, input: request.input ?? {}, executor: request.executor, concurrency, maxCalls, ...reuseAcrossRuns ? { reuseAcrossRuns: true } : {} }; const requestHash = hash(repair ? { ...definition, repair } : definition); const existing = this.store.byRequest(request.requestId); if (existing) { @@ -14294,7 +14305,7 @@ var Engine = class extends EventEmitter { const script = request.script ?? run.script, input = request.input ?? run.input, topology = assertValidDependencies(previewTopology(script, input)); check(input && typeof input === "object" && !Array.isArray(input), "input \u5FC5\u987B\u4E3A JSON object"); boundedJSON(input); - const name = request.name ?? run.name, executor = request.executor ?? run.executor, concurrency = request.concurrency ?? run.concurrency, maxCalls = request.maxCalls ?? run.maxCalls; + const name = request.name ?? run.name, executor = request.executor ?? run.executor, concurrency = request.concurrency ?? run.concurrency, maxCalls = request.maxCalls ?? run.maxCalls, reuseAcrossRuns = request.reuseAcrossRuns === void 0 ? run.reuseAcrossRuns : (check(typeof request.reuseAcrossRuns === "boolean", "reuseAcrossRuns \u5FC5\u987B\u4E3A\u5E03\u5C14"), request.reuseAcrossRuns); check(typeof name === "string" && name.trim().length > 0 && name.length <= 120, "\u540D\u79F0\u987B\u4E3A 1\u2013120 \u5B57\u7B26"); check(["demo", "mcode"].includes(executor), "executor \u987B\u4E3A demo \u6216 mcode"); check(Number.isInteger(concurrency) && concurrency >= 1 && concurrency <= 16, "\u5E76\u53D1\u8303\u56F4 1\u201316"); @@ -14310,7 +14321,7 @@ var Engine = class extends EventEmitter { check(repair && typeof request.reason === "string" && request.reason.trim() && request.reason.length <= 2e3, "\u8BF7\u63D0\u4F9B\u4FEE\u590D\u539F\u56E0\uFF08\u6700\u591A 2000 \u5B57\u7B26\uFF09"); repair = { ...repair, reason: request.reason.trim() }; } - Object.assign(run, ...repair ? [{ repair }] : [], resolveLimits(request, runLimits(run)), metadata, { name, script, input, executor, concurrency, maxCalls, topology, scriptHash: hash(script), revision: run.revision + 1 }); + Object.assign(run, ...repair ? [{ repair }] : [], resolveLimits(request, runLimits(run)), metadata, { name, script, input, executor, concurrency, maxCalls, reuseAcrossRuns, topology, scriptHash: hash(script), revision: run.revision + 1 }); this.save(run); this.emitEvent(id2, "run.updated", { revision: run.revision }); return this.snapshot(id2); @@ -14545,7 +14556,8 @@ var Engine = class extends EventEmitter { check(old.requestHash === hash(payload), "checkpoint \u53C2\u6570\u51B2\u7A81"); return old.output; } - const step = { id: key, kind: "checkpoint", status: "succeeded", output: payload.value, requestHash: hash(payload), label: payload.id, dependsOn: [], createdAt: Date.now() }; + const checkpointHash = hash(payload); + const step = { id: key, kind: "checkpoint", status: "succeeded", output: payload.value, requestHash: checkpointHash, lineageHash: hash({ requestHash: checkpointHash, deps: [] }), label: payload.id, dependsOn: [], createdAt: Date.now() }; this.store.saveStep(ctx.run.id, step); this.emitEvent(ctx.run.id, "checkpoint", { stepId: key }); return payload.value; @@ -14575,6 +14587,11 @@ var Engine = class extends EventEmitter { check(cached2.hash === requestHash, "\u91CD\u590D step id \u53C2\u6570\u51B2\u7A81"); return cached2.promise; } + const ctxHash = ctx.contextHash ?? (ctx.contextHash = hash({ workspace: ctx.run.workspace, input: ctx.run.input, executor: ctx.run.executor, fingerprints: ctx.run.fingerprints })); + const lineageHash = hash({ requestHash, deps: deps.map((id2) => { + const dep = this.store.step(ctx.run.id, id2); + return { id: id2, lineageHash: dep?.lineageHash ?? null, outputHash: dep?.status === "succeeded" ? hash(dep.output ?? null) : null }; + }) }); 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) => { const dep = this.store.step(ctx.run.id, id2); @@ -14598,17 +14615,58 @@ var Engine = class extends EventEmitter { usageHistory: [], sessionId: void 0, turnId: void 0, - reusedFrom: { runId: repair.sourceRunId, stepId: spec.id, endedAt: candidate.endedAt ?? null } + contextHash: ctxHash, + reusedFrom: { runId: repair.sourceRunId, stepId: spec.id, endedAt: candidate.endedAt ?? null }, + // The first producer survives repair chains: R2/R3 relay the output but + // only the original execution produced it. + originalProducer: candidate.originalProducer ?? (candidate.reusedFrom ? { ...candidate.reusedFrom } : { runId: repair.sourceRunId, stepId: spec.id, endedAt: candidate.endedAt ?? null }) }; this.store.saveStep(ctx.run.id, step2); this.emitEvent(ctx.run.id, "step.reused", { stepId: step2.id, sourceRunId: repair.sourceRunId }); return Promise.resolve({ status: "succeeded", output: step2.output, cached: true }); } } + const depsAllAdopted = deps.every((id2) => { + const dep = this.store.step(ctx.run.id, id2); + return dep?.kind === "checkpoint" || dep?.reusedFrom; + }); + if (ctx.run.reuseAcrossRuns && !previous && depsAllAdopted && !(ctx.run.executor === "mcode" && !spec.model)) { + for (const candidate2 of this.store.findCrossRunReuse({ contextHash: ctxHash, requestHash, lineageHash, excludeRunId: ctx.run.id })) { + let valid = true; + try { + if (validateOutput) valid = validateOutput(candidate2.step.output); + } catch { + valid = false; + } + if (!valid) continue; + const source = candidate2.step; + const step2 = { + ...source, + attempt: 0, + createdAt: Date.now(), + startedAt: null, + endedAt: source.endedAt ?? Date.now(), + usage: null, + usageHistory: [], + sessionId: void 0, + turnId: void 0, + contextHash: ctxHash, + // planId must follow the CURRENT dispatch: keeping the source run's planId + // duplicates/mislabels the node against this run's topology (the repair + // path already restamps it). + ...typeof planId === "string" ? { planId } : { planId: void 0 }, + reusedFrom: { runId: candidate2.runId, stepId: candidate2.stepId, endedAt: source.endedAt ?? null, crossRun: true }, + originalProducer: source.originalProducer ?? (source.reusedFrom ? { ...source.reusedFrom } : { runId: candidate2.runId, stepId: candidate2.stepId, endedAt: source.endedAt ?? null, crossRun: true }) + }; + this.store.saveStep(ctx.run.id, step2); + this.emitEvent(ctx.run.id, "step.reused", { stepId: spec.id, sourceRunId: candidate2.runId, crossRun: true }); + return Promise.resolve({ status: "succeeded", output: step2.output, cached: true }); + } + } check(ctx.run.attempts < ctx.run.maxCalls, `\u5DF2\u8FBE\u5230\u5DE5\u4F5C\u6D41 Agent \u603B\u8C03\u7528\u4E0A\u9650 ${ctx.run.maxCalls} \u6B21\uFF08\u5305\u62EC\u6062\u590D\u5C1D\u8BD5\uFF09\uFF0C\u8BF7\u63D0\u9AD8\u603B\u8C03\u7528\u9884\u7B97\u540E\u6062\u590D\u3002`); ctx.run.attempts++; this.save(ctx.run); - const step = { id: spec.id, ...typeof planId === "string" ? { planId } : {}, label: spec.label ?? spec.id, phase: spec.phase ?? null, kind: "agent", dependsOn: deps, requestHash, status: "queued", attempt: (previous?.attempt ?? 0) + 1, createdAt: previous?.createdAt ?? Date.now(), startedAt: null, endedAt: null, prompt: spec.prompt, input: spec.input ?? null, output: null, error: null, errorDetails: null, ...runLimits(ctx.run), timeoutMs: runLimits(ctx.run).stepTimeoutMs, usage: null, usageHistory: [...previous?.usageHistory ?? [], ...previous?.usage ? [previous.usage] : []] }; + const step = { id: spec.id, ...typeof planId === "string" ? { planId } : {}, label: spec.label ?? spec.id, phase: spec.phase ?? null, kind: "agent", dependsOn: deps, requestHash, contextHash: ctxHash, lineageHash, status: "queued", attempt: (previous?.attempt ?? 0) + 1, createdAt: previous?.createdAt ?? Date.now(), startedAt: null, endedAt: null, prompt: spec.prompt, input: spec.input ?? null, output: null, error: null, errorDetails: null, ...runLimits(ctx.run), timeoutMs: runLimits(ctx.run).stepTimeoutMs, usage: null, usageHistory: [...previous?.usageHistory ?? [], ...previous?.usage ? [previous.usage] : []] }; this.store.saveStep(ctx.run.id, step); this.emitEvent(ctx.run.id, "step.queued", { stepId: step.id }); const promise = (async () => { @@ -26429,8 +26487,8 @@ var string3 = { type: "string" }; var id = { runId: string3 }; var TOOLS = [ { name: "workflow_validate", description: "\u9759\u6001\u68C0\u67E5\u811A\u672C\u5E76\u751F\u6210\u7ED3\u6784\u62D3\u6251\uFF0C\u4E0D\u6267\u884C\u811A\u672C\u6216 Agent\u3002DSL: await ctx.phase({id,label}); await ctx.agent({id,label,phase,dependsOn,prompt,input,schema}); ctx.map(items,fn); ctx.log(message,{stepId,phase}); ctx.checkpoint(id,value)\u3002dependsOn \u53EF\u4E3A\u5355\u4E2A ID \u5B57\u7B26\u4E32\u6216 ID \u6570\u7EC4\uFF0C\u63A8\u8350\u6570\u7EC4\uFF1B\u4F9D\u8D56\u987B\u5148\u6210\u529F\u3002agent \u8FD4\u56DE status/output/error\uFF1B\u987B\u663E\u5F0F\u5904\u7406\u5931\u8D25\u3002", inputSchema: obj({ script: string3 }, ["script"]) }, - { name: "workflow_start", description: "\u521B\u5EFA\u5F85\u5BA1\u6838\u5DE5\u4F5C\u6D41\u548C\u7ED3\u6784\u62D3\u6251\uFF0C\u4E0D\u6267\u884C Agent\u3002\u5FC5\u987B\u63D0\u4F9B\u9762\u677F\u8BA9\u7528\u6237\u5BA1\u9605\u3001\u4FEE\u6539\u5E76\u70B9\u51FB\u5F00\u59CB\u6267\u884C\u3002mcode \u6A21\u5F0F\u4F1A\u542F\u52A8\u771F\u5B9E MCode\uFF0C\u4F1A\u4F7F\u7528\u5DF2\u767B\u5F55\u8EAB\u4EFD\u4E0E smart \u6743\u9650\uFF0C\u4E0D\u63D0\u4F9B\u53EA\u8BFB OS \u6C99\u7BB1\u3002demo \u6A21\u5F0F\u4E0D\u8C03\u7528\u6A21\u578B\u3002\u663E\u5F0F requestId \u5E42\u7B49\u3002", inputSchema: obj({ requestId: string3, name: string3, script: string3, input: { type: "object" }, metadata: METADATA_SCHEMA, executor: { enum: ["mcode", "demo"] }, concurrency: { type: "integer", minimum: 1, maximum: 16 }, maxCalls: { type: "integer", minimum: 1, maximum: 100 }, ...LIMIT_SCHEMAS }, ["requestId", "name", "script", "executor"]) }, - { name: "workflow_update", description: "\u4FEE\u6539\u5F85\u5BA1\u6838\u5DE5\u4F5C\u6D41\u7684\u811A\u672C\u3001\u8F93\u5165\u6216\u9884\u7B97\u5E76\u91CD\u5EFA\u62D3\u6251\uFF0C\u4FDD\u5B58\u540E\u4ECD\u5F85\u5BA1\u6838\uFF1Brevision \u5FC5\u987B\u5339\u914D\u5F53\u524D\u7248\u672C\u3002\u4E0D\u53EF\u4FEE\u6539\u5DF2\u5F00\u59CB\u7684\u8FD0\u884C\u3002", inputSchema: obj({ ...id, revision: { type: "integer", minimum: 1 }, reason: { type: "string", maxLength: 2e3 }, reuseStepIds: { type: "array", items: string3, maxItems: 100, uniqueItems: true }, name: string3, script: string3, input: { type: "object" }, metadata: METADATA_SCHEMA, executor: { enum: ["mcode", "demo"] }, concurrency: { type: "integer", minimum: 1, maximum: 16 }, maxCalls: { type: "integer", minimum: 1, maximum: 100 }, ...LIMIT_SCHEMAS }, ["runId", "revision"]) }, + { name: "workflow_start", description: "\u521B\u5EFA\u5F85\u5BA1\u6838\u5DE5\u4F5C\u6D41\u548C\u7ED3\u6784\u62D3\u6251\uFF0C\u4E0D\u6267\u884C Agent\u3002\u5FC5\u987B\u63D0\u4F9B\u9762\u677F\u8BA9\u7528\u6237\u5BA1\u9605\u3001\u4FEE\u6539\u5E76\u70B9\u51FB\u5F00\u59CB\u6267\u884C\u3002mcode \u6A21\u5F0F\u4F1A\u542F\u52A8\u771F\u5B9E MCode\uFF0C\u4F1A\u4F7F\u7528\u5DF2\u767B\u5F55\u8EAB\u4EFD\u4E0E smart \u6743\u9650\uFF0C\u4E0D\u63D0\u4F9B\u53EA\u8BFB OS \u6C99\u7BB1\u3002demo \u6A21\u5F0F\u4E0D\u8C03\u7528\u6A21\u578B\u3002\u663E\u5F0F requestId \u5E42\u7B49\u3002", inputSchema: obj({ requestId: string3, name: string3, script: string3, input: { type: "object" }, metadata: METADATA_SCHEMA, executor: { enum: ["mcode", "demo"] }, concurrency: { type: "integer", minimum: 1, maximum: 16 }, maxCalls: { type: "integer", minimum: 1, maximum: 100 }, reuseAcrossRuns: { type: "boolean", description: "Opt-in: adopt succeeded nodes from prior runs in the same workspace when context and spec hashes match" }, ...LIMIT_SCHEMAS }, ["requestId", "name", "script", "executor"]) }, + { name: "workflow_update", description: "\u4FEE\u6539\u5F85\u5BA1\u6838\u5DE5\u4F5C\u6D41\u7684\u811A\u672C\u3001\u8F93\u5165\u6216\u9884\u7B97\u5E76\u91CD\u5EFA\u62D3\u6251\uFF0C\u4FDD\u5B58\u540E\u4ECD\u5F85\u5BA1\u6838\uFF1Brevision \u5FC5\u987B\u5339\u914D\u5F53\u524D\u7248\u672C\u3002\u4E0D\u53EF\u4FEE\u6539\u5DF2\u5F00\u59CB\u7684\u8FD0\u884C\u3002", inputSchema: obj({ ...id, revision: { type: "integer", minimum: 1 }, reason: { type: "string", maxLength: 2e3 }, reuseStepIds: { type: "array", items: string3, maxItems: 100, uniqueItems: true }, name: string3, script: string3, input: { type: "object" }, metadata: METADATA_SCHEMA, executor: { enum: ["mcode", "demo"] }, concurrency: { type: "integer", minimum: 1, maximum: 16 }, maxCalls: { type: "integer", minimum: 1, maximum: 100 }, reuseAcrossRuns: { type: "boolean", description: "Opt-in: adopt succeeded nodes from prior runs in the same workspace when context and spec hashes match" }, ...LIMIT_SCHEMAS }, ["runId", "revision"]) }, { name: "workflow_repair", description: "\u57FA\u4E8E\u505C\u6B62\u540E\u7684\u8FD0\u884C\u521B\u5EFA\u4FEE\u590D\u8349\u7A3F\uFF0C\u4FDD\u7559\u6E90\u8FD0\u884C\uFF1B\u63D0\u4F9B\u5B8C\u6574\u4FEE\u590D\u811A\u672C\u3001\u5931\u8D25\u539F\u56E0\u4E0E sourceUpdatedAt\u3002\u663E\u5F0F reuseStepIds \u4EC5\u9009\u62E9\u786E\u8BA4\u4ECD\u9002\u7528\u7684\u6210\u529F\u8282\u70B9\uFF0C\u9ED8\u8BA4\u4E0D\u590D\u7528\u3002\u8FD0\u884C\u65F6\u91CD\u65B0\u6821\u9A8C\u8F93\u5165\u3001\u6587\u4EF6\u3001\u53C2\u6570\u4E0E\u4F9D\u8D56\uFF1B\u53D8\u66F4\u6216\u91CD\u8DD1\u7684\u4E0A\u6E38\u4F7F\u4E0B\u6E38\u5931\u6548\u3002\u5FC5\u987B\u6253\u5F00\u9762\u677F\u4EA4\u7528\u6237\u5BA1\u6838\u540E\u5F00\u59CB\uFF0C\u4E0D\u80FD\u81EA\u52A8\u6267\u884C\u3002", inputSchema: obj({ ...id, requestId: string3, sourceUpdatedAt: { type: "integer" }, script: string3, reason: { type: "string", maxLength: 2e3 }, reuseStepIds: { type: "array", items: string3, maxItems: 100, uniqueItems: true }, input: { type: "object" }, ...LIMIT_SCHEMAS, maxCalls: { type: "integer", minimum: 1, maximum: 100 } }, ["runId", "requestId", "sourceUpdatedAt", "script", "reason"]) }, { name: "workflow_status", description: "\u8BFB\u53D6\u8FD0\u884C\u72B6\u6001\u3001\u9636\u6BB5\u548C\u8282\u70B9\uFF1B\u8F93\u51FA\u4E0D\u542B\u5B8C\u6574 prompt/result\u3002\u65E0 runId \u65F6\u5217\u51FA\u6700\u8FD1\u8FD0\u884C\u3002", inputSchema: obj(id) }, { name: "workflow_results", description: "\u5206\u9875\u8BFB\u53D6\u8282\u70B9\u7ED3\u679C\uFF1B\u7EC8\u6001\u62A5\u544A\u4E0E\u5931\u8D25\u660E\u786E\u5206\u5F00\u3002", inputSchema: obj({ ...id, includeDefinition: { type: "boolean" }, offset: { type: "integer", minimum: 0 }, limit: { type: "integer", minimum: 1, maximum: 20 } }, ["runId"]) }, diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/src/engine.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/src/engine.mjs index aca3ae0..a482e4a 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/src/engine.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/src/engine.mjs @@ -44,7 +44,9 @@ export class Engine extends EventEmitter { check(Number.isInteger(concurrency)&&concurrency>=1&&concurrency<=16,'并发范围 1–16');check(Number.isInteger(maxCalls)&&maxCalls>=1&&maxCalls<=100,'调用数范围 1–100'); const limits=resolveLimits(request,this.defaults); const metadata=request.metadata===undefined?{}:{metadata:normalizeMetadata(request.metadata)}; - 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); + // Absent unless explicitly true: an always-present default key would change requestHash and break legacy idempotent replays. + const reuseAcrossRuns=request.reuseAcrossRuns===undefined?false:(check(typeof request.reuseAcrossRuns==='boolean','reuseAcrossRuns 必须为布尔'),request.reuseAcrossRuns); + const definition={...limits,...metadata,name:request.name,script:request.script,input:request.input??{},executor:request.executor,concurrency,maxCalls,...(reuseAcrossRuns?{reuseAcrossRuns:true}:{})};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??{})); check(!this.closing,'服务正在关闭'); @@ -71,14 +73,14 @@ export class Engine extends EventEmitter { check(Number.isInteger(request.revision)&&request.revision===run.revision,'审核版本已更新，请刷新后再修改'); const script=request.script??run.script,input=request.input??run.input,topology=assertValidDependencies(previewTopology(script,input)); check(input&&typeof input==='object'&&!Array.isArray(input),'input 必须为 JSON object');boundedJSON(input); - const name=request.name??run.name,executor=request.executor??run.executor,concurrency=request.concurrency??run.concurrency,maxCalls=request.maxCalls??run.maxCalls; + const name=request.name??run.name,executor=request.executor??run.executor,concurrency=request.concurrency??run.concurrency,maxCalls=request.maxCalls??run.maxCalls,reuseAcrossRuns=request.reuseAcrossRuns===undefined?run.reuseAcrossRuns:(check(typeof request.reuseAcrossRuns==='boolean','reuseAcrossRuns 必须为布尔'),request.reuseAcrossRuns); check(typeof name==='string'&&name.trim().length>0&&name.length<=120,'名称须为 1–120 字符');check(['demo','mcode'].includes(executor),'executor 须为 demo 或 mcode'); check(Number.isInteger(concurrency)&&concurrency>=1&&concurrency<=16,'并发范围 1–16');check(Number.isInteger(maxCalls)&&maxCalls>=1&&maxCalls<=100,'调用数范围 1–100'); const metadata=request.metadata===undefined?{}:{metadata:normalizeMetadata(request.metadata)}; let repair=run.repair; if(request.reuseStepIds!==undefined){const ids=request.reuseStepIds;check(repair&&Array.isArray(ids)&&ids.length<=100&&new Set(ids).size===ids.length&&ids.every(id=>typeof id==='string'&&(repair.candidateStepIds??repair.reuseStepIds).includes(id)),'只能选择修复草稿已冻结的候选节点');repair={...repair,reuseStepIds:ids};} if(request.reason!==undefined){check(repair&&typeof request.reason==='string'&&request.reason.trim()&&request.reason.length<=2000,'请提供修复原因（最多 2000 字符）');repair={...repair,reason:request.reason.trim()};} - Object.assign(run,...(repair?[{repair}]:[]),resolveLimits(request,runLimits(run)),metadata,{name,script,input,executor,concurrency,maxCalls,topology,scriptHash:hash(script),revision:run.revision+1}); + Object.assign(run,...(repair?[{repair}]:[]),resolveLimits(request,runLimits(run)),metadata,{name,script,input,executor,concurrency,maxCalls,reuseAcrossRuns,topology,scriptHash:hash(script),revision:run.revision+1}); this.save(run);this.emitEvent(id,'run.updated',{revision:run.revision});return this.snapshot(id); } saveTemplate(id,{name,revision}={}) { @@ -164,7 +166,8 @@ export class Engine extends EventEmitter { for(const key of ['stepId','phase'])if(payload[key]!==undefined){check(typeof payload[key]==='string'&&/^[A-Za-z0-9_:./-]{1,150}$/.test(payload[key]),'日志关联 ID 无效');detail[key]=payload[key];} ctx.run.latestProgress={...detail,time:Date.now()};this.save(ctx.run);this.emitEvent(ctx.run.id,'log',detail);return; } - if(method==='checkpoint'){check(typeof payload.id==='string'&&payload.id.length<=100,'checkpoint 必须有 ID');const key=`checkpoint:${payload.id}`,old=this.store.step(ctx.run.id,key);if(old){check(old.requestHash===hash(payload),'checkpoint 参数冲突');return old.output;}const step={id:key,kind:'checkpoint',status:'succeeded',output:payload.value,requestHash:hash(payload),label:payload.id,dependsOn:[],createdAt:Date.now()};this.store.saveStep(ctx.run.id,step);this.emitEvent(ctx.run.id,'checkpoint',{stepId:key});return payload.value;} + if(method==='checkpoint'){check(typeof payload.id==='string'&&payload.id.length<=100,'checkpoint 必须有 ID');const key=`checkpoint:${payload.id}`,old=this.store.step(ctx.run.id,key);if(old){check(old.requestHash===hash(payload),'checkpoint 参数冲突');return old.output;}const checkpointHash=hash(payload);// Checkpoints recompute every run; their lineage hash covers the recomputed value so a changed value breaks downstream cross-run lineage. +const step={id:key,kind:'checkpoint',status:'succeeded',output:payload.value,requestHash:checkpointHash,lineageHash:hash({requestHash:checkpointHash,deps:[]}),label:payload.id,dependsOn:[],createdAt:Date.now()};this.store.saveStep(ctx.run.id,step);this.emitEvent(ctx.run.id,'checkpoint',{stepId:key});return payload.value;} throw new Error('未知脚本操作'); } agent(ctx,spec,planId){ @@ -182,6 +185,14 @@ export class Engine extends EventEmitter { const requestHash=hash(spec),previous=this.store.step(ctx.run.id,spec.id),cached=ctx.calls.get(spec.id); if(previous){check(previous.requestHash===requestHash,`步骤 ${spec.id} 使用了不同参数，恢复已停止`);if(previous.status==='succeeded')return Promise.resolve({status:'succeeded',output:previous.output,cached:true});} if(cached){check(cached.hash===requestHash,'重复 step id 参数冲突');return cached.promise;} + // Computed once per dispatch attempt, before any candidate lookup. contextHash is + // stamped on every new step so the store can filter cross-run candidates in SQL + // before LIMIT; lineageHash pins the node to its own spec plus, for each + // succeeded dependency (in dependsOn order), that dependency's lineage hash + // AND output hash — so a changed, rerun, or differently-outcomed upstream + // invalidates downstream candidates even when the downstream spec is unchanged. + const ctxHash=ctx.contextHash??(ctx.contextHash=hash({workspace:ctx.run.workspace,input:ctx.run.input,executor:ctx.run.executor,fingerprints:ctx.run.fingerprints})); + const lineageHash=hash({requestHash,deps:deps.map(id=>{const dep=this.store.step(ctx.run.id,id);return {id,lineageHash:dep?.lineageHash??null,outputHash:dep?.status==='succeeded'?hash(dep.output??null):null};})}); 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}) @@ -191,13 +202,48 @@ export class Engine extends EventEmitter { ||(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, - reusedFrom:{runId:repair.sourceRunId,stepId:spec.id,endedAt:candidate.endedAt??null}}; + 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,contextHash:ctxHash, + reusedFrom:{runId:repair.sourceRunId,stepId:spec.id,endedAt:candidate.endedAt??null}, + // The first producer survives repair chains: R2/R3 relay the output but + // only the original execution produced it. + originalProducer:candidate.originalProducer??(candidate.reusedFrom?{...candidate.reusedFrom}:{runId:repair.sourceRunId,stepId:spec.id,endedAt:candidate.endedAt??null})}; this.store.saveStep(ctx.run.id,step);this.emitEvent(ctx.run.id,'step.reused',{stepId:step.id,sourceRunId:repair.sourceRunId}); return Promise.resolve({status:'succeeded',output:step.output,cached:true});} } + // A freshly executed agent dependency has unproven execution identity: spec and + // output hashes cannot see changed filesystem effects (same return value, different + // written content). Downstream adoption is therefore only sound when every agent + // dependency was itself adopted/reused; checkpoints recompute deterministically and + // their value hash is already bound into lineage. + const depsAllAdopted=deps.every(id=>{const dep=this.store.step(ctx.run.id,id);return dep?.kind==='checkpoint'||dep?.reusedFrom;}); + if(ctx.run.reuseAcrossRuns&&!previous&&depsAllAdopted&&!(ctx.run.executor==='mcode'&&!spec.model)){ + // Cross-run reuse: adopt an earlier run's stored result when the node spec, its + // upstream lineage and the run context (workspace, input, executor, tracked + // files) all hash identically. All three keys are stamped on candidate steps at + // creation, so the store filters in SQL before LIMIT. mcode nodes without an + // explicit model are never candidates: the effective model comes from the CLI + // environment and is not part of the match key. + for(const candidate of this.store.findCrossRunReuse({contextHash:ctxHash,requestHash,lineageHash,excludeRunId:ctx.run.id})){ + let valid=true;try{if(validateOutput)valid=validateOutput(candidate.step.output);}catch{valid=false;} + if(!valid)continue; + const source=candidate.step; + // lineageHash is adopted from the source (dependencies already matched, so the + // lineage is equivalent); contextHash is restamped with this run's. Provenance + // always records the immediate source in reusedFrom and the first producer in + // originalProducer, so chained adoptions stay consistent with the emitted event. + const step={...source,attempt:0,createdAt:Date.now(),startedAt:null,endedAt:source.endedAt??Date.now(),usage:null,usageHistory:[],sessionId:undefined,turnId:undefined,contextHash:ctxHash, + // planId must follow the CURRENT dispatch: keeping the source run's planId + // duplicates/mislabels the node against this run's topology (the repair + // path already restamps it). + ...(typeof planId==='string'?{planId}:{planId:undefined}), + reusedFrom:{runId:candidate.runId,stepId:candidate.stepId,endedAt:source.endedAt??null,crossRun:true}, + originalProducer:source.originalProducer??(source.reusedFrom?{...source.reusedFrom}:{runId:candidate.runId,stepId:candidate.stepId,endedAt:source.endedAt??null,crossRun:true})}; + this.store.saveStep(ctx.run.id,step);this.emitEvent(ctx.run.id,'step.reused',{stepId:spec.id,sourceRunId:candidate.runId,crossRun:true}); + return Promise.resolve({status:'succeeded',output:step.output,cached:true}); + } + } check(ctx.run.attempts{ let release;try{release=await this.acquire(ctx.controller.signal,ctx,step.id);ctx.controller.signal.throwIfAborted();step.status='running';step.startedAt=Date.now();this.store.saveStep(ctx.run.id,step);this.emitEvent(ctx.run.id,'step.started',{stepId:step.id}); diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs index cebb8e6..915142f 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs @@ -47,6 +47,11 @@ export class Store { list() {return this.db.prepare("SELECT body FROM runs ORDER BY CASE WHEN json_extract(body,'$.status') IN ('running','queued','stopping','pausing') THEN 0 WHEN json_extract(body,'$.status')='needs_attention' THEN 1 ELSE 2 END, rowid DESC LIMIT 100").all().map(r=>JSON.parse(r.body));} step(runId,id) {const r=this.db.prepare('SELECT body FROM steps WHERE runId=? AND id=?').get(runId,id);return r?JSON.parse(r.body):null;} steps(runId) {return this.db.prepare('SELECT body FROM steps WHERE runId=? ORDER BY rowid').all(runId).map(r=>JSON.parse(r.body));} + // All match keys (contextHash, lineageHash) are stamped on the step body at + // creation, so filtering happens in SQL and LIMIT applies after the full match. + // Rows without the stamped hashes (legacy runs) never match: cross-run reuse is + // an opt-in feature and older steps are not candidates. + findCrossRunReuse({contextHash,requestHash,lineageHash,excludeRunId,limit=20}) {return this.db.prepare("SELECT runId,body AS stepBody FROM steps WHERE runId<>? AND json_extract(body,'$.kind')='agent' AND json_extract(body,'$.status')='succeeded' AND json_extract(body,'$.requestHash')=? AND json_extract(body,'$.contextHash')=? AND json_extract(body,'$.lineageHash')=? ORDER BY rowid DESC LIMIT ?").all(excludeRunId,requestHash,contextHash,lineageHash,limit).map(r=>{const step=JSON.parse(r.stepBody);return {runId:r.runId,stepId:step.id,step};});} saveStep(runId,step) {this.db.prepare('INSERT INTO steps VALUES(?,?,?) ON CONFLICT(runId,id) DO UPDATE SET body=excluded.body').run(runId,step.id,JSON.stringify(step));} repairCandidate(runId,id) {const r=this.db.prepare('SELECT body FROM repair_cache WHERE runId=? AND id=?').get(runId,id);return r?JSON.parse(r.body):null;} saveRepairCandidate(runId,step) {this.db.prepare('INSERT INTO repair_cache VALUES(?,?,?)').run(runId,step.id,JSON.stringify(step));} diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/src/tools.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/src/tools.mjs index daa3b54..1c181bf 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/src/tools.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/src/tools.mjs @@ -10,8 +10,8 @@ const obj=(properties,required=[])=>({type:'object',properties,required,addition const string={type:'string'};const id={runId:string}; export const TOOLS=[ {name:'workflow_validate',description:'静态检查脚本并生成结构拓扑，不执行脚本或 Agent。DSL: await ctx.phase({id,label}); await ctx.agent({id,label,phase,dependsOn,prompt,input,schema}); ctx.map(items,fn); ctx.log(message,{stepId,phase}); ctx.checkpoint(id,value)。dependsOn 可为单个 ID 字符串或 ID 数组，推荐数组；依赖须先成功。agent 返回 status/output/error；须显式处理失败。',inputSchema:obj({script:string},['script'])}, - {name:'workflow_start',description:'创建待审核工作流和结构拓扑，不执行 Agent。必须提供面板让用户审阅、修改并点击开始执行。mcode 模式会启动真实 MCode，会使用已登录身份与 smart 权限，不提供只读 OS 沙箱。demo 模式不调用模型。显式 requestId 幂等。',inputSchema:obj({requestId:string,name:string,script:string,input:{type:'object'},metadata:METADATA_SCHEMA,executor:{enum:['mcode','demo']},concurrency:{type:'integer',minimum:1,maximum:16},maxCalls:{type:'integer',minimum:1,maximum:100},...LIMIT_SCHEMAS},['requestId','name','script','executor'])}, - {name:'workflow_update',description:'修改待审核工作流的脚本、输入或预算并重建拓扑，保存后仍待审核；revision 必须匹配当前版本。不可修改已开始的运行。',inputSchema:obj({...id,revision:{type:'integer',minimum:1},reason:{type:'string',maxLength:2000},reuseStepIds:{type:'array',items:string,maxItems:100,uniqueItems:true},name:string,script:string,input:{type:'object'},metadata:METADATA_SCHEMA,executor:{enum:['mcode','demo']},concurrency:{type:'integer',minimum:1,maximum:16},maxCalls:{type:'integer',minimum:1,maximum:100},...LIMIT_SCHEMAS},['runId','revision'])}, + {name:'workflow_start',description:'创建待审核工作流和结构拓扑，不执行 Agent。必须提供面板让用户审阅、修改并点击开始执行。mcode 模式会启动真实 MCode，会使用已登录身份与 smart 权限，不提供只读 OS 沙箱。demo 模式不调用模型。显式 requestId 幂等。',inputSchema:obj({requestId:string,name:string,script:string,input:{type:'object'},metadata:METADATA_SCHEMA,executor:{enum:['mcode','demo']},concurrency:{type:'integer',minimum:1,maximum:16},maxCalls:{type:'integer',minimum:1,maximum:100},reuseAcrossRuns:{type:'boolean',description:'Opt-in: adopt succeeded nodes from prior runs in the same workspace when context and spec hashes match'},...LIMIT_SCHEMAS},['requestId','name','script','executor'])}, + {name:'workflow_update',description:'修改待审核工作流的脚本、输入或预算并重建拓扑，保存后仍待审核；revision 必须匹配当前版本。不可修改已开始的运行。',inputSchema:obj({...id,revision:{type:'integer',minimum:1},reason:{type:'string',maxLength:2000},reuseStepIds:{type:'array',items:string,maxItems:100,uniqueItems:true},name:string,script:string,input:{type:'object'},metadata:METADATA_SCHEMA,executor:{enum:['mcode','demo']},concurrency:{type:'integer',minimum:1,maximum:16},maxCalls:{type:'integer',minimum:1,maximum:100},reuseAcrossRuns:{type:'boolean',description:'Opt-in: adopt succeeded nodes from prior runs in the same workspace when context and spec hashes match'},...LIMIT_SCHEMAS},['runId','revision'])}, {name:'workflow_repair',description:'基于停止后的运行创建修复草稿，保留源运行；提供完整修复脚本、失败原因与 sourceUpdatedAt。显式 reuseStepIds 仅选择确认仍适用的成功节点，默认不复用。运行时重新校验输入、文件、参数与依赖；变更或重跑的上游使下游失效。必须打开面板交用户审核后开始，不能自动执行。',inputSchema:obj({...id,requestId:string,sourceUpdatedAt:{type:'integer'},script:string,reason:{type:'string',maxLength:2000},reuseStepIds:{type:'array',items:string,maxItems:100,uniqueItems:true},input:{type:'object'},...LIMIT_SCHEMAS,maxCalls:{type:'integer',minimum:1,maximum:100}},['runId','requestId','sourceUpdatedAt','script','reason'])}, {name:'workflow_status',description:'读取运行状态、阶段和节点；输出不含完整 prompt/result。无 runId 时列出最近运行。',inputSchema:obj(id)}, {name:'workflow_results',description:'分页读取节点结果；终态报告与失败明确分开。',inputSchema:obj({...id,includeDefinition:{type:'boolean'},offset:{type:'integer',minimum:0},limit:{type:'integer',minimum:1,maximum:20}},['runId'])},