diff --git a/LESSONS.md b/LESSONS.md index ae3dab42e..ca91ae291 100644 --- a/LESSONS.md +++ b/LESSONS.md @@ -665,6 +665,10 @@ Seen 1× while implementing #2179. Reusing old group aggregation would accept ca #2179 integration regression: `git rev-parse MERGE_HEAD` returns only the first parent of an octopus merge. Validate every bounded registered MERGE_HEAD entry before sealing; an admitted first source cannot conceal an unapproved second parent. Preserve the staged index on rejection. +### 2026-11-03 — Migration must distinguish unfinished import from missing current data + +An interrupted prepared SQL migration may recover its verified import; an active database that disappears must fail closed, not silently revive the old import. Publish a newer-version JSON guard before initialization, validate one authority, and retain a fenced resumable rollback marker across partial moves. A transaction commit must precede Effects acknowledgement without a fallible post-commit read masking that proof. + ### 2026-11-03 — Public Chief cannot borrow its approver's identity @@ -747,3 +751,7 @@ Seen 1× (composed actual DSH native input failure). A canonical claim keeps the ### 2026-11-03 — Fence old-plan decisions as well as external effects Seen 1× (late Task requirement response regression). Stopping on the next scheduler pass alone leaves a window for an old turn to transfer/split. Bind current claims to the Task requirement revision and recheck after authorization; use actual old Attempt snapshot revisions for compatible prior claims. Keep publication evidence fences too. + +### 2026-11-03 — State maps do not encode chronology + +Seen 1× (actual SQLite native return probe). Two valid Attempts survived, but UUID-key SQL iteration put the successor first and an insertion-order assertion compared it to itself. Use persisted predecessor references for chronology; preserve explicit history positions. Storage order must not decide current execution or evidence. diff --git a/PROJECT.md b/PROJECT.md index ab060305b..14726c16a 100644 --- a/PROJECT.md +++ b/PROJECT.md @@ -52,60 +52,7 @@ Detailed S1–S6 implementation/recovery history is [archived verbatim](docs/his ## Historical upstream Project State -## Mission - -FlyBot is a local team of long-lived AI Bots that runs on DeepSeek Harness (DSH), in your own -browser, with your own model keys. One person talks to one Bot — the **Chief** — and the team takes -the request apart and comes back with one answer. The Bots keep their own memory, working folder, -model and schedules between turns, so work outlives a single chat. - -The project's mission is to keep that team reliable and inspectable: every Bot acts on evidence, -stays inside its goal, and its result can be checked by a person afterwards. The product is the app -in this repository plus the patches that pin it to upstream `ds-bot`; it is not a hosting service. - -`AGENTS.md` states the two halves the code is built from: pristine upstream `ds-bot/` (never edited), -and `core/` — `ds-bot/` plus FlyBot's ordered patches — which is what ships and what the tests run. -The shipped features are the DSH plugin in `core/` and the peer plugins `plugins/social`, -`plugins/tree`, `plugins/timeline` and `plugins/workflow`. - -## Owner and maintainer roles - -Today the project has one owner and a small number of human maintainers; the Bots work under the -owner's delegation. This is the state of the repository as of 2026-10-29, not the target model of -[#1550](https://github.com/ThinkFlowLab/FlyBot/issues/1550). - -| Role | Who | Decides | -| --- | --- | --- | -| **Owner** | Danyang, GitHub [`Birfy`](https://github.com/Birfy) (repository admin) | Mission, governance policy, roles and membership, project Bots, compute and resources, budgets, autonomy boundaries, and every matter reserved to the owner in [docs/governance.md](docs/governance.md) | -| **Maintainer** (human) | People with repository write access — today the owner and [`hsliuustc0106`](https://github.com/hsliuustc0106) | Major decisions under [docs/governance.md](docs/governance.md): roadmap and architecture changes, changes to the governing documents, multi-subsystem contracts. Records a verdict through the review policy, never through an admin merge | -| **Chief** (AI) | The owner's Main Bot, writing as `Birfy` | Plans and routes work, splits a request into role tasks, **classifies** each decision as ordinary / major / severe with its rationale, convenes a discussion when one is needed, and escalates to the owner. Not the ultimate authority and not the owner | -| **Routine Bots** (AI) | `Triage`, `Dev`, `Dev2`–`Dev4`, `Reviewer`, `Reviewer2`, `Release`, `SelfTest`, `Docs`, `Maintainer` — the names in [`writers`](.github/agent/review-policy.json) | Only their routine's job: see [Routines](AGENTS.md#routines). A Bot records a conclusion only where its routine owns that conclusion, and the implementer never writes the verdict on its own change | -| **Contributors** (human) | Anyone who opens an issue or a pull request | Proposals, bug reports, code. No default approval right; a contribution merges only through the gates in [AGENTS.md](AGENTS.md) | - -Deliberately **not** claimed here, because it is not shipped: a Maintainer console, a Member role, a -Project Assistant, per-person GitHub attribution for Bots, and the multi-human authority contract in -[docs/subsystems/team.md](docs/subsystems/team.md) (planned; team mode stays disabled until its -authorization gates pass). Every Bot still writes as the one shared `Birfy` identity, so a -`Written-by:` line is a claim the gates can check for presence and name, not a signature. - -## Resources covered - -- **This repository** — `https://github.com/ThinkFlowLab/FlyBot`, the source of record. `main` is - the live branch; pull requests merge into it only through the merge queue. -- **Code** — `core/` (shipped plugin, source of truth), `patches/ds-bot/` and `core/client.js` - (generated from `core/`), `plugins/*` (peer plugins), `scripts/` and `core/scripts/`. -- **Rules and knowledge** — `AGENTS.md` (operating contract), `CONTRIBUTING.md`, `LESSONS.md` - (observed failures and promotion to standards), `docs/` (architecture, standards, runbooks), - `decisions/` (dated decisions), `PROJECT.md` and `ROADMAP.md` (this page and the plan). -- **Automation** — `.github/workflows/` and `.github/scripts/` (CI, risk tier, review gate, merge - gate, enqueue guard), `.github/agent/` (risk rules and the review policy). -- **Authoritative runtime state elsewhere** — GitHub issues, pull requests and CI runs are the - authority for work state; task boards, approvals and per-Bot memory live in FlyBot's own storage, - never in this repository. `docs/` pages are informative only where a source above decides - something. -- **Not in this repository** — the upstream `ds-bot` project (pinned in `ds-bot.pin.json`, read-only - here) and the self-hosted Bot Social server used by the social plugin. The fork - `Birfy/FlyBot` is an archive: nothing merges into it. +Historical upstream mission, named roles and resource inventory are [preserved verbatim](docs/history/2026-11-03-upstream-project-inventory.md). Current fork ownership and capabilities follow the active state and dated decisions above. The governing and automation boundaries below remain active. ## Governing boundaries @@ -372,4 +319,10 @@ New shared Tasks have Chief admission and versioned requirements. Existing Task/ #2179 follow-up: Chief dependency waivers/substitutions retain exact outcome versions and explicit judgement evidence. Accepted child code enters the existing durable integration gates before parent execution; source admission is not merging or parent acceptance. Actual Git multi-parent sealing tests use deterministic child acceptance fixtures. Full real-provider parallel collaboration and native physical restoration remain pending. +### 2026-11-03 — Transactional state (#2182) + +Opt-in SQL retains the original Store/services and saves domain facts plus Effects atomically. JSON becomes a newer-version migration guard, with verified import/recovery and offline export/rollback; no dual writable snapshots. Histories are normalized and large immutable payloads hash-referenced. Routine state updates/views make no additional model calls or Project-position changes. Full-state hydration/scanning remains; storage growth is distinct from #2177 context budgets. No live migration or provider-loop claim. + Composition checkpoint: current role governance/native exit, bounded context and canonical Task waits now share one execution chain. Native input fences logical role/physical worker separately; Task requirement revision changes stop the old Attempt through existing pool release. Actual composed/native acceptance and remaining API/storage compatibility are still under verification, not deployment. + +SQLite composition now has an actual DSH native return probe: canonical Chief intake, persisted activation/exit, original role context and single-resource handoff are read from committed SQL snapshots. Inference remains deterministic; no production migration or full-provider workflow acceptance is claimed. diff --git a/ROADMAP.md b/ROADMAP.md index 9e371cc6d..d1e2b6d46 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -448,6 +448,8 @@ New shared Tasks have Chief admission and versioned requirements. Existing Task/ #2179 follow-up: Chief dependency waivers/substitutions retain exact outcome versions and explicit judgement evidence. Accepted child code enters the existing durable integration gates before parent execution; source admission is not merging or parent acceptance. Actual Git multi-parent sealing tests use deterministic child acceptance fixtures. Full real-provider parallel collaboration and native physical restoration remain pending. +#2182 implements opt-in transactional SQLite through the existing Store, exact-identity migration, process-death commit tests and recoverable offline rollback. Retain configured Chief semantic decisions while routine facts/queries remain deterministic. Public scoped event/receipt service #2185, legacy entrypoint cutover #2186 and final real deployment/README #2189 remain subsequent stages; current production state is unchanged. + Native prior-host reconciliation now reuses verified supervisor process-group facts; native v2 registered-work/lineage and shared adapter-exit regressions guard against premature initiative transfer. In-flight current-rights abort and composed-stack acceptance remain active. Canonical ongoing role/resource authority checks now reuse native step/SDK input and the existing scheduler + pool stop. Frozen role context remains stable; current revoked rights stop execution, with unknown exit held. Compose descendants and verify parent Task resume/API projections next. diff --git a/changelog.d/2182.added.md b/changelog.d/2182.added.md new file mode 100644 index 000000000..d9343241e --- /dev/null +++ b/changelog.d/2182.added.md @@ -0,0 +1 @@ +Add opt-in transactional SQLite state with atomic Effects, normalized history, hash-verified payloads, guarded JSON migration and recoverable offline export/rollback; retain existing Project APIs and deterministic state updates. Verify canonical Chief/native Task-role continuation, activation and exit proofs against committed SQLite snapshots without relying on map insertion order. diff --git a/core/scripts/state-storage.mjs b/core/scripts/state-storage.mjs new file mode 100644 index 000000000..b85015282 --- /dev/null +++ b/core/scripts/state-storage.mjs @@ -0,0 +1,25 @@ +// Offline private operator interface; no account credentials are accepted here. +import { resolve } from 'node:path' +import { readFileSync } from 'node:fs' +import { exportSqliteState, rollbackSqliteState } from '../src/host/project-state-migration.js' + +try { + const [command, ...args] = process.argv.slice(2), options = {} + if (!['inspect', 'export', 'rollback'].includes(command)) throw new Error('Usage: node core/scripts/state-storage.mjs inspect|export|rollback --home PATH [--migration ID] [--output PATH] [--resume]') + for (let i = 0; i < args.length; i++) { + const key = args[i] + if (!['--home', '--migration', '--output', '--resume'].includes(key) || Object.hasOwn(options, key)) throw new Error('Unknown or duplicate migration option') + if (key === '--resume') options[key] = true + else if (!args[i + 1] || args[i + 1].startsWith('--')) throw new Error('Migration option requires a value') + else options[key] = args[++i] + } + if (!options['--home']) throw new Error('An explicit private --home is required') + const home = resolve(options['--home']) + let result + if (command === 'inspect') { + const guard = JSON.parse(readFileSync(resolve(home, 'state.json'), 'utf8')) + result = { backend: guard.storage === 'sqlite' ? 'sqlite' : 'json', phase: guard.phase ?? null, migrationId: guard.migration?.id ?? null } + } else if (command === 'export') result = exportSqliteState({ home, expectedMigrationId: options['--migration'], output: options['--output'] ? resolve(options['--output']) : null }) + else result = rollbackSqliteState({ home, expectedMigrationId: options['--migration'], resume: options['--resume'] === true }) + process.stdout.write(JSON.stringify(result) + '\n') +} catch (error) { process.stderr.write(`State storage operation refused: ${error.message}\n`); process.exitCode = 1 } diff --git a/core/src/host/project-state-migration.js b/core/src/host/project-state-migration.js new file mode 100644 index 000000000..775aa3757 --- /dev/null +++ b/core/src/host/project-state-migration.js @@ -0,0 +1,84 @@ +// Offline operator tooling. No endpoint grants migration privileges to a Bot. +import { createHash, randomUUID } from 'node:crypto' +import { existsSync, readFileSync, renameSync } from 'node:fs' +import { join } from 'node:path' +import { homeLock } from './home-lock.js' +import { openSqliteState } from './project-state-sqlite.js' +import { writeDurableStateFile } from './project-state-storage.js' + +const digest = text => createHash('sha256').update(text).digest('hex') +const fail = message => Object.assign(new Error(message), { code: 'bot/state-migration' }) +function exclusively(home, action) { + const path = join(home, 'state.lock') + if (existsSync(path) && JSON.parse(readFileSync(path, 'utf8'))?.pid === process.pid) throw fail('Stop the current FlyBot host before migration') + const lock = homeLock(path) + if (lock.claim() !== undefined) throw fail('Stop the active FlyBot host before offline migration') + try { return action() } finally { lock.release() } +} +function guardOf(home, expectedMigrationId) { + const guard = JSON.parse(readFileSync(join(home, 'state.json'), 'utf8')) + if (guard.version !== 3 || guard.storage !== 'sqlite' || guard.phase !== 'active' || !expectedMigrationId || guard.migration?.id !== expectedMigrationId) throw fail('The active SQLite migration ID changed; inspect before exporting or rolling back') + return guard +} +function snapshot(home, expectedMigrationId) { + const guard = guardOf(home, expectedMigrationId) + const db = openSqliteState({ path: join(home, 'state.sqlite'), readOnly: true }) + try { + const metadata = db.metadata() + if (metadata.migration?.id !== guard.migration.id || metadata.migration?.sha256 !== guard.migration.sha256) throw fail('SQLite and guard identities differ') + const state = db.read() + if (state.version !== 2) throw fail('Unsupported exported state version') + guardOf(home, expectedMigrationId) + return { text: JSON.stringify(state, null, 2), revision: metadata.revision } + } finally { db.close() } +} +export function exportSqliteState({ home, output, expectedMigrationId }) { + if (!output) throw fail('Choose a private export output file') + return exclusively(home, () => { + const current = snapshot(home, expectedMigrationId) + writeDurableStateFile(output, current.text, { exclusive: true }) + return { backend: 'sqlite', revision: current.revision, sha256: digest(current.text), output } + }) +} +export function rollbackSqliteState({ home, expectedMigrationId, resume = false }) { + return exclusively(home, () => { + const markerPath = join(home, 'state-storage-rollback.json'), path = join(home, 'state.sqlite'), statePath = join(home, 'state.json') + const previous = existsSync(markerPath) ? JSON.parse(readFileSync(markerPath, 'utf8')) : null + let marker + if (previous?.phase === 'prepared') { + if (!resume || previous.migrationId !== expectedMigrationId) throw fail('A prepared rollback exists; explicitly resume its exact migration ID') + marker = previous + } else { + if (resume && previous?.phase === 'completed' && previous.migrationId === expectedMigrationId) return { ...previous, already: true } + if (resume) throw fail('No matching prepared rollback exists') + const current = snapshot(home, expectedMigrationId), id = randomUUID() + marker = { version: 'flybot-storage-rollback/v1', phase: 'prepared', id, migrationId: expectedMigrationId, revision: current.revision, + exportFile: `state.rollback-${id}.json`, archiveFile: `state.sqlite.rollback-${id}`, sha256: digest(current.text), preparedAt: Date.now() } + writeDurableStateFile(join(home, marker.exportFile), current.text, { exclusive: true }) + writeDurableStateFile(markerPath, JSON.stringify(marker, null, 2)) + } + if (marker.version !== 'flybot-storage-rollback/v1' || !/^[a-f0-9-]{36}$/.test(marker.id) || marker.archiveFile !== `state.sqlite.rollback-${marker.id}` || marker.exportFile !== `state.rollback-${marker.id}.json` || !/^[a-f0-9]{64}$/.test(marker.sha256)) throw fail('Rollback recovery marker is malformed') + const text = readFileSync(join(home, marker.exportFile), 'utf8') + if (digest(text) !== marker.sha256 || JSON.parse(text).version !== 2) throw fail('Rollback export hash changed') + const currentText = readFileSync(statePath, 'utf8') + if (digest(currentText) !== marker.sha256) { + guardOf(home, expectedMigrationId) + if (existsSync(path)) { + const current = snapshot(home, expectedMigrationId) + if (current.revision !== marker.revision || digest(current.text) !== marker.sha256) throw fail('SQL state changed after rollback preparation') + } else if (!existsSync(join(home, marker.archiveFile))) throw fail('Rollback database archive is missing') + // Keep DB/WAL/SHM names paired. A crash between moves leaves the marker and + // verified export; resume finishes it before any runtime can write again. + for (const suffix of ['', '-wal', '-shm']) { + const source = path + suffix, target = join(home, marker.archiveFile) + suffix + if (!existsSync(source)) continue + if (existsSync(target)) throw fail('Rollback archive already contains a conflicting file') + renameSync(source, target) + } + writeDurableStateFile(statePath, text) + } + marker = { ...marker, phase: 'completed', completedAt: Date.now() } + writeDurableStateFile(markerPath, JSON.stringify(marker, null, 2)) + return { ...marker, backend: 'json', removeStateStorageSqliteConfig: true } + }) +} diff --git a/core/src/host/project-state-sqlite.js b/core/src/host/project-state-sqlite.js new file mode 100644 index 000000000..9924ec87a --- /dev/null +++ b/core/src/host/project-state-sqlite.js @@ -0,0 +1,223 @@ +// Same runtime records, transactional durable authority. No model, dispatcher or +// second queue lives here. History rows and immutable payloads bound SQL records. +import { createRequire } from 'node:module' +import { createHash, randomUUID } from 'node:crypto' +import { constants, chmodSync, closeSync, existsSync, fsyncSync, fstatSync, linkSync, lstatSync, mkdirSync, openSync, readSync, realpathSync, unlinkSync, writeFileSync } from 'node:fs' +import { dirname, join, resolve } from 'node:path' + +const require = createRequire(import.meta.url) +const APPLICATION_ID = 0x464c5942, SCHEMA = 1, FILE_THRESHOLD = 65536, MAX_PAYLOAD = 16 * 1024 * 1024 +const historyNames = new Set(['history', 'waitHistory', 'requirementsHistory', 'results', 'roleSessionHistory', 'resumptions', 'versions']) +const hash = value => createHash('sha256').update(value).digest('hex') +const fail = message => Object.assign(new Error(message), { code: 'bot/state-sqlite' }) +const record = value => value !== null && typeof value === 'object' && !Array.isArray(value) +const put = (object, key, value) => Object.defineProperty(object, key, { value, enumerable: true, writable: true, configurable: true }) + +export function requireSqliteRuntime() { + const [major, minor] = process.versions.node.split('.').map(Number) + if (major < 22 || major === 22 && minor < 18) throw fail('SQLite state requires Node.js22.18 or newer; JSON storage has not been migrated') + return require('node:sqlite').DatabaseSync +} + +function normalize(value) { + const histories = [] + function visit(row, path) { + if (Array.isArray(row) && historyNames.has(path.at(-1))) { + row.forEach((item, position) => histories.push({ path: JSON.stringify(path), position, text: JSON.stringify(item) })) + return [] + } + if (Array.isArray(row)) return row.map((item, index) => visit(item, [...path, index])) + if (!record(row)) return row + const result = {} + for (const [key, item] of Object.entries(row)) put(result, key, visit(item, [...path, key])) + return result + } + return { document: JSON.stringify(visit(value, [])), fingerprint: hash(JSON.stringify(value)), histories } +} + +function snapshotRecords(snapshot) { + const records = [], roots = {} + for (const [key, value] of Object.entries(snapshot)) { + if (key === '$root') throw fail('Reserved state namespace') + put(roots, key, record(value) ? 'map' : 'value') + if (record(value)) for (const [id, row] of Object.entries(value)) records.push({ namespace: key, id, ...normalize(row) }) + else records.push({ namespace: '$root', id: key, ...normalize(value) }) + } + return { roots: JSON.stringify(roots), records } +} + +export function openSqliteState({ path, payloadRoot = join(dirname(path), 'state-payloads'), readOnly = false, initial, migration = null }) { + path = resolve(path); payloadRoot = resolve(payloadRoot) + let info + try { info = lstatSync(path) } catch (error) { if (error.code !== 'ENOENT') throw error } + if (info && (!info.isFile() || info.isSymbolicLink())) throw fail('SQLite state must be a registered regular file') + if (!readOnly) mkdirSync(dirname(path), { recursive: true, mode: 0o700 }) + const DatabaseSync = requireSqliteRuntime() + if (!readOnly && !info) { const fd = openSync(path, 'wx', 0o600); closeSync(fd) } + const db = new DatabaseSync(path, { readOnly, timeout: 5000, enableForeignKeyConstraints: true, enableDoubleQuotedStringLiterals: false, allowExtension: false }) + let closed = false, expectedRevision = null + const scalar = sql => Object.values(db.prepare(sql).get())[0] + const rollback = () => { try { db.exec('ROLLBACK') } catch { /* No transaction remains after certain SQLite failures. */ } } + const meta = key => db.prepare('SELECT value FROM state_meta WHERE key=?').get(key)?.value + const setMeta = (key, value) => db.prepare('INSERT INTO state_meta(key,value) VALUES (?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value').run(key, String(value)) + function ensurePayloadRoot(create = false) { + if (!existsSync(payloadRoot) && !create) throw fail('State payload directory is missing') + if (!existsSync(payloadRoot)) mkdirSync(payloadRoot, { recursive: true, mode: 0o700 }) + if (!lstatSync(payloadRoot).isDirectory() || lstatSync(payloadRoot).isSymbolicLink() || realpathSync(payloadRoot) !== payloadRoot) throw fail('State payload directory binding changed') + } + function readPayload(row) { + let text = row.document + if (row.storage === 'file') { + if (!/^[a-f0-9]{64}$/.test(row.digest)) throw fail('Invalid state payload reference') + ensurePayloadRoot() + const file = join(payloadRoot, row.digest + '.json'), info = lstatSync(file) + if (!info.isFile() || info.isSymbolicLink() || info.size > MAX_PAYLOAD) throw fail('Invalid bounded state payload file') + const fd = openSync(file, constants.O_RDONLY | (constants.O_NOFOLLOW ?? 0)) + try { + const current = fstatSync(fd) + if (current.ino !== info.ino || current.dev !== info.dev) throw fail('State payload changed during read') + const buffer = Buffer.alloc(MAX_PAYLOAD + 1), size = readSync(fd, buffer, 0, buffer.length, 0) + if (size > MAX_PAYLOAD) throw fail('State payload exceeds its read budget') + text = buffer.subarray(0, size).toString('utf8') + } finally { closeSync(fd) } + } else if (row.storage !== 'inline' || typeof text !== 'string') throw fail('Invalid state payload storage') + if (Buffer.byteLength(text) > MAX_PAYLOAD || hash(text) !== row.digest) throw fail('State payload content hash changed') + return JSON.parse(text) + } + function payload(text) { + const digest = hash(text), size = Buffer.byteLength(text) + if (size > MAX_PAYLOAD) throw fail('State record exceeds bounded payload size; retain a file reference instead') + if (size <= FILE_THRESHOLD) return { storage: 'inline', document: text, digest } + ensurePayloadRoot(true) + const file = join(payloadRoot, digest + '.json') + if (!existsSync(file)) { + const temp = join(payloadRoot, `.${randomUUID()}.tmp`), fd = openSync(temp, 'wx', 0o600) + try { writeFileSync(fd, text, 'utf8'); fsyncSync(fd) } finally { closeSync(fd) } + try { linkSync(temp, file) } catch (error) { if (error.code !== 'EEXIST') throw error } + finally { unlinkSync(temp) } + const directory = openSync(payloadRoot, 'r') + try { fsyncSync(directory) } finally { closeSync(directory) } + } + readPayload({ storage: 'file', document: null, digest }) + return { storage: 'file', document: null, digest } + } + function write(snapshot, { initializing = false } = {}) { + if (closed || readOnly) throw fail('SQLite state is closed or read-only') + if (!record(snapshot) || snapshot.version !== 2) throw fail('SQLite needs a validated version-2 state') + const prepared = snapshotRecords(snapshot) + if (!initializing) db.exec('BEGIN IMMEDIATE') + try { + const revision = Number(meta('revision') ?? 0) + if (revision !== expectedRevision) throw fail('SQLite state revision conflict; reload before changing facts') + const previous = new Map(db.prepare('SELECT namespace,id,fingerprint FROM state_records').all().map(row => [JSON.stringify([row.namespace, row.id]), row])) + const changes = [] + const upsert = db.prepare('INSERT INTO state_records(namespace,id,document,storage,digest,fingerprint) VALUES (?,?,?,?,?,?) ON CONFLICT(namespace,id) DO UPDATE SET document=excluded.document,storage=excluded.storage,digest=excluded.digest,fingerprint=excluded.fingerprint') + const historyUpsert = db.prepare('INSERT INTO state_history(namespace,id,path,position,document,storage,digest) VALUES (?,?,?,?,?,?,?) ON CONFLICT(namespace,id,path,position) DO UPDATE SET document=excluded.document,storage=excluded.storage,digest=excluded.digest') + for (const row of prepared.records) { + const key = JSON.stringify([row.namespace, row.id]), prior = previous.get(key) + previous.delete(key) + if (prior?.fingerprint === row.fingerprint) continue + const data = payload(row.document) + upsert.run(row.namespace, row.id, data.document, data.storage, data.digest, row.fingerprint) + const oldHistory = new Map(db.prepare('SELECT path,position,digest FROM state_history WHERE namespace=? AND id=?').all(row.namespace, row.id).map(item => [JSON.stringify([item.path, item.position]), item])) + for (const entry of row.histories) { + const key = JSON.stringify([entry.path, entry.position]), old = oldHistory.get(key), digest = hash(entry.text) + oldHistory.delete(key) + if (old?.digest === digest) continue + const data = payload(entry.text) + historyUpsert.run(row.namespace, row.id, entry.path, entry.position, data.document, data.storage, data.digest) + } + for (const item of oldHistory.values()) db.prepare('DELETE FROM state_history WHERE namespace=? AND id=? AND path=? AND position=?').run(row.namespace, row.id, item.path, item.position) + changes.push({ namespace: row.namespace, id: row.id, operation: 'upsert' }) + } + for (const row of previous.values()) { + db.prepare('DELETE FROM state_records WHERE namespace=? AND id=?').run(row.namespace, row.id) + changes.push({ namespace: row.namespace, id: row.id, operation: 'delete' }) + } + const rootsChanged = meta('roots') !== prepared.roots + const next = revision + (changes.length || rootsChanged ? 1 : 0) + if (rootsChanged) setMeta('roots', prepared.roots) + if (next !== revision) { + setMeta('revision', next) + for (const row of changes.sort((a, b) => a.namespace.localeCompare(b.namespace) || a.id.localeCompare(b.id))) db.prepare('INSERT INTO state_changes(revision,namespace,id,operation) VALUES (?,?,?,?)').run(next, row.namespace, row.id, row.operation) + } + const lastSequence = Number(scalar('SELECT COALESCE(MAX(sequence),0) FROM state_changes')) + if (!initializing) db.exec('COMMIT') + // No fallible IO between COMMIT and the Store's Effects acknowledgement. + expectedRevision = next + return { revision: next, changed: changes.length, lastSequence } + } catch (error) { if (!initializing) rollback(); throw error } + } + try { + const schema = Number(scalar('PRAGMA user_version')), app = Number(scalar('PRAGMA application_id')) + const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'").all() + const fresh = schema === 0 && app === 0 && tables.length === 0 + if (!fresh && (schema !== SCHEMA || app !== APPLICATION_ID)) throw fail('Unsupported or foreign SQLite state schema/version') + if (fresh && (readOnly || initial === undefined)) throw fail('SQLite initialization needs validated initial state') + if (!readOnly) { + chmodSync(path, 0o600) + db.exec('PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; PRAGMA foreign_keys=ON; PRAGMA busy_timeout=5000') + if (scalar('PRAGMA journal_mode') !== 'wal' || Number(scalar('PRAGMA synchronous')) !== 2 || Number(scalar('PRAGMA foreign_keys')) !== 1) throw fail('Required SQLite durability configuration was not applied') + } + if (fresh) { + db.exec(`BEGIN IMMEDIATE; + CREATE TABLE state_meta(key TEXT PRIMARY KEY,value TEXT NOT NULL) STRICT; + CREATE TABLE state_records(namespace TEXT NOT NULL,id TEXT NOT NULL,document TEXT,storage TEXT NOT NULL CHECK(storage IN ('inline','file')),digest TEXT NOT NULL,fingerprint TEXT NOT NULL,PRIMARY KEY(namespace,id)) STRICT; + CREATE TABLE state_history(namespace TEXT NOT NULL,id TEXT NOT NULL,path TEXT NOT NULL,position INTEGER NOT NULL,document TEXT,storage TEXT NOT NULL CHECK(storage IN ('inline','file')),digest TEXT NOT NULL,PRIMARY KEY(namespace,id,path,position),FOREIGN KEY(namespace,id) REFERENCES state_records(namespace,id) ON DELETE CASCADE) STRICT; + CREATE TABLE state_changes(sequence INTEGER PRIMARY KEY AUTOINCREMENT,revision INTEGER NOT NULL,namespace TEXT NOT NULL,id TEXT NOT NULL,operation TEXT NOT NULL CHECK(operation IN ('upsert','delete')),UNIQUE(revision,namespace,id)) STRICT; + PRAGMA application_id=${APPLICATION_ID}; PRAGMA user_version=${SCHEMA};`) + expectedRevision = 0 + setMeta('migration', JSON.stringify(migration)) + write(initial, { initializing: true }) + db.exec('COMMIT') + } else expectedRevision = Number(meta('revision')) + if (!Number.isSafeInteger(expectedRevision) || expectedRevision < 1) throw fail('SQLite state revision is malformed') + } catch (error) { rollback(); db.close(); throw error } + function read() { + if (closed) throw fail('SQLite state is closed') + db.exec('BEGIN') + try { + const revision = Number(meta('revision')) + if (!Number.isSafeInteger(revision) || revision < 1) throw fail('SQLite state revision is malformed') + const roots = JSON.parse(meta('roots')), state = {}, values = new Map() + for (const [key, kind] of Object.entries(roots)) { + if (!['map', 'value'].includes(kind)) throw fail('SQLite state root descriptor changed') + if (kind === 'map') put(state, key, {}) + } + for (const row of db.prepare('SELECT * FROM state_records ORDER BY namespace,id').all()) { + const value = readPayload(row) + values.set(JSON.stringify([row.namespace, row.id]), value) + if (row.namespace === '$root') { + if (roots[row.id] !== 'value') throw fail('SQLite value root scope changed') + put(state, row.id, value) + } else { + if (!Object.hasOwn(state, row.namespace) || roots[row.namespace] !== 'map') throw fail('SQLite record root scope changed') + put(state[row.namespace], row.id, value) + } + } + for (const row of db.prepare('SELECT * FROM state_history ORDER BY namespace,id,path,position').all()) { + let target = values.get(JSON.stringify([row.namespace, row.id])) + const path = JSON.parse(row.path) + if (!Array.isArray(path) || !path.length) throw fail('SQLite history path changed') + for (const key of path) { + if (target === null || typeof target !== 'object' || !Object.hasOwn(target, key)) throw fail('SQLite history owner changed') + target = target[key] + } + if (!Array.isArray(target) || target.length !== row.position) throw fail('SQLite history sequence is malformed') + target.push(readPayload(row)) + } + for (const row of db.prepare('SELECT namespace,id,fingerprint FROM state_records').all()) if (hash(JSON.stringify(values.get(JSON.stringify([row.namespace, row.id])))) !== row.fingerprint) throw fail('SQLite record/history fingerprint changed') + db.exec('COMMIT'); expectedRevision = revision + return state + } catch (error) { rollback(); throw error } + } + return { read, write, stamp: () => String(meta('revision')), + metadata: () => ({ schema: SCHEMA, revision: Number(meta('revision')), migration: JSON.parse(meta('migration')), journalMode: scalar('PRAGMA journal_mode'), synchronous: Number(scalar('PRAGMA synchronous')) }), + changes({ after = 0, limit = 50 } = {}) { + if (!Number.isSafeInteger(after) || after < 0 || !Number.isSafeInteger(limit) || limit < 1 || limit > 100) throw fail('Invalid SQLite change cursor or limit') + const items = db.prepare('SELECT sequence,revision,namespace,id,operation FROM state_changes WHERE sequence>? ORDER BY sequence LIMIT ?').all(after, limit) + return { items: items.map(row => ({ ...row })), next: items.at(-1)?.sequence ?? after } + }, + close() { if (!closed) { closed = true; db.close() } }, + } +} diff --git a/core/src/host/project-state-storage.js b/core/src/host/project-state-storage.js new file mode 100644 index 000000000..fd9b43690 --- /dev/null +++ b/core/src/host/project-state-storage.js @@ -0,0 +1,133 @@ +// Migration bridge for the original Store. state.json becomes a newer-version +// guard, not a second writable mirror; its validated import is retained separately. +import { createHash, randomUUID } from 'node:crypto' +import { constants, closeSync, existsSync, fsyncSync, lstatSync, linkSync, mkdirSync, openSync, readFileSync, renameSync, unlinkSync, writeFileSync } from 'node:fs' +import { dirname, join } from 'node:path' +import { openSqliteState, requireSqliteRuntime } from './project-state-sqlite.js' + +const digest = value => createHash('sha256').update(value).digest('hex') +const fail = message => Object.assign(new Error(message), { code: 'bot/state-storage' }) +export function writeDurableStateFile(path, text, { exclusive = false } = {}) { + mkdirSync(dirname(path), { recursive: true, mode: 0o700 }) + const temp = `${path}.${randomUUID()}.tmp`, fd = openSync(temp, 'wx', 0o600) + try { writeFileSync(fd, text, 'utf8'); fsyncSync(fd) } finally { closeSync(fd) } + try { + if (exclusive) { + try { linkSync(temp, path) } + catch (error) { + if (error.code !== 'EEXIST') throw error + const info = lstatSync(path) + if (!info.isFile() || info.isSymbolicLink() || readFileSync(path, 'utf8') !== text) throw fail('Existing migration backup content changed') + } + } else renameSync(temp, path) + } finally { if (existsSync(temp)) unlinkSync(temp) } + const directory = openSync(dirname(path), 'r') + try { fsyncSync(directory) } finally { closeSync(directory) } +} + +export function createProjectStateStorage({ home, statePath = join(home, 'state.json'), mode, validate, ensureWriter }) { + if (mode !== undefined && !['json', 'sqlite'].includes(mode)) throw fail('stateStorage must be json or sqlite') + const path = join(home, 'state.sqlite') + let connection, connectionReadOnly, guard, selected = mode === 'sqlite', knownRevision = null + function checkRollback() { + const marker = join(home, 'state-storage-rollback.json') + if (existsSync(marker) && JSON.parse(readFileSync(marker, 'utf8')).phase === 'prepared') throw fail('Storage rollback is pending; resume the offline migration tool before starting FlyBot') + } + function readGuard() { + if (!existsSync(statePath)) return null + const info = lstatSync(statePath) + if ((selected || existsSync(path)) && (!info.isFile() || info.isSymbolicLink())) throw fail('State migration source must be a regular file') + let parsed + try { parsed = JSON.parse(readFileSync(statePath, 'utf8')) } + catch (error) { if (selected || existsSync(path)) throw fail(`Cannot migrate malformed JSON state: ${error.message}`); return null } + if (parsed?.storage !== 'sqlite') return null + if (!info.isFile() || info.isSymbolicLink()) throw fail('State migration source must be a regular file') + const proof = parsed.migration + if (!['prepared', 'active'].includes(parsed.phase)) throw fail('SQLite migration phase is malformed') + if (parsed.version !== 3 || !proof || typeof proof.id !== 'string' || !/^[a-f0-9]{64}$/.test(proof.sha256 ?? '') || proof.backup !== `state.sqlite-import-${proof.sha256}.json`) throw fail('SQLite migration guard is malformed') + return parsed + } + function importState(proof) { + const file = join(home, proof.backup), info = lstatSync(file) + if (!info.isFile() || info.isSymbolicLink() || info.size > 128 * 1024 * 1024) throw fail('SQLite import backup is not a bounded regular file') + const fd = openSync(file, constants.O_RDONLY | (constants.O_NOFOLLOW ?? 0)) + let text + try { text = readFileSync(fd, 'utf8') } finally { closeSync(fd) } + if (digest(text) !== proof.sha256) throw fail('SQLite import backup hash changed') + const state = JSON.parse(text) + validate(state) + return state + } + function open(writer, initial) { + if (connection && connectionReadOnly === !writer) return connection + connection?.close(); connection = undefined + const candidate = openSqliteState({ path, readOnly: !writer, ...(initial ? { initial } : {}), migration: guard.migration }) + const migration = candidate.metadata().migration + if (migration?.id !== guard.migration.id || migration?.sha256 !== guard.migration.sha256) { + candidate.close(); throw fail('SQLite database and migration guard identify different authorities') + } + connection = candidate; connectionReadOnly = !writer + return candidate + } + function activate() { + if (guard.phase === 'active') return + ensureWriter() + guard = { ...guard, phase: 'active' } + writeDurableStateFile(statePath, JSON.stringify(guard, null, 2)) + } + function assertGuard() { + const current = readGuard() + if (!current || current.phase !== 'active' || current.migration.id !== guard?.migration.id || current.migration.sha256 !== guard.migration.sha256) throw fail('SQLite storage authority changed; reload before writing') + } + return { + get sqlite() { return selected }, + async read({ writer, readLegacy }) { + checkRollback() + const current = readGuard() + if (!current && !existsSync(path)) return { state: await readLegacy(), sqlite: false } + selected = true + if (mode === 'json') throw fail('SQLite is the current authority; export and roll back explicitly before selecting JSON') + if (!current) throw fail('SQLite state is missing its migration guard; refusing a second JSON authority') + guard = current + const initial = importState(guard.migration) + if (!existsSync(path) && guard.phase === 'active') throw fail('The authoritative SQLite database is missing; do not import an outdated backup automatically') + if (!existsSync(path) && !writer) return { state: initial, sqlite: true } + if (writer) ensureWriter() + const db = open(writer, writer ? initial : undefined), state = db.read() + validate(state) + knownRevision = db.stamp() + if (writer) activate() + return { state, sqlite: true } + }, + initialize(state) { + if (!selected || connection) return false + ensureWriter(); validate(state); requireSqliteRuntime() + guard = readGuard() + if (!guard) { + const text = JSON.stringify(state), sha256 = digest(text), backup = `state.sqlite-import-${sha256}.json` + writeDurableStateFile(join(home, backup), text, { exclusive: true }) + guard = { version: 3, storage: 'sqlite', phase: 'prepared', migration: { id: randomUUID(), backup, sha256, importedAt: Date.now() } } + // Guard first: a crash before SQL initialization remains recoverable from + // the verified backup and an older JSON-only binary refuses to write. + writeDurableStateFile(statePath, JSON.stringify(guard, null, 2)) + } + const db = open(true, importState(guard.migration)) + knownRevision = db.stamp() + activate() + return true + }, + write(snapshot) { + if (!selected || !guard) throw fail('SQLite storage has not been initialized') + ensureWriter(); checkRollback(); assertGuard() + const db = open(true) + if (knownRevision !== db.stamp()) throw fail('SQLite state revision conflict; refresh before taking over') + const result = db.write(typeof snapshot === 'string' ? JSON.parse(snapshot) : snapshot) + knownRevision = String(result.revision) + return result + }, + stamp() { return connection ? `sqlite:${guard.migration.id}:${connection.stamp()}` : null }, + status() { return { backend: selected ? 'sqlite' : 'json', ready: Boolean(connection), ...(connection ? connection.metadata() : {}) } }, + changes(options) { if (!connection) throw fail('SQLite storage is not initialized'); return connection.changes(options) }, + close() { connection?.close(); connection = undefined }, + } +} diff --git a/core/src/host/store.js b/core/src/host/store.js index 42e3a97c9..5308f6133 100644 --- a/core/src/host/store.js +++ b/core/src/host/store.js @@ -2,6 +2,7 @@ // host on the same data from writing it, and the indexes that map a Session to the Bot // or group chat it belongs to. import { install as installProjects } from './project-service.js' +import { createProjectStateStorage } from './project-state-storage.js' import { constants, mkdirSync, statSync } from 'node:fs' import { copyFile, mkdir, readFile, rename, unlink, writeFile } from 'node:fs/promises' import { join } from 'node:path' @@ -56,8 +57,11 @@ export function install(rt) { let stopped = false // mtime and size of the state.json this host last read or wrote. let seen = '' + let storage const stamp = () => { try { + const current = storage?.stamp() + if (current !== null && current !== undefined) return current const { mtimeMs, size } = statSync(statePath) return `${mtimeMs}:${size}` } catch { @@ -79,6 +83,7 @@ export function install(rt) { } } const readOnly = () => { + if (storage?.sqlite && lockBroken) return 'SQLite state requires an intact exclusive home lock; fix locking and reload.' if (stuck !== undefined) return `The Bot team file ${statePath} could not be read (${stuck}). FlyBot does not change it; fix or move the file and reload.` if (holder === undefined) return undefined const where = holder.pid ? `pid ${holder.pid}${holder.host ? ` on ${holder.host}` : ''}` : 'unknown process' @@ -161,7 +166,7 @@ export function install(rt) { return undefined } - const readFileState = async () => { + const readJsonState = async () => { let contents try { contents = await readFile(statePath, 'utf8') @@ -221,6 +226,30 @@ export function install(rt) { return undefined } + const validateSqliteState = parsed => { + if (parsed?.version !== STATE_VERSION) throw new Error('Unsupported SQLite state version') + const problem = ownershipProblem(parsed) + if (problem !== undefined) throw new Error(`Malformed SQLite ownership: ${problem}`) + } + storage = createProjectStateStorage({ home, statePath, mode: config.stateStorage, validate: validateSqliteState, + ensureWriter() { if (lockBroken || holder !== undefined || !lock.owns()) throw Object.assign(new Error('SQLite state needs the current exclusive home lock'), { code: 'ELOCKLOST' }) } }) + const readFileState = async () => { + try { + const result = await storage.read({ writer: holder === undefined && !lockBroken, readLegacy: readJsonState }) + if (result.sqlite) { stuck = undefined; seen = stamp() } + return result.state + } catch (error) { + stuck = error.message; seen = stamp() + warn('ds-bot: durable state was refused: %s', error) + return undefined + } + } + const initializeStorage = () => { + if (holder !== undefined || stuck !== undefined) return false + try { return storage.initialize(state) } + catch (error) { stuck = error.message; warn('ds-bot: SQLite migration was refused: %s', error); return false } + } + // Fill `state` from the saved file and bring an older team up to date. Returns the // groups whose Session titles still need the new name. const adopt = (parsed) => { @@ -405,7 +434,8 @@ export function install(rt) { if (holder !== undefined) warn('ds-bot: %s', readOnly()) const { migrated, stale } = adopt(await readFileState()) if (holder !== undefined) return - if (migrated && stuck === undefined) { + const initialized = initializeStorage() + if ((migrated || initialized) && stuck === undefined) { // A failed write must not poison `loaded`: the migrated state stays in // memory and the next save() writes it. try { @@ -438,6 +468,7 @@ export function install(rt) { announceTakeover(previous) warnedReadOnly = false adopt(await readFileState()) + initializeStorage() return } } else if (!lockBroken && !lock.owns()) { @@ -451,11 +482,12 @@ export function install(rt) { // written while no host held it. warnedReadOnly = false adopt(await readFileState()) + initializeStorage() return } warn('ds-bot: %s', readOnly()) } - if ((stuck !== undefined || holder !== undefined) && stamp() !== seen) adopt(await readFileState()) + if ((stuck !== undefined || holder !== undefined) && stamp() !== seen) { adopt(await readFileState()); initializeStorage() } } const warnReadOnly = () => { @@ -526,9 +558,12 @@ export function install(rt) { } } ensureLock() - await writeFile(temp, snapshot, 'utf8') - ensureLock() - await rename(temp, statePath) + if (storage.sqlite) storage.write(snapshot) + else { + await writeFile(temp, snapshot, 'utf8') + ensureLock() + await rename(temp, statePath) + } rt.projectEffects?.saved(effectSnapshot) seen = stamp() await journal?.record(snapshot) @@ -571,7 +606,7 @@ export function install(rt) { ctx.effect(() => () => { stopped = true - return writing.finally(() => lock.release()) + return writing.finally(() => { try { storage.close() } finally { lock.release() } }) }, 'ds-bot: state lock') const botOf = id => (id === undefined ? undefined : ownEntry(state.bots, id)) @@ -658,6 +693,7 @@ export function install(rt) { Object.assign(rt, { state, load, save, refresh, readOnly, mutate, savesQueued, savesSettled, + stateStorage: { status: () => storage.status(), changes: options => storage.changes(options) }, botOf, roomOf, isMain, chatOf, selfOf, isBotSession, chatOwners, groupOwners, pastOwners, groupSessionsOf, dropGroupSession, findBot, findRooms, isAdminOf, diff --git a/core/test/project-state-runtime.test.js b/core/test/project-state-runtime.test.js new file mode 100644 index 000000000..27bde6ef3 --- /dev/null +++ b/core/test/project-state-runtime.test.js @@ -0,0 +1,46 @@ +import assert from 'node:assert/strict' +import { mkdtempSync, writeFileSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { createRuntime } from '../index.js' +import { openSqliteState } from '../src/host/project-state-sqlite.js' +import { fakeContext } from './fake-ctx.js' + +test('original Project APIs persist Task/intake intent in SQL without model calls or positioning changes', async t => { + const home = mkdtempSync(join(tmpdir(), 'flybot-sql-api-')), contexts = [] + writeFileSync(join(home, 'state.json'), JSON.stringify({ version: 2, revision: 1, members: { 'owner-1': { id: 'owner-1', designatedChiefId: 'chief' } }, mainBotIds: ['chief'], bots: { chief: { id: 'chief', sessionId: 'chief', name: 'Chief', ownerId: 'owner-1', projectId: 'p' } }, projects: { p: { id: 'p', name: 'Existing Project', status: 'active', ownerUserId: 'owner-1', goal: 'Retain the agreed product goal', scope: 'Original Project scope', roles: { members: [], maintainers: [] } } }, tasks: {} })) + let modelCalls = 0 + async function start(stateStorage) { + const ctx = fakeContext() + Object.assign(ctx, { llm: { listProviders: () => [], listModels: async () => [], generate: async () => { ++modelCalls; throw Error('No inference expected') } }, workspaceRegistry: { get: () => undefined, archiveSession: async () => {} }, sessionController: { resolveAgent: async () => ({ agent: undefined }) }, sessions: { flush: async () => {} } }) + contexts.push(ctx) + const rt = createRuntime(ctx, { home, projectWorkbench: true, ...(stateStorage ? { stateStorage } : {}) }) + await rt.load() + assert.equal(rt.readOnly(), undefined) + return { ctx, rt } + } + t.after(async () => { for (const ctx of contexts.reverse()) await ctx.dispose(); rmSync(home, { recursive: true, force: true, maxRetries: 5 }) }) + const first = await start('sqlite'), actor = { memberId: 'owner-1' } + const task = await first.rt.taskEndpoint('task-create', { botId: 'chief', projectId: 'p', title: 'Improve backend query', criteria: 'Query regression passes', impact: 'Improve Project observability' }, actor) + assert.equal(task.projectTask.admission.status, 'proposed') + await first.rt.taskEndpoint('task-update', { id: task.id, note: 'Keep the agreed scope and required evidence' }, actor) + const body = { version: 'project-development/v1', kind: 'query', name: 'tasks', projectId: 'p', payload: { limit: 10 } } + const view = await first.rt.projectEndpoint('project-development', body, actor) + assert.equal(view.result.tasks[0].taskId, task.id) + assert.equal(view.result.tasks[0].waitingFor, 'admission') + const database = openSqliteState({ path: join(home, 'state.sqlite'), readOnly: true }), disk = database.read(); database.close() + assert.equal(disk.tasks[task.id].note, 'Keep the agreed scope and required evidence') + const intake = Object.values(disk.projectEffects).find(row => row.kind === 'task.intake' && row.taskId === task.id) + assert.ok(intake) + assert.equal(first.rt.projectEffects.intentCommitted({ id: intake.id, generation: intake.generation }), true) + assert.equal(disk.projects.p.goal, 'Retain the agreed product goal') + assert.equal(disk.projects.p.scope, 'Original Project scope') + assert.equal(modelCalls, 0) + await first.ctx.dispose() + const second = await start() + assert.equal(second.rt.stateStorage.status().backend, 'sqlite') + assert.equal((await second.rt.projectEndpoint('project-development', body, actor)).result.tasks[0].taskId, task.id) + assert.equal(second.rt.state.tasks[task.id].projectTask.admission.status, 'proposed') + assert.equal(modelCalls, 0) +}) diff --git a/core/test/project-state-sqlite.test.js b/core/test/project-state-sqlite.test.js new file mode 100644 index 000000000..54cb2cffa --- /dev/null +++ b/core/test/project-state-sqlite.test.js @@ -0,0 +1,130 @@ +import assert from 'node:assert/strict' +import { mkdtempSync, rmSync, readFileSync, writeFileSync, statSync, symlinkSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { DatabaseSync } from 'node:sqlite' +import { openSqliteState } from '../src/host/project-state-sqlite.js' + +function fixture(t) { + const home = mkdtempSync(join(tmpdir(), 'flybot-sql-state-')) + const options = { path: join(home, 'state.sqlite'), payloadRoot: join(home, 'state-payloads') } + const connections = [] + t.after(() => { for (const db of connections.reverse()) db.close(); rmSync(home, { recursive: true, force: true }) }) + const open = more => { const db = openSqliteState({ ...options, ...more }); connections.push(db); return db } + return { home, options, open } +} +const initial = () => ({ version: 2, revision: 1, members: { owner: { id: 'owner' } }, tasks: { task: { id: 'task', status: 'working', collaboration: { history: [{ id: 'offer-1' }] } } }, projectEffects: { effect: { id: 'effect', status: 'pending' } } }) + +test('SQLite roundtrips identities, normalizes history and commits only changed rows', t => { + const f = fixture(t), state = initial(), db = f.open({ initial: state }) + assert.deepEqual(db.read(), state) + assert.equal(db.metadata().journalMode, 'wal') + assert.equal(db.metadata().synchronous, 2) + const reader = new DatabaseSync(f.options.path, { readOnly: true }) + t.after(() => reader.close()) + assert.equal(reader.prepare('SELECT count(*) AS n FROM state_history').get().n, 1) + assert.equal(JSON.parse(reader.prepare("SELECT document FROM state_records WHERE namespace='tasks' AND id='task'").get().document).collaboration.history.length, 0) + assert.equal(db.write(state).changed, 0) + state.tasks.task.collaboration.history.push({ id: 'offer-2' }) + state.projectEffects.effect.status = 'done' + const before = db.changes({ limit: 100 }).next + const result = db.write(state) + assert.equal(result.changed, 2) + assert.deepEqual(db.read(), state) + assert.deepEqual(db.changes({ after: before }).items.map(({ namespace, id, operation }) => ({ namespace, id, operation })), [{ namespace: 'projectEffects', id: 'effect', operation: 'upsert' }, { namespace: 'tasks', id: 'task', operation: 'upsert' }]) +}) + +test('an interrupted SQL transaction commits neither Task nor Effect and can retry', t => { + const f = fixture(t), state = initial(), db = f.open({ initial: state }) + const control = new DatabaseSync(f.options.path) + t.after(() => control.close()) + control.exec("CREATE TRIGGER fail_effect BEFORE UPDATE ON state_records WHEN NEW.namespace='projectEffects' BEGIN SELECT RAISE(ABORT,'fixture transaction failure'); END") + state.tasks.task.status = 'accepted'; state.projectEffects.effect.status = 'done' + assert.throws(() => db.write(state), /fixture transaction failure/) + assert.deepEqual(f.open({ readOnly: true }).read(), initial()) + control.exec('DROP TRIGGER fail_effect') + db.write(state) + assert.deepEqual(f.open({ readOnly: true }).read(), state) +}) + +test('stale database revisions cannot overwrite a different writer; readonly connections never commit', t => { + const f = fixture(t), state = initial(), first = f.open({ initial: state }), stale = f.open({}) + stale.read() + state.tasks.task.status = 'blocked'; first.write(state) + assert.throws(() => stale.write(initial()), /revision conflict/) + const observer = f.open({ readOnly: true }) + assert.equal(observer.read().tasks.task.status, 'blocked') + assert.throws(() => observer.write(initial()), /read.only/i) + assert.equal(observer.stamp(), first.stamp()) +}) + +test('large records use immutable hashed files and fail closed on content damage', t => { + const f = fixture(t), state = initial() + state.tasks.large = { id: 'large', payload: 'bounded data '.repeat(10000) } + const db = f.open({ initial: state }), inspector = new DatabaseSync(f.options.path, { readOnly: true }) + t.after(() => inspector.close()) + const row = inspector.prepare("SELECT storage,document,digest FROM state_records WHERE namespace='tasks' AND id='large'").get() + assert.equal(row.storage, 'file'); assert.equal(row.document, null) + assert.deepEqual(JSON.parse(readFileSync(join(f.options.payloadRoot, row.digest + '.json'), 'utf8')), state.tasks.large) + assert.deepEqual(db.read(), state) + writeFileSync(join(f.options.payloadRoot, row.digest + '.json'), 'corrupt payload') + assert.throws(() => db.read(), /hash changed/) +}) + +test('future or foreign SQLite schemas refuse to initialize and preserve existing tables', t => { + const f = fixture(t), raw = new DatabaseSync(f.options.path) + raw.exec('CREATE TABLE unrelated(value TEXT); INSERT INTO unrelated VALUES (\'preserve\'); PRAGMA user_version=999') + raw.close() + assert.throws(() => f.open({ initial: initial() }), /schema|version/i) + const check = new DatabaseSync(f.options.path, { readOnly: true }) + assert.equal(check.prepare('SELECT value FROM unrelated').get().value, 'preserve'); check.close() +}) + +test('actual process death before and after COMMIT preserves the correct atomic Task/Effect boundary', async t => { + const { spawn } = await import('node:child_process'), { once } = await import('node:events') + for (const committed of [false, true]) { + const f = fixture(t), seed = f.open({ initial: initial() }) + seed.close() + const source = `import { DatabaseSync } from 'node:sqlite'; + import { writeSync } from 'node:fs'; + import { openSqliteState } from ${JSON.stringify(new URL('../src/host/project-state-sqlite.js', import.meta.url).href)}; + const options=JSON.parse(process.argv[1]); + if(${committed}){const db=openSqliteState(options);const state=db.read();state.tasks.task.status='accepted';state.projectEffects.effect.status='done';db.write(state)} + else{const db=new DatabaseSync(options.path);db.exec("BEGIN IMMEDIATE;UPDATE state_records SET document='{}' WHERE namespace IN ('tasks','projectEffects')")} + writeSync(1,'boundary-ready\\n');setInterval(()=>{},1000);` + const child = spawn(process.execPath, ['--input-type=module', '-e', source, JSON.stringify(f.options)], { stdio: ['ignore', 'pipe', 'pipe'] }) + t.after(() => { if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL') }) + const closed = once(child, 'close') + let output = '' + while (!output.includes('boundary-ready')) { + const [chunk] = await once(child.stdout, 'data', { signal: AbortSignal.timeout(10000) }) + output += chunk.toString() + } + child.kill('SIGKILL') + const [, signal] = await closed + assert.equal(signal, 'SIGKILL') + const state = f.open({ readOnly: true }).read() + assert.equal(state.tasks.task.status, committed ? 'accepted' : 'working') + assert.equal(state.projectEffects.effect.status, committed ? 'done' : 'pending') + } +}) + + +test('database symlinks, prototype-shaped keys and bounded change cursors respect storage boundaries', t => { + const f = fixture(t), state = initial() + Object.defineProperty(state.tasks, '__proto__', { value: { id: '__proto__', history: [{ id: 'owned-history' }] }, enumerable: true }) + const db = f.open({ initial: state }), copy = db.read() + assert.equal(Object.hasOwn(copy.tasks, '__proto__'), true) + assert.equal(copy.tasks.__proto__.history[0].id, 'owned-history') + assert.equal(Object.getPrototypeOf(copy.tasks), Object.prototype) + assert.equal(statSync(f.options.path).mode & 0o777, 0o600) + for (const options of [{ limit: 0 }, { limit: 101 }, { after: -1 }, { after: '0' }]) assert.throws(() => db.changes(options)) + const first = db.changes({ limit: 1 }), second = db.changes({ after: first.next, limit: 1 }) + assert.equal(first.items.length, 1); assert.equal(second.items.length, 1) + assert.ok(second.next > first.next) + db.close() + const linked = join(f.home, 'linked.sqlite') + symlinkSync(f.options.path, linked) + assert.throws(() => openSqliteState({ path: linked }), /regular file/) +}) diff --git a/core/test/project-state-storage.test.js b/core/test/project-state-storage.test.js new file mode 100644 index 000000000..3444bbfc3 --- /dev/null +++ b/core/test/project-state-storage.test.js @@ -0,0 +1,201 @@ +import assert from 'node:assert/strict' +import { createHash } from 'node:crypto' +import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs' +import { hostname, tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { DatabaseSync } from 'node:sqlite' +import { install } from '../src/host/store.js' +import { createProjectEffects } from '../src/host/project-effects.js' + +function fixture(t, saved) { + const home = mkdtempSync(join(tmpdir(), 'flybot-sql-migration-')), statePath = join(home, 'state.json'), runtimes = [] + if (saved !== undefined) writeFileSync(statePath, typeof saved === 'string' ? saved : JSON.stringify(saved)) + async function start(mode) { + const disposers = [], warnings = [], rt = { home, statePath, config: { ...(mode ? { stateStorage: mode } : {}) }, warn: (...args) => warnings.push(args), autoRoomName: () => 'Group', retitleRoom: async () => {}, + ctx: { effect(fn) { disposers.push(fn()) } }, + } + install(rt) + rt.projectEffects = createProjectEffects(rt) + let closed = false + const close = async () => { if (!closed) { closed = true; await rt.projectEffects.close(); for (const fn of disposers.reverse()) await fn?.() } } + runtimes.push(close) + await rt.load() + return { rt, close, warnings } + } + t.after(async () => { for (const close of runtimes.reverse()) await close(); rmSync(home, { recursive: true, force: true }) }) + return { home, statePath, start, path: join(home, 'state.sqlite') } +} +const saved = () => ({ version: 2, revision: 7, members: { owner: { id: 'owner', designatedChiefId: 'chief' } }, mainBotIds: ['chief'], bots: { chief: { id: 'chief', ownerId: 'owner', sessionId: 'chief-session', name: 'Chief' } }, projects: { p: { id: 'p', ownerUserId: 'owner', status: 'active', roles: { members: [], maintainers: [] } } }, tasks: { task: { id: 'task', projectId: 'p', title: 'Retain current work', status: 'working', collaboration: { history: [{ id: 'offer', status: 'accepted' }], claim: { id: 'claim', generation: 7 } } } }, projectTaskSessions: { role: { logicalSessionId: 'role', bindings: [{ backendSessionId: 'sdk-thread' }] } }, projectEffects: {} }) + +test('Store migration preserves working identities, fences JSON writes and automatically reopens SQL authority', async t => { + const f = fixture(t, saved()), first = await f.start('sqlite') + assert.equal(first.rt.readOnly(), undefined) + assert.equal(first.rt.stateStorage.status().backend, 'sqlite') + const guardText = readFileSync(f.statePath, 'utf8'), guard = JSON.parse(guardText) + assert.equal(guard.version, 3); assert.equal(guard.phase, 'active') + assert.equal(createHash('sha256').update(readFileSync(join(f.home, guard.migration.backup))).digest('hex'), guard.migration.sha256) + first.rt.state.tasks.task.status = 'blocked' + await first.rt.save() + assert.equal(readFileSync(f.statePath, 'utf8'), guardText) + await first.close() + const second = await f.start() + assert.equal(second.rt.stateStorage.status().backend, 'sqlite') + assert.equal(second.rt.state.tasks.task.status, 'blocked') + assert.equal(second.rt.state.tasks.task.collaboration.claim.id, 'claim') + assert.equal(second.rt.state.tasks.task.collaboration.history[0].id, 'offer') + assert.equal(second.rt.state.bots.chief.sessionId, 'chief-session') + assert.equal(second.rt.state.projectTaskSessions.role.bindings[0].backendSessionId, 'sdk-thread') + await second.close() + const forbidden = await f.start('json') + assert.match(forbidden.rt.readOnly(), /SQLite.*authority|roll back/) + assert.equal(readFileSync(f.statePath, 'utf8'), guardText) +}) + +test('malformed or invalid-identity JSON cannot migrate into an empty SQL replacement', async t => { + for (const source of ['{broken', { ...saved(), bots: { chief: { id: 'chief', ownerId: 'unknown' } } }]) { + const f = fixture(t, source), original = readFileSync(f.statePath, 'utf8'), current = await f.start('sqlite') + assert.ok(current.rt.readOnly()) + assert.equal(existsSync(f.path), false) + assert.equal(readFileSync(f.statePath, 'utf8'), original) + } +}) + +test('prepared migration recovers exact backup; loss of an active database never imports stale state', async t => { + const f = fixture(t, saved()), first = await f.start('sqlite') + await first.close() + const guard = JSON.parse(readFileSync(f.statePath, 'utf8')) + for (const suffix of ['', '-wal', '-shm']) rmSync(f.path + suffix, { force: true }) + const missing = await f.start('sqlite') + assert.match(missing.rt.readOnly(), /authoritative SQLite database is missing/) + assert.equal(existsSync(f.path), false) + await missing.close() + writeFileSync(f.statePath, JSON.stringify({ ...guard, phase: 'prepared' })) + const recovered = await f.start() + assert.equal(recovered.rt.readOnly(), undefined) + assert.equal(recovered.rt.state.tasks.task.collaboration.claim.id, 'claim') + assert.equal(JSON.parse(readFileSync(f.statePath, 'utf8')).phase, 'active') +}) + +test('SQL state and Effect commit together, recover after restart and acknowledge only a committed transaction', async t => { + const f = fixture(t, saved()), first = await f.start('sqlite'), { rt } = first + let executions = 0 + rt.projectEffects.register('fixture.execute', { run: async () => ({ count: ++executions }) }) + const spec = { kind: 'fixture.execute', key: 'once', projectId: 'p', taskId: 'task', payload: { claimId: 'claim' } } + const token = rt.projectEffects.stage(spec) + rt.state.tasks.task.status = 'queued' + const db = new DatabaseSync(f.path) + db.exec("CREATE TRIGGER fail_effect BEFORE INSERT ON state_records WHEN NEW.namespace='projectEffects' BEGIN SELECT RAISE(ABORT,'fixture effect write failure'); END") + await assert.rejects(rt.projectEffects.persist(token), /fixture effect write failure/) + await rt.projectEffects.drain() + assert.equal(executions, 0); assert.equal(rt.projectEffects.intentCommitted(token), false) + assert.equal(JSON.parse(db.prepare("SELECT document FROM state_records WHERE namespace='tasks' AND id='task'").get().document).status, 'working') + db.exec('DROP TRIGGER fail_effect'); db.close() + await rt.projectEffects.persist(token) + assert.equal(rt.projectEffects.intentCommitted(token), true) + await first.close() + const second = await f.start() + second.rt.projectEffects.register('fixture.execute', { run: async () => ({ count: ++executions }) }) + second.rt.projectEffects.recover() + await second.rt.projectEffects.drain() + await second.rt.projectEffects.drain() + assert.equal(executions, 1) + assert.equal(second.rt.state.projectEffects[token.id].status, 'done') + assert.equal(second.rt.state.tasks.task.status, 'queued') +}) + +test('readonly Store does not acknowledge Effects and refresh adopts current SQL state before takeover', async t => { + const f = fixture(t, saved()), first = await f.start('sqlite') + await first.close() + writeFileSync(join(f.home, 'state.lock'), JSON.stringify({ pid: process.ppid, host: hostname(), token: 'external-holder' })) + const second = await f.start() + assert.match(second.rt.readOnly(), /another dsh/) + second.rt.projectEffects.register('fixture.readonly', { run: async () => { throw new Error('Must not execute') } }) + // stage itself is denied on read-only hosts; a no-op save cannot grant durability. + assert.throws(() => second.rt.projectEffects.stage({ kind: 'fixture.readonly', key: 'denied', projectId: 'p', payload: {} }), /writable/) + const before = second.rt.state.revision + await second.rt.save() + assert.equal(second.rt.state.revision, before) + rmSync(join(f.home, 'state.lock')) + await second.rt.refresh() + assert.equal(second.rt.readOnly(), undefined) + second.rt.state.tasks.task.status = 'blocked' + await second.rt.save() + assert.equal(second.rt.stateStorage.status().backend, 'sqlite') +}) + +test('offline export and rollback retain latest facts and reopen JSON without two writable authorities', async t => { + const { exportSqliteState, rollbackSqliteState } = await import('../src/host/project-state-migration.js') + const f = fixture(t, saved()), first = await f.start('sqlite') + first.rt.state.tasks.task.status = 'blocked' + await first.rt.save() + const guard = JSON.parse(readFileSync(f.statePath, 'utf8')), output = join(f.home, 'private-export.json') + assert.throws(() => exportSqliteState({ home: f.home, output, expectedMigrationId: guard.migration.id }), /Stop the current/) + await first.close() + const exported = exportSqliteState({ home: f.home, output, expectedMigrationId: guard.migration.id }) + assert.equal(JSON.parse(readFileSync(output, 'utf8')).tasks.task.status, 'blocked') + assert.equal(exported.sha256, createHash('sha256').update(readFileSync(output)).digest('hex')) + assert.throws(() => rollbackSqliteState({ home: f.home, expectedMigrationId: 'outdated' }), /ID changed/) + const result = rollbackSqliteState({ home: f.home, expectedMigrationId: guard.migration.id }) + assert.equal(result.backend, 'json'); assert.equal(existsSync(f.path), false) + assert.equal(existsSync(join(f.home, result.archiveFile)), true) + assert.equal(rollbackSqliteState({ home: f.home, expectedMigrationId: guard.migration.id, resume: true }).already, true) + const second = await f.start() + assert.equal(second.rt.readOnly(), undefined) + assert.equal(second.rt.stateStorage.status().backend, 'json') + assert.equal(second.rt.state.tasks.task.status, 'blocked') + assert.equal(second.rt.state.tasks.task.collaboration.claim.id, 'claim') +}) + +test('prepared rollback blocks runtime and resumes after its database move without reviving import state', async t => { + const { renameSync } = await import('node:fs'), { randomUUID } = await import('node:crypto') + const { rollbackSqliteState } = await import('../src/host/project-state-migration.js') + const { openSqliteState } = await import('../src/host/project-state-sqlite.js') + const f = fixture(t, saved()), first = await f.start('sqlite') + first.rt.state.tasks.task.status = 'blocked'; await first.rt.save(); await first.close() + const guard = JSON.parse(readFileSync(f.statePath, 'utf8')), db = openSqliteState({ path: f.path, readOnly: true }) + const text = JSON.stringify(db.read(), null, 2), revision = db.metadata().revision, id = randomUUID(); db.close() + const marker = { version: 'flybot-storage-rollback/v1', phase: 'prepared', id, migrationId: guard.migration.id, revision, exportFile: `state.rollback-${id}.json`, archiveFile: `state.sqlite.rollback-${id}`, sha256: createHash('sha256').update(text).digest('hex') } + writeFileSync(join(f.home, marker.exportFile), text) + writeFileSync(join(f.home, 'state-storage-rollback.json'), JSON.stringify(marker)) + renameSync(f.path, join(f.home, marker.archiveFile)) + const blocked = await f.start() + assert.match(blocked.rt.readOnly(), /rollback is pending/) + await blocked.close() + assert.throws(() => rollbackSqliteState({ home: f.home, expectedMigrationId: guard.migration.id }), /explicitly resume/) + rollbackSqliteState({ home: f.home, expectedMigrationId: guard.migration.id, resume: true }) + const recovered = await f.start() + assert.equal(recovered.rt.readOnly(), undefined) + assert.equal(recovered.rt.stateStorage.status().backend, 'json') + assert.equal(recovered.rt.state.tasks.task.status, 'blocked') +}) + +test('migration CLI exposes only metadata and requires explicit paths and source identity', async t => { + const { execFileSync } = await import('node:child_process') + const f = fixture(t, saved()), first = await f.start('sqlite'); await first.close() + const script = new URL('../scripts/state-storage.mjs', import.meta.url).pathname + const inspect = JSON.parse(execFileSync(process.execPath, [script, 'inspect', '--home', f.home], { encoding: 'utf8' })) + assert.equal(inspect.backend, 'sqlite'); assert.equal(inspect.phase, 'active') + assert.equal(JSON.stringify(inspect).includes('chief-session'), false) + const output = join(f.home, 'cli-export.json') + const exported = JSON.parse(execFileSync(process.execPath, [script, 'export', '--home', f.home, '--migration', inspect.migrationId, '--output', output], { encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'] })) + assert.equal(exported.output, output) + assert.equal(JSON.parse(readFileSync(output, 'utf8')).tasks.task.id, 'task') + assert.throws(() => execFileSync(process.execPath, [script, 'export', '--home', f.home, '--output', output], { stdio: 'pipe' })) + assert.throws(() => execFileSync(process.execPath, [script, 'inspect', '--home', f.home, '--unexpected'], { stdio: 'pipe' })) +}) + +test('stale readonly takeover cannot overwrite newer SQL state without refresh', async t => { + const { openSqliteState } = await import('../src/host/project-state-sqlite.js') + const f = fixture(t, saved()), first = await f.start('sqlite'); await first.close() + writeFileSync(join(f.home, 'state.lock'), JSON.stringify({ pid: process.ppid, host: hostname(), token: 'external-holder' })) + const observer = await f.start() + const writer = openSqliteState({ path: f.path }), current = writer.read() + current.tasks.task.status = 'blocked'; writer.write(current); writer.close() + rmSync(join(f.home, 'state.lock')) + observer.rt.state.tasks.task.status = 'accepted' + await assert.rejects(observer.rt.save(), /revision conflict/) + await observer.close() + const recovered = await f.start() + assert.equal(recovered.rt.state.tasks.task.status, 'blocked') +}) diff --git a/dev/demo/native-pools.mjs b/dev/demo/native-pools.mjs index dcc6114fb..ed0faddb4 100644 --- a/dev/demo/native-pools.mjs +++ b/dev/demo/native-pools.mjs @@ -34,7 +34,7 @@ export function nativePoolModel() { } return { observed, reply } } -export async function checkNativePools({ api, saved, waitFor, repository, model }) { +export async function checkNativePools({ api, saved, waitFor, repository, model, storage = 'json' }) { const invitation = await api('auth-invite') const human = await api('auth-redeem', { token: invitation.token, displayName: 'Native pool owner' }) const call = (endpoint, body = {}) => api(endpoint, body, human.sessionToken) @@ -68,6 +68,12 @@ export async function checkNativePools({ api, saved, waitFor, repository, model assert.ok(model.observed.resumedContext) assert.equal(model.observed.developerTurns, 2); assert.equal(model.observed.testerTurns, 2) assert.ok(Object.values(snapshot.projectNativeWorkers).every(row => row.status === 'quiescent' && row.releaseProof?.kind === 'native-agent-quiescent')) + for (const row of Object.values(snapshot.projectNativeWorkers)) { + const activation = snapshot.projectEffects[row.activationEffect.id] + assert.equal(activation.kind, 'native.activate') + assert.equal(activation.payload.workerId, row.id) + assert.equal(activation.payload.sessionId, row.sessionId) + } const leases = Object.values(snapshot.projectResourceLeases).filter(row => row.projectId === project.id).sort((a, b) => a.createdAt - b.createdAt) assert.equal(leases.length, 4) assert.ok(leases.every(row => row.status === 'released' && row.releaseProof?.kind === 'native-agent-quiescent')) @@ -75,12 +81,14 @@ export async function checkNativePools({ api, saved, waitFor, repository, model const current = snapshot.projectAttempts[snapshot.tasks[task.id].deliveryAttemptId] assert.equal(current.evidence.tests.passed, true) assert.equal(current.evidence.tests.sha, current.candidateSha) - assert.notEqual(attempts[0].candidateSha, current.candidateSha) + const previous = attempts.find(row => row.id === current.feedback.previousAttemptId) + assert.ok(previous, 'The successor points to its actual prior Attempt independently of map ordering') + assert.notEqual(previous.candidateSha, current.candidateSha) assert.notEqual(snapshot.tasks[task.id].status, 'accepted', 'Chief has not adjudicated completion') assert.match(readFileSync(join(repository, 'code.mjs'), 'utf8'), /a-b/, 'The base repository stays intact') await call('project-attempt-stop', { attemptId: current.id }) for (const attempt of attempts) await call('project-attempt-cleanup', { attemptId: attempt.id }) - return { milestone: '2184-native-pools', runtime: 'actual DSH', model: 'deterministic mock', + return { milestone: '2184-native-pools', runtime: 'actual DSH', storage, model: 'deterministic mock', checks: ['actual Chief admission before native pool allocation', 'native per-Bot pool return/handoff', 'same Task and original physical role Session', 'fresh carrier identities and same cwd', 'prior conversation present only in retained context', 'actual two code revisions and configured host tests', 'native quiescence and single-resource release before downstream execution', 'Chief arbitration inbox without fake closure', 'disposable cleanup'], notRun: ['real model', 'Reviewer and forge publication in this continuation probe', 'Chief acceptance', 'native crash recovery'] } } diff --git a/dev/demo/run.mjs b/dev/demo/run.mjs index 032a8cd40..e44195fa3 100644 --- a/dev/demo/run.mjs +++ b/dev/demo/run.mjs @@ -13,6 +13,7 @@ import { serveWorkbench } from './serve.mjs' import { checkWorkbench } from './workbench.mjs' import { checkQueue, provisionQueueGate } from './queue.mjs' import { agentReply, checkAgents } from './agents.mjs' +import { openSqliteState } from '../../core/src/host/project-state-sqlite.js' import { checkDelivery, deliveryReply, provisionDeliveryFixture } from './delivery.mjs' import { checkShared } from './shared.mjs' import { verifyArtifact } from './artifact.mjs' @@ -45,6 +46,7 @@ async function waitFor(read, label, milliseconds = 90_000) { async function main() { const nativePools = process.argv.includes('--native-pools') + const sqlite = process.argv.includes('--sqlite') const nativeModel = nativePools ? nativePoolModel() : null const agents = process.argv.includes('--agents') const serve = process.argv.includes('--workbench-serve') @@ -110,7 +112,7 @@ async function main() { ...(workbench && !live ? { projectResources: { 'demo-local': { type: 'local', root: directory, profileRoot: join(directory, 'private-sdk-profiles'), capacity: 2, allowedProjectOwnerIds: ['owner-1'], adminMemberIds: ['owner-1'], harnesses: ['codex'], profiles: { 'owner-1': { codex: { home: join(directory, 'sdk-home'), configDir: join(directory, 'sdk-config'), model: 'fixture-sdk-probe', baseUrl: 'http://127.0.0.1:1' } } } } } } : {}), ...(workbench ? { projectDemoMode: realModel ? 'real-provider' : 'deterministic-fixture' } : {}), ...(shared ? { controlPlane: { origin: shared.origin, ownerSubject: shared.ownerSubject } } : {}), - ...(delivery ? { projectDispatch: nativePools ? 'pools' : 'workflow', projectRepositories: { [githubRepository ?? 'example/delivery-fixture']: { ...(githubRepository ? { github: { credential: 'gh-cli', configDir: githubConfigDir, ...(codex ? { login: codex.login } : {}) } } : {}), path: workspace, testCommand: queueGate?.command ?? (live || codex ? [process.execPath, golden] : [process.execPath, '--test', 'code.test.mjs']), allowedPaths: ['code.mjs', 'code.test.mjs'] }, ...(live ? { 'example/intentional-failure': { path: workspace, testCommand: [process.execPath, golden, '--force-failure'], allowedPaths: ['code.mjs', 'code.test.mjs'] } } : {}) } } : {}) } + ...(sqlite ? { stateStorage: 'sqlite' } : {}), ...(delivery ? { projectDispatch: nativePools ? 'pools' : 'workflow', projectRepositories: { [githubRepository ?? 'example/delivery-fixture']: { ...(githubRepository ? { github: { credential: 'gh-cli', configDir: githubConfigDir, ...(codex ? { login: codex.login } : {}) } } : {}), path: workspace, testCommand: queueGate?.command ?? (live || codex ? [process.execPath, golden] : [process.execPath, '--test', 'code.test.mjs']), allowedPaths: ['code.mjs', 'code.test.mjs'] }, ...(live ? { 'example/intentional-failure': { path: workspace, testCommand: [process.execPath, golden, '--force-failure'], allowedPaths: ['code.mjs', 'code.test.mjs'] } } : {}) } } : {}) } writeFileSync(patch, readFileSync(join(root, 'docker/web.patch.yml'), 'utf8') + (live ? '\n- id: llm-deepseek\n config: { thinking: disabled, reasoningEffort: off, maxTokens: 8192 }\n' : '') + `\n- id: bot\n config: ${JSON.stringify(botConfig)}\n`, { mode: 0o600 }) } const child = spawn(process.execPath, [cli, '--profile', 'web', '--patch', patch, @@ -119,7 +121,12 @@ async function main() { let childError child.on('error', error => { childError = error }) const statePath = join(home, 'bot/state.json') - const saved = () => existsSync(statePath) ? json(statePath) : null + const saved = () => { + if (!existsSync(statePath)) return null + if (!sqlite) return json(statePath) + const database = openSqliteState({ path: join(home, 'bot/state.sqlite'), readOnly: true }) + try { return database.read() } finally { database.close() } + } const teamWorkspace = join(home, 'bot/workspace') let keep = false const stop = signal => { @@ -179,7 +186,7 @@ async function main() { return } if (nativePools) { - console.log(JSON.stringify(await checkNativePools({ api, saved, waitFor, repository: workspace, model: nativeModel }))) + console.log(JSON.stringify(await checkNativePools({ api, saved, waitFor, repository: workspace, model: nativeModel, storage: sqlite ? 'sqlite' : 'json' }))) return } if (delivery) { diff --git a/dev/demo/test/runtime.test.mjs b/dev/demo/test/runtime.test.mjs index b11668f13..2917aac4f 100644 --- a/dev/demo/test/runtime.test.mjs +++ b/dev/demo/test/runtime.test.mjs @@ -87,3 +87,20 @@ test('real DSH resumes native Task-role context after a pool return and serializ assert.ok(report.notRun.includes('real model')) assert.deepEqual(readdirSync(directory), [], 'Disposable native workers, profiles and worktrees must be removed') }) + +test('actual DSH native return persists Task-role activation, release and continuation through SQLite', { timeout: 150_000 }, t => { + const directory = mkdtempSync(join(tmpdir(), 'flybot-native-sqlite-runtime-')) + t.after(() => rmSync(directory, { recursive: true, force: true })) + const result = spawnSync(process.execPath, [resolve(import.meta.dirname, '../run.mjs'), '--native-pools', '--sqlite'], { + encoding: 'utf8', timeout: 140_000, env: { ...process.env, TMPDIR: directory, M0_REPORT: '' }, + }) + assert.equal(result.status, 0, result.stderr) + const report = JSON.parse(result.stdout) + assert.equal(report.storage, 'sqlite') + assert.equal(report.runtime, 'actual DSH') + assert.ok(report.checks.includes('actual Chief admission before native pool allocation')) + assert.ok(report.checks.includes('same Task and original physical role Session')) + assert.ok(report.checks.includes('native quiescence and single-resource release before downstream execution')) + assert.ok(report.notRun.includes('real model')) + assert.deepEqual(readdirSync(directory), [], 'Disposable SQLite state, workers and worktrees must be removed') +}) diff --git a/docs/architecture.md b/docs/architecture.md index e9ff3710f..1e0f53980 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -176,6 +176,14 @@ The [shared context contract](subsystems/project-context.md) separates fixed rol Project Task contracts extend original Task records with Chief admission, versioned requirements, independent child dependencies and a parent continuation. Existing Effects commit proposals/waits and dispatch after persistence; existing pools serialize the single parent resume. Child completion never accepts the parent or approves a code merge. See [Project Task lifecycle](subsystems/project-task-lifecycle.md). +## Runtime storage convergence (#2182) + +The original Store can persist its canonical domain records and Effects in one +SQLite transaction while retaining existing service objects. Legacy JSON is selected +until explicitly migrated; a versioned guard prevents two writable authorities. +History rows and immutable payload files separate storage growth from bounded model +context. See [transactional storage and offline migration](subsystems/project-state-storage.md). + Project role governance (#2171) separates resident Chief proposals, Maintainer/Owner decisions, the current revisioned role catalog, and archived definitions/audit. Durable Effects proof gates visible changes; logical role keys and execution capability classes remain distinct. See [role governance](subsystems/project-agents.md#project-role-governance-2171). diff --git a/docs/development-service-api.md b/docs/development-service-api.md index 4114b6791..bd1c1e049 100644 --- a/docs/development-service-api.md +++ b/docs/development-service-api.md @@ -537,3 +537,13 @@ resumePoint. Public HTTP cannot supply a role claim. Parent Task views report parent performs its own integration and final verification. For lifecycle fields, current limitations and compatibility, see [Project Task lifecycle](subsystems/project-task-lifecycle.md). + +## Persistence configuration (#2182) + +Plugin configuration `stateStorage: sqlite` selects the original Store's SQL backend +(Node.js22.18+, tested22.22.3). Project commands, authorization and query envelopes +remain unchanged; state and Effects share one commit. Existing JSON remains default +until migrated. Runtime storage diagnostics/change IDs are internal ports, not a +public raw-database API or authenticated event feed. See the [migration/rollback +contract](subsystems/project-state-storage.md). Public bounded receipts/events are +completed separately in #2185; simplified full deployment instructions remain #2189. diff --git a/docs/history/2026-11-03-upstream-project-inventory.md b/docs/history/2026-11-03-upstream-project-inventory.md new file mode 100644 index 000000000..9618a0e91 --- /dev/null +++ b/docs/history/2026-11-03-upstream-project-inventory.md @@ -0,0 +1,60 @@ +# Historical upstream Project inventory + +Preserved verbatim from the explicitly historical portion of PROJECT.md. Current fork authorization and implementation status remain in active memory. Automation boundaries remain active in PROJECT.md without alteration. + +## Mission + +FlyBot is a local team of long-lived AI Bots that runs on DeepSeek Harness (DSH), in your own +browser, with your own model keys. One person talks to one Bot — the **Chief** — and the team takes +the request apart and comes back with one answer. The Bots keep their own memory, working folder, +model and schedules between turns, so work outlives a single chat. + +The project's mission is to keep that team reliable and inspectable: every Bot acts on evidence, +stays inside its goal, and its result can be checked by a person afterwards. The product is the app +in this repository plus the patches that pin it to upstream `ds-bot`; it is not a hosting service. + +`AGENTS.md` states the two halves the code is built from: pristine upstream `ds-bot/` (never edited), +and `core/` — `ds-bot/` plus FlyBot's ordered patches — which is what ships and what the tests run. +The shipped features are the DSH plugin in `core/` and the peer plugins `plugins/social`, +`plugins/tree`, `plugins/timeline` and `plugins/workflow`. + +## Owner and maintainer roles + +Today the project has one owner and a small number of human maintainers; the Bots work under the +owner's delegation. This is the state of the repository as of 2026-10-29, not the target model of +[#1550](https://github.com/ThinkFlowLab/FlyBot/issues/1550). + +| Role | Who | Decides | +| --- | --- | --- | +| **Owner** | Danyang, GitHub [`Birfy`](https://github.com/Birfy) (repository admin) | Mission, governance policy, roles and membership, project Bots, compute and resources, budgets, autonomy boundaries, and every matter reserved to the owner in [docs/governance.md](docs/governance.md) | +| **Maintainer** (human) | People with repository write access — today the owner and [`hsliuustc0106`](https://github.com/hsliuustc0106) | Major decisions under [docs/governance.md](docs/governance.md): roadmap and architecture changes, changes to the governing documents, multi-subsystem contracts. Records a verdict through the review policy, never through an admin merge | +| **Chief** (AI) | The owner's Main Bot, writing as `Birfy` | Plans and routes work, splits a request into role tasks, **classifies** each decision as ordinary / major / severe with its rationale, convenes a discussion when one is needed, and escalates to the owner. Not the ultimate authority and not the owner | +| **Routine Bots** (AI) | `Triage`, `Dev`, `Dev2`–`Dev4`, `Reviewer`, `Reviewer2`, `Release`, `SelfTest`, `Docs`, `Maintainer` — the names in [`writers`](.github/agent/review-policy.json) | Only their routine's job: see [Routines](AGENTS.md#routines). A Bot records a conclusion only where its routine owns that conclusion, and the implementer never writes the verdict on its own change | +| **Contributors** (human) | Anyone who opens an issue or a pull request | Proposals, bug reports, code. No default approval right; a contribution merges only through the gates in [AGENTS.md](AGENTS.md) | + +Deliberately **not** claimed here, because it is not shipped: a Maintainer console, a Member role, a +Project Assistant, per-person GitHub attribution for Bots, and the multi-human authority contract in +[docs/subsystems/team.md](docs/subsystems/team.md) (planned; team mode stays disabled until its +authorization gates pass). Every Bot still writes as the one shared `Birfy` identity, so a +`Written-by:` line is a claim the gates can check for presence and name, not a signature. + +## Resources covered + +- **This repository** — `https://github.com/ThinkFlowLab/FlyBot`, the source of record. `main` is + the live branch; pull requests merge into it only through the merge queue. +- **Code** — `core/` (shipped plugin, source of truth), `patches/ds-bot/` and `core/client.js` + (generated from `core/`), `plugins/*` (peer plugins), `scripts/` and `core/scripts/`. +- **Rules and knowledge** — `AGENTS.md` (operating contract), `CONTRIBUTING.md`, `LESSONS.md` + (observed failures and promotion to standards), `docs/` (architecture, standards, runbooks), + `decisions/` (dated decisions), `PROJECT.md` and `ROADMAP.md` (this page and the plan). +- **Automation** — `.github/workflows/` and `.github/scripts/` (CI, risk tier, review gate, merge + gate, enqueue guard), `.github/agent/` (risk rules and the review policy). +- **Authoritative runtime state elsewhere** — GitHub issues, pull requests and CI runs are the + authority for work state; task boards, approvals and per-Bot memory live in FlyBot's own storage, + never in this repository. `docs/` pages are informative only where a source above decides + something. +- **Not in this repository** — the upstream `ds-bot` project (pinned in `ds-bot.pin.json`, read-only + here) and the self-hosted Bot Social server used by the social plugin. The fork + `Birfy/FlyBot` is an archive: nothing merges into it. + +End of preserved historical inventory. Active governing and automation boundaries remain in PROJECT.md. diff --git a/docs/subsystems/core-host.md b/docs/subsystems/core-host.md index 509ff97dc..4ae3252ea 100644 --- a/docs/subsystems/core-host.md +++ b/docs/subsystems/core-host.md @@ -968,3 +968,7 @@ Personal-agent connection authentication (#2183) installs after authz and wraps New shared Project Task construction is reused through a trusted unsaved staging seam for parallel children. Project contract transitions go through the internal delivery capability; generic board transitions and the legacy task driver cannot take over admission or final judgement. Persisted pre-contract personal/group Tasks retain their original behavior. A failed Attempt leaves a new Project Task blocked with its source Attempt/reason; it does not close the commitment or erase Chief admission. A later active Attempt clears that execution block. Pre-contract legacy Task failure semantics remain unchanged. + +## Transactional persistence (#2182) + +The original Store supports opt-in `stateStorage: sqlite` through the [storage bridge](project-state-storage.md). It retains the same runtime objects, save chain, Project services and home lock, and acknowledges Effects after actual SQL COMMIT. Locked JSON remains the default until selected; after migration the newer-version JSON guard fences old writers and SQL remains authoritative across configuration omission. Invalid migration never creates a fresh empty team. No extra model call or periodic task is introduced. diff --git a/docs/subsystems/demo-baseline.md b/docs/subsystems/demo-baseline.md index 804871279..8b9b5b441 100644 --- a/docs/subsystems/demo-baseline.md +++ b/docs/subsystems/demo-baseline.md @@ -176,3 +176,5 @@ Quota assertions use an operator-owned verification gate in the disposable test `node dev/demo/run.mjs --native-pools` adds actual DSH single-resource Developer/Tester return/context restoration acceptance with a deterministic model. The marker is emitted only into the first Developer conversation outside the durable handoff JSON; the second request must find it in retained history, absent fresh system/input. It verifies fresh carriers, same physical Session/cwd, actual second code revision/host tests, lease intervals and Chief arbitration without fake closure; cleanup stops its disposable Attempt and removes worktrees. The required runtime suite now includes this fifth case. Initial failures exposed implicit [object Object] delivery deduplication and DSH archived-session admission; retained private evidence distinguishes both from the passing probe. + +`run.mjs --native-pools --sqlite` runs the same canonical Chief/pool/context/exit scenario with the real transactional Store; `saved()` reads independent committed SQLite snapshots. Its sixth runtime regression verifies durable activation markers and cleans the disposable database and worktrees. Attempt chronology uses persisted predecessor references, not map key/insertion order. This is deterministic verification, not the backend deployment method. diff --git a/docs/subsystems/project-state-storage.md b/docs/subsystems/project-state-storage.md new file mode 100644 index 000000000..650b79bb4 --- /dev/null +++ b/docs/subsystems/project-state-storage.md @@ -0,0 +1,143 @@ +--- +subsystem: project-state-storage +owned_paths: + - core/src/host/project-state-sqlite.js + - core/src/host/project-state-storage.js + - core/src/host/project-state-migration.js + - core/scripts/state-storage.mjs + - core/test/project-state-sqlite.test.js + - core/test/project-state-storage.test.js + - core/test/project-state-runtime.test.js +--- + +# Transactional runtime state + +The original Store and domain services remain the only runtime objects. Set the +FlyBot plugin configuration `stateStorage: sqlite` to opt into SQLite. Existing +JSON deployments remain JSON until selected. SQLite requires Node.js22.18 or newer; +this implementation is tested on Node.js22.22.3 with SQLite3.51.3. No separate server, +message queue, model call or scheduler is introduced. This PR does not migrate a +running instance. Stop the instance, back up its private home, select the option, +and restart through its existing host. General backend deployment remains #2189. + +## Commit boundary + +The existing save chain captures the full in-memory JSON-compatible snapshot and +Effects watermark at call time. SQLite applies changed records, removed records, +normalized histories and Effect intents in one `BEGIN IMMEDIATE` transaction. +Unique record/history/change keys and a database commit-revision compare-and-set +prevent stale snapshots from overwriting another writer. The original exclusive +home lock remains required; a broken lock does not authorize SQL writes. A read-only +host cannot acknowledge Effects. `projectEffects.saved` runs only after successful +COMMIT, with no fallible database read between COMMIT and that acknowledgement. +External execution remains after commit through existing adapters. + +WAL, synchronous FULL and foreign keys are configured and read back. Database and +payload files are private; external credentials are not introduced. Actual process +kill tests cover before-COMMIT rollback and after-COMMIT Task/Effect survival. They +do not simulate physical power loss or guarantee a filesystem/device honors fsync. +Use a local filesystem compatible with SQLite WAL, not a network-mounted database. +See the [Node.js22.22.3 SQLite API](https://raw.githubusercontent.com/nodejs/node/v22.22.3/doc/api/sqlite.md), +[SQLite synchronous settings](https://www.sqlite.org/pragma.html#pragma_synchronous) +and [WAL constraints](https://www.sqlite.org/wal.html). + +## Layout and growth + +- `state.sqlite`: metadata, per-entity records, history rows and ordered change IDs. +- `state_records`: namespace/ID primary key; immutable content hash and complete + record fingerprint. Root descriptors preserve empty maps and scalar values. +- `state_history`: individual ordered entries for history, requirementsHistory, + waitHistory, results, roleSessionHistory, resumptions and versions arrays. A new + history item does not rewrite previous history items. +- `state_changes`: monotonic sequence, commit revision, namespace/ID and operation; + no private record bodies. Its bounded internal query is not a public event feed. + Authenticated Project-scoped HTTP/events remain #2185. +- `state-payloads/.json`: immutable, hash-verified payloads above64KiB; + individual payloads are limited to16MiB. Files are fsynced before their transaction + can reference them. Failed commits may leave harmless unreferenced files; this + implementation does not automatically delete history or collect those files. +- Existing logs, artifacts, SDK histories and workspaces stay in their existing + filesystem locations. Database records retain their references and hashes. + +Compatibility still hydrates the full runtime state. Each save scans/normalizes its +snapshot and compares fingerprints; this is not constant-time or lazy SQL execution. +It reduces changed disk records and separates append histories. History retention, +in-memory residency and model context budgets are different concerns. Bounded +role context, source retrieval and summary versions stay in #2177; changing the +storage engine does not load all history into a model or solve context overflow. + +## Migration and one writable authority + +1. Validate the original JSON schema and ownership through the original Store, + applying its existing migrations and retaining identities. Invalid JSON or + ownership refuses migration; it cannot create an empty SQL replacement. +2. Save a private immutable normalized import with SHA256. Atomically publish a + version3 `state.json` guard identifying the import and migration ID, phase + `prepared`. JSON-only older binaries refuse that newer version. +3. Initialize state, schema and migration metadata in one SQL transaction. Verify + the DB/guard identity and activate the guard. Thereafter `state.json` is a guard, + not a second writable snapshot; omitting the option on restart still selects SQL. +4. A prepared migration can recover from its verified import. A missing **active** + database refuses to import an old backup automatically. Foreign/future schema, + changed backup, wrong guard ID, invalid ownership or unknown authority fails closed. + +The current implementation migrates the complete original team state into one SQL +transaction domain, including legacy users/Bots/rooms, Project records and pending +Effects; it does not run JSON and SQL as independently writable partial stores. +Working Task, Handoff, Attempt, SDK binding, logical Session and Effect IDs survive. +An old read-only host taking over must refresh current SQL state; a stale raw save +is rejected by revision instead of overwriting newer facts. + +## Private offline operator commands + +All hosts using this home must be stopped for export/rollback. The tool enforces the +exclusive home lock; do not manually copy a live main DB and omit its WAL. + +```sh +node core/scripts/state-storage.mjs inspect --home /private/flybot-home +node core/scripts/state-storage.mjs export --home /private/flybot-home --migration MIGRATION_ID --output /private/export.json +node core/scripts/state-storage.mjs rollback --home /private/flybot-home --migration MIGRATION_ID +``` + +Inspect returns metadata only. Use its exact migration ID as a stale-source guard. +Export preserves the current committed version2 state in a private file, not the +old import. Rollback keeps that export and archives DB/WAL/SHM together, atomically +publishes the latest JSON state and completes a rollback marker. Remove the explicit +`stateStorage: sqlite` option before restarting on JSON; otherwise it intentionally +opts into a fresh SQL migration again. No authorization, evidence or history is reset. + +A prepared rollback blocks runtime writes and startup adoption. After interruption: + +```sh +node core/scripts/state-storage.mjs rollback --home /private/flybot-home --migration MIGRATION_ID --resume +``` + +The tool checks the export hash, exact migration, archived paths and unchanged commit +revision, completes partial moves and publishes JSON once. It will not overwrite a +conflicting archive or silently revert newer SQL data. Current SQL writers also check +the storage guard, so an old connection cannot keep writing after authority changes. +The tool is not exposed to Bot roles or HTTP callers. + +## State semantics and release scope + +Existing authorized structured commands update formal facts deterministically. +Queries derive views from those records; they do not call a model, rewrite Project +goal/scope, or append to PROJECT.md/ROADMAP.md. Chief admission, dependency judgement +and final completion remain the existing semantic decision points. Inspections keep +their configurable scheduled-task policy. Summaries and human documents cannot +replace permissions, execution ownership or matching candidate evidence. + +Comparison checked against upstream main986399246ff8c26bd418978785ce1199dbd239ca +and delivery741bf1ef4aa199f0c7656095b32db3ab8e529a6a: main has the original core +Store but not this delivery branch's Project development service or Effects module. +The delivery base already had locked JSON snapshots and post-save Effects; it did +not have SQLite. This issue's PR stacks on #2179/4826723cc and adds SQL storage and +migration only. It does not merge upstream main, deploy, change control-plane/UI or +claim real-provider parallel Task development validation. Public event/receipt and +history service boundaries remain #2185; legacy entrypoint cutover remains #2186. + +## Composed native lifecycle verification + +The actual DSH native pool return probe also runs with `stateStorage: sqlite`. Its read side loads committed snapshots through a separate read-only SQLite connection, proving canonical Chief admission, native activation intent, retained Task-role context/parts, physical exit proofs and nonoverlapping resource release persist across role handoffs. It uses deterministic model transport and actual code/host tests, not provider inference or production migration. JSON remains the default. The dev probe's `--sqlite` flag is verification tooling; deployment documentation remains #2189. + +State maps have no semantic insertion order. The probe resolves the previous Attempt through the successor's `feedback.previousAttemptId`; SQLite map iteration exposed an invalid earlier assertion that assumed the first UUID record was oldest. Immutable history arrays preserve explicit positions. No domain fact or history was lost by that diagnostic failure. diff --git a/patches/ds-bot/0002-flybot-api-host.patch b/patches/ds-bot/0002-flybot-api-host.patch index 2a433e31f..0279e2580 100644 --- a/patches/ds-bot/0002-flybot-api-host.patch +++ b/patches/ds-bot/0002-flybot-api-host.patch @@ -699,12 +699,13 @@ diff --git a/src/host/roster.js b/src/host/roster.js diff --git a/src/host/store.js b/src/host/store.js --- a/src/host/store.js +++ b/src/host/store.js -@@ -1,21 +1,50 @@ +@@ -1,21 +1,51 @@ // Store: the saved team (state.json), its migrations, the lock that keeps a second dsh // host on the same data from writing it, and the indexes that map a Session to the Bot // or group chat it belongs to. -import { mkdirSync, statSync } from 'node:fs' +import { install as installProjects } from './project-service.js' ++import { createProjectStateStorage } from './project-state-storage.js' +import { constants, mkdirSync, statSync } from 'node:fs' import { copyFile, mkdir, readFile, rename, unlink, writeFile } from 'node:fs/promises' import { join } from 'node:path' @@ -754,17 +755,31 @@ diff --git a/src/host/store.js b/src/host/store.js const EMPTY = JSON.stringify(state) // The other dsh host that holds state.lock while this one only reads (see home-lock.js); -@@ -50,12 +50,88 @@ export function install(rt) { +@@ -27,8 +27,11 @@ export function install(rt) { + let stopped = false + // mtime and size of the state.json this host last read or wrote. + let seen = '' ++ let storage + const stamp = () => { + try { ++ const current = storage?.stamp() ++ if (current !== null && current !== undefined) return current + const { mtimeMs, size } = statSync(statePath) + return `${mtimeMs}:${size}` + } catch { +@@ -50,13 +50,90 @@ export function install(rt) { } } const readOnly = () => { - if (stuck !== undefined) return `The Bot team file ${statePath} could not be read (${stuck}). DS Bot does not change it; fix or move the file and reload.` ++ if (storage?.sqlite && lockBroken) return 'SQLite state requires an intact exclusive home lock; fix locking and reload.' + if (stuck !== undefined) return `The Bot team file ${statePath} could not be read (${stuck}). FlyBot does not change it; fix or move the file and reload.` if (holder === undefined) return undefined const where = holder.pid ? `pid ${holder.pid}${holder.host ? ` on ${holder.host}` : ''}` : 'unknown process' return `The Bot team in ${home} is open in another dsh (Web or Desktop, ${where}). This window shows the team but cannot change it; close the other one and reload.` } +- const readFileState = async () => { + // A version-1 team predates member records: it carries no `members` and no per-Bot + // `ownerId`. Ownership fields on it are not something the migration can trust, so the + // file is refused (fail closed) instead of being guessed at. @@ -841,10 +856,11 @@ diff --git a/src/host/store.js b/src/host/store.js + return undefined + } + - const readFileState = async () => { ++ const readJsonState = async () => { let contents try { -@@ -87,10 +87,32 @@ export function install(rt) { + contents = await readFile(statePath, 'utf8') +@@ -87,13 +87,59 @@ export function install(rt) { return undefined } if (parsed?.version === 1) { @@ -860,10 +876,9 @@ diff --git a/src/host/store.js b/src/host/store.js + warn('ds-bot: unreadable state %s: %s', statePath, backupProblem) + return undefined + } - stuck = undefined - return parsed - } -- stuck = 'saved by a newer DS Bot; update DS Bot' ++ stuck = undefined ++ return parsed ++ } + if (parsed?.version === STATE_VERSION) { + const problem = ownershipProblem(parsed) + if (problem !== undefined) { @@ -871,13 +886,41 @@ diff --git a/src/host/store.js b/src/host/store.js + warn('ds-bot: state %s was refused: %s', statePath, problem) + return undefined + } -+ stuck = undefined -+ return parsed -+ } + stuck = undefined + return parsed + } +- stuck = 'saved by a newer DS Bot; update DS Bot' + stuck = 'saved by a newer FlyBot; update FlyBot' return undefined } ++ const validateSqliteState = parsed => { ++ if (parsed?.version !== STATE_VERSION) throw new Error('Unsupported SQLite state version') ++ const problem = ownershipProblem(parsed) ++ if (problem !== undefined) throw new Error(`Malformed SQLite ownership: ${problem}`) ++ } ++ storage = createProjectStateStorage({ home, statePath, mode: config.stateStorage, validate: validateSqliteState, ++ ensureWriter() { if (lockBroken || holder !== undefined || !lock.owns()) throw Object.assign(new Error('SQLite state needs the current exclusive home lock'), { code: 'ELOCKLOST' }) } }) ++ const readFileState = async () => { ++ try { ++ const result = await storage.read({ writer: holder === undefined && !lockBroken, readLegacy: readJsonState }) ++ if (result.sqlite) { stuck = undefined; seen = stamp() } ++ return result.state ++ } catch (error) { ++ stuck = error.message; seen = stamp() ++ warn('ds-bot: durable state was refused: %s', error) ++ return undefined ++ } ++ } ++ const initializeStorage = () => { ++ if (holder !== undefined || stuck !== undefined) return false ++ try { return storage.initialize(state) } ++ catch (error) { stuck = error.message; warn('ds-bot: SQLite migration was refused: %s', error); return false } ++ } ++ + // Fill `state` from the saved file and bring an older team up to date. Returns the + // groups whose Session titles still need the new name. + const adopt = (parsed) => { @@ -104,6 +104,20 @@ export function install(rt) { // Teams saved before the DSH Agent existed carry no `agents`; corrupt values reset. if (state.agents === null || typeof state.agents !== 'object' || Array.isArray(state.agents)) state.agents = {} @@ -923,7 +966,7 @@ diff --git a/src/host/store.js b/src/host/store.js // LEGACY: Bots saved with a lab ran on that lab's default model; the model they // last ran on becomes their own, so dropping the lab changes nothing for them. if ('lab' in bot) { -@@ -161,12 +161,94 @@ export function install(rt) { +@@ -161,18 +161,101 @@ export function install(rt) { migrated = true } // Unnamed groups from before names followed their members may still list old names. @@ -1019,7 +1062,15 @@ diff --git a/src/host/store.js b/src/host/store.js const load = () => loaded ??= (async () => { claimLock() if (holder !== undefined) warn('ds-bot: %s', readOnly()) -@@ -185,14 +185,24 @@ export function install(rt) { + const { migrated, stale } = adopt(await readFileState()) + if (holder !== undefined) return +- if (migrated && stuck === undefined) { ++ const initialized = initializeStorage() ++ if ((migrated || initialized) && stuck === undefined) { + // A failed write must not poison `loaded`: the migrated state stays in + // memory and the next save() writes it. + try { +@@ -185,16 +185,27 @@ export function install(rt) { for (const room of stale) void rt.retitleRoom(room) })() @@ -1044,7 +1095,24 @@ diff --git a/src/host/store.js b/src/host/store.js + announceTakeover(previous) warnedReadOnly = false adopt(await readFileState()) ++ initializeStorage() return + } + } else if (!lockBroken && !lock.owns()) { +@@ -208,11 +208,12 @@ export function install(rt) { + // written while no host held it. + warnedReadOnly = false + adopt(await readFileState()) ++ initializeStorage() + return + } + warn('ds-bot: %s', readOnly()) + } +- if ((stuck !== undefined || holder !== undefined) && stamp() !== seen) adopt(await readFileState()) ++ if ((stuck !== undefined || holder !== undefined) && stamp() !== seen) { adopt(await readFileState()); initializeStorage() } + } + + const warnReadOnly = () => { @@ -235,13 +235,40 @@ export function install(rt) { return warnReadOnly() } @@ -1086,10 +1154,19 @@ diff --git a/src/host/store.js b/src/host/store.js const snapshot = JSON.stringify(state, null, 2) const run = writing.then(async () => { const temp = `${statePath}.${process.pid}.tmp` -@@ -259,10 +259,12 @@ export function install(rt) { - await writeFile(temp, snapshot, 'utf8') +@@ -256,13 +256,18 @@ export function install(rt) { + } + } ensureLock() - await rename(temp, statePath) +- await writeFile(temp, snapshot, 'utf8') +- ensureLock() +- await rename(temp, statePath) ++ if (storage.sqlite) storage.write(snapshot) ++ else { ++ await writeFile(temp, snapshot, 'utf8') ++ ensureLock() ++ await rename(temp, statePath) ++ } + rt.projectEffects?.saved(effectSnapshot) seen = stamp() + await journal?.record(snapshot) @@ -1130,7 +1207,8 @@ diff --git a/src/host/store.js b/src/host/store.js + ctx.effect(() => () => { stopped = true - return writing.finally(() => lock.release()) +- return writing.finally(() => lock.release()) ++ return writing.finally(() => { try { storage.close() } finally { lock.release() } }) }, 'ds-bot: state lock') - const botOf = id => (id === undefined ? undefined : state.bots[id]) @@ -1163,7 +1241,7 @@ diff --git a/src/host/store.js b/src/host/store.js rt.cancelApprovals?.(sessionId) // A plain archive refuses a running Session, which would go on working for a group // that no longer has it. -@@ -335,23 +335,28 @@ export function install(rt) { +@@ -335,23 +335,29 @@ export function install(rt) { } const findBot = (ref) => { if (ref === undefined) return undefined @@ -1189,6 +1267,7 @@ diff --git a/src/host/store.js b/src/host/store.js Object.assign(rt, { - state, load, save, refresh, readOnly, + state, load, save, refresh, readOnly, mutate, savesQueued, savesSettled, ++ stateStorage: { status: () => storage.status(), changes: options => storage.changes(options) }, botOf, roomOf, isMain, chatOf, selfOf, isBotSession, chatOwners, groupOwners, pastOwners, groupSessionsOf, dropGroupSession, findBot, findRooms, isAdminOf, diff --git a/patches/ds-bot/files/scripts/state-storage.mjs b/patches/ds-bot/files/scripts/state-storage.mjs new file mode 100644 index 000000000..b85015282 --- /dev/null +++ b/patches/ds-bot/files/scripts/state-storage.mjs @@ -0,0 +1,25 @@ +// Offline private operator interface; no account credentials are accepted here. +import { resolve } from 'node:path' +import { readFileSync } from 'node:fs' +import { exportSqliteState, rollbackSqliteState } from '../src/host/project-state-migration.js' + +try { + const [command, ...args] = process.argv.slice(2), options = {} + if (!['inspect', 'export', 'rollback'].includes(command)) throw new Error('Usage: node core/scripts/state-storage.mjs inspect|export|rollback --home PATH [--migration ID] [--output PATH] [--resume]') + for (let i = 0; i < args.length; i++) { + const key = args[i] + if (!['--home', '--migration', '--output', '--resume'].includes(key) || Object.hasOwn(options, key)) throw new Error('Unknown or duplicate migration option') + if (key === '--resume') options[key] = true + else if (!args[i + 1] || args[i + 1].startsWith('--')) throw new Error('Migration option requires a value') + else options[key] = args[++i] + } + if (!options['--home']) throw new Error('An explicit private --home is required') + const home = resolve(options['--home']) + let result + if (command === 'inspect') { + const guard = JSON.parse(readFileSync(resolve(home, 'state.json'), 'utf8')) + result = { backend: guard.storage === 'sqlite' ? 'sqlite' : 'json', phase: guard.phase ?? null, migrationId: guard.migration?.id ?? null } + } else if (command === 'export') result = exportSqliteState({ home, expectedMigrationId: options['--migration'], output: options['--output'] ? resolve(options['--output']) : null }) + else result = rollbackSqliteState({ home, expectedMigrationId: options['--migration'], resume: options['--resume'] === true }) + process.stdout.write(JSON.stringify(result) + '\n') +} catch (error) { process.stderr.write(`State storage operation refused: ${error.message}\n`); process.exitCode = 1 } diff --git a/patches/ds-bot/files/src/host/project-state-migration.js b/patches/ds-bot/files/src/host/project-state-migration.js new file mode 100644 index 000000000..775aa3757 --- /dev/null +++ b/patches/ds-bot/files/src/host/project-state-migration.js @@ -0,0 +1,84 @@ +// Offline operator tooling. No endpoint grants migration privileges to a Bot. +import { createHash, randomUUID } from 'node:crypto' +import { existsSync, readFileSync, renameSync } from 'node:fs' +import { join } from 'node:path' +import { homeLock } from './home-lock.js' +import { openSqliteState } from './project-state-sqlite.js' +import { writeDurableStateFile } from './project-state-storage.js' + +const digest = text => createHash('sha256').update(text).digest('hex') +const fail = message => Object.assign(new Error(message), { code: 'bot/state-migration' }) +function exclusively(home, action) { + const path = join(home, 'state.lock') + if (existsSync(path) && JSON.parse(readFileSync(path, 'utf8'))?.pid === process.pid) throw fail('Stop the current FlyBot host before migration') + const lock = homeLock(path) + if (lock.claim() !== undefined) throw fail('Stop the active FlyBot host before offline migration') + try { return action() } finally { lock.release() } +} +function guardOf(home, expectedMigrationId) { + const guard = JSON.parse(readFileSync(join(home, 'state.json'), 'utf8')) + if (guard.version !== 3 || guard.storage !== 'sqlite' || guard.phase !== 'active' || !expectedMigrationId || guard.migration?.id !== expectedMigrationId) throw fail('The active SQLite migration ID changed; inspect before exporting or rolling back') + return guard +} +function snapshot(home, expectedMigrationId) { + const guard = guardOf(home, expectedMigrationId) + const db = openSqliteState({ path: join(home, 'state.sqlite'), readOnly: true }) + try { + const metadata = db.metadata() + if (metadata.migration?.id !== guard.migration.id || metadata.migration?.sha256 !== guard.migration.sha256) throw fail('SQLite and guard identities differ') + const state = db.read() + if (state.version !== 2) throw fail('Unsupported exported state version') + guardOf(home, expectedMigrationId) + return { text: JSON.stringify(state, null, 2), revision: metadata.revision } + } finally { db.close() } +} +export function exportSqliteState({ home, output, expectedMigrationId }) { + if (!output) throw fail('Choose a private export output file') + return exclusively(home, () => { + const current = snapshot(home, expectedMigrationId) + writeDurableStateFile(output, current.text, { exclusive: true }) + return { backend: 'sqlite', revision: current.revision, sha256: digest(current.text), output } + }) +} +export function rollbackSqliteState({ home, expectedMigrationId, resume = false }) { + return exclusively(home, () => { + const markerPath = join(home, 'state-storage-rollback.json'), path = join(home, 'state.sqlite'), statePath = join(home, 'state.json') + const previous = existsSync(markerPath) ? JSON.parse(readFileSync(markerPath, 'utf8')) : null + let marker + if (previous?.phase === 'prepared') { + if (!resume || previous.migrationId !== expectedMigrationId) throw fail('A prepared rollback exists; explicitly resume its exact migration ID') + marker = previous + } else { + if (resume && previous?.phase === 'completed' && previous.migrationId === expectedMigrationId) return { ...previous, already: true } + if (resume) throw fail('No matching prepared rollback exists') + const current = snapshot(home, expectedMigrationId), id = randomUUID() + marker = { version: 'flybot-storage-rollback/v1', phase: 'prepared', id, migrationId: expectedMigrationId, revision: current.revision, + exportFile: `state.rollback-${id}.json`, archiveFile: `state.sqlite.rollback-${id}`, sha256: digest(current.text), preparedAt: Date.now() } + writeDurableStateFile(join(home, marker.exportFile), current.text, { exclusive: true }) + writeDurableStateFile(markerPath, JSON.stringify(marker, null, 2)) + } + if (marker.version !== 'flybot-storage-rollback/v1' || !/^[a-f0-9-]{36}$/.test(marker.id) || marker.archiveFile !== `state.sqlite.rollback-${marker.id}` || marker.exportFile !== `state.rollback-${marker.id}.json` || !/^[a-f0-9]{64}$/.test(marker.sha256)) throw fail('Rollback recovery marker is malformed') + const text = readFileSync(join(home, marker.exportFile), 'utf8') + if (digest(text) !== marker.sha256 || JSON.parse(text).version !== 2) throw fail('Rollback export hash changed') + const currentText = readFileSync(statePath, 'utf8') + if (digest(currentText) !== marker.sha256) { + guardOf(home, expectedMigrationId) + if (existsSync(path)) { + const current = snapshot(home, expectedMigrationId) + if (current.revision !== marker.revision || digest(current.text) !== marker.sha256) throw fail('SQL state changed after rollback preparation') + } else if (!existsSync(join(home, marker.archiveFile))) throw fail('Rollback database archive is missing') + // Keep DB/WAL/SHM names paired. A crash between moves leaves the marker and + // verified export; resume finishes it before any runtime can write again. + for (const suffix of ['', '-wal', '-shm']) { + const source = path + suffix, target = join(home, marker.archiveFile) + suffix + if (!existsSync(source)) continue + if (existsSync(target)) throw fail('Rollback archive already contains a conflicting file') + renameSync(source, target) + } + writeDurableStateFile(statePath, text) + } + marker = { ...marker, phase: 'completed', completedAt: Date.now() } + writeDurableStateFile(markerPath, JSON.stringify(marker, null, 2)) + return { ...marker, backend: 'json', removeStateStorageSqliteConfig: true } + }) +} diff --git a/patches/ds-bot/files/src/host/project-state-sqlite.js b/patches/ds-bot/files/src/host/project-state-sqlite.js new file mode 100644 index 000000000..9924ec87a --- /dev/null +++ b/patches/ds-bot/files/src/host/project-state-sqlite.js @@ -0,0 +1,223 @@ +// Same runtime records, transactional durable authority. No model, dispatcher or +// second queue lives here. History rows and immutable payloads bound SQL records. +import { createRequire } from 'node:module' +import { createHash, randomUUID } from 'node:crypto' +import { constants, chmodSync, closeSync, existsSync, fsyncSync, fstatSync, linkSync, lstatSync, mkdirSync, openSync, readSync, realpathSync, unlinkSync, writeFileSync } from 'node:fs' +import { dirname, join, resolve } from 'node:path' + +const require = createRequire(import.meta.url) +const APPLICATION_ID = 0x464c5942, SCHEMA = 1, FILE_THRESHOLD = 65536, MAX_PAYLOAD = 16 * 1024 * 1024 +const historyNames = new Set(['history', 'waitHistory', 'requirementsHistory', 'results', 'roleSessionHistory', 'resumptions', 'versions']) +const hash = value => createHash('sha256').update(value).digest('hex') +const fail = message => Object.assign(new Error(message), { code: 'bot/state-sqlite' }) +const record = value => value !== null && typeof value === 'object' && !Array.isArray(value) +const put = (object, key, value) => Object.defineProperty(object, key, { value, enumerable: true, writable: true, configurable: true }) + +export function requireSqliteRuntime() { + const [major, minor] = process.versions.node.split('.').map(Number) + if (major < 22 || major === 22 && minor < 18) throw fail('SQLite state requires Node.js22.18 or newer; JSON storage has not been migrated') + return require('node:sqlite').DatabaseSync +} + +function normalize(value) { + const histories = [] + function visit(row, path) { + if (Array.isArray(row) && historyNames.has(path.at(-1))) { + row.forEach((item, position) => histories.push({ path: JSON.stringify(path), position, text: JSON.stringify(item) })) + return [] + } + if (Array.isArray(row)) return row.map((item, index) => visit(item, [...path, index])) + if (!record(row)) return row + const result = {} + for (const [key, item] of Object.entries(row)) put(result, key, visit(item, [...path, key])) + return result + } + return { document: JSON.stringify(visit(value, [])), fingerprint: hash(JSON.stringify(value)), histories } +} + +function snapshotRecords(snapshot) { + const records = [], roots = {} + for (const [key, value] of Object.entries(snapshot)) { + if (key === '$root') throw fail('Reserved state namespace') + put(roots, key, record(value) ? 'map' : 'value') + if (record(value)) for (const [id, row] of Object.entries(value)) records.push({ namespace: key, id, ...normalize(row) }) + else records.push({ namespace: '$root', id: key, ...normalize(value) }) + } + return { roots: JSON.stringify(roots), records } +} + +export function openSqliteState({ path, payloadRoot = join(dirname(path), 'state-payloads'), readOnly = false, initial, migration = null }) { + path = resolve(path); payloadRoot = resolve(payloadRoot) + let info + try { info = lstatSync(path) } catch (error) { if (error.code !== 'ENOENT') throw error } + if (info && (!info.isFile() || info.isSymbolicLink())) throw fail('SQLite state must be a registered regular file') + if (!readOnly) mkdirSync(dirname(path), { recursive: true, mode: 0o700 }) + const DatabaseSync = requireSqliteRuntime() + if (!readOnly && !info) { const fd = openSync(path, 'wx', 0o600); closeSync(fd) } + const db = new DatabaseSync(path, { readOnly, timeout: 5000, enableForeignKeyConstraints: true, enableDoubleQuotedStringLiterals: false, allowExtension: false }) + let closed = false, expectedRevision = null + const scalar = sql => Object.values(db.prepare(sql).get())[0] + const rollback = () => { try { db.exec('ROLLBACK') } catch { /* No transaction remains after certain SQLite failures. */ } } + const meta = key => db.prepare('SELECT value FROM state_meta WHERE key=?').get(key)?.value + const setMeta = (key, value) => db.prepare('INSERT INTO state_meta(key,value) VALUES (?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value').run(key, String(value)) + function ensurePayloadRoot(create = false) { + if (!existsSync(payloadRoot) && !create) throw fail('State payload directory is missing') + if (!existsSync(payloadRoot)) mkdirSync(payloadRoot, { recursive: true, mode: 0o700 }) + if (!lstatSync(payloadRoot).isDirectory() || lstatSync(payloadRoot).isSymbolicLink() || realpathSync(payloadRoot) !== payloadRoot) throw fail('State payload directory binding changed') + } + function readPayload(row) { + let text = row.document + if (row.storage === 'file') { + if (!/^[a-f0-9]{64}$/.test(row.digest)) throw fail('Invalid state payload reference') + ensurePayloadRoot() + const file = join(payloadRoot, row.digest + '.json'), info = lstatSync(file) + if (!info.isFile() || info.isSymbolicLink() || info.size > MAX_PAYLOAD) throw fail('Invalid bounded state payload file') + const fd = openSync(file, constants.O_RDONLY | (constants.O_NOFOLLOW ?? 0)) + try { + const current = fstatSync(fd) + if (current.ino !== info.ino || current.dev !== info.dev) throw fail('State payload changed during read') + const buffer = Buffer.alloc(MAX_PAYLOAD + 1), size = readSync(fd, buffer, 0, buffer.length, 0) + if (size > MAX_PAYLOAD) throw fail('State payload exceeds its read budget') + text = buffer.subarray(0, size).toString('utf8') + } finally { closeSync(fd) } + } else if (row.storage !== 'inline' || typeof text !== 'string') throw fail('Invalid state payload storage') + if (Buffer.byteLength(text) > MAX_PAYLOAD || hash(text) !== row.digest) throw fail('State payload content hash changed') + return JSON.parse(text) + } + function payload(text) { + const digest = hash(text), size = Buffer.byteLength(text) + if (size > MAX_PAYLOAD) throw fail('State record exceeds bounded payload size; retain a file reference instead') + if (size <= FILE_THRESHOLD) return { storage: 'inline', document: text, digest } + ensurePayloadRoot(true) + const file = join(payloadRoot, digest + '.json') + if (!existsSync(file)) { + const temp = join(payloadRoot, `.${randomUUID()}.tmp`), fd = openSync(temp, 'wx', 0o600) + try { writeFileSync(fd, text, 'utf8'); fsyncSync(fd) } finally { closeSync(fd) } + try { linkSync(temp, file) } catch (error) { if (error.code !== 'EEXIST') throw error } + finally { unlinkSync(temp) } + const directory = openSync(payloadRoot, 'r') + try { fsyncSync(directory) } finally { closeSync(directory) } + } + readPayload({ storage: 'file', document: null, digest }) + return { storage: 'file', document: null, digest } + } + function write(snapshot, { initializing = false } = {}) { + if (closed || readOnly) throw fail('SQLite state is closed or read-only') + if (!record(snapshot) || snapshot.version !== 2) throw fail('SQLite needs a validated version-2 state') + const prepared = snapshotRecords(snapshot) + if (!initializing) db.exec('BEGIN IMMEDIATE') + try { + const revision = Number(meta('revision') ?? 0) + if (revision !== expectedRevision) throw fail('SQLite state revision conflict; reload before changing facts') + const previous = new Map(db.prepare('SELECT namespace,id,fingerprint FROM state_records').all().map(row => [JSON.stringify([row.namespace, row.id]), row])) + const changes = [] + const upsert = db.prepare('INSERT INTO state_records(namespace,id,document,storage,digest,fingerprint) VALUES (?,?,?,?,?,?) ON CONFLICT(namespace,id) DO UPDATE SET document=excluded.document,storage=excluded.storage,digest=excluded.digest,fingerprint=excluded.fingerprint') + const historyUpsert = db.prepare('INSERT INTO state_history(namespace,id,path,position,document,storage,digest) VALUES (?,?,?,?,?,?,?) ON CONFLICT(namespace,id,path,position) DO UPDATE SET document=excluded.document,storage=excluded.storage,digest=excluded.digest') + for (const row of prepared.records) { + const key = JSON.stringify([row.namespace, row.id]), prior = previous.get(key) + previous.delete(key) + if (prior?.fingerprint === row.fingerprint) continue + const data = payload(row.document) + upsert.run(row.namespace, row.id, data.document, data.storage, data.digest, row.fingerprint) + const oldHistory = new Map(db.prepare('SELECT path,position,digest FROM state_history WHERE namespace=? AND id=?').all(row.namespace, row.id).map(item => [JSON.stringify([item.path, item.position]), item])) + for (const entry of row.histories) { + const key = JSON.stringify([entry.path, entry.position]), old = oldHistory.get(key), digest = hash(entry.text) + oldHistory.delete(key) + if (old?.digest === digest) continue + const data = payload(entry.text) + historyUpsert.run(row.namespace, row.id, entry.path, entry.position, data.document, data.storage, data.digest) + } + for (const item of oldHistory.values()) db.prepare('DELETE FROM state_history WHERE namespace=? AND id=? AND path=? AND position=?').run(row.namespace, row.id, item.path, item.position) + changes.push({ namespace: row.namespace, id: row.id, operation: 'upsert' }) + } + for (const row of previous.values()) { + db.prepare('DELETE FROM state_records WHERE namespace=? AND id=?').run(row.namespace, row.id) + changes.push({ namespace: row.namespace, id: row.id, operation: 'delete' }) + } + const rootsChanged = meta('roots') !== prepared.roots + const next = revision + (changes.length || rootsChanged ? 1 : 0) + if (rootsChanged) setMeta('roots', prepared.roots) + if (next !== revision) { + setMeta('revision', next) + for (const row of changes.sort((a, b) => a.namespace.localeCompare(b.namespace) || a.id.localeCompare(b.id))) db.prepare('INSERT INTO state_changes(revision,namespace,id,operation) VALUES (?,?,?,?)').run(next, row.namespace, row.id, row.operation) + } + const lastSequence = Number(scalar('SELECT COALESCE(MAX(sequence),0) FROM state_changes')) + if (!initializing) db.exec('COMMIT') + // No fallible IO between COMMIT and the Store's Effects acknowledgement. + expectedRevision = next + return { revision: next, changed: changes.length, lastSequence } + } catch (error) { if (!initializing) rollback(); throw error } + } + try { + const schema = Number(scalar('PRAGMA user_version')), app = Number(scalar('PRAGMA application_id')) + const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'").all() + const fresh = schema === 0 && app === 0 && tables.length === 0 + if (!fresh && (schema !== SCHEMA || app !== APPLICATION_ID)) throw fail('Unsupported or foreign SQLite state schema/version') + if (fresh && (readOnly || initial === undefined)) throw fail('SQLite initialization needs validated initial state') + if (!readOnly) { + chmodSync(path, 0o600) + db.exec('PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; PRAGMA foreign_keys=ON; PRAGMA busy_timeout=5000') + if (scalar('PRAGMA journal_mode') !== 'wal' || Number(scalar('PRAGMA synchronous')) !== 2 || Number(scalar('PRAGMA foreign_keys')) !== 1) throw fail('Required SQLite durability configuration was not applied') + } + if (fresh) { + db.exec(`BEGIN IMMEDIATE; + CREATE TABLE state_meta(key TEXT PRIMARY KEY,value TEXT NOT NULL) STRICT; + CREATE TABLE state_records(namespace TEXT NOT NULL,id TEXT NOT NULL,document TEXT,storage TEXT NOT NULL CHECK(storage IN ('inline','file')),digest TEXT NOT NULL,fingerprint TEXT NOT NULL,PRIMARY KEY(namespace,id)) STRICT; + CREATE TABLE state_history(namespace TEXT NOT NULL,id TEXT NOT NULL,path TEXT NOT NULL,position INTEGER NOT NULL,document TEXT,storage TEXT NOT NULL CHECK(storage IN ('inline','file')),digest TEXT NOT NULL,PRIMARY KEY(namespace,id,path,position),FOREIGN KEY(namespace,id) REFERENCES state_records(namespace,id) ON DELETE CASCADE) STRICT; + CREATE TABLE state_changes(sequence INTEGER PRIMARY KEY AUTOINCREMENT,revision INTEGER NOT NULL,namespace TEXT NOT NULL,id TEXT NOT NULL,operation TEXT NOT NULL CHECK(operation IN ('upsert','delete')),UNIQUE(revision,namespace,id)) STRICT; + PRAGMA application_id=${APPLICATION_ID}; PRAGMA user_version=${SCHEMA};`) + expectedRevision = 0 + setMeta('migration', JSON.stringify(migration)) + write(initial, { initializing: true }) + db.exec('COMMIT') + } else expectedRevision = Number(meta('revision')) + if (!Number.isSafeInteger(expectedRevision) || expectedRevision < 1) throw fail('SQLite state revision is malformed') + } catch (error) { rollback(); db.close(); throw error } + function read() { + if (closed) throw fail('SQLite state is closed') + db.exec('BEGIN') + try { + const revision = Number(meta('revision')) + if (!Number.isSafeInteger(revision) || revision < 1) throw fail('SQLite state revision is malformed') + const roots = JSON.parse(meta('roots')), state = {}, values = new Map() + for (const [key, kind] of Object.entries(roots)) { + if (!['map', 'value'].includes(kind)) throw fail('SQLite state root descriptor changed') + if (kind === 'map') put(state, key, {}) + } + for (const row of db.prepare('SELECT * FROM state_records ORDER BY namespace,id').all()) { + const value = readPayload(row) + values.set(JSON.stringify([row.namespace, row.id]), value) + if (row.namespace === '$root') { + if (roots[row.id] !== 'value') throw fail('SQLite value root scope changed') + put(state, row.id, value) + } else { + if (!Object.hasOwn(state, row.namespace) || roots[row.namespace] !== 'map') throw fail('SQLite record root scope changed') + put(state[row.namespace], row.id, value) + } + } + for (const row of db.prepare('SELECT * FROM state_history ORDER BY namespace,id,path,position').all()) { + let target = values.get(JSON.stringify([row.namespace, row.id])) + const path = JSON.parse(row.path) + if (!Array.isArray(path) || !path.length) throw fail('SQLite history path changed') + for (const key of path) { + if (target === null || typeof target !== 'object' || !Object.hasOwn(target, key)) throw fail('SQLite history owner changed') + target = target[key] + } + if (!Array.isArray(target) || target.length !== row.position) throw fail('SQLite history sequence is malformed') + target.push(readPayload(row)) + } + for (const row of db.prepare('SELECT namespace,id,fingerprint FROM state_records').all()) if (hash(JSON.stringify(values.get(JSON.stringify([row.namespace, row.id])))) !== row.fingerprint) throw fail('SQLite record/history fingerprint changed') + db.exec('COMMIT'); expectedRevision = revision + return state + } catch (error) { rollback(); throw error } + } + return { read, write, stamp: () => String(meta('revision')), + metadata: () => ({ schema: SCHEMA, revision: Number(meta('revision')), migration: JSON.parse(meta('migration')), journalMode: scalar('PRAGMA journal_mode'), synchronous: Number(scalar('PRAGMA synchronous')) }), + changes({ after = 0, limit = 50 } = {}) { + if (!Number.isSafeInteger(after) || after < 0 || !Number.isSafeInteger(limit) || limit < 1 || limit > 100) throw fail('Invalid SQLite change cursor or limit') + const items = db.prepare('SELECT sequence,revision,namespace,id,operation FROM state_changes WHERE sequence>? ORDER BY sequence LIMIT ?').all(after, limit) + return { items: items.map(row => ({ ...row })), next: items.at(-1)?.sequence ?? after } + }, + close() { if (!closed) { closed = true; db.close() } }, + } +} diff --git a/patches/ds-bot/files/src/host/project-state-storage.js b/patches/ds-bot/files/src/host/project-state-storage.js new file mode 100644 index 000000000..fd9b43690 --- /dev/null +++ b/patches/ds-bot/files/src/host/project-state-storage.js @@ -0,0 +1,133 @@ +// Migration bridge for the original Store. state.json becomes a newer-version +// guard, not a second writable mirror; its validated import is retained separately. +import { createHash, randomUUID } from 'node:crypto' +import { constants, closeSync, existsSync, fsyncSync, lstatSync, linkSync, mkdirSync, openSync, readFileSync, renameSync, unlinkSync, writeFileSync } from 'node:fs' +import { dirname, join } from 'node:path' +import { openSqliteState, requireSqliteRuntime } from './project-state-sqlite.js' + +const digest = value => createHash('sha256').update(value).digest('hex') +const fail = message => Object.assign(new Error(message), { code: 'bot/state-storage' }) +export function writeDurableStateFile(path, text, { exclusive = false } = {}) { + mkdirSync(dirname(path), { recursive: true, mode: 0o700 }) + const temp = `${path}.${randomUUID()}.tmp`, fd = openSync(temp, 'wx', 0o600) + try { writeFileSync(fd, text, 'utf8'); fsyncSync(fd) } finally { closeSync(fd) } + try { + if (exclusive) { + try { linkSync(temp, path) } + catch (error) { + if (error.code !== 'EEXIST') throw error + const info = lstatSync(path) + if (!info.isFile() || info.isSymbolicLink() || readFileSync(path, 'utf8') !== text) throw fail('Existing migration backup content changed') + } + } else renameSync(temp, path) + } finally { if (existsSync(temp)) unlinkSync(temp) } + const directory = openSync(dirname(path), 'r') + try { fsyncSync(directory) } finally { closeSync(directory) } +} + +export function createProjectStateStorage({ home, statePath = join(home, 'state.json'), mode, validate, ensureWriter }) { + if (mode !== undefined && !['json', 'sqlite'].includes(mode)) throw fail('stateStorage must be json or sqlite') + const path = join(home, 'state.sqlite') + let connection, connectionReadOnly, guard, selected = mode === 'sqlite', knownRevision = null + function checkRollback() { + const marker = join(home, 'state-storage-rollback.json') + if (existsSync(marker) && JSON.parse(readFileSync(marker, 'utf8')).phase === 'prepared') throw fail('Storage rollback is pending; resume the offline migration tool before starting FlyBot') + } + function readGuard() { + if (!existsSync(statePath)) return null + const info = lstatSync(statePath) + if ((selected || existsSync(path)) && (!info.isFile() || info.isSymbolicLink())) throw fail('State migration source must be a regular file') + let parsed + try { parsed = JSON.parse(readFileSync(statePath, 'utf8')) } + catch (error) { if (selected || existsSync(path)) throw fail(`Cannot migrate malformed JSON state: ${error.message}`); return null } + if (parsed?.storage !== 'sqlite') return null + if (!info.isFile() || info.isSymbolicLink()) throw fail('State migration source must be a regular file') + const proof = parsed.migration + if (!['prepared', 'active'].includes(parsed.phase)) throw fail('SQLite migration phase is malformed') + if (parsed.version !== 3 || !proof || typeof proof.id !== 'string' || !/^[a-f0-9]{64}$/.test(proof.sha256 ?? '') || proof.backup !== `state.sqlite-import-${proof.sha256}.json`) throw fail('SQLite migration guard is malformed') + return parsed + } + function importState(proof) { + const file = join(home, proof.backup), info = lstatSync(file) + if (!info.isFile() || info.isSymbolicLink() || info.size > 128 * 1024 * 1024) throw fail('SQLite import backup is not a bounded regular file') + const fd = openSync(file, constants.O_RDONLY | (constants.O_NOFOLLOW ?? 0)) + let text + try { text = readFileSync(fd, 'utf8') } finally { closeSync(fd) } + if (digest(text) !== proof.sha256) throw fail('SQLite import backup hash changed') + const state = JSON.parse(text) + validate(state) + return state + } + function open(writer, initial) { + if (connection && connectionReadOnly === !writer) return connection + connection?.close(); connection = undefined + const candidate = openSqliteState({ path, readOnly: !writer, ...(initial ? { initial } : {}), migration: guard.migration }) + const migration = candidate.metadata().migration + if (migration?.id !== guard.migration.id || migration?.sha256 !== guard.migration.sha256) { + candidate.close(); throw fail('SQLite database and migration guard identify different authorities') + } + connection = candidate; connectionReadOnly = !writer + return candidate + } + function activate() { + if (guard.phase === 'active') return + ensureWriter() + guard = { ...guard, phase: 'active' } + writeDurableStateFile(statePath, JSON.stringify(guard, null, 2)) + } + function assertGuard() { + const current = readGuard() + if (!current || current.phase !== 'active' || current.migration.id !== guard?.migration.id || current.migration.sha256 !== guard.migration.sha256) throw fail('SQLite storage authority changed; reload before writing') + } + return { + get sqlite() { return selected }, + async read({ writer, readLegacy }) { + checkRollback() + const current = readGuard() + if (!current && !existsSync(path)) return { state: await readLegacy(), sqlite: false } + selected = true + if (mode === 'json') throw fail('SQLite is the current authority; export and roll back explicitly before selecting JSON') + if (!current) throw fail('SQLite state is missing its migration guard; refusing a second JSON authority') + guard = current + const initial = importState(guard.migration) + if (!existsSync(path) && guard.phase === 'active') throw fail('The authoritative SQLite database is missing; do not import an outdated backup automatically') + if (!existsSync(path) && !writer) return { state: initial, sqlite: true } + if (writer) ensureWriter() + const db = open(writer, writer ? initial : undefined), state = db.read() + validate(state) + knownRevision = db.stamp() + if (writer) activate() + return { state, sqlite: true } + }, + initialize(state) { + if (!selected || connection) return false + ensureWriter(); validate(state); requireSqliteRuntime() + guard = readGuard() + if (!guard) { + const text = JSON.stringify(state), sha256 = digest(text), backup = `state.sqlite-import-${sha256}.json` + writeDurableStateFile(join(home, backup), text, { exclusive: true }) + guard = { version: 3, storage: 'sqlite', phase: 'prepared', migration: { id: randomUUID(), backup, sha256, importedAt: Date.now() } } + // Guard first: a crash before SQL initialization remains recoverable from + // the verified backup and an older JSON-only binary refuses to write. + writeDurableStateFile(statePath, JSON.stringify(guard, null, 2)) + } + const db = open(true, importState(guard.migration)) + knownRevision = db.stamp() + activate() + return true + }, + write(snapshot) { + if (!selected || !guard) throw fail('SQLite storage has not been initialized') + ensureWriter(); checkRollback(); assertGuard() + const db = open(true) + if (knownRevision !== db.stamp()) throw fail('SQLite state revision conflict; refresh before taking over') + const result = db.write(typeof snapshot === 'string' ? JSON.parse(snapshot) : snapshot) + knownRevision = String(result.revision) + return result + }, + stamp() { return connection ? `sqlite:${guard.migration.id}:${connection.stamp()}` : null }, + status() { return { backend: selected ? 'sqlite' : 'json', ready: Boolean(connection), ...(connection ? connection.metadata() : {}) } }, + changes(options) { if (!connection) throw fail('SQLite storage is not initialized'); return connection.changes(options) }, + close() { connection?.close(); connection = undefined }, + } +} diff --git a/patches/ds-bot/files/test/project-state-runtime.test.js b/patches/ds-bot/files/test/project-state-runtime.test.js new file mode 100644 index 000000000..27bde6ef3 --- /dev/null +++ b/patches/ds-bot/files/test/project-state-runtime.test.js @@ -0,0 +1,46 @@ +import assert from 'node:assert/strict' +import { mkdtempSync, writeFileSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { createRuntime } from '../index.js' +import { openSqliteState } from '../src/host/project-state-sqlite.js' +import { fakeContext } from './fake-ctx.js' + +test('original Project APIs persist Task/intake intent in SQL without model calls or positioning changes', async t => { + const home = mkdtempSync(join(tmpdir(), 'flybot-sql-api-')), contexts = [] + writeFileSync(join(home, 'state.json'), JSON.stringify({ version: 2, revision: 1, members: { 'owner-1': { id: 'owner-1', designatedChiefId: 'chief' } }, mainBotIds: ['chief'], bots: { chief: { id: 'chief', sessionId: 'chief', name: 'Chief', ownerId: 'owner-1', projectId: 'p' } }, projects: { p: { id: 'p', name: 'Existing Project', status: 'active', ownerUserId: 'owner-1', goal: 'Retain the agreed product goal', scope: 'Original Project scope', roles: { members: [], maintainers: [] } } }, tasks: {} })) + let modelCalls = 0 + async function start(stateStorage) { + const ctx = fakeContext() + Object.assign(ctx, { llm: { listProviders: () => [], listModels: async () => [], generate: async () => { ++modelCalls; throw Error('No inference expected') } }, workspaceRegistry: { get: () => undefined, archiveSession: async () => {} }, sessionController: { resolveAgent: async () => ({ agent: undefined }) }, sessions: { flush: async () => {} } }) + contexts.push(ctx) + const rt = createRuntime(ctx, { home, projectWorkbench: true, ...(stateStorage ? { stateStorage } : {}) }) + await rt.load() + assert.equal(rt.readOnly(), undefined) + return { ctx, rt } + } + t.after(async () => { for (const ctx of contexts.reverse()) await ctx.dispose(); rmSync(home, { recursive: true, force: true, maxRetries: 5 }) }) + const first = await start('sqlite'), actor = { memberId: 'owner-1' } + const task = await first.rt.taskEndpoint('task-create', { botId: 'chief', projectId: 'p', title: 'Improve backend query', criteria: 'Query regression passes', impact: 'Improve Project observability' }, actor) + assert.equal(task.projectTask.admission.status, 'proposed') + await first.rt.taskEndpoint('task-update', { id: task.id, note: 'Keep the agreed scope and required evidence' }, actor) + const body = { version: 'project-development/v1', kind: 'query', name: 'tasks', projectId: 'p', payload: { limit: 10 } } + const view = await first.rt.projectEndpoint('project-development', body, actor) + assert.equal(view.result.tasks[0].taskId, task.id) + assert.equal(view.result.tasks[0].waitingFor, 'admission') + const database = openSqliteState({ path: join(home, 'state.sqlite'), readOnly: true }), disk = database.read(); database.close() + assert.equal(disk.tasks[task.id].note, 'Keep the agreed scope and required evidence') + const intake = Object.values(disk.projectEffects).find(row => row.kind === 'task.intake' && row.taskId === task.id) + assert.ok(intake) + assert.equal(first.rt.projectEffects.intentCommitted({ id: intake.id, generation: intake.generation }), true) + assert.equal(disk.projects.p.goal, 'Retain the agreed product goal') + assert.equal(disk.projects.p.scope, 'Original Project scope') + assert.equal(modelCalls, 0) + await first.ctx.dispose() + const second = await start() + assert.equal(second.rt.stateStorage.status().backend, 'sqlite') + assert.equal((await second.rt.projectEndpoint('project-development', body, actor)).result.tasks[0].taskId, task.id) + assert.equal(second.rt.state.tasks[task.id].projectTask.admission.status, 'proposed') + assert.equal(modelCalls, 0) +}) diff --git a/patches/ds-bot/files/test/project-state-sqlite.test.js b/patches/ds-bot/files/test/project-state-sqlite.test.js new file mode 100644 index 000000000..54cb2cffa --- /dev/null +++ b/patches/ds-bot/files/test/project-state-sqlite.test.js @@ -0,0 +1,130 @@ +import assert from 'node:assert/strict' +import { mkdtempSync, rmSync, readFileSync, writeFileSync, statSync, symlinkSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { DatabaseSync } from 'node:sqlite' +import { openSqliteState } from '../src/host/project-state-sqlite.js' + +function fixture(t) { + const home = mkdtempSync(join(tmpdir(), 'flybot-sql-state-')) + const options = { path: join(home, 'state.sqlite'), payloadRoot: join(home, 'state-payloads') } + const connections = [] + t.after(() => { for (const db of connections.reverse()) db.close(); rmSync(home, { recursive: true, force: true }) }) + const open = more => { const db = openSqliteState({ ...options, ...more }); connections.push(db); return db } + return { home, options, open } +} +const initial = () => ({ version: 2, revision: 1, members: { owner: { id: 'owner' } }, tasks: { task: { id: 'task', status: 'working', collaboration: { history: [{ id: 'offer-1' }] } } }, projectEffects: { effect: { id: 'effect', status: 'pending' } } }) + +test('SQLite roundtrips identities, normalizes history and commits only changed rows', t => { + const f = fixture(t), state = initial(), db = f.open({ initial: state }) + assert.deepEqual(db.read(), state) + assert.equal(db.metadata().journalMode, 'wal') + assert.equal(db.metadata().synchronous, 2) + const reader = new DatabaseSync(f.options.path, { readOnly: true }) + t.after(() => reader.close()) + assert.equal(reader.prepare('SELECT count(*) AS n FROM state_history').get().n, 1) + assert.equal(JSON.parse(reader.prepare("SELECT document FROM state_records WHERE namespace='tasks' AND id='task'").get().document).collaboration.history.length, 0) + assert.equal(db.write(state).changed, 0) + state.tasks.task.collaboration.history.push({ id: 'offer-2' }) + state.projectEffects.effect.status = 'done' + const before = db.changes({ limit: 100 }).next + const result = db.write(state) + assert.equal(result.changed, 2) + assert.deepEqual(db.read(), state) + assert.deepEqual(db.changes({ after: before }).items.map(({ namespace, id, operation }) => ({ namespace, id, operation })), [{ namespace: 'projectEffects', id: 'effect', operation: 'upsert' }, { namespace: 'tasks', id: 'task', operation: 'upsert' }]) +}) + +test('an interrupted SQL transaction commits neither Task nor Effect and can retry', t => { + const f = fixture(t), state = initial(), db = f.open({ initial: state }) + const control = new DatabaseSync(f.options.path) + t.after(() => control.close()) + control.exec("CREATE TRIGGER fail_effect BEFORE UPDATE ON state_records WHEN NEW.namespace='projectEffects' BEGIN SELECT RAISE(ABORT,'fixture transaction failure'); END") + state.tasks.task.status = 'accepted'; state.projectEffects.effect.status = 'done' + assert.throws(() => db.write(state), /fixture transaction failure/) + assert.deepEqual(f.open({ readOnly: true }).read(), initial()) + control.exec('DROP TRIGGER fail_effect') + db.write(state) + assert.deepEqual(f.open({ readOnly: true }).read(), state) +}) + +test('stale database revisions cannot overwrite a different writer; readonly connections never commit', t => { + const f = fixture(t), state = initial(), first = f.open({ initial: state }), stale = f.open({}) + stale.read() + state.tasks.task.status = 'blocked'; first.write(state) + assert.throws(() => stale.write(initial()), /revision conflict/) + const observer = f.open({ readOnly: true }) + assert.equal(observer.read().tasks.task.status, 'blocked') + assert.throws(() => observer.write(initial()), /read.only/i) + assert.equal(observer.stamp(), first.stamp()) +}) + +test('large records use immutable hashed files and fail closed on content damage', t => { + const f = fixture(t), state = initial() + state.tasks.large = { id: 'large', payload: 'bounded data '.repeat(10000) } + const db = f.open({ initial: state }), inspector = new DatabaseSync(f.options.path, { readOnly: true }) + t.after(() => inspector.close()) + const row = inspector.prepare("SELECT storage,document,digest FROM state_records WHERE namespace='tasks' AND id='large'").get() + assert.equal(row.storage, 'file'); assert.equal(row.document, null) + assert.deepEqual(JSON.parse(readFileSync(join(f.options.payloadRoot, row.digest + '.json'), 'utf8')), state.tasks.large) + assert.deepEqual(db.read(), state) + writeFileSync(join(f.options.payloadRoot, row.digest + '.json'), 'corrupt payload') + assert.throws(() => db.read(), /hash changed/) +}) + +test('future or foreign SQLite schemas refuse to initialize and preserve existing tables', t => { + const f = fixture(t), raw = new DatabaseSync(f.options.path) + raw.exec('CREATE TABLE unrelated(value TEXT); INSERT INTO unrelated VALUES (\'preserve\'); PRAGMA user_version=999') + raw.close() + assert.throws(() => f.open({ initial: initial() }), /schema|version/i) + const check = new DatabaseSync(f.options.path, { readOnly: true }) + assert.equal(check.prepare('SELECT value FROM unrelated').get().value, 'preserve'); check.close() +}) + +test('actual process death before and after COMMIT preserves the correct atomic Task/Effect boundary', async t => { + const { spawn } = await import('node:child_process'), { once } = await import('node:events') + for (const committed of [false, true]) { + const f = fixture(t), seed = f.open({ initial: initial() }) + seed.close() + const source = `import { DatabaseSync } from 'node:sqlite'; + import { writeSync } from 'node:fs'; + import { openSqliteState } from ${JSON.stringify(new URL('../src/host/project-state-sqlite.js', import.meta.url).href)}; + const options=JSON.parse(process.argv[1]); + if(${committed}){const db=openSqliteState(options);const state=db.read();state.tasks.task.status='accepted';state.projectEffects.effect.status='done';db.write(state)} + else{const db=new DatabaseSync(options.path);db.exec("BEGIN IMMEDIATE;UPDATE state_records SET document='{}' WHERE namespace IN ('tasks','projectEffects')")} + writeSync(1,'boundary-ready\\n');setInterval(()=>{},1000);` + const child = spawn(process.execPath, ['--input-type=module', '-e', source, JSON.stringify(f.options)], { stdio: ['ignore', 'pipe', 'pipe'] }) + t.after(() => { if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL') }) + const closed = once(child, 'close') + let output = '' + while (!output.includes('boundary-ready')) { + const [chunk] = await once(child.stdout, 'data', { signal: AbortSignal.timeout(10000) }) + output += chunk.toString() + } + child.kill('SIGKILL') + const [, signal] = await closed + assert.equal(signal, 'SIGKILL') + const state = f.open({ readOnly: true }).read() + assert.equal(state.tasks.task.status, committed ? 'accepted' : 'working') + assert.equal(state.projectEffects.effect.status, committed ? 'done' : 'pending') + } +}) + + +test('database symlinks, prototype-shaped keys and bounded change cursors respect storage boundaries', t => { + const f = fixture(t), state = initial() + Object.defineProperty(state.tasks, '__proto__', { value: { id: '__proto__', history: [{ id: 'owned-history' }] }, enumerable: true }) + const db = f.open({ initial: state }), copy = db.read() + assert.equal(Object.hasOwn(copy.tasks, '__proto__'), true) + assert.equal(copy.tasks.__proto__.history[0].id, 'owned-history') + assert.equal(Object.getPrototypeOf(copy.tasks), Object.prototype) + assert.equal(statSync(f.options.path).mode & 0o777, 0o600) + for (const options of [{ limit: 0 }, { limit: 101 }, { after: -1 }, { after: '0' }]) assert.throws(() => db.changes(options)) + const first = db.changes({ limit: 1 }), second = db.changes({ after: first.next, limit: 1 }) + assert.equal(first.items.length, 1); assert.equal(second.items.length, 1) + assert.ok(second.next > first.next) + db.close() + const linked = join(f.home, 'linked.sqlite') + symlinkSync(f.options.path, linked) + assert.throws(() => openSqliteState({ path: linked }), /regular file/) +}) diff --git a/patches/ds-bot/files/test/project-state-storage.test.js b/patches/ds-bot/files/test/project-state-storage.test.js new file mode 100644 index 000000000..3444bbfc3 --- /dev/null +++ b/patches/ds-bot/files/test/project-state-storage.test.js @@ -0,0 +1,201 @@ +import assert from 'node:assert/strict' +import { createHash } from 'node:crypto' +import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs' +import { hostname, tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { DatabaseSync } from 'node:sqlite' +import { install } from '../src/host/store.js' +import { createProjectEffects } from '../src/host/project-effects.js' + +function fixture(t, saved) { + const home = mkdtempSync(join(tmpdir(), 'flybot-sql-migration-')), statePath = join(home, 'state.json'), runtimes = [] + if (saved !== undefined) writeFileSync(statePath, typeof saved === 'string' ? saved : JSON.stringify(saved)) + async function start(mode) { + const disposers = [], warnings = [], rt = { home, statePath, config: { ...(mode ? { stateStorage: mode } : {}) }, warn: (...args) => warnings.push(args), autoRoomName: () => 'Group', retitleRoom: async () => {}, + ctx: { effect(fn) { disposers.push(fn()) } }, + } + install(rt) + rt.projectEffects = createProjectEffects(rt) + let closed = false + const close = async () => { if (!closed) { closed = true; await rt.projectEffects.close(); for (const fn of disposers.reverse()) await fn?.() } } + runtimes.push(close) + await rt.load() + return { rt, close, warnings } + } + t.after(async () => { for (const close of runtimes.reverse()) await close(); rmSync(home, { recursive: true, force: true }) }) + return { home, statePath, start, path: join(home, 'state.sqlite') } +} +const saved = () => ({ version: 2, revision: 7, members: { owner: { id: 'owner', designatedChiefId: 'chief' } }, mainBotIds: ['chief'], bots: { chief: { id: 'chief', ownerId: 'owner', sessionId: 'chief-session', name: 'Chief' } }, projects: { p: { id: 'p', ownerUserId: 'owner', status: 'active', roles: { members: [], maintainers: [] } } }, tasks: { task: { id: 'task', projectId: 'p', title: 'Retain current work', status: 'working', collaboration: { history: [{ id: 'offer', status: 'accepted' }], claim: { id: 'claim', generation: 7 } } } }, projectTaskSessions: { role: { logicalSessionId: 'role', bindings: [{ backendSessionId: 'sdk-thread' }] } }, projectEffects: {} }) + +test('Store migration preserves working identities, fences JSON writes and automatically reopens SQL authority', async t => { + const f = fixture(t, saved()), first = await f.start('sqlite') + assert.equal(first.rt.readOnly(), undefined) + assert.equal(first.rt.stateStorage.status().backend, 'sqlite') + const guardText = readFileSync(f.statePath, 'utf8'), guard = JSON.parse(guardText) + assert.equal(guard.version, 3); assert.equal(guard.phase, 'active') + assert.equal(createHash('sha256').update(readFileSync(join(f.home, guard.migration.backup))).digest('hex'), guard.migration.sha256) + first.rt.state.tasks.task.status = 'blocked' + await first.rt.save() + assert.equal(readFileSync(f.statePath, 'utf8'), guardText) + await first.close() + const second = await f.start() + assert.equal(second.rt.stateStorage.status().backend, 'sqlite') + assert.equal(second.rt.state.tasks.task.status, 'blocked') + assert.equal(second.rt.state.tasks.task.collaboration.claim.id, 'claim') + assert.equal(second.rt.state.tasks.task.collaboration.history[0].id, 'offer') + assert.equal(second.rt.state.bots.chief.sessionId, 'chief-session') + assert.equal(second.rt.state.projectTaskSessions.role.bindings[0].backendSessionId, 'sdk-thread') + await second.close() + const forbidden = await f.start('json') + assert.match(forbidden.rt.readOnly(), /SQLite.*authority|roll back/) + assert.equal(readFileSync(f.statePath, 'utf8'), guardText) +}) + +test('malformed or invalid-identity JSON cannot migrate into an empty SQL replacement', async t => { + for (const source of ['{broken', { ...saved(), bots: { chief: { id: 'chief', ownerId: 'unknown' } } }]) { + const f = fixture(t, source), original = readFileSync(f.statePath, 'utf8'), current = await f.start('sqlite') + assert.ok(current.rt.readOnly()) + assert.equal(existsSync(f.path), false) + assert.equal(readFileSync(f.statePath, 'utf8'), original) + } +}) + +test('prepared migration recovers exact backup; loss of an active database never imports stale state', async t => { + const f = fixture(t, saved()), first = await f.start('sqlite') + await first.close() + const guard = JSON.parse(readFileSync(f.statePath, 'utf8')) + for (const suffix of ['', '-wal', '-shm']) rmSync(f.path + suffix, { force: true }) + const missing = await f.start('sqlite') + assert.match(missing.rt.readOnly(), /authoritative SQLite database is missing/) + assert.equal(existsSync(f.path), false) + await missing.close() + writeFileSync(f.statePath, JSON.stringify({ ...guard, phase: 'prepared' })) + const recovered = await f.start() + assert.equal(recovered.rt.readOnly(), undefined) + assert.equal(recovered.rt.state.tasks.task.collaboration.claim.id, 'claim') + assert.equal(JSON.parse(readFileSync(f.statePath, 'utf8')).phase, 'active') +}) + +test('SQL state and Effect commit together, recover after restart and acknowledge only a committed transaction', async t => { + const f = fixture(t, saved()), first = await f.start('sqlite'), { rt } = first + let executions = 0 + rt.projectEffects.register('fixture.execute', { run: async () => ({ count: ++executions }) }) + const spec = { kind: 'fixture.execute', key: 'once', projectId: 'p', taskId: 'task', payload: { claimId: 'claim' } } + const token = rt.projectEffects.stage(spec) + rt.state.tasks.task.status = 'queued' + const db = new DatabaseSync(f.path) + db.exec("CREATE TRIGGER fail_effect BEFORE INSERT ON state_records WHEN NEW.namespace='projectEffects' BEGIN SELECT RAISE(ABORT,'fixture effect write failure'); END") + await assert.rejects(rt.projectEffects.persist(token), /fixture effect write failure/) + await rt.projectEffects.drain() + assert.equal(executions, 0); assert.equal(rt.projectEffects.intentCommitted(token), false) + assert.equal(JSON.parse(db.prepare("SELECT document FROM state_records WHERE namespace='tasks' AND id='task'").get().document).status, 'working') + db.exec('DROP TRIGGER fail_effect'); db.close() + await rt.projectEffects.persist(token) + assert.equal(rt.projectEffects.intentCommitted(token), true) + await first.close() + const second = await f.start() + second.rt.projectEffects.register('fixture.execute', { run: async () => ({ count: ++executions }) }) + second.rt.projectEffects.recover() + await second.rt.projectEffects.drain() + await second.rt.projectEffects.drain() + assert.equal(executions, 1) + assert.equal(second.rt.state.projectEffects[token.id].status, 'done') + assert.equal(second.rt.state.tasks.task.status, 'queued') +}) + +test('readonly Store does not acknowledge Effects and refresh adopts current SQL state before takeover', async t => { + const f = fixture(t, saved()), first = await f.start('sqlite') + await first.close() + writeFileSync(join(f.home, 'state.lock'), JSON.stringify({ pid: process.ppid, host: hostname(), token: 'external-holder' })) + const second = await f.start() + assert.match(second.rt.readOnly(), /another dsh/) + second.rt.projectEffects.register('fixture.readonly', { run: async () => { throw new Error('Must not execute') } }) + // stage itself is denied on read-only hosts; a no-op save cannot grant durability. + assert.throws(() => second.rt.projectEffects.stage({ kind: 'fixture.readonly', key: 'denied', projectId: 'p', payload: {} }), /writable/) + const before = second.rt.state.revision + await second.rt.save() + assert.equal(second.rt.state.revision, before) + rmSync(join(f.home, 'state.lock')) + await second.rt.refresh() + assert.equal(second.rt.readOnly(), undefined) + second.rt.state.tasks.task.status = 'blocked' + await second.rt.save() + assert.equal(second.rt.stateStorage.status().backend, 'sqlite') +}) + +test('offline export and rollback retain latest facts and reopen JSON without two writable authorities', async t => { + const { exportSqliteState, rollbackSqliteState } = await import('../src/host/project-state-migration.js') + const f = fixture(t, saved()), first = await f.start('sqlite') + first.rt.state.tasks.task.status = 'blocked' + await first.rt.save() + const guard = JSON.parse(readFileSync(f.statePath, 'utf8')), output = join(f.home, 'private-export.json') + assert.throws(() => exportSqliteState({ home: f.home, output, expectedMigrationId: guard.migration.id }), /Stop the current/) + await first.close() + const exported = exportSqliteState({ home: f.home, output, expectedMigrationId: guard.migration.id }) + assert.equal(JSON.parse(readFileSync(output, 'utf8')).tasks.task.status, 'blocked') + assert.equal(exported.sha256, createHash('sha256').update(readFileSync(output)).digest('hex')) + assert.throws(() => rollbackSqliteState({ home: f.home, expectedMigrationId: 'outdated' }), /ID changed/) + const result = rollbackSqliteState({ home: f.home, expectedMigrationId: guard.migration.id }) + assert.equal(result.backend, 'json'); assert.equal(existsSync(f.path), false) + assert.equal(existsSync(join(f.home, result.archiveFile)), true) + assert.equal(rollbackSqliteState({ home: f.home, expectedMigrationId: guard.migration.id, resume: true }).already, true) + const second = await f.start() + assert.equal(second.rt.readOnly(), undefined) + assert.equal(second.rt.stateStorage.status().backend, 'json') + assert.equal(second.rt.state.tasks.task.status, 'blocked') + assert.equal(second.rt.state.tasks.task.collaboration.claim.id, 'claim') +}) + +test('prepared rollback blocks runtime and resumes after its database move without reviving import state', async t => { + const { renameSync } = await import('node:fs'), { randomUUID } = await import('node:crypto') + const { rollbackSqliteState } = await import('../src/host/project-state-migration.js') + const { openSqliteState } = await import('../src/host/project-state-sqlite.js') + const f = fixture(t, saved()), first = await f.start('sqlite') + first.rt.state.tasks.task.status = 'blocked'; await first.rt.save(); await first.close() + const guard = JSON.parse(readFileSync(f.statePath, 'utf8')), db = openSqliteState({ path: f.path, readOnly: true }) + const text = JSON.stringify(db.read(), null, 2), revision = db.metadata().revision, id = randomUUID(); db.close() + const marker = { version: 'flybot-storage-rollback/v1', phase: 'prepared', id, migrationId: guard.migration.id, revision, exportFile: `state.rollback-${id}.json`, archiveFile: `state.sqlite.rollback-${id}`, sha256: createHash('sha256').update(text).digest('hex') } + writeFileSync(join(f.home, marker.exportFile), text) + writeFileSync(join(f.home, 'state-storage-rollback.json'), JSON.stringify(marker)) + renameSync(f.path, join(f.home, marker.archiveFile)) + const blocked = await f.start() + assert.match(blocked.rt.readOnly(), /rollback is pending/) + await blocked.close() + assert.throws(() => rollbackSqliteState({ home: f.home, expectedMigrationId: guard.migration.id }), /explicitly resume/) + rollbackSqliteState({ home: f.home, expectedMigrationId: guard.migration.id, resume: true }) + const recovered = await f.start() + assert.equal(recovered.rt.readOnly(), undefined) + assert.equal(recovered.rt.stateStorage.status().backend, 'json') + assert.equal(recovered.rt.state.tasks.task.status, 'blocked') +}) + +test('migration CLI exposes only metadata and requires explicit paths and source identity', async t => { + const { execFileSync } = await import('node:child_process') + const f = fixture(t, saved()), first = await f.start('sqlite'); await first.close() + const script = new URL('../scripts/state-storage.mjs', import.meta.url).pathname + const inspect = JSON.parse(execFileSync(process.execPath, [script, 'inspect', '--home', f.home], { encoding: 'utf8' })) + assert.equal(inspect.backend, 'sqlite'); assert.equal(inspect.phase, 'active') + assert.equal(JSON.stringify(inspect).includes('chief-session'), false) + const output = join(f.home, 'cli-export.json') + const exported = JSON.parse(execFileSync(process.execPath, [script, 'export', '--home', f.home, '--migration', inspect.migrationId, '--output', output], { encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'] })) + assert.equal(exported.output, output) + assert.equal(JSON.parse(readFileSync(output, 'utf8')).tasks.task.id, 'task') + assert.throws(() => execFileSync(process.execPath, [script, 'export', '--home', f.home, '--output', output], { stdio: 'pipe' })) + assert.throws(() => execFileSync(process.execPath, [script, 'inspect', '--home', f.home, '--unexpected'], { stdio: 'pipe' })) +}) + +test('stale readonly takeover cannot overwrite newer SQL state without refresh', async t => { + const { openSqliteState } = await import('../src/host/project-state-sqlite.js') + const f = fixture(t, saved()), first = await f.start('sqlite'); await first.close() + writeFileSync(join(f.home, 'state.lock'), JSON.stringify({ pid: process.ppid, host: hostname(), token: 'external-holder' })) + const observer = await f.start() + const writer = openSqliteState({ path: f.path }), current = writer.read() + current.tasks.task.status = 'blocked'; writer.write(current); writer.close() + rmSync(join(f.home, 'state.lock')) + observer.rt.state.tasks.task.status = 'accepted' + await assert.rejects(observer.rt.save(), /revision conflict/) + await observer.close() + const recovered = await f.start() + assert.equal(recovered.rt.state.tasks.task.status, 'blocked') +})