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 new file mode 100644 index 00000000..a7585043 --- /dev/null +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/REVIEW-NOTES.md @@ -0,0 +1,56 @@ +# 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. + +## 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: + +- `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..a874b36e --- /dev/null +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/REVIEW-REPLY.md @@ -0,0 +1,30 @@ +## 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, +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..1f104e03 --- /dev/null +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/durability.check.mjs @@ -0,0 +1,264 @@ +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, TOOLS } 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(); } + }); +}); + +// 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('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('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); + 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/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/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/integrity.check.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/checks/integrity.check.mjs index a510165a..2b95e6ef 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.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});} +}); + 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/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 0b42f464..e2d9dd9b 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=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); @@ -7754,6 +7757,9 @@ 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 = 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); @@ -7769,58 +7775,140 @@ var Store = class { } transaction(fn) { if (this.txDepth) return fn(); - this.txDepth = 1; + this.recover("\u4E8B\u52A1"); + 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"); + let open4 = true; + try { + this.db.exec("ROLLBACK"); + } catch { + } + if (!this.db.isTransaction) 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. + // 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])); + 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() { + 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; + 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"); + this.txDepth = 1; + try { + this.saveSetting("event_sequence_lease", limit); + this.db.exec("COMMIT"); + } catch (error2) { + try { + this.db.exec("ROLLBACK"); + } catch { + } + throw error2; } finally { this.txDepth = 0; } + this.eventSequenceLimit = limit; + } + allocateEventSequence() { + this.reserveEventSequences(); + 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 @@ -7828,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; } @@ -7848,14 +7939,61 @@ 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); + 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 = { ...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 || this.closing) 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 = []; + 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)); + 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.restoreVolatile(batch); + 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) })); + 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 [...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"); @@ -7877,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}`; @@ -7910,15 +8050,43 @@ 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.db.close(); - this.releaseLock(); + if (this.closing) return; + this.closing = true; + const errors = []; + try { + this.flushVolatile(); + } catch (error2) { + errors.push(error2); + } + try { + this.recover("\u5173\u95ED"); + } 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"); } }; @@ -26556,12 +26724,19 @@ 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"]) }, { 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 }; } @@ -26607,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.isInteger(t) && t >= 0 && t <= 25e3 && Number.isInteger(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": @@ -26673,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") { @@ -27779,10 +27954,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;roundr)); 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/main.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/src/main.mjs index de57baca..527366bc 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/src/main.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/src/main.mjs @@ -89,7 +89,11 @@ if(values.stdio&&process.env.MCODE_WORKFLOW_CHILD==='1'){ const temp=endpointPath+'.'+process.pid+'.tmp';await writeFile(temp,JSON.stringify({pid:process.pid,url:panel.url,workspace,serviceProtocol:2}),{mode:0o600});await rename(temp,endpointPath); }catch(e){await engine?.close();await panel?.close();store.close();throw e;} process.stdout.write(`Workflow Studio: ${panel.url}\n`); - let closing=false;async function close(){if(closing)return;closing=true;await engine.close();await panel.close();store.close();process.exitCode=0;} + let closing=false;async function close(){if(closing)return;closing=true;const errors=[]; + for(const cleanup of [()=>engine.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 5ac9fc16..7daedb35 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/src/store.mjs @@ -13,13 +13,24 @@ 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; + // 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); @@ -28,6 +39,9 @@ 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=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); @@ -35,29 +49,140 @@ 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;}} - templates() {return this.db.prepare('SELECT body FROM templates ORDER BY rowid DESC').all().map(r=>JSON.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));} + 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 + // 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{} + if(!this.db.isTransaction)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. + // 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])); + 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() { + // 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) {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()};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=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={...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||this.closing)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=[]; + 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)); + 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){ + // 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;} + } + 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 [...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 // anchors whatever the table held when the chain first appears. Once a head @@ -67,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 @@ -95,6 +223,18 @@ 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.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);} + // 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]; + if(errors.length)throw new AggregateError(errors,'Store shutdown failed'); + } } diff --git a/plugins/hetaoBackend/mcode-dynamic-workflows/src/tools.mjs b/plugins/hetaoBackend/mcode-dynamic-workflows/src/tools.mjs index 09731d5f..a801f8f6 100644 --- a/plugins/hetaoBackend/mcode-dynamic-workflows/src/tools.mjs +++ b/plugins/hetaoBackend/mcode-dynamic-workflows/src/tools.mjs @@ -15,12 +15,19 @@ 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'])}, {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.isInteger(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));