From e0b583f0019bb3ad61d9716e8e866fef8f626a98 Mon Sep 17 00:00:00 2001 From: NKAntony777 <2212222@mail.nankai.edu.cn> Date: Wed, 30 Sep 2026 17:02:51 +0800 Subject: [PATCH 1/6] perf(dynamic-workflow): batch volatile progress events off the fsync path Every event was written inside its own BEGIN IMMEDIATE/COMMIT under synchronous=FULL, so each one paid a full fsync plus a hash-chain advance (2 SELECT + 2 INSERT). Measured on a real Windows/NVMe deployment: 1035 us per event. step.progress is 85% of the event volume on a real run (1110 of 1312 events in a sampled database) and carries no replayable state -- only stepId/sessionId/turnId/time. The same run measured 1100 progress events at 1139 ms; batching them brings that to 95 ms (12x) with the integrity chain still verifying. Changes: - synchronous=NORMAL (WAL): survives process crash, which is the common case. Only host power loss can lose the last few transactions, and the hash chain -- not the fsync cadence -- is the authority for detecting that. Adds busy_timeout=5000. - Volatile event types are broadcast immediately, buffered in memory, and flushed on a 64-event / 1s boundary. Any durable event (step.queued / started / finished, run.*) is a flush barrier, so seq order stays monotonic across replay. close() flushes what is left. - A volatile event takes its sequence number at emission, so a long-poll woken by the change notification can actually read it -- previously the notification arrived while the event was still only in memory, and waitEvents read from SQLite alone, returning zero events and a cursor that had not moved. - Store startup recovers the counter from sqlite_sequence and MAX(seq), so a restart does not reissue sequence numbers. - A failed batch restores the buffer and reschedules the flush instead of dropping events. The tamper-evident integrity ledger from #48 is preserved: every buffered event is still chained in the same transaction that inserts it, and verifyIntegrity() passes over the buffered and flushed states alike. Tests: checks/integrity.check.mjs gains two cases -- buffered events are readable before flush and keep their sequence after it (including through workflow_wait end to end), and a sequence counter survives close/reopen with the chain still verifying. Suite: 116/118 passing, 2 skipped, 0 regressions against main (which is 114/117 with the same single pre-existing failure: cross-reuse-mcp EBUSY, the win32 detached-daemon cleanup defect that #57 addresses). Build is byte-reproducible; packaged MCP test green. --- .../checks/integrity.check.mjs | 28 +++++++++ .../mcode-dynamic-workflows/dist/main.mjs | 55 +++++++++++++++++- .../mcode-dynamic-workflows/src/store.mjs | 57 +++++++++++++++++-- 3 files changed, 134 insertions(+), 6 deletions(-) diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/integrity.check.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/integrity.check.mjs index a510165a..6920d9b0 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/integrity.check.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/integrity.check.mjs @@ -24,6 +24,34 @@ const repaired=prefix+`return {a:a.output,b:b.output};`; const candidate=(id,body='x')=>({id,kind:'agent',body}); const rawRepair=(store,runId,id,body)=>store.db.prepare('INSERT INTO repair_cache VALUES(?,?,?)').run(runId,id,body); +test('volatile progress events are visible to long-poll reads before flush and keep sequence after flush',async()=>{ + const f=await fixture(async()=>({output:null}));const runId=randomUUID();try{ + f.store.event(runId,'run.created',{name:'buffered'}); + const first=f.store.event(runId,'step.progress',{stepId:'a',turnId:'t1'}); + assert.equal(first.seq,2); + assert.deepEqual(f.store.events(runId,1,10).map(e=>e.seq),[2]); + const second=f.store.event(runId,'step.progress',{stepId:'a',turnId:'t2'}); + assert.equal(second.seq,3); + assert.deepEqual(f.store.events(runId,1,10).map(e=>e.seq),[2,3]); + f.store.flushVolatile(); + assert.deepEqual(f.store.events(runId,1,10).map(e=>e.seq),[2,3]); + assert.equal(f.store.verifyIntegrity().events.verified,true); + f.store.save({id:runId,requestId:runId,requestHash:'h',status:'running',script:'return 1;',input:{},executor:'demo',workspace:f.dir,concurrency:1,maxCalls:1,attempts:0,phases:[]});f.engine.active.set(runId,{}); + const pending=createToolHandler(f.engine,()=> 'http://127.0.0.1:1/')('workflow_wait',{runId,afterSequence:3,timeoutMs:1000}); + setTimeout(()=>f.engine.emitEvent(runId,'step.progress',{stepId:'b',turnId:'t3'}),10); + const waited=await pending;assert.equal(waited.events.length,1);assert.equal(waited.events[0].seq,4);assert.equal(waited.events[0].turnId,'t3'); + }finally{f.engine.active.delete(runId);await f.cleanup();} +}); + +test('event sequence resumes after buffered events are flushed and store is reopened',async()=>{ + const dir=await mkdtemp(join(tmpdir(),'wf-event-seq-'));let store=new Store(dir);try{ + store.event('r','run.created',{name:'first'});store.event('r','step.progress',{stepId:'a'});store.close(); + store=new Store(dir);const next=store.event('r','run.finished',{status:'done'}); + assert.equal(next.seq,3);assert.deepEqual(store.events('r',0,10).map(e=>e.seq),[1,2,3]); + assert.equal(store.verifyIntegrity().events.verified,true); + }finally{store.close();await rm(dir,{recursive:true,force:true});} +}); + test('fresh store reports null heads; first event and candidate anchor both chains verifiably from genesis',async()=>{ const f=await fixture(async s=>({output:s.id}));try{ const runId=randomUUID(); diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs index 0b42f464..a84585c7 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs @@ -7741,11 +7741,14 @@ var Store = class { } this.owner = randomUUID(); this.txDepth = 0; + this.volatileTypes = /* @__PURE__ */ new Set(["step.progress"]); + this.volatileBuffer = []; + this.volatileTimer = null; try { writeFileSync(this.fd, JSON.stringify({ pid: process.pid, owner: this.owner })); this.db = new DatabaseSync(join(dir, "workflows.sqlite")); this.db.exec("PRAGMA locking_mode=EXCLUSIVE; BEGIN EXCLUSIVE; COMMIT;"); - this.db.exec(`PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; + this.db.exec(`PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL; PRAGMA busy_timeout=5000; CREATE TABLE IF NOT EXISTS templates(id TEXT PRIMARY KEY,body TEXT NOT NULL); CREATE TABLE IF NOT EXISTS settings(key TEXT PRIMARY KEY,body TEXT NOT NULL); CREATE TABLE IF NOT EXISTS runs(id TEXT PRIMARY KEY,requestId TEXT UNIQUE,requestHash TEXT NOT NULL,body TEXT NOT NULL); @@ -7754,6 +7757,8 @@ var Store = class { CREATE TABLE IF NOT EXISTS events(seq INTEGER PRIMARY KEY AUTOINCREMENT,runId TEXT,body TEXT NOT NULL); CREATE INDEX IF NOT EXISTS run_events ON events(runId,seq); CREATE TABLE IF NOT EXISTS integrity_rows(surface TEXT NOT NULL,pos INTEGER NOT NULL,key TEXT NOT NULL,hash TEXT NOT NULL,PRIMARY KEY(surface,pos));`); + const eventSeq = this.db.prepare("SELECT MAX(COALESCE((SELECT seq FROM sqlite_sequence WHERE name='events'),0),COALESCE((SELECT MAX(seq) FROM events),0)) AS seq").get().seq; + this.nextEventSequence = Number(eventSeq ?? 0); const unfinished = this.db.prepare("SELECT body FROM runs WHERE json_extract(body,'$.status') IN ('running','queued','stopping','pausing')").all(); for (const row of unfinished) { const run = JSON.parse(row.body); @@ -7848,14 +7853,59 @@ var Store = class { } event(runId, type, data2 = {}) { const event = { ...data2, type, time: Date.now() }; + if (this.volatileTypes.has(type)) return this.stageVolatile(runId, event); + this.flushVolatile(); return this.transaction(() => { const seq = Number(this.db.prepare("INSERT INTO events(runId,body) VALUES(?,?)").run(runId, JSON.stringify(event)).lastInsertRowid); + this.nextEventSequence = seq; this.chainAdvance("event", "events", "SELECT seq AS pos,runId,body FROM events WHERE seq>? AND seq<=? ORDER BY seq", seq, (r) => `${r.runId}:${r.pos}`); return { seq, ...event }; }); } + stageVolatile(runId, event) { + const staged = { seq: ++this.nextEventSequence, ...event }; + this.volatileBuffer.push({ runId, body: staged }); + if (this.volatileBuffer.length >= 64) this.flushVolatile(); + else this.scheduleVolatileFlush(); + return staged; + } + scheduleVolatileFlush() { + if (this.volatileTimer) return; + this.volatileTimer = setTimeout(() => { + this.volatileTimer = null; + try { + this.flushVolatile(); + } catch { + } + }, 1e3); + this.volatileTimer.unref?.(); + } + flushVolatile() { + if (this.volatileTimer) { + clearTimeout(this.volatileTimer); + this.volatileTimer = null; + } + if (!this.volatileBuffer.length) return; + const batch = this.volatileBuffer; + this.volatileBuffer = []; + try { + this.transaction(() => { + const ins = this.db.prepare("INSERT INTO events(seq,runId,body) VALUES(?,?,?)"); + for (const { runId, body } of batch) { + ins.run(body.seq, runId, JSON.stringify(body)); + this.chainAdvance("event", "events", "SELECT seq AS pos,runId,body FROM events WHERE seq>? AND seq<=? ORDER BY seq", body.seq, (r) => `${r.runId}:${r.pos}`); + } + }); + } catch (error2) { + this.volatileBuffer = [...batch, ...this.volatileBuffer]; + this.scheduleVolatileFlush(); + throw error2; + } + } events(runId, after = 0, limit = 150) { - return this.db.prepare("SELECT seq,body FROM events WHERE runId=? AND seq>? ORDER BY seq LIMIT ?").all(runId, after, limit).map((e) => ({ seq: e.seq, ...JSON.parse(e.body) })); + const persisted = this.db.prepare("SELECT seq,body FROM events WHERE runId=? AND seq>? ORDER BY seq LIMIT ?").all(runId, after, limit).map((e) => ({ seq: e.seq, ...JSON.parse(e.body) })); + const buffered = this.volatileBuffer.filter((e) => e.runId === runId && e.body.seq > after).map((e) => e.body); + return [...persisted, ...buffered].sort((a, b2) => a.seq - b2.seq).slice(0, limit); } rowHash(prev, kind, key, body) { return createHash("sha256").update(`${prev}:${kind}:${key}:${body}`).digest("hex"); @@ -7917,6 +7967,7 @@ var Store = class { } } close() { + this.flushVolatile(); this.db.close(); this.releaseLock(); } diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs index 5ac9fc16..c63a52ac 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs @@ -13,13 +13,27 @@ export class Store { unlinkSync(this.lock); this.fd=openSync(this.lock,'wx',0o600); } this.owner=randomUUID();this.txDepth=0; + // PERF: volatile event types are broadcast to live listeners immediately but + // only flushed to the events table on flushVolatile()/close(). Measured cost + // of one persisted progress event was ~960us under synchronous=FULL; a run + // emits ~1100 of them, so persisting each one is ~1s of pure stall for data + // that carries no replayable state. Durable types (step.queued/started/ + // finished, run.*) are still written synchronously so resume stays correct. + this.volatileTypes=new Set(['step.progress']); + this.volatileBuffer=[]; + this.volatileTimer=null; try { writeFileSync(this.fd,JSON.stringify({pid:process.pid,owner:this.owner})); this.db=new DatabaseSync(join(dir,'workflows.sqlite')); // The kernel-held SQLite lock is authoritative if stale lockfile reclamation // races with another starter. Keep it for this service connection's lifetime. this.db.exec('PRAGMA locking_mode=EXCLUSIVE; BEGIN EXCLUSIVE; COMMIT;'); - this.db.exec(`PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; + // PERF: synchronous=FULL fsyncs on every commit. With ~960us per commit on + // Windows NVMe, the per-event fsync dominated all other engine work. WAL + // + NORMAL still survives process crash (the common case); only a host + // power loss can lose the last few transactions, and the integrity chain + // is the authority for detecting that, not the fsync cadence. + this.db.exec(`PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL; PRAGMA busy_timeout=5000; CREATE TABLE IF NOT EXISTS templates(id TEXT PRIMARY KEY,body TEXT NOT NULL); CREATE TABLE IF NOT EXISTS settings(key TEXT PRIMARY KEY,body TEXT NOT NULL); CREATE TABLE IF NOT EXISTS runs(id TEXT PRIMARY KEY,requestId TEXT UNIQUE,requestHash TEXT NOT NULL,body TEXT NOT NULL); @@ -28,6 +42,8 @@ export class Store { CREATE TABLE IF NOT EXISTS events(seq INTEGER PRIMARY KEY AUTOINCREMENT,runId TEXT,body TEXT NOT NULL); CREATE INDEX IF NOT EXISTS run_events ON events(runId,seq); CREATE TABLE IF NOT EXISTS integrity_rows(surface TEXT NOT NULL,pos INTEGER NOT NULL,key TEXT NOT NULL,hash TEXT NOT NULL,PRIMARY KEY(surface,pos));`); + const eventSeq=this.db.prepare("SELECT MAX(COALESCE((SELECT seq FROM sqlite_sequence WHERE name='events'),0),COALESCE((SELECT MAX(seq) FROM events),0)) AS seq").get().seq; + this.nextEventSequence=Number(eventSeq??0); // Recovery must inspect every unfinished run, not just the dashboard page. const unfinished=this.db.prepare("SELECT body FROM runs WHERE json_extract(body,'$.status') IN ('running','queued','stopping','pausing')").all(); for(const row of unfinished) {const run=JSON.parse(row.body); @@ -56,8 +72,41 @@ export class Store { 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.transaction(()=>{const rowid=Number(this.db.prepare('INSERT INTO repair_cache VALUES(?,?,?)').run(runId,step.id,JSON.stringify(step)).lastInsertRowid);this.chainAdvance('repair','repair','SELECT rowid AS pos,runId,id,body FROM repair_cache WHERE rowid>? AND rowid<=? ORDER BY rowid',rowid,r=>`${r.runId}/${r.id}`);});} - event(runId,type,data={}) {const event={...data,type,time:Date.now()};return this.transaction(()=>{const seq=Number(this.db.prepare('INSERT INTO events(runId,body) VALUES(?,?)').run(runId,JSON.stringify(event)).lastInsertRowid);this.chainAdvance('event','events','SELECT seq AS pos,runId,body FROM events WHERE seq>? AND seq<=? ORDER BY seq',seq,r=>`${r.runId}:${r.pos}`);return {seq,...event};});} - events(runId,after=0,limit=150) {return this.db.prepare('SELECT seq,body FROM events WHERE runId=? AND seq>? ORDER BY seq LIMIT ?').all(runId,after,limit).map(e=>({seq:e.seq,...JSON.parse(e.body)}));} + event(runId,type,data={}) {const event={...data,type,time:Date.now()}; + if(this.volatileTypes.has(type))return this.stageVolatile(runId,event); + // A durable event is the flush barrier: everything buffered before it must + // reach the table first, so seq order stays monotonic on replay. + this.flushVolatile(); + return this.transaction(()=>{const seq=Number(this.db.prepare('INSERT INTO events(runId,body) VALUES(?,?)').run(runId,JSON.stringify(event)).lastInsertRowid);this.nextEventSequence=seq;this.chainAdvance('event','events','SELECT seq AS pos,runId,body FROM events WHERE seq>? AND seq<=? ORDER BY seq',seq,r=>`${r.runId}:${r.pos}`);return {seq,...event};});} + stageVolatile(runId,event){ + // Assign sequence at emission time so live notifications and long-poll reads + // refer to the same event even before its durable batch flush. + const staged={seq:++this.nextEventSequence,...event}; + this.volatileBuffer.push({runId,body:staged}); + if(this.volatileBuffer.length>=64)this.flushVolatile(); + else this.scheduleVolatileFlush(); + return staged; + } + scheduleVolatileFlush(){ + if(this.volatileTimer)return; + this.volatileTimer=setTimeout(()=>{this.volatileTimer=null;try{this.flushVolatile();}catch{/* batch restored; a later event/close retries */}},1000); + this.volatileTimer.unref?.(); + } + flushVolatile(){ + if(this.volatileTimer){clearTimeout(this.volatileTimer);this.volatileTimer=null;} + if(!this.volatileBuffer.length)return; + const batch=this.volatileBuffer;this.volatileBuffer=[]; + try{ + this.transaction(()=>{const ins=this.db.prepare('INSERT INTO events(seq,runId,body) VALUES(?,?,?)'); + for(const {runId,body} of batch){ins.run(body.seq,runId,JSON.stringify(body)); + this.chainAdvance('event','events','SELECT seq AS pos,runId,body FROM events WHERE seq>? AND seq<=? ORDER BY seq',body.seq,r=>`${r.runId}:${r.pos}`);}}); + }catch(error){this.volatileBuffer=[...batch,...this.volatileBuffer];this.scheduleVolatileFlush();throw error;} + } + events(runId,after=0,limit=150) { + const persisted=this.db.prepare('SELECT seq,body FROM events WHERE runId=? AND seq>? ORDER BY seq LIMIT ?').all(runId,after,limit).map(e=>({seq:e.seq,...JSON.parse(e.body)})); + const buffered=this.volatileBuffer.filter(e=>e.runId===runId&&e.body.seq>after).map(e=>e.body); + return [...persisted,...buffered].sort((a,b)=>a.seq-b.seq).slice(0,limit); + } rowHash(prev,kind,key,body) {return createHash('sha256').update(`${prev}:${kind}:${key}:${body}`).digest('hex');} // Bulk adoption of pre-existing rows is an initial-creation behavior only: it // anchors whatever the table held when the chain first appears. Once a head @@ -96,5 +145,5 @@ export class Store { repair:face('repair','repair','repair_cache','rowid','SELECT runId,id,body FROM repair_cache WHERE rowid=?',row=>`${row.runId}/${row.id}`)}; } releaseLock() {closeSync(this.fd);try{if(JSON.parse(readFileSync(this.lock,'utf8')).owner===this.owner)unlinkSync(this.lock);}catch{}} - close() {this.db.close();this.releaseLock();} + close() {this.flushVolatile();this.db.close();this.releaseLock();} } From a4f382a70d127bd93a8d4a4c59ef5b27d1668264 Mon Sep 17 00:00:00 2001 From: NKAntony777 <2212222@mail.nankai.edu.cn> Date: Fri, 9 Oct 2026 18:43:51 +0800 Subject: [PATCH 2/6] fix(workflows): preserve FULL durability and crash-safe batched event cursors --- .../mcode-dynamic-workflows/REVIEW-NOTES.md | 44 +++++ .../mcode-dynamic-workflows/REVIEW-REPLY.md | 20 +++ .../checks/cross-reuse-mcp.check.mjs | 6 +- .../checks/durability.check.mjs | 156 ++++++++++++++++++ .../fixtures/crash-before-progress-flush.mjs | 16 ++ .../checks/integrity.check.mjs | 2 +- .../mcode-dynamic-workflows/dist/main.mjs | 90 ++++++++-- .../scripts/benchmark-durable-batching.mjs | 34 ++++ .../mcode-dynamic-workflows/src/main.mjs | 6 +- .../mcode-dynamic-workflows/src/store.mjs | 51 ++++-- 10 files changed, 392 insertions(+), 33 deletions(-) create mode 100644 plugins/hetaoBackend/mcode-dynamic-workflows/REVIEW-NOTES.md create mode 100644 plugins/hetaoBackend/mcode-dynamic-workflows/REVIEW-REPLY.md create mode 100644 plugins/hetaoBackend/mcode-dynamic-workflows/checks/durability.check.mjs create mode 100644 plugins/hetaoBackend/mcode-dynamic-workflows/checks/fixtures/crash-before-progress-flush.mjs create mode 100644 plugins/hetaoBackend/mcode-dynamic-workflows/scripts/benchmark-durable-batching.mjs diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/REVIEW-NOTES.md b/plugins/hetaoBackend/mcode-dynamic-workflows/REVIEW-NOTES.md new file mode 100644 index 00000000..7a6ea888 --- /dev/null +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/REVIEW-NOTES.md @@ -0,0 +1,44 @@ +# PR #67: durable progress batching + +The change preserves immediate progress delivery while preventing acknowledged cursors from +being reused after a process crash. SQLite stays in WAL mode with `synchronous=FULL`. + +- Reserve 1024 sequence numbers in a separately committed high-watermark transaction before + exposing a cursor. Recovery skips unused numbers; gaps are valid. Reservation failures expose + no cursor. Lease exhaustion within an existing transaction fails rather than publishing an + unreserved sequence. +- Durable events flush preceding progress and keep their own synchronous transaction. +- Failed batch commits restore the ordered buffer and roll back ledger updates, including a + nested flush when its outer transaction fails. +- Shutdown attempts all cleanup. A failed flush still closes SQLite and releases the owner lock; + the service reports failure through stderr and exit status. +- The public MCP test now stops its detached service before removing the temporary directory, + fixing its Windows `EBUSY` cleanup failure without changing service lifetime behavior. + +## Local verification, October 9, 2026 + +Windows, Node 24.12.0: + +- `npm test`: **124 passed, 2 skipped, 0 failed** (126 tests). +- `node --test checks/integrity.check.mjs checks/durability.check.mjs test/package.test.mjs`: + **25 passed**. Includes a killed child whose memory-only progress was acknowledged by + `workflow_wait`, reopen/cursor checks, FULL pragmas, commit injection and failed-close cleanup. +- Package build regenerated `dist/main.mjs`; unrelated generated output was restored. + +## Repeatable performance comparison + +Run `node scripts/benchmark-durable-batching.mjs` from this plugin directory. +The baseline reads Store source from repository commit `6481e4a`. Both modes require FULL and +verify all 1100 event rows and the integrity chain after writing. Five rounds on the same host: + +| Mode | Times (ms) | Median (ms) | +| --- | --- | --- | +| FULL, commit every progress event | 2988.755, 2911.603, 2863.510, 3312.873, 3098.278 | 2988.755 | +| FULL, safe batching | 217.803, 198.446, 361.923, 163.422, 143.395 | 198.446 | + +This microbenchmark is **15.06x faster**. It measures event persistence overhead, not whole +workflow/model latency. Progress may disappear if the process dies before its batch is committed; +durable state retains FULL semantics and future sequence numbers exceed the lost acknowledged +cursor. SIGKILL/transaction injection is not a physical power-loss test. + +GitHub CI execution is separate from these local results; a pending approval is not a test pass. diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/REVIEW-REPLY.md b/plugins/hetaoBackend/mcode-dynamic-workflows/REVIEW-REPLY.md new file mode 100644 index 00000000..593c0d22 --- /dev/null +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/REVIEW-REPLY.md @@ -0,0 +1,20 @@ +Thanks for identifying the durability issues. SQLite WAL is restored to synchronous=FULL. +Progress batching remains, but sequences are now leased through an independently committed +high watermark before publishing a cursor. After a crash, unused lease values are skipped, +so workflow_wait(afterSequence) cannot skip future durable events due to cursor reuse. + +Failed batch commits restore the ordered buffer and roll back ledger writes, including nested +flushes in a failed outer transaction. Shutdown attempts every cleanup step; a flush error +still closes SQLite/releases the owner lock and is reported through stderr/exit status. + +Local verification: 124 tests passed, 2 platform skips; 25 focused integrity/durability/package +tests passed, including a killed process after workflow_wait acknowledged memory-only progress. +The Windows public MCP test also now stops its detached service before temporary-directory +cleanup. Build output is updated. + +The repeatable 1100-event, five-round comparison against commit 6481e4a checks FULL, row counts +and ledger integrity in both modes: median 2989 ms with per-event commits versus 198 ms with +safe batching (15.06x). This measures persistence overhead, not end-to-end workflow latency. +REVIEW-NOTES.md documents that progress can still be lost before batch commit, while cursors +are never reused and durable state retains FULL semantics. No physical power-loss or GitHub +CI pass is claimed from these local tests. 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 index a036360a..f3f0cf74 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/cross-reuse-mcp.check.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/cross-reuse-mcp.check.mjs @@ -1,4 +1,5 @@ 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'; +import {execFile} from 'node:child_process';import {promisify} from 'node:util'; 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'}); @@ -10,5 +11,8 @@ test('packaged MCP advertises reuseAcrossRuns and accepts it through the public 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});} + }finally{await client.close();await transport.close(); + // stdio is a client of a detached owner; release SQLite before Windows cleanup. + await promisify(execFile)(process.execPath,[resolve('dist/main.mjs'),'--stop-service','--settings',join(dir,'settings.json')],{windowsHide:true,timeout:15000}); + await rm(dir,{recursive:true,force:true,maxRetries:10,retryDelay:100});} }); diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/durability.check.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/durability.check.mjs new file mode 100644 index 00000000..6cf4a436 --- /dev/null +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/durability.check.mjs @@ -0,0 +1,156 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { fork } from 'node:child_process'; +import { once } from 'node:events'; +import { mkdtemp, rm, access } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { Store } from '../src/store.mjs'; +import { Engine } from '../src/engine.mjs'; +import { createToolHandler } from '../src/tools.mjs'; + +async function temporary(fn) { + const dir = await mkdtemp(join(tmpdir(), 'wf-durability-')); + try { await fn(dir); } finally { await rm(dir, { recursive: true, force: true }); } +} + +function failNextCommit(store) { + const exec = store.db.exec.bind(store.db); + let fail = true; + store.db.exec = sql => { + if (sql === 'COMMIT' && fail) { fail = false; throw new Error('injected commit failure'); } + return exec(sql); + }; +} + +test('kill-before-flush workflow_wait cursor is never reused after restart', { timeout: 15000 }, async () => { + await temporary(async dir => { + const child = fork(new URL('./fixtures/crash-before-progress-flush.mjs', import.meta.url), [dir], { silent: true }); + let stderr = ''; + child.stderr.on('data', data => { stderr += data; }); + const exited = once(child, 'exit'); + try { + const [message] = await Promise.race([ + once(child, 'message'), + exited.then(() => { throw new Error(`worker exited before cursor: ${stderr}`); }), + ]); + assert.equal(message.persisted, 1); + assert.equal(message.result.events[0].type, 'step.progress'); + const cursor = message.result.nextSequence; + child.kill('SIGKILL'); + await exited; + const store = new Store(dir); + const engine = new Engine(store, { workspace: dir }); + try { + const durable = engine.emitEvent('crash-run', 'run.finished', { status: 'done' }); + const result = await createToolHandler(engine, () => '')('workflow_wait', { + runId: 'crash-run', afterSequence: cursor, timeoutMs: 0, + }); + assert.equal(result.events.length, 1); + assert.equal(result.events[0].type, 'run.finished'); + assert.ok(result.nextSequence > cursor); + assert.equal(store.verifyIntegrity().events.verified, true); + } finally { await engine.close(); store.close(); } + } finally { if (child.exitCode === null && child.signalCode === null) { child.kill('SIGKILL'); await exited; } } + }); +}); + +test('FULL durability remains enabled for state, repair and batched event writes', async () => { + await temporary(async dir => { + const store = new Store(dir); + try { + assert.equal(store.db.prepare('PRAGMA synchronous').get().synchronous, 2); + store.save({ id: 'r', requestId: 'r', requestHash: 'h', status: 'succeeded' }); + store.saveRepairCandidate('r', { id: 'a', kind: 'agent' }); + store.event('r', 'run.created'); + store.event('r', 'step.progress'); + store.flushVolatile(); + assert.equal(store.db.prepare('PRAGMA synchronous').get().synchronous, 2); + assert.equal(store.verifyIntegrity().events.verified, true); + assert.equal(store.verifyIntegrity().repair.verified, true); + } finally { store.close(); } + }); +}); + +test('failed sequence reservation publishes no cursor and can be retried', async () => { + await temporary(async dir => { + const store = new Store(dir); + try { + failNextCommit(store); + assert.throws(() => store.event('r', 'step.progress'), /injected commit failure/); + assert.equal(store.volatileBuffer.length, 0); + assert.equal(store.nextEventSequence, 0); + assert.equal(store.setting('event_sequence_lease'), undefined); + assert.equal(store.event('r', 'step.progress').seq, 1); + } finally { store.close(); } + }); +}); + +test('batch commit failure rolls back the ledger and restores ordered events for retry', async () => { + await temporary(async dir => { + const store = new Store(dir); + try { + store.event('r', 'run.created'); + const heads = store.integrityHeads(); + store.event('r', 'step.progress'); + store.event('r', 'step.progress'); + failNextCommit(store); + assert.throws(() => store.flushVolatile(), /injected commit failure/); + assert.deepEqual(store.integrityHeads(), heads); + assert.equal(store.db.prepare('SELECT COUNT(*) AS n FROM events').get().n, 1); + assert.deepEqual(store.events('r').map(e => e.seq), [1, 2, 3]); + store.event('r', 'run.finished'); + assert.deepEqual(store.events('r').map(e => e.seq), [1, 2, 3, 4]); + assert.equal(store.verifyIntegrity().events.verified, true); + } finally { store.close(); } + }); +}); + +test('close releases SQLite and owner locks even when volatile flush fails', async () => { + await temporary(async dir => { + const store = new Store(dir); + store.event('r', 'run.created'); + const cursor = store.event('r', 'step.progress').seq; + failNextCommit(store); + assert.throws(() => store.close(), /injected commit failure/); + assert.equal(store.volatileTimer, null); + assert.equal(store.fd, undefined); + assert.throws(() => store.db.prepare('SELECT 1')); + await assert.rejects(access(join(dir, 'owner.lock')), { code: 'ENOENT' }); + assert.doesNotThrow(() => store.close()); + const reopened = new Store(dir); + try { + assert.ok(reopened.event('r', 'run.finished').seq > cursor); + assert.equal(reopened.verifyIntegrity().events.verified, true); + } finally { reopened.close(); } + }); +}); + +test('lease rollover keeps every progress cursor monotonic and ledger verified', async () => { + await temporary(async dir => { + const store = new Store(dir); + try { + for (let seq = 1; seq <= 2050; seq++) assert.equal(store.event('r', 'step.progress').seq, seq); + store.flushVolatile(); + assert.equal(store.setting('event_sequence_lease'), 3072); + assert.equal(store.verifyIntegrity().events.verified, true); + } finally { store.close(); } + }); +}); + +test('outer durable transaction rollback restores a nested progress flush', async () => { + await temporary(async dir => { + const store = new Store(dir); + try { + store.event('r', 'run.created'); + const progress = store.event('r', 'step.progress'); + failNextCommit(store); + assert.throws(() => store.transaction(() => store.event('r', 'run.finished')), /injected commit failure/); + assert.equal(store.db.prepare('SELECT COUNT(*) AS n FROM events').get().n, 1); + assert.equal(store.volatileBuffer[0].body.seq, progress.seq); + store.event('r', 'run.finished'); + assert.deepEqual(store.events('r').map(e => e.type), ['run.created', 'step.progress', 'run.finished']); + assert.equal(store.verifyIntegrity().events.verified, true); + } finally { store.close(); } + }); +}); diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/fixtures/crash-before-progress-flush.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/fixtures/crash-before-progress-flush.mjs new file mode 100644 index 00000000..b90cc00e --- /dev/null +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/fixtures/crash-before-progress-flush.mjs @@ -0,0 +1,16 @@ +import { Store } from '../../src/store.mjs'; +import { Engine } from '../../src/engine.mjs'; +import { createToolHandler } from '../../src/tools.mjs'; + +const store = new Store(process.argv[2]); +const engine = new Engine(store, { workspace: process.argv[2] }); +store.save({ id: 'crash-run', requestId: 'crash-run', requestHash: 'h', status: 'running', phases: [], script: 'return 1;', input: {}, executor: 'demo', workspace: process.argv[2] }); +engine.emitEvent('crash-run', 'run.created'); +engine.emitEvent('crash-run', 'step.progress', { stepId: 'a' }); +// The parent kills this process with an acknowledged, still-memory-only cursor. +clearTimeout(store.volatileTimer); +const result = await createToolHandler(engine, () => '')('workflow_wait', { + runId: 'crash-run', afterSequence: 1, timeoutMs: 0, +}); +process.send({ result, persisted: store.db.prepare('SELECT COUNT(*) AS n FROM events').get().n }); +setInterval(() => {}, 1000); diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/integrity.check.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/integrity.check.mjs index 6920d9b0..2b95e6ef 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/integrity.check.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/integrity.check.mjs @@ -47,7 +47,7 @@ test('event sequence resumes after buffered events are flushed and store is reop const dir=await mkdtemp(join(tmpdir(),'wf-event-seq-'));let store=new Store(dir);try{ store.event('r','run.created',{name:'first'});store.event('r','step.progress',{stepId:'a'});store.close(); store=new Store(dir);const next=store.event('r','run.finished',{status:'done'}); - assert.equal(next.seq,3);assert.deepEqual(store.events('r',0,10).map(e=>e.seq),[1,2,3]); + assert.ok(next.seq>2);assert.deepEqual(store.events('r',0,10).map(e=>e.seq),[1,2,next.seq]); assert.equal(store.verifyIntegrity().events.verified,true); }finally{store.close();await rm(dir,{recursive:true,force:true});} }); diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs index a84585c7..0a029a3f 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs @@ -7748,7 +7748,7 @@ var Store = class { writeFileSync(this.fd, JSON.stringify({ pid: process.pid, owner: this.owner })); this.db = new DatabaseSync(join(dir, "workflows.sqlite")); this.db.exec("PRAGMA locking_mode=EXCLUSIVE; BEGIN EXCLUSIVE; COMMIT;"); - this.db.exec(`PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL; PRAGMA busy_timeout=5000; + this.db.exec(`PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; PRAGMA busy_timeout=5000; CREATE TABLE IF NOT EXISTS templates(id TEXT PRIMARY KEY,body TEXT NOT NULL); CREATE TABLE IF NOT EXISTS settings(key TEXT PRIMARY KEY,body TEXT NOT NULL); CREATE TABLE IF NOT EXISTS runs(id TEXT PRIMARY KEY,requestId TEXT UNIQUE,requestHash TEXT NOT NULL,body TEXT NOT NULL); @@ -7758,7 +7758,8 @@ var Store = class { CREATE INDEX IF NOT EXISTS run_events ON events(runId,seq); CREATE TABLE IF NOT EXISTS integrity_rows(surface TEXT NOT NULL,pos INTEGER NOT NULL,key TEXT NOT NULL,hash TEXT NOT NULL,PRIMARY KEY(surface,pos));`); const eventSeq = this.db.prepare("SELECT MAX(COALESCE((SELECT seq FROM sqlite_sequence WHERE name='events'),0),COALESCE((SELECT MAX(seq) FROM events),0)) AS seq").get().seq; - this.nextEventSequence = Number(eventSeq ?? 0); + this.nextEventSequence = Math.max(Number(eventSeq ?? 0), this.setting("event_sequence_lease") ?? 0); + this.eventSequenceLimit = this.nextEventSequence; const unfinished = this.db.prepare("SELECT body FROM runs WHERE json_extract(body,'$.status') IN ('running','queued','stopping','pausing')").all(); for (const row of unfinished) { const run = JSON.parse(row.body); @@ -7774,18 +7775,45 @@ var Store = class { } transaction(fn) { if (this.txDepth) return fn(); - this.txDepth = 1; + this.reserveEventSequences(); this.db.exec("BEGIN IMMEDIATE"); + this.txDepth = 1; + this.txVolatile = []; try { const r = fn(); this.db.exec("COMMIT"); return r; } catch (e) { this.db.exec("ROLLBACK"); + const restored = new Map([...this.txVolatile, ...this.volatileBuffer].map((item) => [item.body.seq, item])); + this.volatileBuffer = [...restored.values()].sort((a, b2) => a.body.seq - b2.body.seq); + if (this.volatileBuffer.length) this.scheduleVolatileFlush(); throw e; } finally { this.txDepth = 0; + this.txVolatile = null; + } + } + reserveEventSequences() { + const persisted = Number(this.db.prepare("SELECT COALESCE((SELECT seq FROM sqlite_sequence WHERE name='events'),0) AS seq").get().seq); + this.nextEventSequence = Math.max(this.nextEventSequence, persisted); + if (this.nextEventSequence < this.eventSequenceLimit) return; + if (this.txDepth) throw new Error("Event sequence lease exhausted inside a transaction"); + const limit = this.nextEventSequence + 1024; + if (!Number.isSafeInteger(limit)) throw new Error("Event sequence exhausted"); + this.db.exec("BEGIN IMMEDIATE"); + try { + this.saveSetting("event_sequence_lease", limit); + this.db.exec("COMMIT"); + } catch (error2) { + this.db.exec("ROLLBACK"); + throw error2; } + this.eventSequenceLimit = limit; + } + allocateEventSequence() { + this.reserveEventSequences(); + return ++this.nextEventSequence; } templates() { return this.db.prepare("SELECT body FROM templates ORDER BY rowid DESC").all().map((r) => JSON.parse(r.body)); @@ -7856,21 +7884,21 @@ var Store = class { if (this.volatileTypes.has(type)) return this.stageVolatile(runId, event); this.flushVolatile(); return this.transaction(() => { - const seq = Number(this.db.prepare("INSERT INTO events(runId,body) VALUES(?,?)").run(runId, JSON.stringify(event)).lastInsertRowid); - this.nextEventSequence = seq; + const seq = this.allocateEventSequence(); + this.db.prepare("INSERT INTO events(seq,runId,body) VALUES(?,?,?)").run(seq, runId, JSON.stringify(event)); this.chainAdvance("event", "events", "SELECT seq AS pos,runId,body FROM events WHERE seq>? AND seq<=? ORDER BY seq", seq, (r) => `${r.runId}:${r.pos}`); return { seq, ...event }; }); } stageVolatile(runId, event) { - const staged = { seq: ++this.nextEventSequence, ...event }; + const staged = { ...event, seq: this.allocateEventSequence() }; this.volatileBuffer.push({ runId, body: staged }); if (this.volatileBuffer.length >= 64) this.flushVolatile(); else this.scheduleVolatileFlush(); return staged; } scheduleVolatileFlush() { - if (this.volatileTimer) return; + if (this.volatileTimer || this.closing) return; this.volatileTimer = setTimeout(() => { this.volatileTimer = null; try { @@ -7888,6 +7916,7 @@ var Store = class { if (!this.volatileBuffer.length) return; const batch = this.volatileBuffer; this.volatileBuffer = []; + if (this.txDepth) this.txVolatile.push(...batch); try { this.transaction(() => { const ins = this.db.prepare("INSERT INTO events(seq,runId,body) VALUES(?,?,?)"); @@ -7960,16 +7989,38 @@ var Store = class { }; } releaseLock() { - closeSync(this.fd); + if (this.fd === void 0) return; try { - if (JSON.parse(readFileSync(this.lock, "utf8")).owner === this.owner) unlinkSync(this.lock); - } catch { + closeSync(this.fd); + } finally { + this.fd = void 0; + try { + if (JSON.parse(readFileSync(this.lock, "utf8")).owner === this.owner) unlinkSync(this.lock); + } catch { + } } } close() { - this.flushVolatile(); - this.db.close(); - this.releaseLock(); + if (this.closing) return; + this.closing = true; + const errors = []; + try { + this.flushVolatile(); + } catch (error2) { + errors.push(error2); + } + try { + this.db.close(); + } catch (error2) { + errors.push(error2); + } + try { + this.releaseLock(); + } catch (error2) { + errors.push(error2); + } + if (errors.length === 1) throw errors[0]; + if (errors.length) throw new AggregateError(errors, "Store shutdown failed"); } }; @@ -27830,10 +27881,15 @@ if (values.stdio && process.env.MCODE_WORKFLOW_CHILD === "1") { async function close() { if (closing) return; closing = true; - await engine.close(); - await panel.close(); - store.close(); - process.exitCode = 0; + const errors = []; + for (const cleanup of [() => engine.close(), () => panel.close(), () => store.close()]) try { + await cleanup(); + } catch (error2) { + errors.push(error2); + } + process.exitCode = errors.length ? 1 : 0; + if (errors.length) process.stderr.write(`Workflow shutdown failed: ${errors.map((e) => e.message).join("; ")} +`); } process.once("SIGINT", () => void close()); process.once("SIGTERM", () => void close()); diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/scripts/benchmark-durable-batching.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/scripts/benchmark-durable-batching.mjs new file mode 100644 index 00000000..89be9412 --- /dev/null +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/scripts/benchmark-durable-batching.mjs @@ -0,0 +1,34 @@ +import assert from 'node:assert/strict'; +import { execFileSync } from 'node:child_process'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { performance } from 'node:perf_hooks'; +import { Store } from '../src/store.mjs'; + +const file='plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs'; +const reference=process.argv[2]??'6481e4a'; +const source=execFileSync('git',['show',`${reference}:${file}`],{encoding:'utf8'}); +const {Store:Baseline}=await import(`data:text/javascript;base64,${Buffer.from(source).toString('base64')}`); +const rounds=5, count=1100; +const median=values=>[...values].sort((a,b)=>a-b)[Math.floor(values.length/2)]; +const result={baseline:reference,events:count,rounds}; +for(const [name,Constructor] of [['full_per_event',Baseline],['full_batched',Store]]){ + const times=[]; + for(let round=0;roundengine.close(),()=>panel.close(),()=>store.close()])try{await cleanup();}catch(error){errors.push(error);} + process.exitCode=errors.length?1:0; + if(errors.length)process.stderr.write(`Workflow shutdown failed: ${errors.map(e=>e.message).join('; ')}\n`); + } process.once('SIGINT',()=>void close());process.once('SIGTERM',()=>void close()); } } diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs index c63a52ac..44d115b8 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs @@ -28,12 +28,9 @@ export class Store { // The kernel-held SQLite lock is authoritative if stale lockfile reclamation // races with another starter. Keep it for this service connection's lifetime. this.db.exec('PRAGMA locking_mode=EXCLUSIVE; BEGIN EXCLUSIVE; COMMIT;'); - // PERF: synchronous=FULL fsyncs on every commit. With ~960us per commit on - // Windows NVMe, the per-event fsync dominated all other engine work. WAL - // + NORMAL still survives process crash (the common case); only a host - // power loss can lose the last few transactions, and the integrity chain - // is the authority for detecting that, not the fsync cadence. - this.db.exec(`PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL; PRAGMA busy_timeout=5000; + // Batching reduces commit count without weakening durable workflow writes. + // A hash chain cannot detect loss of a complete row/head transaction. + this.db.exec(`PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; PRAGMA busy_timeout=5000; CREATE TABLE IF NOT EXISTS templates(id TEXT PRIMARY KEY,body TEXT NOT NULL); CREATE TABLE IF NOT EXISTS settings(key TEXT PRIMARY KEY,body TEXT NOT NULL); CREATE TABLE IF NOT EXISTS runs(id TEXT PRIMARY KEY,requestId TEXT UNIQUE,requestHash TEXT NOT NULL,body TEXT NOT NULL); @@ -43,7 +40,8 @@ export class Store { CREATE INDEX IF NOT EXISTS run_events ON events(runId,seq); CREATE TABLE IF NOT EXISTS integrity_rows(surface TEXT NOT NULL,pos INTEGER NOT NULL,key TEXT NOT NULL,hash TEXT NOT NULL,PRIMARY KEY(surface,pos));`); const eventSeq=this.db.prepare("SELECT MAX(COALESCE((SELECT seq FROM sqlite_sequence WHERE name='events'),0),COALESCE((SELECT MAX(seq) FROM events),0)) AS seq").get().seq; - this.nextEventSequence=Number(eventSeq??0); + this.nextEventSequence=Math.max(Number(eventSeq??0),this.setting('event_sequence_lease')??0); + this.eventSequenceLimit=this.nextEventSequence; // Recovery must inspect every unfinished run, not just the dashboard page. const unfinished=this.db.prepare("SELECT body FROM runs WHERE json_extract(body,'$.status') IN ('running','queued','stopping','pausing')").all(); for(const row of unfinished) {const run=JSON.parse(row.body); @@ -51,7 +49,25 @@ export class Store { } }catch(error){this.db?.close();this.releaseLock();throw error;} } - transaction(fn) {if(this.txDepth)return fn();this.txDepth=1;this.db.exec('BEGIN IMMEDIATE');try{const r=fn();this.db.exec('COMMIT');return r;}catch(e){this.db.exec('ROLLBACK');throw e;}finally{this.txDepth=0;}} + transaction(fn) {if(this.txDepth)return fn();this.reserveEventSequences();this.db.exec('BEGIN IMMEDIATE');this.txDepth=1;this.txVolatile=[]; + try{const r=fn();this.db.exec('COMMIT');return r;} + catch(e){this.db.exec('ROLLBACK');const restored=new Map([...this.txVolatile,...this.volatileBuffer].map(item=>[item.body.seq,item]));this.volatileBuffer=[...restored.values()].sort((a,b)=>a.body.seq-b.body.seq);if(this.volatileBuffer.length)this.scheduleVolatileFlush();throw e;} + finally{this.txDepth=0;this.txVolatile=null;}} + reserveEventSequences() { + const persisted=Number(this.db.prepare("SELECT COALESCE((SELECT seq FROM sqlite_sequence WHERE name='events'),0) AS seq").get().seq); + this.nextEventSequence=Math.max(this.nextEventSequence,persisted); + if(this.nextEventSequenceJSON.parse(r.body));} template(id) {const r=this.db.prepare('SELECT body FROM templates WHERE id=?').get(id);return r?JSON.parse(r.body):null;} saveTemplate(value) {this.db.prepare('INSERT INTO templates VALUES(?,?)').run(value.id,JSON.stringify(value));} @@ -77,18 +93,18 @@ export class Store { // A durable event is the flush barrier: everything buffered before it must // reach the table first, so seq order stays monotonic on replay. this.flushVolatile(); - return this.transaction(()=>{const seq=Number(this.db.prepare('INSERT INTO events(runId,body) VALUES(?,?)').run(runId,JSON.stringify(event)).lastInsertRowid);this.nextEventSequence=seq;this.chainAdvance('event','events','SELECT seq AS pos,runId,body FROM events WHERE seq>? AND seq<=? ORDER BY seq',seq,r=>`${r.runId}:${r.pos}`);return {seq,...event};});} + return this.transaction(()=>{const seq=this.allocateEventSequence();this.db.prepare('INSERT INTO events(seq,runId,body) VALUES(?,?,?)').run(seq,runId,JSON.stringify(event));this.chainAdvance('event','events','SELECT seq AS pos,runId,body FROM events WHERE seq>? AND seq<=? ORDER BY seq',seq,r=>`${r.runId}:${r.pos}`);return {seq,...event};});} stageVolatile(runId,event){ // Assign sequence at emission time so live notifications and long-poll reads // refer to the same event even before its durable batch flush. - const staged={seq:++this.nextEventSequence,...event}; + const staged={...event,seq:this.allocateEventSequence()}; this.volatileBuffer.push({runId,body:staged}); if(this.volatileBuffer.length>=64)this.flushVolatile(); else this.scheduleVolatileFlush(); return staged; } scheduleVolatileFlush(){ - if(this.volatileTimer)return; + if(this.volatileTimer||this.closing)return; this.volatileTimer=setTimeout(()=>{this.volatileTimer=null;try{this.flushVolatile();}catch{/* batch restored; a later event/close retries */}},1000); this.volatileTimer.unref?.(); } @@ -96,6 +112,7 @@ export class Store { if(this.volatileTimer){clearTimeout(this.volatileTimer);this.volatileTimer=null;} if(!this.volatileBuffer.length)return; const batch=this.volatileBuffer;this.volatileBuffer=[]; + if(this.txDepth)this.txVolatile.push(...batch); try{ this.transaction(()=>{const ins=this.db.prepare('INSERT INTO events(seq,runId,body) VALUES(?,?,?)'); for(const {runId,body} of batch){ins.run(body.seq,runId,JSON.stringify(body)); @@ -144,6 +161,14 @@ export class Store { return {events:face('event','events','events','seq','SELECT runId,body FROM events WHERE seq=?',(row,pos)=>`${row.runId}:${pos}`), repair:face('repair','repair','repair_cache','rowid','SELECT runId,id,body FROM repair_cache WHERE rowid=?',row=>`${row.runId}/${row.id}`)}; } - releaseLock() {closeSync(this.fd);try{if(JSON.parse(readFileSync(this.lock,'utf8')).owner===this.owner)unlinkSync(this.lock);}catch{}} - close() {this.flushVolatile();this.db.close();this.releaseLock();} + releaseLock() {if(this.fd===undefined)return;try{closeSync(this.fd);}finally{this.fd=undefined;try{if(JSON.parse(readFileSync(this.lock,'utf8')).owner===this.owner)unlinkSync(this.lock);}catch{}}} + close() { + if(this.closing)return;this.closing=true; + const errors=[]; + try{this.flushVolatile();}catch(error){errors.push(error);} + try{this.db.close();}catch(error){errors.push(error);} + try{this.releaseLock();}catch(error){errors.push(error);} + if(errors.length===1)throw errors[0]; + if(errors.length)throw new AggregateError(errors,'Store shutdown failed'); + } } From 8d57c9a34ee285fc4d65f08a98ac8be06b6e815e Mon Sep 17 00:00:00 2001 From: NKAntony777 <2212222@mail.nankai.edu.cn> Date: Sat, 10 Oct 2026 11:02:49 +0800 Subject: [PATCH 3/6] fix(workflows): reject unsafe event cursors and reconcile ambiguous COMMIT workflow_wait validated its cursor with Number.isInteger, which accepts 1e100. A packaged MCP call with afterSequence: 1e100 returned nextSequence: 1e100 and no events, and every real future sequence was then filtered out for that client permanently. Require Number.isSafeInteger and give the schema an explicit Number.MAX_SAFE_INTEGER maximum; the regression covers 1e100, MAX_SAFE_INTEGER + 1, Infinity and a negative cursor. flushVolatile restored the whole batch on any exception, so a COMMIT that reached SQLite but reported an error afterwards left the same rows both persisted and buffered: events() exposed duplicate sequence numbers and a later close() failed with UNIQUE constraint failed: events.seq. A thrown error is not evidence the rows are absent, so the table now decides. restoreVolatile merges keyed by seq and drops rows already present; the probe is skipped inside an enclosing transaction, where the rows are only visible and an outer ROLLBACK would otherwise drop them silently. Adds after-COMMIT ambiguity and unsafe-cursor regressions. 126 checks pass, and the packaged MCP test passes when run standalone: the two suites must not run concurrently because they contend for the same dashboard port. --- .../checks/durability.check.mjs | 62 ++++++++++++++++++- .../mcode-dynamic-workflows/dist/main.mjs | 31 +++++++--- .../mcode-dynamic-workflows/src/store.mjs | 32 +++++++++- .../mcode-dynamic-workflows/src/tools.mjs | 4 +- 4 files changed, 117 insertions(+), 12 deletions(-) diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/durability.check.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/durability.check.mjs index 6cf4a436..2b030256 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/durability.check.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/durability.check.mjs @@ -7,7 +7,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { Store } from '../src/store.mjs'; import { Engine } from '../src/engine.mjs'; -import { createToolHandler } from '../src/tools.mjs'; +import { createToolHandler, TOOLS } from '../src/tools.mjs'; async function temporary(fn) { const dir = await mkdtemp(join(tmpdir(), 'wf-durability-')); @@ -154,3 +154,63 @@ test('outer durable transaction rollback restores a nested progress flush', asyn } finally { store.close(); } }); }); + +// The write reached SQLite but the caller still saw a failure afterwards. The +// outcome is only knowable from the table, so a thrown error must not restore +// rows that are already durable — that is what produced duplicate sequences and +// the later UNIQUE constraint failure during close(). +test('after-COMMIT failure does not duplicate rows that already reached SQLite', async () => { + await temporary(async dir => { + const store = new Store(dir); + try { + store.event('r', 'run.created'); + store.event('r', 'step.progress'); + const second = store.event('r', 'step.progress'); + // Commit lands durably, then the failure is reported to the caller. + const exec = store.db.exec.bind(store.db); + let once = true; + store.db.exec = sql => { + const result = exec(sql); + if (sql === 'COMMIT' && once) { once = false; throw new Error('injected post-commit failure'); } + return result; + }; + assert.throws(() => store.flushVolatile(), /injected post-commit failure/); + store.db.exec = exec; + const rows = store.db.prepare('SELECT seq FROM events ORDER BY seq').all().map(r => r.seq); + assert.deepEqual(rows, [1, 2, 3]); + assert.deepEqual(store.events('r').map(e => e.seq), [1, 2, 3]); + assert.equal(store.volatileBuffer.filter(item => rows.includes(item.body.seq)).length, 0); + // A later flush must not hit UNIQUE constraint failed: events.seq. + assert.doesNotThrow(() => store.flushVolatile()); + assert.doesNotThrow(() => store.close()); + const reopened = new Store(dir); + try { + assert.deepEqual(reopened.events('r').map(e => e.seq), [1, 2, 3]); + assert.equal(reopened.verifyIntegrity().events.verified, true); + } finally { reopened.close(); } + } finally { store.close(); } + }); +}); + +test('workflow_wait rejects cursors that would skip every future event', async () => { + await temporary(async dir => { + const store = new Store(dir); + const engine = new Engine(store, { workspace: dir }); + try { + store.save({ id: 'cursor-run', requestId: 'cursor-run', requestHash: 'h', status: 'running', phases: [], script: 'return 1;', input: {}, executor: 'demo', workspace: dir }); + engine.emitEvent('cursor-run', 'run.created'); + store.flushVolatile(); + const call = createToolHandler(engine, () => ''); + const wait = schema => TOOLS.find(t => t.name === 'workflow_wait').inputSchema.properties[schema]; + assert.equal(wait('afterSequence').maximum, Number.MAX_SAFE_INTEGER); + for (const cursor of [1e100, Number.MAX_SAFE_INTEGER + 1, Infinity, -1]) { + await assert.rejects(call('workflow_wait', { runId: 'cursor-run', afterSequence: cursor, timeoutMs: 0 }), /等待参数无效/); + } + const accepted = await call('workflow_wait', { runId: 'cursor-run', afterSequence: Number.MAX_SAFE_INTEGER, timeoutMs: 0 }); + assert.equal(accepted.nextSequence, Number.MAX_SAFE_INTEGER); + engine.emitEvent('cursor-run', 'run.finished'); + const live = await call('workflow_wait', { runId: 'cursor-run', afterSequence: 1, timeoutMs: 0 }); + assert.equal(live.events.at(-1).type, 'run.finished'); + } finally { await engine.close(); store.close(); } + }); +}); diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs index 0a029a3f..bb7a7afe 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs @@ -7784,16 +7784,33 @@ var Store = class { this.db.exec("COMMIT"); return r; } catch (e) { - this.db.exec("ROLLBACK"); - const restored = new Map([...this.txVolatile, ...this.volatileBuffer].map((item) => [item.body.seq, item])); - this.volatileBuffer = [...restored.values()].sort((a, b2) => a.body.seq - b2.body.seq); - if (this.volatileBuffer.length) this.scheduleVolatileFlush(); + let open4 = true; + try { + this.db.exec("ROLLBACK"); + } catch { + open4 = false; + } + this.restoreVolatile([...this.txVolatile, ...this.volatileBuffer], !open4); throw e; } finally { this.txDepth = 0; this.txVolatile = null; } } + // A COMMIT can fail after SQLite has already made the rows durable, so a + // thrown error is not evidence that the batch is absent. The events table — + // not the control flow — decides what still needs retrying: re-emitting an + // already persisted row duplicates its sequence number and later fails the + // events.seq primary key during close(). Keyed merge keeps one row per seq. + // Inside an enclosing transaction the probe is skipped: those rows are only + // visible, not committed, and the outer ROLLBACK would silently drop them. + restoreVolatile(items, probe = !this.txDepth) { + if (!items.length) return; + const exists = probe ? this.db.prepare("SELECT 1 AS present FROM events WHERE seq=?") : null; + const restored = new Map([...items, ...this.volatileBuffer].filter((item) => !exists?.get(item.body.seq)).map((item) => [item.body.seq, item])); + this.volatileBuffer = [...restored.values()].sort((a, b2) => a.body.seq - b2.body.seq); + if (this.volatileBuffer.length) this.scheduleVolatileFlush(); + } reserveEventSequences() { const persisted = Number(this.db.prepare("SELECT COALESCE((SELECT seq FROM sqlite_sequence WHERE name='events'),0) AS seq").get().seq); this.nextEventSequence = Math.max(this.nextEventSequence, persisted); @@ -7926,7 +7943,7 @@ var Store = class { } }); } catch (error2) { - this.volatileBuffer = [...batch, ...this.volatileBuffer]; + this.restoreVolatile(batch, this.txDepth ? false : void 0); this.scheduleVolatileFlush(); throw error2; } @@ -26658,7 +26675,7 @@ var TOOLS = [ { 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\uFF0C\u9ED8\u8BA4\u8FD4\u56DE\u6570\u7EC4\uFF08\u65E2\u6709\u5F62\u72B6\u4E0D\u53D8\uFF09\u3002verifyIntegrity:true \u65F6\u6539\u8FD4 {runs,integrityHeads,integrity} \u5BF9\u8C61\u5F62\u5E76\u5168\u91CF\u91CD\u7B97\u4E24\u6761\u5B8C\u6574\u6027\u94FE\uFF1B\u6821\u9A8C\u8986\u76D6\u5DF2\u951A\u5B9A\u524D\u7F00\uFF0C\u4EFB\u4F55\u672A\u951A\u5B9A\u884C fail-closed\uFF08unchained>0 \u5373 verified:false\uFF09\u3002\u5B8C\u6574\u6027\u662F\u5E93\u5185\u7BE1\u6539\u8BC1\u636E\u4FE1\u53F7\uFF0C\u4E0D\u662F\u5BF9\u6297\u80FD\u91CD\u5199\u6574\u5E93\u8005\u7684\u4FE1\u4EFB\u951A\u3002", inputSchema: obj({ ...id, verifyIntegrity: { type: "boolean", description: "\u5168\u91CF\u91CD\u7B97\u5B8C\u6574\u6027\u94FE\uFF0C\u8FD4\u56DE {runs,integrityHeads,integrity} \u5BF9\u8C61\u5F62\uFF08\u9ED8\u8BA4\u4E3A\u7EAF\u6570\u7EC4\uFF09" } }) }, { 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"]) }, - { name: "workflow_wait", description: "\u6309\u4E8B\u4EF6\u6E38\u6807\u7B49\u5F85\u53D8\u5316\uFF0C\u6700\u957F25\u79D2\u3002\u9700\u8981\u7EE7\u7EED\u5173\u6CE8\u65F6\u4F7F\u7528\u8FD4\u56DE\u7684nextSequence\u3002", inputSchema: obj({ ...id, afterSequence: { type: "integer", minimum: 0 }, timeoutMs: { type: "integer", minimum: 0, maximum: 25e3 } }, ["runId"]) }, + { name: "workflow_wait", description: "\u6309\u4E8B\u4EF6\u6E38\u6807\u7B49\u5F85\u53D8\u5316\uFF0C\u6700\u957F25\u79D2\u3002\u9700\u8981\u7EE7\u7EED\u5173\u6CE8\u65F6\u4F7F\u7528\u8FD4\u56DE\u7684nextSequence\u3002", inputSchema: obj({ ...id, afterSequence: { type: "integer", minimum: 0, maximum: Number.MAX_SAFE_INTEGER }, timeoutMs: { type: "integer", minimum: 0, maximum: 25e3 } }, ["runId"]) }, { name: "workflow_cancel", description: "\u53D6\u6D88\u672C\u63D2\u4EF6\u5DE5\u4F5C\u6D41\uFF0C\u7B49\u5F85\u5728\u9014 exec \u9000\u51FA\uFF1B\u4E0D\u53D6\u6D88\u5176\u4ED6 MCode \u4F1A\u8BDD\u3002", inputSchema: obj(id, ["runId"]) }, { name: "workflow_pause", description: "\u505C\u6B62\u6D3E\u53D1\u5E76\u4E2D\u65AD\u5728\u9014\u8C03\u7528\uFF0C\u4FDD\u7559\u5DF2\u5B8C\u6210\u8282\u70B9\uFF0C\u53EF\u6062\u590D\u3002", inputSchema: obj(id, ["runId"]) }, { name: "workflow_resume", description: "\u539F\u811A\u672C\u4E0E\u539F\u8F93\u5165\u6062\u590D\uFF0C\u590D\u7528\u5DF2\u6210\u529F\u8282\u70B9\u3002\u53EF\u8C03\u6574 maxSteps/stepTimeoutMs/runTimeoutMs/maxCalls \u540E\u91CD\u8BD5\uFF0C\u6210\u529F\u8282\u70B9\u590D\u7528\uFF0C\u5931\u8D25\u8282\u70B9\u4ECE\u5934\u6267\u884C\u3002\u5F02\u5E38\u9000\u51FA\u9700\u8981\u7528\u6237\u5148\u786E\u8BA4\u65E7 Agent \u5DF2\u505C\u6B62\u3002", inputSchema: obj({ ...id, confirmStopped: { type: "boolean" }, ...LIMIT_SCHEMAS, maxCalls: { type: "integer", minimum: 1, maximum: 100 } }, ["runId"]) }, @@ -26710,7 +26727,7 @@ function createToolHandler(engine, getURL) { } case "workflow_wait": { const t = args.timeoutMs ?? 25e3, a = args.afterSequence ?? 0; - check(Number.isInteger(t) && t >= 0 && t <= 25e3 && Number.isInteger(a) && a >= 0, "\u7B49\u5F85\u53C2\u6570\u65E0\u6548"); + check(Number.isSafeInteger(t) && t >= 0 && t <= 25e3 && Number.isSafeInteger(a) && a >= 0, "\u7B49\u5F85\u53C2\u6570\u65E0\u6548"); return waitEvents(engine, args.runId, a, t); } case "workflow_cancel": diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs index 44d115b8..ab5452c0 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs @@ -51,8 +51,31 @@ export class Store { } transaction(fn) {if(this.txDepth)return fn();this.reserveEventSequences();this.db.exec('BEGIN IMMEDIATE');this.txDepth=1;this.txVolatile=[]; try{const r=fn();this.db.exec('COMMIT');return r;} - catch(e){this.db.exec('ROLLBACK');const restored=new Map([...this.txVolatile,...this.volatileBuffer].map(item=>[item.body.seq,item]));this.volatileBuffer=[...restored.values()].sort((a,b)=>a.body.seq-b.body.seq);if(this.volatileBuffer.length)this.scheduleVolatileFlush();throw e;} + catch(e){ + // A committed transaction cannot be rolled back, so a failing ROLLBACK is + // the signal that COMMIT may already have reached the table. Only then is + // the events table consulted before re-queueing anything. + let open=true; + try{this.db.exec('ROLLBACK');}catch{open=false;} + this.restoreVolatile([...this.txVolatile,...this.volatileBuffer],!open); + throw e;} finally{this.txDepth=0;this.txVolatile=null;}} + // A COMMIT can fail after SQLite has already made the rows durable, so a + // thrown error is not evidence that the batch is absent. The events table — + // not the control flow — decides what still needs retrying: re-emitting an + // already persisted row duplicates its sequence number and later fails the + // events.seq primary key during close(). Keyed merge keeps one row per seq. + // Inside an enclosing transaction the probe is skipped: those rows are only + // visible, not committed, and the outer ROLLBACK would silently drop them. + restoreVolatile(items,probe=!this.txDepth){ + if(!items.length)return; + const exists=probe?this.db.prepare('SELECT 1 AS present FROM events WHERE seq=?'):null; + const restored=new Map([...items,...this.volatileBuffer] + .filter(item=>!exists?.get(item.body.seq)) + .map(item=>[item.body.seq,item])); + this.volatileBuffer=[...restored.values()].sort((a,b)=>a.body.seq-b.body.seq); + if(this.volatileBuffer.length)this.scheduleVolatileFlush(); + } reserveEventSequences() { const persisted=Number(this.db.prepare("SELECT COALESCE((SELECT seq FROM sqlite_sequence WHERE name='events'),0) AS seq").get().seq); this.nextEventSequence=Math.max(this.nextEventSequence,persisted); @@ -117,7 +140,12 @@ export class Store { this.transaction(()=>{const ins=this.db.prepare('INSERT INTO events(seq,runId,body) VALUES(?,?,?)'); for(const {runId,body} of batch){ins.run(body.seq,runId,JSON.stringify(body)); this.chainAdvance('event','events','SELECT seq AS pos,runId,body FROM events WHERE seq>? AND seq<=? ORDER BY seq',body.seq,r=>`${r.runId}:${r.pos}`);}}); - }catch(error){this.volatileBuffer=[...batch,...this.volatileBuffer];this.scheduleVolatileFlush();throw error;} + }catch(error){ + // transaction() already reconciled the ambiguous case; re-check the table + // here so a batch whose rows did reach SQLite is never re-emitted. + this.restoreVolatile(batch,this.txDepth?false:undefined); + this.scheduleVolatileFlush(); + throw error;} } events(runId,after=0,limit=150) { const persisted=this.db.prepare('SELECT seq,body FROM events WHERE runId=? AND seq>? ORDER BY seq LIMIT ?').all(runId,after,limit).map(e=>({seq:e.seq,...JSON.parse(e.body)})); diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/src/tools.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/src/tools.mjs index 09731d5f..c8bda2f8 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/src/tools.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/src/tools.mjs @@ -15,7 +15,7 @@ export const TOOLS=[ {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 时列出最近运行，默认返回数组（既有形状不变）。verifyIntegrity:true 时改返 {runs,integrityHeads,integrity} 对象形并全量重算两条完整性链；校验覆盖已锚定前缀，任何未锚定行 fail-closed（unchained>0 即 verified:false）。完整性是库内篡改证据信号，不是对抗能重写整库者的信任锚。',inputSchema:obj({...id,verifyIntegrity:{type:'boolean',description:'全量重算完整性链，返回 {runs,integrityHeads,integrity} 对象形（默认为纯数组）'}})}, {name:'workflow_results',description:'分页读取节点结果；终态报告与失败明确分开。',inputSchema:obj({...id,includeDefinition:{type:'boolean'},offset:{type:'integer',minimum:0},limit:{type:'integer',minimum:1,maximum:20}},['runId'])}, - {name:'workflow_wait',description:'按事件游标等待变化，最长25秒。需要继续关注时使用返回的nextSequence。',inputSchema:obj({...id,afterSequence:{type:'integer',minimum:0},timeoutMs:{type:'integer',minimum:0,maximum:25000}},['runId'])}, + {name:'workflow_wait',description:'按事件游标等待变化，最长25秒。需要继续关注时使用返回的nextSequence。',inputSchema:obj({...id,afterSequence:{type:'integer',minimum:0,maximum:Number.MAX_SAFE_INTEGER},timeoutMs:{type:'integer',minimum:0,maximum:25000}},['runId'])}, {name:'workflow_cancel',description:'取消本插件工作流，等待在途 exec 退出；不取消其他 MCode 会话。',inputSchema:obj(id,['runId'])}, {name:'workflow_pause',description:'停止派发并中断在途调用，保留已完成节点，可恢复。',inputSchema:obj(id,['runId'])}, {name:'workflow_resume',description:'原脚本与原输入恢复，复用已成功节点。可调整 maxSteps/stepTimeoutMs/runTimeoutMs/maxCalls 后重试，成功节点复用，失败节点从头执行。异常退出需要用户先确认旧 Agent 已停止。',inputSchema:obj({...id,confirmStopped:{type:'boolean'},...LIMIT_SCHEMAS,maxCalls:{type:'integer',minimum:1,maximum:100}},['runId'])}, @@ -31,7 +31,7 @@ export function createToolHandler(engine,getURL){return async(name,args={})=>{ case 'workflow_repair':return summary(await engine.repair(args.runId,args)); case 'workflow_status':if(args.runId)return summary(engine.snapshot(args.runId));if(args.verifyIntegrity===true)return {runs:engine.store.list().map(r=>summary(r)),integrityHeads:engine.store.integrityHeads(),integrity:engine.store.verifyIntegrity()};return engine.store.list().map(r=>summary(r)); case 'workflow_results':{const r=engine.snapshot(args.runId);const offset=args.offset??0,limit=args.limit??10;check(Number.isInteger(offset)&&offset>=0&&Number.isInteger(limit)&&limit>=1&&limit<=20,'分页参数无效');return {status:r.status,updatedAt:r.updatedAt,...(args.includeDefinition?{definition:templateDefinition(r)}:{}),result:r.result,steps:r.steps.slice(offset,offset+limit).map(({prompt,input,...s})=>s),nextOffset:offset+limit=0&&t<=25000&&Number.isInteger(a)&&a>=0,'等待参数无效');return waitEvents(engine,args.runId,a,t);} + case 'workflow_wait':{const t=args.timeoutMs??25000,a=args.afterSequence??0;check(Number.isSafeInteger(t)&&t>=0&&t<=25000&&Number.isSafeInteger(a)&&a>=0,'等待参数无效');return waitEvents(engine,args.runId,a,t);} case 'workflow_cancel':return summary(await engine.stop(args.runId)); case 'workflow_pause':return summary(await engine.stop(args.runId,'paused')); case 'workflow_resume':return summary(await engine.resume(args.runId,args)); From 21a7f0dc5fb6f946dd22f49b3102679432053d81 Mon Sep 17 00:00:00 2001 From: NKAntony777 <2212222@mail.nankai.edu.cn> Date: Sat, 10 Oct 2026 11:51:12 +0800 Subject: [PATCH 4/6] fix(workflows): decide COMMIT durability from isTransaction, not ROLLBACK failure A failing ROLLBACK was treated as proof that COMMIT had already landed. That inference is wrong: a rollback can fail for other reasons (busy/locked) and leave the transaction open. Probing the events table in that state reads the connection's own uncommitted rows as durable and drops them from the retry buffer, losing progress events that were never persisted. Ask SQLite instead. isTransaction reports whether a transaction is still open, and only once it has genuinely ended does a row on this connection mean committed. restoreVolatile now defaults its probe to that condition instead of the unrelated txDepth counter. flushVolatile owns the reconciliation for its own batch: the batch leaves volatileBuffer before transaction() runs, so transaction() cannot see it. Restoring it in transaction()'s catch instead would double-restore against a buffer that no longer holds it. Adds a regression that injects COMMIT and ROLLBACK failures together and asserts the uncommitted rows stay queued. 127 checks pass plus the packaged MCP test standalone. --- .../checks/durability.check.mjs | 25 +++++++++++++++++ .../mcode-dynamic-workflows/src/store.mjs | 28 ++++++++++++------- 2 files changed, 43 insertions(+), 10 deletions(-) diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/durability.check.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/durability.check.mjs index 2b030256..66c2350a 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/durability.check.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/durability.check.mjs @@ -192,6 +192,31 @@ test('after-COMMIT failure does not duplicate rows that already reached SQLite', }); }); +test('failed rollback inside a still-open transaction does not drop the batch', async () => { + await temporary(async dir => { + const store = new Store(dir); + try { + store.event('r', 'run.created'); + const first = store.event('r', 'step.progress'); + const second = store.event('r', 'step.progress'); + // COMMIT fails and ROLLBACK also fails, but the transaction is still open. + // The rows are visible on this connection and NOT durable. Treating the + // table as evidence of a commit would silently discard them from the + // retry buffer, so the outcome must be decided by isTransaction instead. + const exec = store.db.exec.bind(store.db); + store.db.exec = sql => { + if (sql === 'COMMIT' || sql === 'ROLLBACK') throw new Error('injected io failure'); + return exec(sql); + }; + assert.throws(() => store.flushVolatile(), /injected io failure/); + store.db.exec = exec; + assert.equal(store.db.isTransaction, true, 'precondition: transaction still open'); + const buffered = store.volatileBuffer.map(item => item.body.seq).sort((a, b) => a - b); + assert.deepEqual(buffered, [first.seq, second.seq], 'uncommitted rows must stay queued for retry'); + } finally { store.db.exec('ROLLBACK'); store.close(); } + }); +}); + test('workflow_wait rejects cursors that would skip every future event', async () => { await temporary(async dir => { const store = new Store(dir); diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs index ab5452c0..5c33001e 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs @@ -52,11 +52,16 @@ export class Store { transaction(fn) {if(this.txDepth)return fn();this.reserveEventSequences();this.db.exec('BEGIN IMMEDIATE');this.txDepth=1;this.txVolatile=[]; try{const r=fn();this.db.exec('COMMIT');return r;} catch(e){ - // A committed transaction cannot be rolled back, so a failing ROLLBACK is - // the signal that COMMIT may already have reached the table. Only then is - // the events table consulted before re-queueing anything. + // Whether the batch is durable must be read from SQLite, never inferred + // from the fact that ROLLBACK threw: a rollback can fail for reasons other + // than "already committed" (busy/locked), leaving the transaction open. + // Probing such a transaction would read its own uncommitted rows as + // durable and silently drop them from the retry buffer. isTransaction + // answers the real question — only a genuinely ended transaction makes a + // row on this connection mean "committed". let open=true; - try{this.db.exec('ROLLBACK');}catch{open=false;} + try{this.db.exec('ROLLBACK');}catch{} + if(!this.db.isTransaction)open=false; this.restoreVolatile([...this.txVolatile,...this.volatileBuffer],!open); throw e;} finally{this.txDepth=0;this.txVolatile=null;}} @@ -65,9 +70,10 @@ export class Store { // not the control flow — decides what still needs retrying: re-emitting an // already persisted row duplicates its sequence number and later fails the // events.seq primary key during close(). Keyed merge keeps one row per seq. - // Inside an enclosing transaction the probe is skipped: those rows are only - // visible, not committed, and the outer ROLLBACK would silently drop them. - restoreVolatile(items,probe=!this.txDepth){ + // The probe is skipped while a transaction is open on this connection: those + // rows are only visible, not committed, and treating them as durable would + // drop them from the buffer that still owes them a write. + restoreVolatile(items,probe=!this.db.isTransaction){ if(!items.length)return; const exists=probe?this.db.prepare('SELECT 1 AS present FROM events WHERE seq=?'):null; const restored=new Map([...items,...this.volatileBuffer] @@ -141,9 +147,11 @@ export class Store { for(const {runId,body} of batch){ins.run(body.seq,runId,JSON.stringify(body)); this.chainAdvance('event','events','SELECT seq AS pos,runId,body FROM events WHERE seq>? AND seq<=? ORDER BY seq',body.seq,r=>`${r.runId}:${r.pos}`);}}); }catch(error){ - // transaction() already reconciled the ambiguous case; re-check the table - // here so a batch whose rows did reach SQLite is never re-emitted. - this.restoreVolatile(batch,this.txDepth?false:undefined); + // The batch left volatileBuffer before transaction() ran, so transaction() + // cannot see it and cannot re-queue it. Reconciliation therefore happens + // here, against isTransaction: only once the transaction has genuinely + // ended does a row on this connection mean "committed" and may be dropped. + this.restoreVolatile(batch); this.scheduleVolatileFlush(); throw error;} } From 3712fbcbc4102528cd1ed1f9a073be95d81753e3 Mon Sep 17 00:00:00 2001 From: NKAntony777 <2212222@mail.nankai.edu.cn> Date: Sat, 10 Oct 2026 13:12:53 +0800 Subject: [PATCH 5/6] fix(workflows): recover an open transaction before the next event flush A failed COMMIT followed by a failed ROLLBACK left SQLite inside a transaction while the store already believed it was idle. The next BEGIN then failed, so close() could not persist the queued batch. Roll that leftover transaction back before reserving sequences. Co-authored-by: Cursor --- .../checks/durability.check.mjs | 23 +++++++++++++++++++ .../mcode-dynamic-workflows/dist/main.mjs | 12 ++++++---- .../mcode-dynamic-workflows/src/store.mjs | 5 ++++ 3 files changed, 35 insertions(+), 5 deletions(-) diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/durability.check.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/durability.check.mjs index 66c2350a..1f104e03 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/durability.check.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/durability.check.mjs @@ -217,6 +217,29 @@ test('failed rollback inside a still-open transaction does not drop the batch', }); }); +test('a later flush commits the batch after rollback itself failed', async () => { + await temporary(async dir => { + const store = new Store(dir); + try { + store.event('r', 'run.created'); + const first = store.event('r', 'step.progress'); + const second = store.event('r', 'step.progress'); + const exec = store.db.exec.bind(store.db); + store.db.exec = sql => { + if (sql === 'COMMIT' || sql === 'ROLLBACK') throw new Error('injected io failure'); + return exec(sql); + }; + assert.throws(() => store.flushVolatile(), /injected io failure/); + store.db.exec = exec; + assert.equal(store.db.isTransaction, true); + assert.doesNotThrow(() => store.flushVolatile()); + assert.equal(store.db.isTransaction, false); + assert.deepEqual(store.db.prepare('SELECT seq FROM events ORDER BY seq').all().map(r => r.seq), [1, first.seq, second.seq]); + assert.equal(store.volatileBuffer.length, 0); + } finally { store.close(); } + }); +}); + test('workflow_wait rejects cursors that would skip every future event', async () => { await temporary(async dir => { const store = new Store(dir); diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs index bb7a7afe..75224556 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs @@ -7788,8 +7788,8 @@ var Store = class { try { this.db.exec("ROLLBACK"); } catch { - open4 = false; } + if (!this.db.isTransaction) open4 = false; this.restoreVolatile([...this.txVolatile, ...this.volatileBuffer], !open4); throw e; } finally { @@ -7802,9 +7802,10 @@ var Store = class { // not the control flow — decides what still needs retrying: re-emitting an // already persisted row duplicates its sequence number and later fails the // events.seq primary key during close(). Keyed merge keeps one row per seq. - // Inside an enclosing transaction the probe is skipped: those rows are only - // visible, not committed, and the outer ROLLBACK would silently drop them. - restoreVolatile(items, probe = !this.txDepth) { + // The probe is skipped while a transaction is open on this connection: those + // rows are only visible, not committed, and treating them as durable would + // drop them from the buffer that still owes them a write. + restoreVolatile(items, probe = !this.db.isTransaction) { if (!items.length) return; const exists = probe ? this.db.prepare("SELECT 1 AS present FROM events WHERE seq=?") : null; const restored = new Map([...items, ...this.volatileBuffer].filter((item) => !exists?.get(item.body.seq)).map((item) => [item.body.seq, item])); @@ -7812,6 +7813,7 @@ var Store = class { if (this.volatileBuffer.length) this.scheduleVolatileFlush(); } reserveEventSequences() { + if (!this.txDepth && this.db.isTransaction) this.db.exec("ROLLBACK"); const persisted = Number(this.db.prepare("SELECT COALESCE((SELECT seq FROM sqlite_sequence WHERE name='events'),0) AS seq").get().seq); this.nextEventSequence = Math.max(this.nextEventSequence, persisted); if (this.nextEventSequence < this.eventSequenceLimit) return; @@ -7943,7 +7945,7 @@ var Store = class { } }); } catch (error2) { - this.restoreVolatile(batch, this.txDepth ? false : void 0); + this.restoreVolatile(batch); this.scheduleVolatileFlush(); throw error2; } diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs index 5c33001e..71528326 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs @@ -83,6 +83,11 @@ export class Store { if(this.volatileBuffer.length)this.scheduleVolatileFlush(); } reserveEventSequences() { + // txDepth 0 with a SQLite transaction still open means an earlier COMMIT + // and ROLLBACK both failed. The queued batch is still in memory; roll the + // leftover transaction back so the next BEGIN can persist it. A rollback + // that itself fails propagates, and the caller keeps the batch queued. + if(!this.txDepth&&this.db.isTransaction)this.db.exec('ROLLBACK'); const persisted=Number(this.db.prepare("SELECT COALESCE((SELECT seq FROM sqlite_sequence WHERE name='events'),0) AS seq").get().seq); this.nextEventSequence=Math.max(this.nextEventSequence,persisted); if(this.nextEventSequence Date: Sun, 11 Oct 2026 14:04:28 +0800 Subject: [PATCH 6/6] fix(workflows): recover residual transactions and validate all cursor entry points --- .github/workflows/dynamic-workflow.yml | 5 + .../mcode-dynamic-workflows/README.md | 4 + .../mcode-dynamic-workflows/REVIEW-NOTES.md | 12 + .../mcode-dynamic-workflows/REVIEW-REPLY.md | 10 + .../checks/http.check.mjs | 42 ++++ .../checks/package.check.mjs | 27 +++ .../checks/residual-transaction.check.mjs | 221 ++++++++++++++++++ .../mcode-dynamic-workflows/dist/main.mjs | 66 +++++- .../scripts/build-web.mjs | 2 +- .../mcode-dynamic-workflows/src/http.mjs | 4 +- .../mcode-dynamic-workflows/src/store.mjs | 75 ++++-- .../mcode-dynamic-workflows/src/tools.mjs | 9 +- 12 files changed, 442 insertions(+), 35 deletions(-) create mode 100644 plugins/hetaoBackend/mcode-dynamic-workflows/checks/residual-transaction.check.mjs diff --git a/.github/workflows/dynamic-workflow.yml b/.github/workflows/dynamic-workflow.yml index 20d15e54..14b13856 100644 --- a/.github/workflows/dynamic-workflow.yml +++ b/.github/workflows/dynamic-workflow.yml @@ -1,6 +1,7 @@ name: Dynamic Workflow on: + workflow_dispatch: pull_request: paths: - 'plugins/hetaoBackend/mcode-dynamic-workflows/**' @@ -27,6 +28,8 @@ jobs: node-version: 22 cache: npm cache-dependency-path: plugins/hetaoBackend/mcode-dynamic-workflows/package-lock.json + - name: Record the exact checked-out revision + run: git rev-parse HEAD - run: npm ci --ignore-scripts --registry=https://registry.npmjs.org - run: npm test - run: npm run build @@ -44,5 +47,7 @@ jobs: - uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0 with: node-version: 22 + - name: Record the exact checked-out revision + run: git rev-parse HEAD - run: npm ci --ignore-scripts --registry=https://registry.npmjs.org - run: node --test checks/process-tree.check.mjs diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/README.md b/plugins/hetaoBackend/mcode-dynamic-workflows/README.md index f671ac8f..b82ca61c 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/README.md +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/README.md @@ -80,6 +80,10 @@ The dev-only source checks use the pinned dependencies. They live in `checks/*.c Verification covers isolated demo execution, approval gating, selective reuse and invalidation, project routing, persistence, parsing, local HTTP protections and UI behavior. Controlled executors are not evidence of live model correctness, account authorization or provider availability. No paid model calls or user research reruns were used for this contribution. +When upgrading, stop each existing project service before starting the updated `dist/main.mjs`; a connected chat may otherwise keep using an older background process. Use the existing service's workspace and data directory with `node dist/main.mjs --stop-service --workspace --data-dir `, then reconnect through the updated package and confirm the dashboard address belongs to the new instance. + +The store rolls back residual failed transactions before subsequent reads or writes. If recovery still fails, the operation reports an error rather than acknowledging a state that could later disappear. HTTP `after` and MCP `afterSequence` share the same nonnegative safe-integer cursor policy; invalid HTTP cursors return a client error before waiting. + ## MCP tools `workflow_validate`, `workflow_start`, `workflow_update`, `workflow_repair`, `workflow_status`, `workflow_results`, `workflow_wait`, `workflow_pause`, `workflow_cancel`, `workflow_resume`, `workflow_dashboard`. diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/REVIEW-NOTES.md b/plugins/hetaoBackend/mcode-dynamic-workflows/REVIEW-NOTES.md index 7a6ea888..a7585043 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/REVIEW-NOTES.md +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/REVIEW-NOTES.md @@ -15,6 +15,18 @@ being reused after a process crash. SQLite stays in WAL mode with `synchronous=F - The public MCP test now stops its detached service before removing the temporary directory, fixing its Windows `EBUSY` cleanup failure without changing service lifetime behavior. +## Follow-up repair and local verification, October 11, 2026 + +These counts were measured on the `codex/pr67-durable-batching` worktree based on `3712fbcbc4102528cd1ed1f9a073be95d81753e3` before committing this revision. Post-commit rechecks are recorded separately; no same-head GitHub CI result is claimed. + +- Residual failed transactions are rolled back before subsequent Store reads/writes, reservation, maintenance and close. Recovery failures are reported, not acknowledged as durable writes. Six dedicated residual-transaction checks cover acknowledged saves, duplicate/uncommitted events, saveStep/saveSetting, engine finish state, flush failure and leasing/FULL semantics. +- HTTP/dashboard `after` and MCP `afterSequence` share `parseEventCursor`. Defaults remain zero; negative, fractional, nonfinite and unsafe integers are rejected before wait/query. Boolean/array/object coercion is rejected too. Actual HTTP and bundled stdio MCP are exercised, including maximum safe integer and legal cursors. +- Web bundle generation normalizes the embedded example's CRLF to LF, so rebuilding on Windows does not drift from Linux committed output. `dist/main.mjs` and `web/app.js` are regenerated; the rebuilt artifacts were stable in a second build. +- Final Windows/Node 24.12.0 checks: **136 passed, 2 skipped, 0 failed** (138); public package check: **1 passed, 0 skipped**. The two existing skips require POSIX signal/process-group behavior. No paid model calls or live account operations were used. +- Local root validation with development `node_modules` present scans third-party README text and fails on a third-party placeholder. The separately staged delivery check excludes development environments/dependencies, matching shipped files; its result is recorded in the handoff evidence. The unrelated root validator is not weakened. + +CI records the checked-out revision and has a manual dispatch entry; the existing source/package and Windows process jobs remain enabled. Exact-head CI remains pending until these repairs are committed and checked on that SHA. Stop old workspace services before reconnecting to the new package; an existing owner process does not reload itself just because dist changed. + ## Local verification, October 9, 2026 Windows, Node 24.12.0: diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/REVIEW-REPLY.md b/plugins/hetaoBackend/mcode-dynamic-workflows/REVIEW-REPLY.md index 593c0d22..a874b36e 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/REVIEW-REPLY.md +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/REVIEW-REPLY.md @@ -1,3 +1,13 @@ +## October 11 follow-up — local worktree evidence only + +Residual failed transactions now recover before subsequent Store entry points can acknowledge state. Six regression checks verify save/saveStep/saveSetting durability, event visibility, consistent engine finish state, failed flush/close reporting and unchanged FULL/sequence leasing behavior. + +HTTP/dashboard and MCP now use the same safe-integer event cursor parser, before querying or waiting. Actual HTTP and bundled MCP reject negative/fractional/nonfinite/oversized values and boolean/array/object coercion; valid/default cursors still work. Runtime assets are rebuilt and embedded-example line endings are normalized for reproducible Windows/Linux output. + +Windows Node 24.12.0: 136 checks passed, 2 POSIX-only checks skipped; public package check passed separately. CI records the checked-out SHA and supports manual dispatch. This revision is committed locally after the worktree checks, with post-commit independent review requested. It remains unpushed; no new GitHub run or same-head approval is claimed. Existing services must be stopped/reconnected before the updated package takes effect. REVIEW-NOTES.md contains limitations and the older historical performance evidence below. + +--- + Thanks for identifying the durability issues. SQLite WAL is restored to synchronous=FULL. Progress batching remains, but sequences are now leased through an independently committed high watermark before publishing a cursor. After a crash, unused lease values are skipped, diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/http.check.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/http.check.mjs index e6964d04..20caae37 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/http.check.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/http.check.mjs @@ -1,2 +1,44 @@ import test from 'node:test';import assert from 'node:assert/strict';import {mkdtemp,rm} from 'node:fs/promises';import {join} from 'node:path';import {tmpdir} from 'node:os';import {createHash} from 'node:crypto';import {REPORT_STYLES} from '../src/reports.mjs';import {Store} from '../src/store.mjs';import {Engine} from '../src/engine.mjs';import {startHTTP} from '../src/http.mjs'; +import {parseEventCursor} from '../src/tools.mjs'; test('HTTP local-origin checks, validation, creation and durable state',async()=>{const dir=await mkdtemp(join(tmpdir(),'wf-http-'));const store=new Store(dir),engine=new Engine(store,{workspace:dir});const http=await startHTTP(engine);const u=new URL(http.url),headers={'X-Workflow-Client':'1','Content-Type':'application/json'};try{const config=await(await fetch(u.origin+'/api/config',{headers})).json();assert.equal(u.hash,'');const page=await fetch(u.origin+'/');const policy=page.headers.get('content-security-policy');assert.ok(policy.includes("style-src 'self' 'sha256-"+createHash('sha256').update(REPORT_STYLES).digest('base64')+"'"));assert.ok(!policy.includes("'unsafe-inline'"));assert.equal((await fetch(u.origin+'/readable.css')).status,200);assert.equal(config.serviceProtocol,2);assert.equal((await fetch(u.origin+'/api/runs',{headers:{...headers,'Sec-Fetch-Site':'cross-site'}})).status,403);assert.equal((await fetch(u.origin+'/api/runs',{method:'OPTIONS',headers:{Origin:'https://example.com','Access-Control-Request-Headers':'x-workflow-client'}})).headers.get('access-control-allow-origin'),null);assert.equal(config.scheduler.limit,8);assert.equal((await fetch(u.origin+'/api/scheduler',{method:'POST',headers,body:JSON.stringify({globalConcurrency:16})})).status,200);assert.equal(engine.globalConcurrency,16);assert.equal((await fetch(u.origin+'/api/scheduler',{method:'POST',headers,body:JSON.stringify({globalConcurrency:0})})).status,400);assert.equal(config.defaults.maxSteps,120);assert.equal(config.defaults.stepTimeoutMs,1800000);assert.equal(typeof config.mcodeAvailable,'boolean');assert.equal((await fetch(u.origin+'/api/runs')).status,403);assert.equal((await fetch(u.origin+'/api/runs',{headers:{...headers,Origin:'https://example.com'}})).status,400);const bad=await fetch(u.origin+'/api/validate',{method:'POST',headers,body:JSON.stringify({script:'await import("fs")'})});assert.equal(bad.status,400);const badDeps=await fetch(u.origin+'/api/validate',{method:'POST',headers,body:JSON.stringify({script:'await ctx.agent({id:"b",prompt:"p",dependsOn:42})'})});assert.equal(badDeps.status,400);assert.match((await badDeps.json()).error,/b.*dependsOn.*number/);const res=await fetch(u.origin+'/api/runs',{method:'POST',headers,body:JSON.stringify({requestId:'h',name:'HTTP',executor:'demo',script:'return {ok:true}'})});assert.equal(res.status,201);const r=await res.json();assert.ok(store.get(r.id));assert.equal(r.status,'pending_review');assert.equal(engine.active.size,0);const edit=await fetch(u.origin+`/api/runs/${r.id}/edit`,{method:'POST',headers,body:JSON.stringify({revision:1,script:'return {edited:true}'})});assert.equal((await edit.json()).revision,2);assert.equal((await fetch(u.origin+`/api/runs/${r.id}/approve`,{method:'POST',headers,body:JSON.stringify({revision:1})})).status,400);assert.equal((await fetch(u.origin+`/api/runs/${r.id}/approve`,{method:'POST',headers,body:JSON.stringify({revision:2})})).status,200);const snapshot=await fetch(u.origin+`/api/runs/${r.id}`,{headers});assert.equal(snapshot.status,200);}finally{await engine.close();await http.close();store.close();await rm(dir,{recursive:true,force:true});}}); +test('HTTP workflow_wait rejects cursors the MCP schema already rejects',async()=>{ + // Illegal cursors must fail before any wait or DB query, not after one. + assert.equal(parseEventCursor(undefined),0); + assert.equal(parseEventCursor(''),0); + assert.equal(parseEventCursor(null),0); + assert.equal(parseEventCursor(0),0); + assert.equal(parseEventCursor(7),7); + assert.equal(parseEventCursor(Number.MAX_SAFE_INTEGER),Number.MAX_SAFE_INTEGER); + // '0x10' is a valid integer representation (16) and intentionally accepted; + // the guard rejects non-integers/negatives/out-of-range, not alternate radix. + for(const bad of[true,false,[],[0],[1],{},-1,-0.5,'-1',0.5,1.5,'1.5','1e100',Number.MAX_SAFE_INTEGER+1,'Infinity','-Infinity','NaN','abc',' 1 2 ',1e100,Infinity,-Infinity,NaN]) + assert.throws(()=>parseEventCursor(bad),/等待参数无效/,`must reject ${String(bad)}`); + const dir=await mkdtemp(join(tmpdir(),'wf-http-cursor-')); + const store=new Store(dir),engine=new Engine(store,{workspace:dir}); + const http=await startHTTP(engine); + const u=new URL(http.url),headers={'X-Workflow-Client':'1','Content-Type':'application/json'}; + try{ + const created=await fetch(u.origin+'/api/runs',{method:'POST',headers,body:JSON.stringify({requestId:'cursor',name:'Cursor',executor:'demo',script:'return {ok:true}'})}); + const run=await created.json(); + // Same matrix over a real listening server, including forms that only exist + // as text on the wire (NaN/Infinity/fractional cannot be JSON numbers). + for(const after of['-1','0.5','1.5','1e100','9007199254740992','Infinity','-Infinity','NaN','abc','%201%202%20']){ + const wait=await fetch(u.origin+`/api/runs/${run.id}/wait?after=${encodeURIComponent(after)}`,{headers}); + assert.equal(wait.status,400,`after=${after} must be rejected`); + assert.match(await wait.text(),/等待参数无效/); + } + const missing=await fetch(u.origin+`/api/runs/${run.id}/wait`,{headers}); + assert.equal(missing.status,200); + const open=await missing.json(); + assert.equal(Number.isSafeInteger(open.nextSequence)&&open.nextSequence>=0,true); + for(const after of['0','1','9007199254740991']){ + const ok=await fetch(u.origin+`/api/runs/${run.id}/wait?after=${after}`,{headers}); + assert.equal(ok.status,200,`after=${after} must be accepted`); + const body=await ok.json(); + assert.equal(Number.isSafeInteger(body.nextSequence),true); + } + // A rejected cursor must not have consumed events or touched the store. + assert.equal(store.get(run.id).status,'pending_review'); + assert.equal(store.verifyIntegrity().events.verified,true); + }finally{await engine.close();await http.close();store.close();await rm(dir,{recursive:true,force:true});} +}); diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/package.check.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/package.check.mjs index ab9cbe94..43520081 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/package.check.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/package.check.mjs @@ -7,3 +7,30 @@ test('second MCP connection proxies the existing owner, child workers expose no }); async function stopService(dir){const {execFile}=await import('node:child_process');const {promisify}=await import('node:util');await promisify(execFile)(process.execPath,[resolve('dist/main.mjs'),'--stop-service','--data-dir',dir,'--workspace',dir]);} + +// The packaged bundle must reject cursors that would skip every future event, +// not just the source parser: the shipped dist/main.mjs is what users run. +test('packaged MCP rejects illegal workflow_wait cursors and accepts valid ones',async()=>{ + const dir=await mkdtemp(join(tmpdir(),'wf-pkg-cursor-'));await writeFile(join(dir,'settings.json'),JSON.stringify({workspace:dir,dataDir:dir})); + const client=new Client({name:'cursor-test',version:'1'}); + const transport=new StdioClientTransport({command:process.execPath,args:[resolve('dist/main.mjs'),'--stdio','--settings',join(dir,'settings.json')],stderr:'pipe'}); + let stderr='';transport.stderr?.on('data',c=>stderr+=c); + try{ + await client.connect(transport); + const start=await client.callTool({name:'workflow_start',arguments:{requestId:'pkg-cursor',name:'Cursor',executor:'demo',script:'return {ok:true};'}}); + assert.ok(!start.isError,start.content?.[0]?.text); + const id=JSON.parse(start.content[0].text).id; + // NaN/Infinity are not JSON-encodable: over stdio they decode as null, which + // the parser treats as the default cursor 0 (never a skip). Their rejection is + // asserted at the parser level and as string forms over HTTP instead. + for(const cursor of[true,false,[],[1],{},-1,1.5,1e100,Number.MAX_SAFE_INTEGER+1,'abc']){ + const r=await client.callTool({name:'workflow_wait',arguments:{runId:id,afterSequence:cursor,timeoutMs:0}}); + assert.match(JSON.stringify(r),/等待参数无效/,`cursor ${cursor} must be rejected`); + } + for(const cursor of[0,1,Number.MAX_SAFE_INTEGER]){ + const r=await client.callTool({name:'workflow_wait',arguments:{runId:id,afterSequence:cursor,timeoutMs:0}}); + assert.ok(!r.isError,`cursor ${cursor} accepted: ${r.content?.[0]?.text}`); + assert.ok(Number.isSafeInteger(JSON.parse(r.content[0].text).nextSequence)); + } + }finally{await client.close();await transport.close();await stopService(dir);await rm(dir,{recursive:true,force:true});} +}); diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/checks/residual-transaction.check.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/residual-transaction.check.mjs new file mode 100644 index 00000000..9c8f1294 --- /dev/null +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/residual-transaction.check.mjs @@ -0,0 +1,221 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtemp, rm, access } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { Store } from '../src/store.mjs'; +import { Engine } from '../src/engine.mjs'; + +async function temporary(fn) { + const dir = await mkdtemp(join(tmpdir(), 'wf-residual-')); + try { await fn(dir); } finally { await rm(dir, { recursive: true, force: true }); } +} + +// Failures are injected only at db.exec so the code under test runs against a +// real node:sqlite connection: BEGIN/ROLLBACK/COMMIT still reach SQLite normally +// except for the chosen statement, which is what an IO fault on commit looks like. +function breakTransactionControl(store, { commit = true, rollback = true } = {}) { + const exec = store.db.exec.bind(store.db); + store.db.exec = sql => { + if (sql === 'COMMIT' && commit) throw new Error('injected commit failure'); + if (sql === 'ROLLBACK' && rollback) throw new Error('injected rollback failure'); + return exec(sql); + }; + return exec; +} + +const run = (status, extra = {}) => ({ id: 'r', requestId: 'r', requestHash: 'h', status, ...extra }); + +test('acknowledged run save survives a residual transaction, recovery and reopen', async () => { + await temporary(async dir => { + const store = new Store(dir); + let exec; + try { + store.save(run('running')); + store.event('r', 'run.created'); + store.event('r', 'step.progress', { message: 'progress' }); + exec = breakTransactionControl(store); + assert.throws(() => store.flushVolatile(), /injected commit failure/); + assert.equal(store.db.isTransaction, true, 'precondition: transaction still open'); + + // While the fault is live nothing may claim durability: writes must either + // recover first or refuse, and reads must not expose the open transaction. + assert.throws(() => store.save(run('running', { marker: 'during-fault' })), /残留事务无法回滚/); + assert.throws(() => store.saveStep('r', { id: 's', kind: 'agent' }), /残留事务无法回滚/); + assert.throws(() => store.saveSetting('during-fault', true), /残留事务无法回滚/); + assert.throws(() => store.saveTemplate({ id: 't', name: 't' }), /残留事务无法回滚/); + assert.throws(() => store.events('r'), /残留事务无法回滚/); + + // Faults clear: the first operation recovers the connection by rolling the + // leftover transaction back, and the row it had acknowledged survives. + store.db.exec = exec; + assert.doesNotThrow(() => store.recover('测试')); + assert.equal(store.db.isTransaction, false); + store.save(run('completed')); + assert.equal(store.get('r').status, 'completed'); + store.event('r', 'run.finished'); + assert.doesNotThrow(() => store.close()); + + const reopened = new Store(dir); + try { + assert.equal(reopened.get('r').status, 'completed', 'acknowledged save survived recovery + reopen'); + assert.equal(reopened.setting('during-fault'), undefined, 'refused writes left nothing behind'); + const events = reopened.events('r'); + assert.deepEqual(events.map(e => e.seq), [...new Set(events.map(e => e.seq))], 'no duplicate seq'); + assert.equal(events.at(-1).type, 'run.finished'); + assert.equal(reopened.verifyIntegrity().events.verified, true); + } finally { reopened.close(); } + } finally { store.db.exec = exec; if (store.db.isOpen !== false) { try { store.close(); } catch {} } } + }); +}); + +test('event reads during a fault never show duplicate or uncommitted sequence numbers', async () => { + await temporary(async dir => { + const store = new Store(dir); + let exec; + try { + store.event('r', 'run.created'); + store.event('r', 'step.progress'); + exec = breakTransactionControl(store); + assert.throws(() => store.flushVolatile(), /injected commit failure/); + assert.equal(store.db.isTransaction, true); + // The uncommitted transaction still holds the progress row on this + // connection; reading it would publish [1,2,2] and lose the queued retry. + assert.throws(() => store.events('r'), /残留事务无法回滚/); + + store.db.exec = exec; + const events = store.events('r'); + assert.deepEqual(events.map(e => e.seq), [1, 2], 'recovered read shows each seq once'); + assert.deepEqual(events.map(e => e.seq), [...new Set(events.map(e => e.seq))]); + assert.equal(store.volatileBuffer.length, 1, 'uncommitted row stays queued for retry'); + } finally { store.db.exec = exec; try { store.close(); } catch {} } + }); +}); + +test('save/saveStep/saveSetting recover first and only then report success', async () => { + await temporary(async dir => { + const store = new Store(dir); + let exec; + try { + store.event('r', 'run.created'); + exec = breakTransactionControl(store); + assert.throws(() => store.event('r', 'run.finished'), /injected commit failure/); + assert.equal(store.db.isTransaction, true, 'precondition: both COMMIT and ROLLBACK failed'); + + // ROLLBACK still broken: every plain write is refused rather than joining + // the doomed transaction and returning a success it cannot honour. + assert.throws(() => store.save(run('succeeded')), /残留事务无法回滚/); + store.db.exec = sql => { if (sql === 'ROLLBACK') throw new Error('injected rollback failure'); return exec(sql); }; + assert.throws(() => store.saveStep('r', { id: 's', kind: 'agent' }), /残留事务无法回滚/); + assert.throws(() => store.saveSetting('k', 1), /残留事务无法回滚/); + assert.throws(() => store.verifyIntegrity(), /残留事务无法回滚/); + + // ROLLBACK restored: the same writes now recover and genuinely persist. + store.db.exec = exec; + store.save(run('succeeded')); + store.saveStep('r', { id: 's', kind: 'agent' }); + store.saveSetting('k', 1); + assert.equal(store.db.isTransaction, false); + assert.equal(store.get('r').status, 'succeeded'); + assert.equal(store.step('r', 's').kind, 'agent'); + assert.equal(store.setting('k'), 1); + assert.equal(store.verifyIntegrity().events.verified, true); + } finally { store.db.exec = exec; try { store.close(); } catch {} } + }); +}); + +test('engine keeps run status and the finish event consistent across a residual transaction', async () => { + await temporary(async dir => { + const store = new Store(dir); + const engine = new Engine(store, { workspace: dir }); + let exec; + try { + store.save(run('running')); + store.event('r', 'run.created'); + store.event('r', 'step.progress', { message: 'progress' }); + exec = breakTransactionControl(store); + assert.throws(() => store.flushVolatile(), /injected commit failure/); + assert.equal(store.db.isTransaction, true, 'precondition: transaction left open'); + + // The real engine ordering: save node, save run, then emit run.finished. + // With ROLLBACK still broken the run save must be refused, not acknowledged. + assert.throws(() => store.saveStep('r', { id: 's', kind: 'agent', status: 'succeeded' }), /残留事务无法回滚/); + assert.throws(() => store.save(run('succeeded')), /残留事务无法回滚/); + + store.db.exec = exec; + const finished = { ...store.get('r'), status: 'succeeded' }; + store.saveStep('r', { id: 's', kind: 'agent', status: 'succeeded' }); + engine.save(finished); + engine.emitEvent('r', 'run.finished', { status: 'succeeded' }); + assert.equal(store.get('r').status, 'succeeded'); + assert.doesNotThrow(() => store.close()); + + const reopened = new Store(dir); + try { + assert.equal(reopened.get('r').status, 'succeeded', 'status did not revert to running'); + const events = reopened.events('r'); + assert.deepEqual(events.map(e => e.seq), [...new Set(events.map(e => e.seq))]); + assert.equal(events.at(-1).type, 'run.finished'); + assert.equal(reopened.step('r', 's').status, 'succeeded'); + assert.equal(reopened.verifyIntegrity().events.verified, true); + } finally { reopened.close(); } + } finally { store.db.exec = exec; await engine.close().catch(() => {}); try { store.close(); } catch {} } + }); +}); + +test('periodic flush failure keeps the batch and close reports it instead of losing it', async () => { + await temporary(async dir => { + const store = new Store(dir); + let exec; + try { + store.event('r', 'run.created'); + const progress = store.event('r', 'step.progress'); + exec = breakTransactionControl(store); + // The scheduled flush swallows its own error, so the batch must survive it + // in the buffer instead of being dropped by a later rollback. + assert.throws(() => store.flushVolatile(), /injected commit failure/); + assert.equal(store.db.isTransaction, true); + assert.deepEqual(store.volatileBuffer.map(i => i.body.seq), [progress.seq], 'periodic batch stays queued'); + + // Explicit close during the fault must surface the problem, and must not + // let db.close() silently discard the open transaction. The residual + // transaction is still open here, so recovery fails and shutdown reports + // it instead of dropping the queued batch without a word. + assert.throws(() => store.close(), error => { + assert.equal(error.constructor, AggregateError); + assert.ok(error.errors.length >= 1); + return error.errors.every(e => /残留事务无法回滚/.test(e.message)); + }); + assert.equal(store.fd, undefined, 'owner lock released even under fault'); + await assert.rejects(access(join(dir, 'owner.lock')), { code: 'ENOENT' }); + + store.db.exec = exec; + const reopened = new Store(dir); + try { + assert.ok(reopened.event('r', 'run.finished').seq > progress.seq, 'a new Store acquires the lock'); + assert.equal(reopened.verifyIntegrity().events.verified, true); + } finally { reopened.close(); } + } finally { store.db.exec = exec; try { store.close(); } catch {} } + }); +}); + +test('durability mode, sequence leasing and batch throughput are unchanged', async () => { + await temporary(async dir => { + const store = new Store(dir); + try { + assert.equal(store.db.prepare('PRAGMA synchronous').get().synchronous, 2, 'FULL synchronous preserved'); + assert.equal(store.db.prepare('PRAGMA busy_timeout').get().timeout, 5000); + store.save(run('running')); + store.saveRepairCandidate('r', { id: 'a', kind: 'agent' }); + store.event('r', 'run.created'); + for (let seq = 2; seq <= 151; seq++) assert.equal(store.event('r', 'step.progress').seq, seq); + store.flushVolatile(); + assert.equal(store.volatileBuffer.length, 0); + assert.equal(store.setting('event_sequence_lease'), 1024, 'leasing intact'); + assert.deepEqual(store.events('r').map(e => e.seq).slice(0, 4), [1, 2, 3, 4]); + assert.equal(store.verifyIntegrity().events.verified, true); + assert.equal(store.verifyIntegrity().repair.verified, true); + assert.equal(store.db.prepare('PRAGMA synchronous').get().synchronous, 2); + } finally { store.close(); } + }); +}); diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs index 75224556..e2d9dd9b 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/dist/main.mjs @@ -7775,6 +7775,7 @@ var Store = class { } transaction(fn) { if (this.txDepth) return fn(); + this.recover("\u4E8B\u52A1"); this.reserveEventSequences(); this.db.exec("BEGIN IMMEDIATE"); this.txDepth = 1; @@ -7812,8 +7813,25 @@ var Store = class { this.volatileBuffer = [...restored.values()].sort((a, b2) => a.body.seq - b2.body.seq); if (this.volatileBuffer.length) this.scheduleVolatileFlush(); } + // Single connection-recovery gate for every read and write path. When both + // COMMIT and ROLLBACK fail, the SQLite transaction stays open but the finally + // block has already cleared txDepth, so the connection looks idle while + // holding uncommitted rows. The next plain write would join that transaction + // and return success for rows a later ROLLBACK discards, and a read would + // publish them as durable. recover() is therefore called before every normal + // path: it rolls a leftover transaction back, and if even that fails it + // rejects the operation loudly instead of acknowledging a write it cannot + // guarantee. txDepth>0 means the open transaction is the current one. + recover(op = "\u64CD\u4F5C") { + if (!this.db || this.txDepth || !this.db.isTransaction) return; + try { + this.db.exec("ROLLBACK"); + } catch (error2) { + throw new Error(`\u72B6\u6001\u5E93\u6B8B\u7559\u4E8B\u52A1\u65E0\u6CD5\u56DE\u6EDA\uFF0C\u672C\u6B21${op}\u5DF2\u62D2\u7EDD\u4EE5\u514D\u4E22\u5931\u5DF2\u786E\u8BA4\u7684\u6570\u636E\uFF0C\u8BF7\u91CD\u542F\u670D\u52A1\u540E\u91CD\u8BD5\uFF1A${error2.message}`); + } + } reserveEventSequences() { - if (!this.txDepth && this.db.isTransaction) this.db.exec("ROLLBACK"); + this.recover("\u4E8B\u4EF6\u5E8F\u53F7\u9884\u7559"); const persisted = Number(this.db.prepare("SELECT COALESCE((SELECT seq FROM sqlite_sequence WHERE name='events'),0) AS seq").get().seq); this.nextEventSequence = Math.max(this.nextEventSequence, persisted); if (this.nextEventSequence < this.eventSequenceLimit) return; @@ -7821,12 +7839,18 @@ var Store = class { const limit = this.nextEventSequence + 1024; if (!Number.isSafeInteger(limit)) throw new Error("Event sequence exhausted"); this.db.exec("BEGIN IMMEDIATE"); + this.txDepth = 1; try { this.saveSetting("event_sequence_lease", limit); this.db.exec("COMMIT"); } catch (error2) { - this.db.exec("ROLLBACK"); + try { + this.db.exec("ROLLBACK"); + } catch { + } throw error2; + } finally { + this.txDepth = 0; } this.eventSequenceLimit = limit; } @@ -7835,44 +7859,56 @@ var Store = class { return ++this.nextEventSequence; } templates() { + this.recover("\u6A21\u677F\u8BFB\u53D6"); return this.db.prepare("SELECT body FROM templates ORDER BY rowid DESC").all().map((r) => JSON.parse(r.body)); } template(id2) { + this.recover("\u6A21\u677F\u8BFB\u53D6"); const r = this.db.prepare("SELECT body FROM templates WHERE id=?").get(id2); return r ? JSON.parse(r.body) : null; } saveTemplate(value) { + this.recover("\u6A21\u677F\u5199\u5165"); this.db.prepare("INSERT INTO templates VALUES(?,?)").run(value.id, JSON.stringify(value)); } deleteTemplate(id2) { + this.recover("\u6A21\u677F\u5220\u9664"); return this.db.prepare("DELETE FROM templates WHERE id=?").run(id2).changes > 0; } setting(key) { + this.recover("\u8BBE\u7F6E\u8BFB\u53D6"); const row = this.db.prepare("SELECT body FROM settings WHERE key=?").get(key); return row ? JSON.parse(row.body) : void 0; } saveSetting(key, value) { + this.recover("\u8BBE\u7F6E\u5199\u5165"); this.db.prepare("INSERT INTO settings VALUES(?,?) ON CONFLICT(key) DO UPDATE SET body=excluded.body").run(key, JSON.stringify(value)); } save(run) { + this.recover("\u8FD0\u884C\u72B6\u6001\u5199\u5165"); this.db.prepare("INSERT INTO runs VALUES(?,?,?,?) ON CONFLICT(id) DO UPDATE SET body=excluded.body").run(run.id, run.requestId, run.requestHash, JSON.stringify(run)); } get(id2) { + this.recover("\u8FD0\u884C\u72B6\u6001\u8BFB\u53D6"); const r = this.db.prepare("SELECT body FROM runs WHERE id=?").get(id2); return r ? JSON.parse(r.body) : null; } byRequest(id2) { + this.recover("\u8FD0\u884C\u72B6\u6001\u8BFB\u53D6"); const r = this.db.prepare("SELECT body FROM runs WHERE requestId=?").get(id2); return r ? JSON.parse(r.body) : null; } list() { + this.recover("\u8FD0\u884C\u5217\u8868\u8BFB\u53D6"); 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, id2) { + this.recover("\u6B65\u9AA4\u8BFB\u53D6"); const r = this.db.prepare("SELECT body FROM steps WHERE runId=? AND id=?").get(runId, id2); return r ? JSON.parse(r.body) : null; } steps(runId) { + this.recover("\u6B65\u9AA4\u8BFB\u53D6"); 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 @@ -7880,15 +7916,18 @@ var Store = class { // 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 }) { + this.recover("\u590D\u7528\u5019\u9009\u8BFB\u53D6"); 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.recover("\u6B65\u9AA4\u5199\u5165"); 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, id2) { + this.recover("\u4FEE\u590D\u5019\u9009\u8BFB\u53D6"); const r = this.db.prepare("SELECT body FROM repair_cache WHERE runId=? AND id=?").get(runId, id2); return r ? JSON.parse(r.body) : null; } @@ -7951,9 +7990,10 @@ var Store = class { } } events(runId, after = 0, limit = 150) { + this.recover("\u4E8B\u4EF6\u8BFB\u53D6"); const persisted = this.db.prepare("SELECT seq,body FROM events WHERE runId=? AND seq>? ORDER BY seq LIMIT ?").all(runId, after, limit).map((e) => ({ seq: e.seq, ...JSON.parse(e.body) })); const buffered = this.volatileBuffer.filter((e) => e.runId === runId && e.body.seq > after).map((e) => e.body); - return [...persisted, ...buffered].sort((a, b2) => a.seq - b2.seq).slice(0, limit); + return [...new Map([...persisted, ...buffered].map((e) => [e.seq, e])).values()].sort((a, b2) => a.seq - b2.seq).slice(0, limit); } rowHash(prev, kind, key, body) { return createHash("sha256").update(`${prev}:${kind}:${key}:${body}`).digest("hex"); @@ -7975,9 +8015,11 @@ var Store = class { this.saveSetting(`integrity_${surface}`, { head: prev, upto: newUpto }); } integrityHeads() { + this.recover("\u5B8C\u6574\u6027\u8BFB\u53D6"); return { events: this.setting("integrity_events") ?? null, repair: this.setting("integrity_repair") ?? null }; } verifyIntegrity() { + this.recover("\u5B8C\u6574\u6027\u6821\u9A8C"); const genesis = "0".repeat(64); const face = (kind, surface, table, posCol, rowSql, keyOf) => { const skey = `integrity_${surface}`; @@ -8028,6 +8070,11 @@ var Store = class { } catch (error2) { errors.push(error2); } + try { + this.recover("\u5173\u95ED"); + } catch (error2) { + errors.push(error2); + } try { this.db.close(); } catch (error2) { @@ -26683,6 +26730,13 @@ var TOOLS = [ { name: "workflow_resume", description: "\u539F\u811A\u672C\u4E0E\u539F\u8F93\u5165\u6062\u590D\uFF0C\u590D\u7528\u5DF2\u6210\u529F\u8282\u70B9\u3002\u53EF\u8C03\u6574 maxSteps/stepTimeoutMs/runTimeoutMs/maxCalls \u540E\u91CD\u8BD5\uFF0C\u6210\u529F\u8282\u70B9\u590D\u7528\uFF0C\u5931\u8D25\u8282\u70B9\u4ECE\u5934\u6267\u884C\u3002\u5F02\u5E38\u9000\u51FA\u9700\u8981\u7528\u6237\u5148\u786E\u8BA4\u65E7 Agent \u5DF2\u505C\u6B62\u3002", inputSchema: obj({ ...id, confirmStopped: { type: "boolean" }, ...LIMIT_SCHEMAS, maxCalls: { type: "integer", minimum: 1, maximum: 100 } }, ["runId"]) }, { name: "workflow_dashboard", description: "\u8FD4\u56DE\u53EF\u6536\u85CF\u7684\u672C\u673A\u53EF\u89C6\u5316\u9762\u677F\u5730\u5740\uFF0C\u65E0\u9700 token\u3002\u670D\u52A1\u72EC\u7ACB\u4E8E\u804A\u5929\u4F1A\u8BDD\uFF0C\u91CD\u542F\u540E\u590D\u7528\u7AEF\u53E3\u3002", inputSchema: obj({}) } ]; +function parseEventCursor(value) { + if (value === void 0 || value === null || value === "") return 0; + check(typeof value === "number" || typeof value === "string", "\u7B49\u5F85\u53C2\u6570\u65E0\u6548"); + const numeric = typeof value === "number" ? value : Number(value); + check(Number.isSafeInteger(numeric) && numeric >= 0, "\u7B49\u5F85\u53C2\u6570\u65E0\u6548"); + return numeric; +} function summary(snapshot) { return { ...snapshot, script: void 0, input: void 0, fingerprints: void 0, requestHash: void 0, steps: snapshot.steps?.map(({ prompt, input, output, rawOutput, requestHash, ...s }) => s), result: void 0 }; } @@ -26728,8 +26782,8 @@ function createToolHandler(engine, getURL) { return { status: r.status, updatedAt: r.updatedAt, ...args.includeDefinition ? { definition: templateDefinition(r) } : {}, result: r.result, steps: r.steps.slice(offset2, offset2 + limit).map(({ prompt, input, ...s }) => s), nextOffset: offset2 + limit < r.steps.length ? offset2 + limit : null }; } case "workflow_wait": { - const t = args.timeoutMs ?? 25e3, a = args.afterSequence ?? 0; - check(Number.isSafeInteger(t) && t >= 0 && t <= 25e3 && Number.isSafeInteger(a) && a >= 0, "\u7B49\u5F85\u53C2\u6570\u65E0\u6548"); + const t = args.timeoutMs ?? 25e3, a = parseEventCursor(args.afterSequence); + check(Number.isSafeInteger(t) && t >= 0 && t <= 25e3, "\u7B49\u5F85\u53C2\u6570\u65E0\u6548"); return waitEvents(engine, args.runId, a, t); } case "workflow_cancel": @@ -26794,7 +26848,7 @@ async function startHTTP(engine, { port = 0, webRoot = new URL("../web/", import if (req.method === "GET" && url.pathname === "/api/runs") return json(engine.store.list().map(({ script, input, result, fingerprints, ...r }) => r)); const match = url.pathname.match(/^\/api\/runs\/([a-f0-9-]+)(?:\/(wait|pause|cancel|resume|edit|approve|repair))?$/); if (match && req.method === "GET") { - if (match[2] === "wait") return json(await waitEvents(engine, match[1], Math.max(0, Number(url.searchParams.get("after")) || 0), 2e4)); + if (match[2] === "wait") return json(await waitEvents(engine, match[1], parseEventCursor(url.searchParams.get("after")), 2e4)); return json(engine.snapshot(match[1])); } if (req.method === "POST") { diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/scripts/build-web.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/scripts/build-web.mjs index 06c63965..ed25b11f 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/scripts/build-web.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/scripts/build-web.mjs @@ -1,6 +1,6 @@ import {contentStyles} from '../web/readable.mjs'; import {readFile,writeFile} from 'node:fs/promises'; import {build} from 'esbuild'; -await build({entryPoints:['web/app.source.mjs'],outfile:'web/app.js',bundle:true,platform:'browser',format:'esm',target:'es2022',define:{DEFAULT_ENGLISH_EXAMPLE:JSON.stringify(await readFile('examples/audit-en.js','utf8'))}}); +await build({entryPoints:['web/app.source.mjs'],outfile:'web/app.js',bundle:true,platform:'browser',format:'esm',target:'es2022',define:{DEFAULT_ENGLISH_EXAMPLE:JSON.stringify((await readFile('examples/audit-en.js','utf8')).replaceAll('\r\n','\n'))}}); await writeFile('web/readable.css',contentStyles); diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/src/http.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/src/http.mjs index e9c065a7..529cd492 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/src/http.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/src/http.mjs @@ -6,7 +6,7 @@ import { resolveMcode } from './availability.mjs'; import { readFile } from 'node:fs/promises'; import { check } from './common.mjs'; import { previewTopology } from './topology.mjs'; -import { waitEvents,createToolHandler } from './tools.mjs'; +import { waitEvents,createToolHandler,parseEventCursor } from './tools.mjs'; export async function startHTTP(engine,{port=0,webRoot=new URL('../web/',import.meta.url),exampleRoot=new URL('../examples/',import.meta.url)}={}){ const reportStyleHash=createHash('sha256').update(REPORT_STYLES).digest('base64'); let origin; @@ -28,7 +28,7 @@ export async function startHTTP(engine,{port=0,webRoot=new URL('../web/',import. if(req.method==='GET'&&url.pathname==='/api/scheduler')return json(engine.schedulerStatus()); if(req.method==='GET'&&url.pathname==='/api/runs')return json(engine.store.list().map(({script,input,result,fingerprints,...r})=>r)); const match=url.pathname.match(/^\/api\/runs\/([a-f0-9-]+)(?:\/(wait|pause|cancel|resume|edit|approve|repair))?$/); - if(match&&req.method==='GET'){if(match[2]==='wait')return json(await waitEvents(engine,match[1],Math.max(0,Number(url.searchParams.get('after'))||0),20000));return json(engine.snapshot(match[1]));} + if(match&&req.method==='GET'){if(match[2]==='wait')return json(await waitEvents(engine,match[1],parseEventCursor(url.searchParams.get('after')),20000));return json(engine.snapshot(match[1]));} if(req.method==='POST'){ check(req.headers['content-type']?.startsWith('application/json'),'需要 application/json');req.setEncoding('utf8');let body='';for await(const chunk of req){body+=chunk;check(Buffer.byteLength(body)<=700_000,'请求过大');}const data=JSON.parse(body||'{}'); if(url.pathname==='/api/templates')return json(engine.saveTemplate(data.runId,data),201); diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs index 71528326..7daedb35 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs @@ -49,7 +49,7 @@ export class Store { } }catch(error){this.db?.close();this.releaseLock();throw error;} } - transaction(fn) {if(this.txDepth)return fn();this.reserveEventSequences();this.db.exec('BEGIN IMMEDIATE');this.txDepth=1;this.txVolatile=[]; + transaction(fn) {if(this.txDepth)return fn();this.recover('事务');this.reserveEventSequences();this.db.exec('BEGIN IMMEDIATE');this.txDepth=1;this.txVolatile=[]; try{const r=fn();this.db.exec('COMMIT');return r;} catch(e){ // Whether the batch is durable must be read from SQLite, never inferred @@ -82,12 +82,22 @@ export class Store { this.volatileBuffer=[...restored.values()].sort((a,b)=>a.body.seq-b.body.seq); if(this.volatileBuffer.length)this.scheduleVolatileFlush(); } + // Single connection-recovery gate for every read and write path. When both + // COMMIT and ROLLBACK fail, the SQLite transaction stays open but the finally + // block has already cleared txDepth, so the connection looks idle while + // holding uncommitted rows. The next plain write would join that transaction + // and return success for rows a later ROLLBACK discards, and a read would + // publish them as durable. recover() is therefore called before every normal + // path: it rolls a leftover transaction back, and if even that fails it + // rejects the operation loudly instead of acknowledging a write it cannot + // guarantee. txDepth>0 means the open transaction is the current one. + recover(op='操作'){if(!this.db||this.txDepth||!this.db.isTransaction)return; + try{this.db.exec('ROLLBACK');} + catch(error){throw new Error(`状态库残留事务无法回滚，本次${op}已拒绝以免丢失已确认的数据，请重启服务后重试：${error.message}`);}} reserveEventSequences() { - // txDepth 0 with a SQLite transaction still open means an earlier COMMIT - // and ROLLBACK both failed. The queued batch is still in memory; roll the - // leftover transaction back so the next BEGIN can persist it. A rollback - // that itself fails propagates, and the caller keeps the batch queued. - if(!this.txDepth&&this.db.isTransaction)this.db.exec('ROLLBACK'); + // A residual open transaction must end before a new BEGIN or the next + // sequence read would run inside it; recover() rolls it back or rejects. + this.recover('事件序号预留'); const persisted=Number(this.db.prepare("SELECT COALESCE((SELECT seq FROM sqlite_sequence WHERE name='events'),0) AS seq").get().seq); this.nextEventSequence=Math.max(this.nextEventSequence,persisted); if(this.nextEventSequenceJSON.parse(r.body));} - template(id) {const r=this.db.prepare('SELECT body FROM templates WHERE id=?').get(id);return r?JSON.parse(r.body):null;} - saveTemplate(value) {this.db.prepare('INSERT INTO templates VALUES(?,?)').run(value.id,JSON.stringify(value));} - deleteTemplate(id) {return this.db.prepare('DELETE FROM templates WHERE id=?').run(id).changes>0;} - setting(key) {const row=this.db.prepare('SELECT body FROM settings WHERE key=?').get(key);return row?JSON.parse(row.body):undefined;} - saveSetting(key,value) {this.db.prepare('INSERT INTO settings VALUES(?,?) ON CONFLICT(key) DO UPDATE SET body=excluded.body').run(key,JSON.stringify(value));} - save(run) {this.db.prepare('INSERT INTO runs VALUES(?,?,?,?) ON CONFLICT(id) DO UPDATE SET body=excluded.body').run(run.id,run.requestId,run.requestHash,JSON.stringify(run));} - get(id) {const r=this.db.prepare('SELECT body FROM runs WHERE id=?').get(id);return r ? JSON.parse(r.body):null;} - byRequest(id) {const r=this.db.prepare('SELECT body FROM runs WHERE requestId=?').get(id);return r ? JSON.parse(r.body):null;} - 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));} + templates() {this.recover('模板读取');return this.db.prepare('SELECT body FROM templates ORDER BY rowid DESC').all().map(r=>JSON.parse(r.body));} + template(id) {this.recover('模板读取');const r=this.db.prepare('SELECT body FROM templates WHERE id=?').get(id);return r?JSON.parse(r.body):null;} + saveTemplate(value) {this.recover('模板写入');this.db.prepare('INSERT INTO templates VALUES(?,?)').run(value.id,JSON.stringify(value));} + deleteTemplate(id) {this.recover('模板删除');return this.db.prepare('DELETE FROM templates WHERE id=?').run(id).changes>0;} + setting(key) {this.recover('设置读取');const row=this.db.prepare('SELECT body FROM settings WHERE key=?').get(key);return row?JSON.parse(row.body):undefined;} + saveSetting(key,value) {this.recover('设置写入');this.db.prepare('INSERT INTO settings VALUES(?,?) ON CONFLICT(key) DO UPDATE SET body=excluded.body').run(key,JSON.stringify(value));} + save(run) {this.recover('运行状态写入');this.db.prepare('INSERT INTO runs VALUES(?,?,?,?) ON CONFLICT(id) DO UPDATE SET body=excluded.body').run(run.id,run.requestId,run.requestHash,JSON.stringify(run));} + get(id) {this.recover('运行状态读取');const r=this.db.prepare('SELECT body FROM runs WHERE id=?').get(id);return r ? JSON.parse(r.body):null;} + byRequest(id) {this.recover('运行状态读取');const r=this.db.prepare('SELECT body FROM runs WHERE requestId=?').get(id);return r ? JSON.parse(r.body):null;} + list() {this.recover('运行列表读取');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) {this.recover('步骤读取');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) {this.recover('步骤读取');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;} + findCrossRunReuse({contextHash,requestHash,lineageHash,excludeRunId,limit=20}) {this.recover('复用候选读取');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.recover('步骤写入');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) {this.recover('修复候选读取');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.transaction(()=>{const rowid=Number(this.db.prepare('INSERT INTO repair_cache VALUES(?,?,?)').run(runId,step.id,JSON.stringify(step)).lastInsertRowid);this.chainAdvance('repair','repair','SELECT rowid AS pos,runId,id,body FROM repair_cache WHERE rowid>? AND rowid<=? ORDER BY rowid',rowid,r=>`${r.runId}/${r.id}`);});} event(runId,type,data={}) {const event={...data,type,time:Date.now()}; if(this.volatileTypes.has(type))return this.stageVolatile(runId,event); @@ -161,9 +174,14 @@ export class Store { throw error;} } events(runId,after=0,limit=150) { + // A leftover transaction would make the table read return its uncommitted + // rows, duplicating a seq the buffer still owes a write ([1,2,2] views). + // recover() ends it first, so the table only holds committed rows and the + // merge below stays a true union of durable and not-yet-flushed events. + this.recover('事件读取'); const persisted=this.db.prepare('SELECT seq,body FROM events WHERE runId=? AND seq>? ORDER BY seq LIMIT ?').all(runId,after,limit).map(e=>({seq:e.seq,...JSON.parse(e.body)})); const buffered=this.volatileBuffer.filter(e=>e.runId===runId&&e.body.seq>after).map(e=>e.body); - return [...persisted,...buffered].sort((a,b)=>a.seq-b.seq).slice(0,limit); + return [...new Map([...persisted,...buffered].map(e=>[e.seq,e])).values()].sort((a,b)=>a.seq-b.seq).slice(0,limit); } rowHash(prev,kind,key,body) {return createHash('sha256').update(`${prev}:${kind}:${key}:${body}`).digest('hex');} // Bulk adoption of pre-existing rows is an initial-creation behavior only: it @@ -174,8 +192,11 @@ export class Store { chainAdvance(kind,surface,sql,newUpto,keyOf) {const tail=this.setting(`integrity_${surface}`);let prev=tail?.head??'0'.repeat(64); const range=tail?`SELECT * FROM (${sql}) WHERE pos=${newUpto}`:sql; for(const r of this.db.prepare(range).all(tail?.upto??0,newUpto)){const k=keyOf(r);prev=this.rowHash(prev,kind,k,r.body);this.db.prepare('INSERT OR REPLACE INTO integrity_rows VALUES(?,?,?,?)').run(surface,r.pos,k,prev);}this.saveSetting(`integrity_${surface}`,{head:prev,upto:newUpto});} - integrityHeads() {return {events:this.setting('integrity_events')??null,repair:this.setting('integrity_repair')??null};} + integrityHeads() {this.recover('完整性读取');return {events:this.setting('integrity_events')??null,repair:this.setting('integrity_repair')??null};} verifyIntegrity() { + // Verification is a durability verdict, so it must not run against the + // uncommitted view a leftover transaction would expose as real state. + this.recover('完整性校验'); // Each ledger link is re-checked against the live row's own identity columns: // the key is re-derived from the row and must equal the recorded key before that // recorded key may take part in any digest recomputation, so re-attributing a @@ -207,6 +228,10 @@ export class Store { if(this.closing)return;this.closing=true; const errors=[]; try{this.flushVolatile();}catch(error){errors.push(error);} + // db.close() silently discards an open transaction, so a residual one must + // end first: closing here would drop acknowledged-but-uncommitted rows + // without ever reporting an error to the caller. + try{this.recover('关闭');}catch(error){errors.push(error);} try{this.db.close();}catch(error){errors.push(error);} try{this.releaseLock();}catch(error){errors.push(error);} if(errors.length===1)throw errors[0]; diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/src/tools.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/src/tools.mjs index c8bda2f8..a801f8f6 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/src/tools.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/src/tools.mjs @@ -21,6 +21,13 @@ export const TOOLS=[ {name:'workflow_resume',description:'原脚本与原输入恢复，复用已成功节点。可调整 maxSteps/stepTimeoutMs/runTimeoutMs/maxCalls 后重试，成功节点复用，失败节点从头执行。异常退出需要用户先确认旧 Agent 已停止。',inputSchema:obj({...id,confirmStopped:{type:'boolean'},...LIMIT_SCHEMAS,maxCalls:{type:'integer',minimum:1,maximum:100}},['runId'])}, {name:'workflow_dashboard',description:'返回可收藏的本机可视化面板地址，无需 token。服务独立于聊天会话，重启后复用端口。',inputSchema:obj({})} ]; +export function parseEventCursor(value){ + if(value===undefined||value===null||value==='')return 0; + check(typeof value==='number'||typeof value==='string','等待参数无效'); + const numeric=typeof value==='number'?value:Number(value); + check(Number.isSafeInteger(numeric)&&numeric>=0,'等待参数无效'); + return numeric; +} export function summary(snapshot){return {...snapshot,script:undefined,input:undefined,fingerprints:undefined,requestHash:undefined,steps:snapshot.steps?.map(({prompt,input,output,rawOutput,requestHash,...s})=>s),result:undefined};} export async function waitEvents(engine,runId,after=0,timeout=25000){engine.snapshot(runId);const get=()=>engine.store.events(runId,after,100);let events=get();if(!events.length&&engine.active.has(runId)&&timeout>0)await new Promise(resolve=>{const done=()=>{clearTimeout(timer);engine.off('change',listener);resolve();};const listener=e=>{if(e.runId===runId)done();};const timer=setTimeout(done,timeout);engine.on('change',listener);if(get().length)done();});events=get();return {events,nextSequence:events.at(-1)?.seq??after,status:engine.store.get(runId).status};} export function createToolHandler(engine,getURL){return async(name,args={})=>{ @@ -31,7 +38,7 @@ export function createToolHandler(engine,getURL){return async(name,args={})=>{ case 'workflow_repair':return summary(await engine.repair(args.runId,args)); case 'workflow_status':if(args.runId)return summary(engine.snapshot(args.runId));if(args.verifyIntegrity===true)return {runs:engine.store.list().map(r=>summary(r)),integrityHeads:engine.store.integrityHeads(),integrity:engine.store.verifyIntegrity()};return engine.store.list().map(r=>summary(r)); case 'workflow_results':{const r=engine.snapshot(args.runId);const offset=args.offset??0,limit=args.limit??10;check(Number.isInteger(offset)&&offset>=0&&Number.isInteger(limit)&&limit>=1&&limit<=20,'分页参数无效');return {status:r.status,updatedAt:r.updatedAt,...(args.includeDefinition?{definition:templateDefinition(r)}:{}),result:r.result,steps:r.steps.slice(offset,offset+limit).map(({prompt,input,...s})=>s),nextOffset:offset+limit=0&&t<=25000&&Number.isSafeInteger(a)&&a>=0,'等待参数无效');return waitEvents(engine,args.runId,a,t);} + case 'workflow_wait':{const t=args.timeoutMs??25000,a=parseEventCursor(args.afterSequence);check(Number.isSafeInteger(t)&&t>=0&&t<=25000,'等待参数无效');return waitEvents(engine,args.runId,a,t);} case 'workflow_cancel':return summary(await engine.stop(args.runId)); case 'workflow_pause':return summary(await engine.stop(args.runId,'paused')); case 'workflow_resume':return summary(await engine.resume(args.runId,args));