diff --git a/CHANGELOG.md b/CHANGELOG.md index 5e85a1915..ea14c27ba 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,8 @@ - **PLAN_REALITY wants a receipt per measured number and the rule itself.** Each clause stating a measured quantity (seconds, %, $, tokens, MB) needs its own file:line, script or command receipt (`unreceipted_numbers=` plus one `NUMBER_UNRECEIPTED:` line each), and a line claiming a decision rule must quote or link it (`unquoted=`). Receipts on dot paths such as `.gstack/ship-policy.json:1` count. `submit` records the check for each plan revision; it never refuses a submit. - **`autoplan/DRIVER.md` is the compact driver.** About a hundred lines generated from the code: the phase order, the artifact contract, the gate protocol and the coordinator contract, with no hook prose. `gstack-autoplan --help` and the first `next` print its path, and any host can read it instead of the full autoplan skill. +- **Six submits at once bind once.** `run.lock` is now created with its owner already written (a staged file linked into place), and an unreadable lock younger than 30 seconds counts as held, so a command that reads another's lock between create and write no longer deletes it and runs alongside it. + ### For contributors - New modules: `lib/autoplan-revise.ts`, `lib/autoplan-migrate.ts`, `lib/autoplan-plan-text.ts`, `lib/autoplan-admission.ts`, `lib/autoplan-scope.ts`, `lib/source-root.ts`, `scripts/resolvers/autoplan-driver.ts`. diff --git a/lib/autoplan-run.ts b/lib/autoplan-run.ts index 1f5eaf364..a3f66fbdc 100644 --- a/lib/autoplan-run.ts +++ b/lib/autoplan-run.ts @@ -14,6 +14,7 @@ * EXECUTION_UNKNOWN, never redispatching them on its own. Thin bin: * bin/gstack-autoplan. */ +import * as crypto from 'node:crypto'; import * as fs from 'node:fs'; import * as os from 'node:os'; import * as path from 'node:path'; @@ -31,28 +32,35 @@ import { outsideConsent } from './outside-consent'; import { familyConflict, modelFamily } from './outside-voice-runner'; import { renderSpend, reserve, settle, summarize, readLedger } from './spend-ledger'; +/** An unreadable run.lock younger than this is a lock some other gstack is still writing, never a stale one. */ +const LOCK_WRITE_GRACE_MS = 30_000; + /** Exclusive ownership for one command: a lock whose pid is alive is never stolen; a dead pid's lock is reclaimed. */ export function withRunLock(out: string, fn: () => T, opts: { steal?: boolean } = {}): T { fs.mkdirSync(out, { recursive: true }); const lock = path.join(out, FILES.lock); const mine = JSON.stringify({ pid: process.pid, host: os.hostname(), at: now() }); + const staged = `${lock}.${process.pid}.${crypto.randomBytes(4).toString('hex')}`; + fs.writeFileSync(staged, mine); for (let i = 0; ; i++) { try { - const fd = fs.openSync(lock, 'wx'); - fs.writeSync(fd, mine); - fs.closeSync(fd); + fs.linkSync(staged, lock); + fs.unlinkSync(staged); break; } catch (e: any) { - if (e.code !== 'EEXIST') throw e; + if (e.code !== 'EEXIST') { fs.rmSync(staged, { force: true }); throw e; } let owner: { pid?: number; host?: string; at?: string } = {}; - try { owner = JSON.parse(fs.readFileSync(lock, 'utf8')); } catch { /* unreadable: treat as stale below */ } + let unreadable = false; + try { owner = JSON.parse(fs.readFileSync(lock, 'utf8')); } catch { unreadable = true; } let alive = false; - if (owner.host === os.hostname() && typeof owner.pid === 'number') { + if (unreadable) { + try { alive = Date.now() - fs.statSync(lock).mtimeMs < LOCK_WRITE_GRACE_MS; } catch { /* gone: retry the link */ } + } else if (owner.host === os.hostname() && typeof owner.pid === 'number') { try { process.kill(owner.pid, 0); alive = true; } catch (k: any) { alive = k.code === 'EPERM'; } } else if (owner.host && owner.host !== os.hostname()) alive = !opts.steal; - if (alive) throw new RunError('RUN_LOCKED', `pid ${owner.pid} on ${owner.host} since ${owner.at}`, 3); + if (alive) { fs.rmSync(staged, { force: true }); throw new RunError('RUN_LOCKED', unreadable ? `${lock} is being written by another command` : `pid ${owner.pid} on ${owner.host} since ${owner.at}`, 3); } try { fs.unlinkSync(lock); } catch { /* raced with its owner's exit */ } - if (i > 50) throw new RunError('RUN_LOCKED', `could not reclaim ${lock}`, 3); + if (i > 50) { fs.rmSync(staged, { force: true }); throw new RunError('RUN_LOCKED', `could not reclaim ${lock}`, 3); } } } try { return fn(); } finally { try { if (fs.readFileSync(lock, 'utf8') === mine) fs.unlinkSync(lock); } catch { /* already gone */ } } diff --git a/test/autoplan-run-concurrency.test.ts b/test/autoplan-run-concurrency.test.ts index 1a23fd1a4..b8ac8aa68 100644 --- a/test/autoplan-run-concurrency.test.ts +++ b/test/autoplan-run-concurrency.test.ts @@ -24,6 +24,22 @@ async function parallel(argvs: string[][]): Promise { + test('a lock caught between create and write is held, not stolen: a fresh empty run.lock refuses RUN_LOCKED; an old one is reclaimed', () => { + const out = startSmallRun(path.join(TMP, 'half-written')); + const lock = path.join(out, 'run.lock'); + const next = () => Bun.spawnSync(['bun', AUTOPLAN, 'next', '--out', out], { cwd: ROOT, env: env(), timeout: 60_000 }); + fs.writeFileSync(lock, ''); + const held = next(); + expect(held.exitCode).toBe(3); + expect(held.stderr.toString()).toMatch(/\(RUN_LOCKED\)$/m); + expect(fs.existsSync(lock)).toBe(true); + const old = new Date(Date.now() - 60_000); + fs.utimesSync(lock, old, old); + const reclaimed = next(); + expect(reclaimed.exitCode, reclaimed.stderr.toString()).toBe(0); + expect(fs.existsSync(lock)).toBe(false); + }); + test('six concurrent submits of the same attempt: exactly one binds, the rest are refused, one journal row, one settlement', async () => { const out = startSmallRun(path.join(TMP, 'submit'), ['--estimate-usd', '1']); const ids = attemptsOf(out, 'ceo');