From d3146681487fa765f90ba4c008563eaccddcbd33 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 18 Sep 2026 01:31:24 -0700 Subject: [PATCH] fix: keep historical state repairs in Doctor (#151418) * fix: keep historical state repairs in Doctor * fix: narrow added state column names before recording them * fix: preserve Doctor repair ownership and prove retained history upgrades * test: match native SQLite locations in ownership race proof * test: distinguish published updater cron-history repairs --- .../integrity-and-recovery.md | 11 + docs/reference/database-schemas/layout.md | 16 +- .../e2e/lib/upgrade-survivor/diagnostics.mjs | 6 + .../legacy-operator-cron-history.mjs | 375 ++++++++++++++++++ scripts/e2e/lib/upgrade-survivor/run.sh | 18 + scripts/pr-lib/wrapper-components.txt | 1 + ...ubagent-completion-admission.store.test.ts | 8 +- .../subagent-registry.store.sqlite.test.ts | 14 +- src/gateway/worker-environments/store.test.ts | 4 +- src/infra/startup-maintenance-required.ts | 1 + .../state-migrations.cron-run-logs.test.ts | 21 +- src/infra/state-migrations.doctor.ts | 2 +- .../state-migrations.plugin-doctor.test.ts | 82 +++- src/infra/state-migrations.plugin-doctor.ts | 11 +- src/infra/state-migrations.test.ts | 50 --- .../openclaw-database-maintenance.test.ts | 2 +- .../openclaw-state-db-acp-replay.test.ts | 160 +++++++- .../openclaw-state-db-binding-targets.test.ts | 5 +- src/state/openclaw-state-db-fast-path.test.ts | 75 +++- src/state/openclaw-state-db-fast-path.ts | 20 +- ...openclaw-state-db-legacy-backfills.test.ts | 77 +++- src/state/openclaw-state-db-maintenance.ts | 86 ++-- src/state/openclaw-state-db-repair.ts | 37 +- .../openclaw-state-db-schema-additive.ts | 71 +++- ...claw-state-db-schema-migration-required.ts | 1 + src/state/openclaw-state-db-schema-runtime.ts | 161 ++++++++ src/state/openclaw-state-db.test.ts | 52 +-- src/state/openclaw-state-db.ts | 141 +------ src/state/openclaw-state-ownership.test.ts | 27 +- src/state/openclaw-state-worker-error.test.ts | 12 + src/state/openclaw-state-worker-error.ts | 2 + .../upgrade-survivor-cron-history.test.ts | 278 +++++++++++++ 32 files changed, 1467 insertions(+), 360 deletions(-) create mode 100644 scripts/e2e/lib/upgrade-survivor/legacy-operator-cron-history.mjs create mode 100644 src/state/openclaw-state-db-schema-runtime.ts create mode 100644 test/scripts/upgrade-survivor-cron-history.test.ts diff --git a/docs/reference/database-schemas/integrity-and-recovery.md b/docs/reference/database-schemas/integrity-and-recovery.md index 05231fd9a1af..a3ef1c498d4c 100644 --- a/docs/reference/database-schemas/integrity-and-recovery.md +++ b/docs/reference/database-schemas/integrity-and-recovery.md @@ -32,6 +32,17 @@ operations, the daily verifier, or explicit maintenance instead of a full scan on each reopen. Schema, ownership, and current write authority are never borrowed from the integrity result. +Shared-state runtime opens and automatic startup preparation converge supported +schema additions and preserve atomic upgrades from older schema versions. A +newly added supported column receives its required content transformation in the +same transaction. Opens do not rerun historical row backfills for columns already +present when the application version changes. Run +`openclaw doctor --fix` during update maintenance to repair historical accounting +or legacy payload fields. A current-schema database that still contains the +retired `cron_run_logs` table requires Doctor before runtime can open it; Doctor +imports its retained history into task runs atomically before removing the table. +Shared-state integrity, schema, version, and ownership checks remain in place. + Schema compatibility preflight can read agent schema headers without a full integrity scan. For ordinary rollback-mode agent databases and complete WAL families, a read-only child reads the schema version and optional writer build in one fresh SQLite transaction, including committed WAL changes, without copying unrelated database contents. Its source-reader lease stays held through native close; cancellation and timeout wait for child closure. Parent-side diagnostics do not open or close the live agent file, preserving the parent's SQLite locks. As with the previous online-backup reader, native SQLite may update SHM read marks or rebuild existing SHM after a quiescent family reopens; the database and WAL contents remain unchanged. These headers are not cached compatibility or integrity proof: full readiness and writable admission retain their existing validation and fresh authority checks. Private snapshots remain necessary inside owner-held source-exclusion or canonical-mutation scopes, for incomplete WAL families whose inspection would create source sidecars, and for rollback journals requiring private recovery. Those cases use the existing snapshot owner and deadline; ordinary inspection errors do not trigger a full-copy fallback. Shared-state preflight is unchanged. `openclaw database preflight` performs the release-local shape comparison for an explicit copied file. The background verifier also scans already-open databases about once daily. diff --git a/docs/reference/database-schemas/layout.md b/docs/reference/database-schemas/layout.md index 932319638d03..5ecef9633941 100644 --- a/docs/reference/database-schemas/layout.md +++ b/docs/reference/database-schemas/layout.md @@ -85,17 +85,21 @@ persisted text field, plus 32 bytes per row. Session totals include their events This is a retained-content estimate, not a limit on SQLite file, page, or WAL size. Older releases counted characters inconsistently, undercounting Unicode and -allowing unchanged metadata writes to drift. The existing app-version upgrade -repair and explicit shared-state schema repair rebuild all derived totals +allowing unchanged metadata writes to drift. Explicit Doctor shared-state +repair rebuilds all derived totals atomically, preserving event JSON text, identifiers, timestamps, and sequence. Repair does not prune history. The next ordinary session write applies the existing caps and eviction order, so corrected Unicode history may trim sooner and use transcript fallback when loaded. -A current-app-version reopen skips this repair. Replacing code without changing -the app version does not repair an already-open or current-version database; -explicit schema repair remains the repair owner for that case. Accounting repair -cannot recover history already evicted by an older writer. See [ACP CLI](/cli/acp). +Normal runtime opens and automatic startup schema preparation leave existing +accounting columns unchanged, including after the application version changes. If +the supported older shape lacks accounting columns, adding them also initializes +their totals in the same transaction. Run +`openclaw doctor --fix` during update maintenance to repair historical accounting. +Supported older-schema upgrades still perform the content transformations needed +to preserve data while changing its schema. Accounting repair cannot recover +history already evicted by an older writer. See [ACP CLI](/cli/acp). ### Meeting transcript tables diff --git a/scripts/e2e/lib/upgrade-survivor/diagnostics.mjs b/scripts/e2e/lib/upgrade-survivor/diagnostics.mjs index abd8e63acb67..f11b1a5c5b58 100644 --- a/scripts/e2e/lib/upgrade-survivor/diagnostics.mjs +++ b/scripts/e2e/lib/upgrade-survivor/diagnostics.mjs @@ -39,6 +39,7 @@ const logNames = [ "workshop-baseline-doctor.json", "workshop-recovered-upgrade.json", "workshop-candidate-doctor.json", + "legacy-operator-cron-history-proof.json", "gateway.log", "gateway.log.doctor", "baseline-service-install.err", @@ -1610,6 +1611,11 @@ function publishedSuccessSummary(artifactRoot, sanitize) { ...(snapshot.scenario === "workshop-doctor-recovery" ? ["workshop-doctor-recovery.json", "baseline-doctor.log", "doctor.log"] : []), + ...(snapshot.scenario === "legacy-operator-state" && + snapshot.updateRestartMode === "manual" && + ["2026.9.3", "2026.9.4"].includes(snapshot.baseline.version) + ? ["legacy-operator-cron-history-proof.json"] + : []), ].map((name) => [name, sanitize(readOwned(artifactRoot, name, name), name)]), ), omissions, diff --git a/scripts/e2e/lib/upgrade-survivor/legacy-operator-cron-history.mjs b/scripts/e2e/lib/upgrade-survivor/legacy-operator-cron-history.mjs new file mode 100644 index 000000000000..5248b3ae6d1c --- /dev/null +++ b/scripts/e2e/lib/upgrade-survivor/legacy-operator-cron-history.mjs @@ -0,0 +1,375 @@ +import assert from "node:assert/strict"; +import { execFileSync } from "node:child_process"; +import { createHash } from "node:crypto"; +import fs from "node:fs"; +import path from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import { pathToFileURL } from "node:url"; +import { isMainThread } from "node:worker_threads"; + +const MIGRATION = "state:cron-run-logs-to-task-runs:v1"; +const FIXTURE_NAME = "legacy-operator-cron-history.json"; +const readJson = (file) => JSON.parse(fs.readFileSync(file, "utf8")); +const hash = (bytes) => createHash("sha256").update(bytes).digest("hex"); +const writeJson = (file, value) => + fs.writeFileSync(file, `${JSON.stringify(value, null, 2)}\n`, { mode: 0o600 }); + +function identity(manifestBytes, buildBytes) { + const manifest = JSON.parse(manifestBytes.toString()); + assert.equal(manifest.name, "openclaw"); + JSON.parse(buildBytes.toString()); + return { + version: manifest.version, + stateSchemaVersion: manifest.openclaw.schemaVersions.state, + manifestSha256: hash(manifestBytes), + buildInfoSha256: hash(buildBytes), + }; +} + +function installedIdentity(root) { + return identity( + fs.readFileSync(path.join(root, "package.json")), + fs.readFileSync(path.join(root, "dist/build-info.json")), + ); +} + +function snapshot(fixture) { + const db = new DatabaseSync(fixture.databasePath, { readOnly: true }); + try { + const legacySchema = db + .prepare("SELECT sql FROM sqlite_schema WHERE type = 'table' AND name = 'cron_run_logs'") + .get()?.sql; + const legacyRows = legacySchema + ? db + .prepare("SELECT * FROM cron_run_logs ORDER BY store_key, job_id, seq") + .all() + .map((row) => Object.assign({}, row)) + : []; + const tasks = db + .prepare("SELECT * FROM task_runs WHERE source_id IN (?, ?) ORDER BY source_id") + .all(...fixture.entries.map((entry) => entry.jobId)) + .map((row) => Object.assign({}, row)); + const migration = db + .prepare("SELECT status, report_json FROM migration_runs WHERE id = ?") + .get(MIGRATION); + return { + stateSchemaVersion: db.prepare("PRAGMA user_version").get().user_version, + legacySchema: legacySchema ?? null, + legacyRows, + legacySha256: hash(JSON.stringify({ legacySchema: legacySchema ?? null, legacyRows })), + tasks, + migration: migration ?? null, + }; + } finally { + db.close(); + } +} + +export function seedCronHistory(stateDir, artifactRoot, baselineRoot, candidateTarball) { + const baseline = installedIdentity(baselineRoot); + assert(["2026.9.3", "2026.9.4"].includes(baseline.version)); + const packed = (name) => + execFileSync("tar", ["-xOf", candidateTarball, `package/${name}`], { + maxBuffer: 1024 * 1024, + }); + const candidate = identity(packed("package.json"), packed("dist/build-info.json")); + assert( + Number.isSafeInteger(candidate.stateSchemaVersion) && + candidate.stateSchemaVersion >= baseline.stateSchemaVersion, + "retained-history candidate must declare a valid nonolder state schema", + ); + assert.notEqual(baseline.buildInfoSha256, candidate.buildInfoSha256); + const jobs = readJson(path.join(artifactRoot, "legacy-operator-baseline.json")).jobs; + assert.equal(jobs.length, 2); + const fixture = { + baseline, + candidate, + databasePath: path.join(stateDir, "state/openclaw.sqlite"), + storeKey: path.resolve(stateDir, "cron/jobs.json"), + entries: jobs.map((job, index) => ({ + jobId: job.id, + action: "finished", + ts: 1_800_000_000_100 + index * 1000, + runAtMs: 1_800_000_000_000 + index * 1000, + durationMs: 100, + runId: `survivor-retained-cron-${index}`, + status: index === 0 ? "ok" : "error", + completionStatus: index === 0 ? "succeeded" : "failed", + deliveryStatus: "not-requested", + summary: `retained cron history ${index}`, + error: index === 1 ? "synthetic retained failure" : undefined, + })), + }; + const db = new DatabaseSync(fixture.databasePath); + try { + assert.equal(db.prepare("PRAGMA user_version").get().user_version, baseline.stateSchemaVersion); + assert.equal(baseline.stateSchemaVersion, baseline.version === "2026.9.4" ? 17 : 16); + // A retained historical table is the specimen; executing a modern cron job + // writes task_runs directly and would never exercise this import boundary. + db.exec(`CREATE TABLE cron_run_logs ( + store_key TEXT NOT NULL, job_id TEXT NOT NULL, seq INTEGER NOT NULL, + ts INTEGER NOT NULL, entry_json TEXT NOT NULL, created_at INTEGER NOT NULL, + PRIMARY KEY (store_key, job_id, seq) + ) STRICT;`); + const insert = db.prepare("INSERT INTO cron_run_logs VALUES (?, ?, ?, ?, ?, ?)"); + for (const entry of fixture.entries) { + insert.run(fixture.storeKey, entry.jobId, 1, entry.ts, JSON.stringify(entry), entry.ts); + } + } finally { + db.close(); + } + fixture.before = snapshot(fixture); + assert.equal(fixture.before.tasks.length, 0, "fixture jobs already have task history"); + writeJson(path.join(artifactRoot, FIXTURE_NAME), fixture); + return fixture; +} + +function assertImported(fixture, state) { + assert.equal(state.legacySchema, null, "retained cron_run_logs table was not retired"); + assert.deepEqual(state.legacyRows, []); + assert.equal(state.tasks.length, fixture.entries.length, "retained cron task count changed"); + for (const entry of fixture.entries) { + const task = state.tasks.find((row) => row.source_id === entry.jobId); + assert(task, "retained cron history was lost"); + const taskId = `cron-runlog-import:${entry.jobId}:${entry.ts}:1`; + for (const [key, expected] of Object.entries({ + task_id: taskId, + runtime: "cron", + source_id: entry.jobId, + run_id: taskId, + task: entry.jobId, + status: entry.completionStatus, + scope_kind: "system", + created_at: entry.runAtMs, + started_at: entry.runAtMs, + ended_at: entry.ts, + last_event_at: entry.ts, + cleanup_after: null, + error: entry.error ?? null, + terminal_summary: entry.summary, + terminal_outcome: entry.status === "ok" ? "succeeded" : null, + delivery_status: "not_applicable", + notify_policy: "silent", + })) { + assert.equal(task[key], expected, `retained cron task changed: ${key}`); + } + assert.deepEqual(JSON.parse(task.detail_json), { + kind: "cron-run", + status: entry.status, + completionStatus: entry.completionStatus, + error: entry.error ?? null, + summary: entry.summary, + storeKey: fixture.storeKey, + deliveryStatus: entry.deliveryStatus, + runId: entry.runId, + runAtMs: entry.runAtMs, + durationMs: entry.durationMs, + }); + } + assert.equal(state.migration?.status, "completed"); + assert.deepEqual(JSON.parse(state.migration.report_json), { + imported: 2, + alreadyMirrored: 0, + malformed: 0, + skipped: false, + }); +} + +function observeUpdateProcess() { + const fixturePath = process.env.OPENCLAW_UPGRADE_SURVIVOR_CRON_HISTORY_FIXTURE; + const observations = process.env.OPENCLAW_UPGRADE_SURVIVOR_ARTIFACT_ROOT; + const command = process.argv[2]; + if (!isMainThread || !fixturePath || !observations || !["doctor", "update"].includes(command)) { + return; + } + const receipt = { + role: command, + pid: process.pid, + parentPid: process.ppid, + startedAtMs: Date.now(), + }; + let fixture; + try { + fixture = readJson(fixturePath); + assert.equal( + fixture.databasePath, + path.join(process.env.OPENCLAW_STATE_DIR, "state/openclaw.sqlite"), + ); + let root = path.dirname(fs.realpathSync(process.argv[1])); + for (let depth = 0; depth < 3; depth++, root = path.dirname(root)) { + if ( + fs.existsSync(path.join(root, "package.json")) && + readJson(path.join(root, "package.json")).name === "openclaw" + ) { + receipt.identity = installedIdentity(root); + break; + } + } + receipt.updateInProgress = process.env.OPENCLAW_UPDATE_IN_PROGRESS === "1"; + receipt.before = snapshot(fixture); + } catch (error) { + receipt.observationError = String(error); + } + const file = path.join(observations, `cron-history-${command}-${process.pid}.json`); + writeJson(file, receipt); + process.once("exit", (exitCode) => { + try { + receipt.after = snapshot(fixture); + } catch (error) { + receipt.observationError = String(error); + } + writeJson(file, { ...receipt, exitCode }); + }); +} + +function assertProcessReceipt(observations, witness, role) { + const processReceipt = readJson( + path.join(observations, "diagnostics", `process-${witness.pid}-exited.json`), + ); + assert.equal(processReceipt.role, role); + assert.equal(processReceipt.pid, witness.pid); + assert.equal(processReceipt.packageVersion, witness.identity.version); + assert.equal(processReceipt.parentPid, witness.parentPid); + assert.equal(processReceipt.exitCode, 0); + assert.equal(witness.observationError, undefined); + assert.equal(witness.exitCode, 0); +} + +function summarizeSnapshot(state) { + if (!state) { + return undefined; + } + return { + stateSchemaVersion: state.stateSchemaVersion, + legacySha256: state.legacySha256, + legacyRows: state.legacyRows.length, + tasks: state.tasks.length, + tasksSha256: hash(JSON.stringify(state.tasks)), + migration: state.migration + ? { + status: state.migration.status, + report_json: state.migration.report_json.slice(0, 160), + } + : null, + }; +} + +export function assertCronHistory(artifactRoot, observations) { + const fixture = readJson(path.join(artifactRoot, FIXTURE_NAME)); + const proofFile = path.join(artifactRoot, "legacy-operator-cron-history-proof.json"); + const proof = { + baseline: fixture.baseline, + candidate: fixture.candidate, + retainedSha256: fixture.before.legacySha256, + }; + let receipts = []; + let current; + try { + receipts = fs + .readdirSync(observations) + .filter((name) => /^cron-history-(?:doctor|update)-\d+\.json$/u.test(name)) + .map((name) => readJson(path.join(observations, name))) + .toSorted((left, right) => left.startedAtMs - right.startedAtMs); + current = snapshot(fixture); + const doctors = receipts.filter( + (receipt) => + receipt.role === "doctor" && + receipt.identity?.buildInfoSha256 === fixture.candidate.buildInfoSha256 && + !receipt.observationError, + ); + let updater; + let witness; + if (fixture.baseline.version === "2026.9.3") { + // The shipped 9.3 updater imports through its normal opener when admitting + // the update ledger, before candidate code runs. Its result must survive Doctor. + updater = receipts.find( + (receipt) => + receipt.role === "update" && + receipt.identity?.buildInfoSha256 === fixture.baseline.buildInfoSha256 && + receipt.before?.legacySha256 === fixture.before.legacySha256, + ); + assert(updater, "published updater never received the unchanged retained cron history"); + assert.deepEqual(updater.identity, fixture.baseline); + assertProcessReceipt(observations, updater, "update"); + assert.equal(updater.before.stateSchemaVersion, fixture.before.stateSchemaVersion); + assert.deepEqual(updater.before.legacyRows, fixture.before.legacyRows); + assert.deepEqual(updater.before.tasks, []); + witness = doctors[0]; + assert(witness, "candidate Doctor was not observed against the live database"); + assertImported(fixture, witness.before); + assert.deepEqual(witness.after.tasks, witness.before.tasks, "cron history changed in Doctor"); + assert.equal(witness.after.migration?.report_json, witness.before.migration.report_json); + } else { + witness = doctors.find( + (receipt) => receipt.before?.legacySha256 === fixture.before.legacySha256, + ); + assert(witness, "candidate Doctor never received the unchanged retained cron history"); + assert.equal(witness.before.stateSchemaVersion, fixture.before.stateSchemaVersion); + assert.deepEqual(witness.before.legacyRows, fixture.before.legacyRows); + assert.deepEqual(witness.before.tasks, []); + } + assert.deepEqual(witness.identity, fixture.candidate); + assert.equal(witness.updateInProgress, true, "Doctor was not an updater child"); + assertProcessReceipt(observations, witness, "doctor"); + assertImported(fixture, witness.after); + assertImported(fixture, current); + assert.equal(witness.after.stateSchemaVersion, fixture.candidate.stateSchemaVersion); + assert.equal(current.stateSchemaVersion, fixture.candidate.stateSchemaVersion); + assert.deepEqual(current.tasks, witness.after.tasks, "cron history changed after Doctor"); + writeJson(proofFile, { + ...proof, + status: "passed", + contract: updater ? "published-updater-import-preserved" : "candidate-doctor-import", + currentSchemaAtDoctorEntry: + witness.before.stateSchemaVersion === fixture.candidate.stateSchemaVersion, + ...(updater + ? { + updater: { + pid: updater.pid, + parentPid: updater.parentPid, + identity: updater.identity, + before: updater.before, + }, + } + : {}), + doctor: witness, + }); + } catch (error) { + writeJson(proofFile, { + ...proof, + status: "failed", + failure: String(error).slice(0, 500), + current: summarizeSnapshot(current), + observationCount: receipts.length, + // This file shares the existing 16 KiB diagnostic publication budget. + observations: receipts.slice(0, 8).map((receipt) => ({ + role: receipt.role, + pid: receipt.pid, + parentPid: receipt.parentPid, + startedAtMs: receipt.startedAtMs, + identity: receipt.identity, + exitCode: receipt.exitCode, + observationError: receipt.observationError?.slice(0, 160), + before: summarizeSnapshot(receipt.before), + after: summarizeSnapshot(receipt.after), + })), + }); + throw error; + } +} + +observeUpdateProcess(); + +if (process.argv[1] && import.meta.url === pathToFileURL(path.resolve(process.argv[1])).href) { + const [command, ...args] = process.argv.slice(2); + if (command === "seed") { + seedCronHistory( + process.env.OPENCLAW_STATE_DIR, + process.env.OPENCLAW_UPGRADE_SURVIVOR_ARTIFACT_ROOT, + ...args, + ); + } else { + assert.equal(command, "assert"); + assertCronHistory(process.env.OPENCLAW_UPGRADE_SURVIVOR_ARTIFACT_ROOT, args[0]); + } +} diff --git a/scripts/e2e/lib/upgrade-survivor/run.sh b/scripts/e2e/lib/upgrade-survivor/run.sh index a1265775cdaf..0f7ca6fb2554 100644 --- a/scripts/e2e/lib/upgrade-survivor/run.sh +++ b/scripts/e2e/lib/upgrade-survivor/run.sh @@ -1465,6 +1465,11 @@ update_candidate() { update_node_options+=" --import=$PWD/scripts/e2e/lib/upgrade-survivor/workshop-doctor-recovery.mjs" update_env+=("OPENCLAW_UPGRADE_SURVIVOR_WORKSHOP_STATE_DIR=$OPENCLAW_STATE_DIR") fi + if [ "$SCENARIO" = "legacy-operator-state" ] && [ "$UPDATE_RESTART_MODE" = "manual" ] && + { [ "${baseline_version:-}" = "2026.9.3" ] || [ "${baseline_version:-}" = "2026.9.4" ]; }; then + update_node_options+=" --import=$PWD/scripts/e2e/lib/upgrade-survivor/legacy-operator-cron-history.mjs" + update_env+=("OPENCLAW_UPGRADE_SURVIVOR_CRON_HISTORY_FIXTURE=$ARTIFACT_ROOT/legacy-operator-cron-history.json") + fi update_env+=( "OPENCLAW_UPGRADE_SURVIVOR_ARTIFACT_ROOT=$observation_root" "NODE_OPTIONS=$update_node_options" @@ -2264,7 +2269,20 @@ if [ "$SCENARIO" = "legacy-operator-state" ]; then seed-legacy-operator-external-plugin fi run_missing_load_path_fixture unavailable +if [ "$SCENARIO" = "legacy-operator-state" ] && [ "$UPDATE_RESTART_MODE" = "manual" ] && + { [ "${baseline_version:-}" = "2026.9.3" ] || [ "${baseline_version:-}" = "2026.9.4" ]; }; then + # Retained history is added after baseline/backup commands so only the real + # updater and its candidate Doctor can consume this migration specimen. + phase seed-retained-cron-history node scripts/e2e/lib/upgrade-survivor/legacy-operator-cron-history.mjs \ + seed "$(package_root)" "$CANDIDATE_SPEC" +fi phase update-candidate update_candidate_for_install_mode +if [ "$SCENARIO" = "legacy-operator-state" ] && [ "$UPDATE_RESTART_MODE" = "manual" ] && + { [ "${baseline_version:-}" = "2026.9.3" ] || [ "${baseline_version:-}" = "2026.9.4" ]; }; then + # Native read-only inspection precedes every candidate CLI/Gateway probe. + phase assert-retained-cron-doctor node scripts/e2e/lib/upgrade-survivor/legacy-operator-cron-history.mjs \ + assert "$last_update_observation_root" +fi run_missing_load_path_fixture post-update if [ "$SCENARIO" = "legacy-operator-state" ]; then phase assert-formerly-bundled-plugin node scripts/e2e/lib/upgrade-survivor/assertions.mjs \ diff --git a/scripts/pr-lib/wrapper-components.txt b/scripts/pr-lib/wrapper-components.txt index 796bf27f29bb..b28ae070c32c 100644 --- a/scripts/pr-lib/wrapper-components.txt +++ b/scripts/pr-lib/wrapper-components.txt @@ -1069,6 +1069,7 @@ src/state/openclaw-state-db-schema-additive.ts src/state/openclaw-state-db-schema-helpers.ts src/state/openclaw-state-db-schema-migration-required.ts src/state/openclaw-state-db-schema-repair.ts +src/state/openclaw-state-db-schema-runtime.ts src/state/openclaw-state-db-schema-v12-foldin.ts src/state/openclaw-state-db-schema-v13-widerow.ts src/state/openclaw-state-db-schema-version.ts diff --git a/src/agents/subagents/completion/subagent-completion-admission.store.test.ts b/src/agents/subagents/completion/subagent-completion-admission.store.test.ts index 2a3e262093a4..a82e5e0ce89a 100644 --- a/src/agents/subagents/completion/subagent-completion-admission.store.test.ts +++ b/src/agents/subagents/completion/subagent-completion-admission.store.test.ts @@ -17,6 +17,7 @@ import { closeOpenClawStateDatabaseAsync, closeOpenClawStateDatabaseForTest, openOpenClawStateDatabase, + repairOpenClawStateDatabaseSchema, type OpenClawStateDatabase, } from "../../../state/openclaw-state-db.js"; import { captureOpenClawStateWorkerContext } from "../../../state/openclaw-state-worker-context.js"; @@ -55,8 +56,6 @@ import { const resumeSubagentRun = vi.hoisted(() => vi.fn()); const tempDirs = useAutoCleanupTempDirTracker(afterEach); -const discardTerminalDelivery = (entry: SubagentRunRecord, completedAt: number) => - SubagentLifecycleController.discardTerminalDelivery(entry, completedAt); vi.mock("../registry/subagent-registry.js", () => ({ resumeSubagentRun })); @@ -797,7 +796,7 @@ describe("atomic subagent completion admission store", () => { }); }); - it("reloads a blocked text completion from SQLite before canonical owner redrive", async () => { + it("repairs a blocked legacy text completion with Doctor before canonical owner redrive", async () => { await withEnvAsync({ OPENCLAW_STATE_DIR: tempDir }, async () => { closeOpenClawStateDatabaseForTest(); database = openOpenClawStateDatabase(); @@ -863,6 +862,7 @@ describe("atomic subagent completion admission store", () => { resetTaskRegistryForTests({ persist: false }); subagentRuns.clear(); closeOpenClawStateDatabaseForTest(); + expect(repairOpenClawStateDatabaseSchema().warnings).toEqual([]); database = openOpenClawStateDatabase(); queueContext = captureOpenClawStateWorkerContext({ path: database.path, @@ -977,7 +977,7 @@ describe("atomic subagent completion admission store", () => { expect(resumeSubagentRun).not.toHaveBeenCalled(); const discardInsideTransaction = vi.fn((entry: SubagentRunRecord, completedAt: number) => { expect(database.db.isTransaction).toBe(true); - discardTerminalDelivery(entry, completedAt); + SubagentLifecycleController.discardTerminalDelivery(entry, completedAt); }); database.db.exec(` CREATE TRIGGER fail_dismissed_task_persist diff --git a/src/agents/subagents/registry/subagent-registry.store.sqlite.test.ts b/src/agents/subagents/registry/subagent-registry.store.sqlite.test.ts index c925574ec8cd..06b5f8da02ae 100644 --- a/src/agents/subagents/registry/subagent-registry.store.sqlite.test.ts +++ b/src/agents/subagents/registry/subagent-registry.store.sqlite.test.ts @@ -12,6 +12,7 @@ import type { DB as OpenClawStateKyselyDatabase } from "../../../state/openclaw- import { closeOpenClawStateDatabaseForTest, openOpenClawStateDatabase, + repairOpenClawStateDatabaseSchema, } from "../../../state/openclaw-state-db.js"; import { withEnvAsync } from "../../../test-utils/env.js"; import { @@ -573,7 +574,7 @@ describe("subagent registry sqlite store", () => { }); }); - it("promotes legacy retained results into canonical completion state once", async () => { + it("preserves legacy retained results until Doctor promotes canonical completion state", async () => { await withTempStateEnv(async () => { const run = createRun({ completion: { required: true, resultText: "NO_REPLY" }, @@ -598,11 +599,22 @@ describe("subagent registry sqlite store", () => { }, }); saveSubagentRegistryToSqlite(new Map([[run.runId, run]])); + const before = openOpenClawStateDatabase() + .db.prepare("SELECT payload_json FROM subagent_runs WHERE run_id = ?") + .get(run.runId); openOpenClawStateDatabase() .db.prepare("UPDATE schema_meta SET app_version = ? WHERE meta_key = 'primary'") .run("2026.7.0"); closeOpenClawStateDatabaseForTest(); + expect( + openOpenClawStateDatabase() + .db.prepare("SELECT payload_json FROM subagent_runs WHERE run_id = ?") + .get(run.runId), + ).toEqual(before); + closeOpenClawStateDatabaseForTest(); + expect(repairOpenClawStateDatabaseSchema().warnings).toEqual([]); + const restored = loadSubagentRegistryFromSqlite().get(run.runId); expect(restored?.completion).toMatchObject({ resultText: "NO_REPLY", diff --git a/src/gateway/worker-environments/store.test.ts b/src/gateway/worker-environments/store.test.ts index 05a46b8bba47..363a6a200e49 100644 --- a/src/gateway/worker-environments/store.test.ts +++ b/src/gateway/worker-environments/store.test.ts @@ -584,8 +584,8 @@ describe("worker environment store", () => { }); it("idempotently ensures desktop_json on an existing state database", () => { - ensureAdditiveStateColumns(database.db); - ensureAdditiveStateColumns(database.db); + ensureAdditiveStateColumns(database.db, "runtime"); + ensureAdditiveStateColumns(database.db, "runtime"); const columns = database.db.prepare("PRAGMA table_info(worker_environments)").all() as Array<{ name: string; }>; diff --git a/src/infra/startup-maintenance-required.ts b/src/infra/startup-maintenance-required.ts index e11d6d43fb71..475a5310c0b9 100644 --- a/src/infra/startup-maintenance-required.ts +++ b/src/infra/startup-maintenance-required.ts @@ -8,6 +8,7 @@ const maintenanceReasons = { "agent-databases-composite-primary-key": "state database schema migration", "audit-events-v2": "state database schema migration", "legacy-workshop-review-index": "state database schema migration", + "legacy-cron-run-logs": "cron run history migration", "legacy-workspace": "workspace setup state migration", "legacy-session-store": "session store migration", } as const; diff --git a/src/infra/state-migrations.cron-run-logs.test.ts b/src/infra/state-migrations.cron-run-logs.test.ts index f6b3593a577b..6f515ec1bd0f 100644 --- a/src/infra/state-migrations.cron-run-logs.test.ts +++ b/src/infra/state-migrations.cron-run-logs.test.ts @@ -7,6 +7,8 @@ import { readCronTaskRunHistoryPage } from "../cron/task-run-history.js"; import { closeOpenClawStateDatabaseForTest, openOpenClawStateDatabase, + repairOpenClawStateDatabaseSchema, + repairOpenClawStateDatabaseSchemaIfNeeded, } from "../state/openclaw-state-db.js"; import { resetTaskRegistryForTests } from "../tasks/task-runtime.test-helpers.js"; import { withOpenClawTestState } from "../test-utils/openclaw-test-state.js"; @@ -14,7 +16,7 @@ import { withOpenClawTestState } from "../test-utils/openclaw-test-state.js"; const CRON_RUN_LOG_TASK_IMPORT_MIGRATION_ID = "state:cron-run-logs-to-task-runs:v1"; describe("cron run-log task import", () => { - it("imports legacy cron history into task runs once at state database open", async () => { + it("preserves legacy cron history on runtime refusal, then Doctor imports it once", async () => { await withOpenClawTestState( { layout: "state-only", prefix: "openclaw-cron-run-log-import-" }, async (state) => { @@ -158,6 +160,23 @@ describe("cron run-log task import", () => { fixture.close(); } + expect(repairOpenClawStateDatabaseSchemaIfNeeded()).toEqual({ + changes: [], + warnings: [expect.stringMatching(/legacy-cron-run-logs.*doctor --fix/u)], + }); + expect(() => openOpenClawStateDatabase()).toThrow(/legacy-cron-run-logs.*doctor --fix/u); + const preserved = new DatabaseSync(databasePath, { readOnly: true }); + try { + expect(preserved.prepare("SELECT COUNT(*) AS count FROM cron_run_logs").get()).toEqual({ + count: 8, + }); + expect(preserved.prepare("SELECT COUNT(*) AS count FROM task_runs").get()).toEqual({ + count: 2, + }); + } finally { + preserved.close(); + } + expect(repairOpenClawStateDatabaseSchema().warnings).toEqual([]); const reopened = openOpenClawStateDatabase(); const report = reopened.db .prepare("SELECT report_json FROM migration_runs WHERE id = ?") diff --git a/src/infra/state-migrations.doctor.ts b/src/infra/state-migrations.doctor.ts index 95e6967abe31..d1bc538be4f4 100644 --- a/src/infra/state-migrations.doctor.ts +++ b/src/infra/state-migrations.doctor.ts @@ -3037,7 +3037,7 @@ export async function prepareLegacyStateDatabaseSchema( createStateSchemaMigrationStep({ stateDir: resolveStateDir(env), env, - mode: "automatic", + mode: "doctor", requiredness: "conditional", }), ]); diff --git a/src/infra/state-migrations.plugin-doctor.test.ts b/src/infra/state-migrations.plugin-doctor.test.ts index bcce32ddd4a1..3b70f6c539ac 100644 --- a/src/infra/state-migrations.plugin-doctor.test.ts +++ b/src/infra/state-migrations.plugin-doctor.test.ts @@ -1,11 +1,16 @@ import fs from "node:fs"; import path from "node:path"; +import { DatabaseSync } from "node:sqlite"; import { afterEach, describe, expect, it, vi } from "vitest"; import type { listPluginDoctorStateMigrationEntries } from "../plugins/doctor-contract-registry.js"; import { closeOpenClawAgentDatabasesForTest } from "../state/openclaw-agent-db.js"; import { closeOpenClawStateDatabaseForTest } from "../state/openclaw-state-db.js"; import { createTrackedTempDirs } from "../test-utils/tracked-temp-dirs.js"; -import { runPostSessionPluginDoctorStateRepairs } from "./state-migrations.plugin-doctor.js"; +import { + autoMigrateLegacyPluginDoctorState, + runPostSessionPluginDoctorStateRepairs, +} from "./state-migrations.plugin-doctor.js"; +import { resetAutoMigrateLegacyStateDirForTest } from "./state-migrations.state-dir.js"; const controls = vi.hoisted(() => ({ entries: [] as ReturnType, @@ -39,12 +44,85 @@ const tempDirs = createTrackedTempDirs(); afterEach(async () => { controls.entries = []; controls.failSettlement = false; + resetAutoMigrateLegacyStateDirForTest(); closeOpenClawAgentDatabasesForTest(); closeOpenClawStateDatabaseForTest(); await tempDirs.cleanup(); }); -describe("plugin Doctor migration settlement", () => { +describe("plugin Doctor migrations", () => { + it("requires explicit Doctor to repair shared schema before plugin migrations", async () => { + const root = await tempDirs.make("openclaw-plugin-doctor-shared-schema-"); + const stateDir = path.join(root, ".openclaw"); + const env = { ...process.env, HOME: root, OPENCLAW_STATE_DIR: stateDir }; + const cfg = {}; + const stateDbPath = path.join(stateDir, "state", "openclaw.sqlite"); + fs.mkdirSync(path.dirname(stateDbPath), { recursive: true }); + const db = new DatabaseSync(stateDbPath); + try { + db.exec(` + CREATE TABLE agent_databases ( + agent_id TEXT PRIMARY KEY, + path TEXT NOT NULL, + schema_version INTEGER NOT NULL, + last_seen_at INTEGER NOT NULL, + size_bytes INTEGER + ); + INSERT INTO agent_databases VALUES ('main', 'agent.sqlite', 1, 10, 20); + `); + } finally { + db.close(); + } + const migrateLegacyState = vi.fn(() => ({ + changes: ["plugin state migrated"], + warnings: [], + })); + controls.entries = [ + { + pluginId: "memory-core", + channelIds: [], + migration: { + id: "memory-core-test", + label: "Memory Core test migration", + detectLegacyState: () => ({ preview: ["plugin state"] }), + migrateLegacyState, + }, + }, + ]; + + await expect( + autoMigrateLegacyPluginDoctorState({ config: cfg, env, homedir: () => root }), + ).rejects.toThrow("agent-databases-composite-primary-key"); + expect(migrateLegacyState).not.toHaveBeenCalled(); + const preserved = new DatabaseSync(stateDbPath, { readOnly: true }); + try { + expect(preserved.prepare("SELECT * FROM agent_databases").all()).toEqual([ + { + agent_id: "main", + path: "agent.sqlite", + schema_version: 1, + last_seen_at: 10, + size_bytes: 20, + }, + ]); + } finally { + preserved.close(); + } + + const result = await autoMigrateLegacyPluginDoctorState({ + config: cfg, + env, + homedir: () => root, + doctorOnlyStateMigrations: true, + }); + + expect(result.warnings).toStrictEqual([]); + expect(result.changes).toContain( + "Migrated shared state agent database registry primary key → agent_id,path", + ); + expect(result.changes).toContain("plugin state migrated"); + expect(migrateLegacyState).toHaveBeenCalledOnce(); + }); it.each([ { name: "reordered", diff --git a/src/infra/state-migrations.plugin-doctor.ts b/src/infra/state-migrations.plugin-doctor.ts index 1247f6192083..e236afc11330 100644 --- a/src/infra/state-migrations.plugin-doctor.ts +++ b/src/infra/state-migrations.plugin-doctor.ts @@ -11,7 +11,10 @@ import { } from "../plugins/doctor-contract-registry.js"; import { withPluginLifecycleLease } from "../plugins/plugin-lifecycle-lease.js"; import { withAgentDatabaseMaintenanceLease } from "../state/openclaw-agent-db.js"; -import { repairOpenClawStateDatabaseSchemaIfNeeded } from "../state/openclaw-state-db.js"; +import { + repairOpenClawStateDatabaseSchema, + repairOpenClawStateDatabaseSchemaIfNeeded, +} from "../state/openclaw-state-db.js"; import { acquireGatewayLock } from "./gateway-lock.js"; import { formatStartupMigrationFailure } from "./state-migrations.messages.js"; import { createPluginDoctorStateMigrationContext } from "./state-migrations.plugin-doctor-context.js"; @@ -520,7 +523,11 @@ export async function autoMigrateLegacyPluginDoctorState(params: { }); const stateDir = resolveStateDir(env, params.homedir ?? os.homedir); const oauthDir = resolveOAuthDir(env, stateDir); - const stateSchema = repairOpenClawStateDatabaseSchemaIfNeeded({ + const prepareStateSchema = + params.doctorOnlyStateMigrations === true + ? repairOpenClawStateDatabaseSchema + : repairOpenClawStateDatabaseSchemaIfNeeded; + const stateSchema = prepareStateSchema({ env: { ...env, OPENCLAW_STATE_DIR: stateDir }, }); const changes = [...stateDirResult.changes, ...stateSchema.changes]; diff --git a/src/infra/state-migrations.test.ts b/src/infra/state-migrations.test.ts index 30c412f0c5d8..eb846dbe59ad 100644 --- a/src/infra/state-migrations.test.ts +++ b/src/infra/state-migrations.test.ts @@ -3826,56 +3826,6 @@ describe("state migrations", () => { await expectMissingPath(restartSentinelPath); }); - it("runs plugin doctor migrations after repairing shared state schema", async () => { - const { root, stateDir, env } = createMigrationContext(await createTempDir()); - const cfg = createConfig(); - const stateDbPath = path.join(stateDir, "state", "openclaw.sqlite"); - await fs.mkdir(path.dirname(stateDbPath), { recursive: true }); - const db = new DatabaseSync(stateDbPath); - try { - db.exec(` - CREATE TABLE agent_databases ( - agent_id TEXT PRIMARY KEY, - path TEXT NOT NULL, - schema_version INTEGER NOT NULL, - last_seen_at INTEGER NOT NULL, - size_bytes INTEGER - ); - INSERT INTO agent_databases VALUES ('main', 'agent.sqlite', 1, 10, 20); - `); - } finally { - db.close(); - } - const migrateLegacyState = vi.fn(() => ({ - changes: ["plugin state migrated"], - warnings: [], - })); - pluginDoctorStateMigrationEntries.entries = [ - { - pluginId: "memory-core", - migration: { - id: "memory-core-test", - label: "Memory Core test migration", - detectLegacyState: () => ({ preview: ["plugin state"] }), - migrateLegacyState, - }, - }, - ]; - - const result = await autoMigrateLegacyPluginDoctorState({ - config: cfg, - env, - homedir: () => root, - }); - - expect(result.warnings).toStrictEqual([]); - expect(result.changes).toContain( - "Migrated shared state agent database registry primary key → agent_id,path", - ); - expect(result.changes).toContain("plugin state migrated"); - expect(migrateLegacyState).toHaveBeenCalledOnce(); - }); - it("previews and repairs the released audit ledger before other state migrations", async () => { const { root, stateDir, env } = createMigrationContext(await createTempDir()); const databasePath = await createLegacyAuditLedger(stateDir); diff --git a/src/state/openclaw-database-maintenance.test.ts b/src/state/openclaw-database-maintenance.test.ts index 02d5c87e1cdd..c5cee9e33873 100644 --- a/src/state/openclaw-database-maintenance.test.ts +++ b/src/state/openclaw-database-maintenance.test.ts @@ -296,7 +296,7 @@ CREATE INDEX IF NOT EXISTS idx_web_push_approval_deliveries_subscription database.exec(`ALTER TABLE "${tableName}" DROP COLUMN "${columnName}";`); } - ensureAdditiveStateColumns(database); + ensureAdditiveStateColumns(database, "runtime"); expect(() => assertOpenClawStateDatabaseForMaintenance(database, { pathname: "global.sqlite", diff --git a/src/state/openclaw-state-db-acp-replay.test.ts b/src/state/openclaw-state-db-acp-replay.test.ts index 63128247f7be..84a4cfa08fb5 100644 --- a/src/state/openclaw-state-db-acp-replay.test.ts +++ b/src/state/openclaw-state-db-acp-replay.test.ts @@ -1,13 +1,14 @@ import path from "node:path"; import { constants, DatabaseSync } from "node:sqlite"; -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { expectAcpReplayUtf8Accounting } from "../acp/event-ledger.test-support.js"; +import * as nodeSqlite from "../infra/node-sqlite.js"; import { withTestDir } from "../test-helpers/temp-dir.js"; -import { isOpenClawStateSchemaFastPathEligible } from "./openclaw-state-db-fast-path.js"; import { closeOpenClawStateDatabaseForTest, openOpenClawStateDatabase, repairOpenClawStateDatabaseSchema, + repairOpenClawStateDatabaseSchemaIfNeeded, } from "./openclaw-state-db.js"; function seedLegacyReplay(db: DatabaseSync) { @@ -65,12 +66,100 @@ function estimates(db: DatabaseSync) { }; } +function withoutHistoricalPayloadReads(pathname: string, operation: () => T): T { + const open = nodeSqlite.openNodeSqliteDatabase; + const opened = new Set(); + const historicalReads: string[] = []; + const historicalColumns = new Set([ + "acp_replay_events.update_json", + "acp_replay_sessions.estimated_bytes", + "subagent_runs.payload_json", + "task_runs.delivery_status", + "operator_approvals.resolution_ref", + "cron_jobs.job_json", + "delivery_queue_entries.entry_json", + ]); + const spy = vi.spyOn(nodeSqlite, "openNodeSqliteDatabase").mockImplementation((...args) => { + const database = open(...args); + if (args[0] === pathname) { + opened.add(database); + database.setAuthorizer((action, table, column) => { + const field = `${table}.${column}`; + if (action === constants.SQLITE_READ && historicalColumns.has(field)) { + historicalReads.push(field); + return constants.SQLITE_DENY; + } + return constants.SQLITE_OK; + }); + } + return database; + }); + try { + const result = operation(); + expect(opened.size).toBeGreaterThan(0); + expect(historicalReads).toEqual([]); + return result; + } finally { + spy.mockRestore(); + for (const database of opened) { + if (database.isOpen) { + database.setAuthorizer(null); + } + } + } +} + describe("ACP replay accounting repair", () => { afterEach(() => closeOpenClawStateDatabaseForTest()); - it.each(["UTF-8", "UTF-16le"])( - "repairs every derived total on app-version reopen without changing canonical %s rows", - async (encoding) => { + it.each(["runtime", "automatic"])( + "populates newly added accounting columns through %s without changing canonical rows", + async (entrance) => { + await withTestDir({ prefix: "openclaw-acp-additive-" }, async (dir) => { + const options = { path: path.join(dir, "state.sqlite") }; + const initial = openOpenClawStateDatabase(options).db; + seedLegacyReplay(initial); + const before = canonicalReplay(initial); + initial.exec(` + ALTER TABLE acp_replay_events DROP COLUMN estimated_bytes; + ALTER TABLE acp_replay_sessions DROP COLUMN estimated_bytes; + `); + closeOpenClawStateDatabaseForTest(); + + if (entrance === "automatic") { + expect(repairOpenClawStateDatabaseSchemaIfNeeded(options).warnings).toEqual([]); + } + const upgraded = openOpenClawStateDatabase(options).db; + expectAcpReplayUtf8Accounting(upgraded); + expect(canonicalReplay(upgraded)).toEqual(before); + closeOpenClawStateDatabaseForTest(); + + const reopened = withoutHistoricalPayloadReads(options.path, () => + openOpenClawStateDatabase(options), + ).db; + expect(reopened.prepare("SELECT total_changes() AS count").get()?.count).toBe(0); + expectAcpReplayUtf8Accounting(reopened); + expect(canonicalReplay(reopened)).toEqual(before); + }); + }, + ); + + it.each( + ["UTF-8", "UTF-16le"].flatMap((encoding) => + [ + "current", + "missing-column", + "ordered-column", + "sandbox-column", + "cron-description", + "index-drift", + ].flatMap((schema) => + ["runtime", "automatic"].map((entrance) => ({ encoding, schema, entrance })), + ), + ), + )( + "leaves $encoding replay repair to Doctor through $entrance with $schema schema", + async ({ encoding, schema, entrance }) => { await withTestDir({ prefix: "openclaw-acp-repair-" }, async (dir) => { const options = { path: path.join(dir, "state.sqlite") }; const seed = new DatabaseSync(options.path); @@ -81,28 +170,63 @@ describe("ACP replay accounting repair", () => { const initial = openOpenClawStateDatabase(options).db; seedLegacyReplay(initial); const before = canonicalReplay(initial); + const oldEstimates = estimates(initial); + if (schema === "missing-column") { + initial.exec("ALTER TABLE claw_installs DROP COLUMN bootstrap_source_path"); + } else if (schema === "ordered-column") { + initial.exec("ALTER TABLE worktrees DROP COLUMN provisioned_paths_json"); + } else if (schema === "sandbox-column") { + initial.exec("ALTER TABLE sandbox_registry_entries DROP COLUMN image"); + } else if (schema === "cron-description") { + initial.exec("ALTER TABLE cron_jobs DROP COLUMN description"); + } else if (schema === "index-drift") { + initial.exec(`DROP INDEX idx_plugin_state_listing; + CREATE INDEX idx_plugin_state_listing + ON plugin_state_entries(plugin_id, namespace, created_at, entry_key);`); + } closeOpenClawStateDatabaseForTest(); + if (entrance === "automatic") { + expect( + withoutHistoricalPayloadReads(options.path, () => + repairOpenClawStateDatabaseSchemaIfNeeded(options), + ).warnings, + ).toEqual([]); + } else { + withoutHistoricalPayloadReads(options.path, () => openOpenClawStateDatabase(options)); + closeOpenClawStateDatabaseForTest(); + } + const inspected = new DatabaseSync(options.path, { readOnly: true }); + try { + expect(estimates(inspected)).toMatchObject({ + sessions: oldEstimates.sessions, + events: oldEstimates.events, + }); + expect(canonicalReplay(inspected)).toEqual(before); + } finally { + inspected.close(); + } + const runtime = withoutHistoricalPayloadReads(options.path, () => + openOpenClawStateDatabase(options), + ).db; + expect(estimates(runtime)).toMatchObject({ + sessions: oldEstimates.sessions, + events: oldEstimates.events, + }); + expect(canonicalReplay(runtime)).toEqual(before); + closeOpenClawStateDatabaseForTest(); + + expect(repairOpenClawStateDatabaseSchema(options).warnings).toEqual([]); const repaired = openOpenClawStateDatabase(options).db; expectAcpReplayUtf8Accounting(repaired); expect(canonicalReplay(repaired)).toEqual(before); const repairedEstimates = estimates(repaired); closeOpenClawStateDatabaseForTest(); - const reopened = openOpenClawStateDatabase(options).db; + const reopened = withoutHistoricalPayloadReads(options.path, () => + openOpenClawStateDatabase(options), + ).db; expect(reopened.prepare("SELECT total_changes() AS count").get()?.count).toBe(0); expect(estimates(reopened)).toEqual(repairedEstimates); - reopened.setAuthorizer((action, table, column) => - action === constants.SQLITE_READ && - table === "acp_replay_events" && - column === "update_json" - ? constants.SQLITE_DENY - : constants.SQLITE_OK, - ); - try { - expect(isOpenClawStateSchemaFastPathEligible(reopened, options.path)).toBe(true); - } finally { - reopened.setAuthorizer(null); - } }); }, ); diff --git a/src/state/openclaw-state-db-binding-targets.test.ts b/src/state/openclaw-state-db-binding-targets.test.ts index 62e7796e4a06..d4f1431cd075 100644 --- a/src/state/openclaw-state-db-binding-targets.test.ts +++ b/src/state/openclaw-state-db-binding-targets.test.ts @@ -271,7 +271,10 @@ describe("conversation binding target migration", () => { migrated.db .prepare("SELECT schema_version, app_version FROM schema_meta WHERE meta_key = 'primary'") .get(), - ).toEqual({ schema_version: OPENCLAW_STATE_SCHEMA_VERSION, app_version: VERSION }); + ).toEqual({ + schema_version: OPENCLAW_STATE_SCHEMA_VERSION, + app_version: migrationPath === "runtime open" ? VERSION : null, + }); expect( migrated.db .prepare( diff --git a/src/state/openclaw-state-db-fast-path.test.ts b/src/state/openclaw-state-db-fast-path.test.ts index 0897d20c9579..6049ceeb2dcf 100644 --- a/src/state/openclaw-state-db-fast-path.test.ts +++ b/src/state/openclaw-state-db-fast-path.test.ts @@ -5,6 +5,7 @@ import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js"; import { closeOpenClawStateDatabaseForTest, openOpenClawStateDatabase, + repairOpenClawStateDatabaseSchema, } from "./openclaw-state-db.js"; const dirs = useAutoCleanupTempDirTracker((cleanup) => @@ -21,23 +22,33 @@ describe("state schema fast-path failure settlement", () => { name: "retains repair after successful rollback", rollbackFails: false, undefinedError: false, + convergenceFails: false, }, { name: "preserves the original error after native close", rollbackFails: true, undefinedError: false, + convergenceFails: false, }, { name: "preserves undefined rejection after native close", rollbackFails: true, undefinedError: true, + convergenceFails: false, }, - ])("$name", ({ rollbackFails, undefinedError }) => { + { + name: "restores foreign-key enforcement after failed schema convergence", + rollbackFails: false, + undefinedError: false, + convergenceFails: true, + }, + ])("$name", ({ rollbackFails, undefinedError, convergenceFails }) => { const env = { OPENCLAW_STATE_DIR: dirs.make("state-fast-path-settlement-") }; const pathname = realpathSync(openOpenClawStateDatabase({ env }).path); closeOpenClawStateDatabaseForTest(); const original = undefinedError ? undefined : new Error("synthetic fast-path COMMIT failure"); const rollbackError = new Error("synthetic fast-path ROLLBACK failure"); + const convergenceError = new Error("synthetic schema BEGIN failure"); // oxlint-disable-next-line typescript/unbound-method -- Fault injection forwards the native method with its exact database receiver. const exec = DatabaseSync.prototype.exec; // oxlint-disable-next-line typescript/unbound-method -- Native close is called with its exact database receiver below. @@ -46,6 +57,7 @@ describe("state schema fast-path failure settlement", () => { []; const selected = new Set(); let injected = false; + let foreignKeysAtClose: unknown; vi.spyOn(DatabaseSync.prototype, "exec").mockImplementation(function (this: DatabaseSync, sql) { if (!selected.size && sql === "BEGIN" && this.location() === pathname) { selected.add(this); @@ -66,10 +78,16 @@ describe("state schema fast-path failure settlement", () => { if (sql === "PRAGMA foreign_keys = OFF;") { events.push({ phase: "fallback", isOpen: this.isOpen }); } + if (convergenceFails && sql === "BEGIN IMMEDIATE") { + throw convergenceError; + } } Reflect.apply(exec, this, [sql]); }); vi.spyOn(DatabaseSync.prototype, "close").mockImplementation(function (this: DatabaseSync) { + if (selected.has(this)) { + foreignKeysAtClose = this.prepare("PRAGMA foreign_keys").get()?.foreign_keys; + } Reflect.apply(close, this, []); if (selected.has(this)) { events.push({ phase: "close", isOpen: this.isOpen }); @@ -84,7 +102,16 @@ describe("state schema fast-path failure settlement", () => { result = { status: "rejected", error }; } expect(injected).toBe(true); - if (rollbackFails) { + if (convergenceFails) { + expect(result).toEqual({ status: "rejected", error: convergenceError }); + expect(foreignKeysAtClose).toBe(1); + expect(events).toEqual([ + { phase: "commit", isOpen: true }, + { phase: "rollback", isOpen: true }, + { phase: "fallback", isOpen: true }, + { phase: "close", isOpen: false }, + ]); + } else if (rollbackFails) { expect(result.status).toBe("rejected"); if (result.status !== "rejected") { throw new Error("Expected the failed native rollback to refuse opening"); @@ -102,6 +129,7 @@ describe("state schema fast-path failure settlement", () => { throw new Error("Expected successful rollback to retain the schema repair fallback"); } expect(result.database.db.isOpen).toBe(true); + expect(result.database.db.prepare("PRAGMA foreign_keys").get()).toEqual({ foreign_keys: 1 }); expect(events).toEqual([ { phase: "commit", isOpen: true }, { phase: "rollback", isOpen: true }, @@ -109,4 +137,47 @@ describe("state schema fast-path failure settlement", () => { ]); } }); + + it.each([false, true])( + "settles Doctor's disabled foreign-key connection (failure=%s)", + (fails) => { + const env = { OPENCLAW_STATE_DIR: dirs.make("state-doctor-foreign-keys-") }; + const pathname = realpathSync(openOpenClawStateDatabase({ env }).path); + closeOpenClawStateDatabaseForTest(); + const originalExec = Object.getOwnPropertyDescriptor(DatabaseSync.prototype, "exec") + ?.value as ((this: DatabaseSync, sql: string) => void) | undefined; + const originalClose = Object.getOwnPropertyDescriptor(DatabaseSync.prototype, "close") + ?.value as ((this: DatabaseSync) => void) | undefined; + if (!originalExec || !originalClose) { + throw new Error("Native SQLite descriptors are unavailable"); + } + const selected = new Set(); + let foreignKeysAtClose: unknown; + vi.spyOn(DatabaseSync.prototype, "exec").mockImplementation( + function (this: DatabaseSync, sql) { + if (sql === "BEGIN IMMEDIATE" && this.location() === pathname) { + selected.add(this); + expect(this.prepare("PRAGMA foreign_keys").get()).toEqual({ foreign_keys: 0 }); + if (fails) { + throw new Error("synthetic Doctor BEGIN failure"); + } + } + originalExec.call(this, sql); + }, + ); + vi.spyOn(DatabaseSync.prototype, "close").mockImplementation(function (this: DatabaseSync) { + if (selected.has(this)) { + foreignKeysAtClose = this.prepare("PRAGMA foreign_keys").get()?.foreign_keys; + } + originalClose.call(this); + }); + + const repaired = repairOpenClawStateDatabaseSchema({ env }); + expect(repaired.warnings).toEqual( + fails ? [expect.stringContaining("synthetic Doctor BEGIN failure")] : [], + ); + expect([...selected].map((database) => database.isOpen)).toEqual([false]); + expect(foreignKeysAtClose).toBe(0); + }, + ); }); diff --git a/src/state/openclaw-state-db-fast-path.ts b/src/state/openclaw-state-db-fast-path.ts index 4f79a6babc84..2c6791e23325 100644 --- a/src/state/openclaw-state-db-fast-path.ts +++ b/src/state/openclaw-state-db-fast-path.ts @@ -8,9 +8,9 @@ import { } from "../infra/sqlite-schema-contract.js"; import { runSqliteDeferredTransactionSync } from "../infra/sqlite-transaction.js"; import { hasLegacyCronRunLogs } from "../infra/state-migrations.cron-run-logs.js"; -import { VERSION } from "../version.js"; import { OPENCLAW_STATE_SCHEMA_VERSION } from "./openclaw-state-db-contract.js"; import { assertOpenClawStateDatabaseForMaintenance } from "./openclaw-state-db-maintenance.js"; +import { OpenClawStateDatabaseSchemaMigrationRequiredError } from "./openclaw-state-db-schema-migration-required.js"; import { assertCanonicalStateSchemaShape, detectOpenClawStateDatabaseSchemaMigrationsFromDatabase, @@ -32,6 +32,7 @@ export function needsOpenClawStateDatabaseSchemaRepair(pathname: string): boolea assertSupportedStateSchemaVersion(database, pathname); const needsRepair = readStateSchemaMigrationVersion(database) !== OPENCLAW_STATE_SCHEMA_VERSION || + hasLegacyCronRunLogs(database) || detectOpenClawStateDatabaseSchemaMigrationsFromDatabase(database, pathname).length > 0; if (!needsRepair) { assertCurrentStateRuntimeSchema(database, pathname); @@ -54,6 +55,13 @@ export function assertCurrentStateRuntimeSchema( assertOpenClawStateDatabaseForMaintenance(database, { pathname }, readTable); } +/** Catalog presence is enough to refuse retired history without reading or rewriting its rows. */ +export function assertNoLegacyStateRuntimeRepair(database: DatabaseSync, pathname: string): void { + if (hasLegacyCronRunLogs(database)) { + throw new OpenClawStateDatabaseSchemaMigrationRequiredError("legacy-cron-run-logs", pathname); + } +} + export function isOpenClawStateSchemaFastPathEligible( database: DatabaseSync, pathname: string, @@ -76,13 +84,7 @@ export function isOpenClawStateSchemaFastPathEligible( if (startupRepairRequired) { return false; } - if (hasLegacyCronRunLogs(database)) { - return false; - } - // app_version commits only after this release's repairs; same-build writes are canonical. - const metadata = database - .prepare("SELECT app_version FROM schema_meta WHERE meta_key = 'primary' LIMIT 1") - .get(); - return metadata?.app_version === VERSION; + assertNoLegacyStateRuntimeRepair(database, pathname); + return true; }); } diff --git a/src/state/openclaw-state-db-legacy-backfills.test.ts b/src/state/openclaw-state-db-legacy-backfills.test.ts index 8b81c2997e3b..b89f81be2302 100644 --- a/src/state/openclaw-state-db-legacy-backfills.test.ts +++ b/src/state/openclaw-state-db-legacy-backfills.test.ts @@ -8,7 +8,10 @@ import { import { closeOpenClawStateDatabaseForTest, openOpenClawStateDatabase, + repairOpenClawStateDatabaseSchema, + repairOpenClawStateDatabaseSchemaIfNeeded, } from "./openclaw-state-db.js"; +import { removePreparedWorkerOwnershipColumns } from "./openclaw-state-schema-v17.test-support.js"; const tempDirs = useAutoCleanupTempDirTracker((cleanup) => { afterEach(() => { @@ -42,8 +45,68 @@ function createDatabase() { }; } -describe("repairLegacySubagentSuspensionReasons", () => { - it("rewrites the shipped reason on open and stays canonical after a second open", () => { +describe("Doctor historical row repair", () => { + it("refuses an automatic older-schema upgrade with invalid foreign keys without repairing rows", () => { + const stateDir = tempDirs.make("openclaw-automatic-upgrade-integrity-"); + const options = { env: { OPENCLAW_STATE_DIR: stateDir } }; + const { db: initial, path: pathname } = openOpenClawStateDatabase(options); + initial.exec("PRAGMA foreign_keys = OFF;"); + initial + .prepare("INSERT INTO task_delivery_state (task_id, requester_origin_json) VALUES (?, ?)") + .run("missing-task", '{"channel":"synthetic"}'); + removePreparedWorkerOwnershipColumns(initial); + initial.exec("PRAGMA user_version = 16; UPDATE schema_meta SET schema_version = 16;"); + closeOpenClawStateDatabaseForTest(); + + expect(repairOpenClawStateDatabaseSchemaIfNeeded(options)).toEqual({ + changes: [], + warnings: [expect.stringMatching(/foreign_key_check failed.*task_delivery_state/iu)], + }); + const preserved = new DatabaseSync(pathname, { readOnly: true }); + try { + expect(preserved.prepare("PRAGMA user_version").get()).toEqual({ user_version: 16 }); + expect( + preserved.prepare("SELECT task_id, requester_origin_json FROM task_delivery_state").all(), + ).toEqual([{ task_id: "missing-task", requester_origin_json: '{"channel":"synthetic"}' }]); + expect( + preserved + .prepare("PRAGMA table_info(worker_environments)") + .all() + .map((row) => row.name), + ).not.toContain("preparation_key"); + } finally { + preserved.close(); + } + }); + + it("leaves retired history on open until Doctor removes it", () => { + const stateDir = tempDirs.make("openclaw-retired-history-"); + const options = { env: { OPENCLAW_STATE_DIR: stateDir } }; + const initial = openOpenClawStateDatabase(options).db; + const transientHistoryTable = ["database", "verifications"].join("_"); + initial.exec(`CREATE TABLE ${transientHistoryTable} (path TEXT PRIMARY KEY) STRICT;`); + initial + .prepare("UPDATE schema_meta SET app_version = ? WHERE meta_key = 'primary'") + .run("2026.7.0"); + closeOpenClawStateDatabaseForTest(); + + const reopened = openOpenClawStateDatabase(options); + expect( + reopened.db + .prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?") + .get(transientHistoryTable), + ).toEqual({ name: transientHistoryTable }); + closeOpenClawStateDatabaseForTest(); + expect(repairOpenClawStateDatabaseSchema(options).warnings).toEqual([]); + const repaired = openOpenClawStateDatabase(options); + expect( + repaired.db + .prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?") + .get(transientHistoryTable), + ).toBeUndefined(); + }); + + it("leaves the shipped reason unchanged on open until Doctor repairs it", () => { const stateDir = tempDirs.make("openclaw-subagent-suspension-backfill-"); const options = { env: { OPENCLAW_STATE_DIR: stateDir } }; const initial = openOpenClawStateDatabase(options); @@ -77,6 +140,16 @@ describe("repairLegacySubagentSuspensionReasons", () => { .run("2026.7.0"); closeOpenClawStateDatabaseForTest(); + const runtime = openOpenClawStateDatabase(options); + expect( + runtime.db + .prepare( + "SELECT json_extract(payload_json, '$.delivery.suspendedReason') AS reason FROM subagent_runs WHERE run_id = ?", + ) + .get(runId), + ).toEqual({ reason: "retry-limit" }); + closeOpenClawStateDatabaseForTest(); + expect(repairOpenClawStateDatabaseSchema(options).warnings).toEqual([]); const firstOpen = openOpenClawStateDatabase(options); const firstStored = firstOpen.db .prepare("SELECT payload_json FROM subagent_runs WHERE run_id = ?") diff --git a/src/state/openclaw-state-db-maintenance.ts b/src/state/openclaw-state-db-maintenance.ts index 82c98640471c..b6786d195c65 100644 --- a/src/state/openclaw-state-db-maintenance.ts +++ b/src/state/openclaw-state-db-maintenance.ts @@ -543,45 +543,57 @@ export function runStateSchemaMigrationTransaction( transactionOptions: SqliteTransactionOptions, prepareSchema?: () => void, ): T { - return runSqliteImmediateTransactionSync( - db, - () => { - // Doctor restores catalog readability before the publication prelude reads it. - prepareSchema?.(); - const publishedVersion = readSqliteUserVersion(db); - const blocker = - publishedVersion < OPENCLAW_STATE_SCHEMA_VERSION - ? readStateSchemaPublicationBlocker(db) - : undefined; - if (!blocker) { - return migrate(); - } - try { - // Check before canonical DDL could recreate the missing publication owner. - if (!tableExists(db, "config_machine_state")) { - throw new Error("Shared state schema publication requires config_machine_state."); + const foreignKeysWereEnabled = + Number(db.prepare("PRAGMA foreign_keys").get()?.foreign_keys) === 1; + // Referenced-table rebuilds require this before BEGIN, including runtime convergence. + if (foreignKeysWereEnabled) { + db.exec("PRAGMA foreign_keys = OFF;"); + } + try { + return runSqliteImmediateTransactionSync( + db, + () => { + // Doctor restores catalog readability before the publication prelude reads it. + prepareSchema?.(); + const publishedVersion = readSqliteUserVersion(db); + const blocker = + publishedVersion < OPENCLAW_STATE_SCHEMA_VERSION + ? readStateSchemaPublicationBlocker(db) + : undefined; + if (!blocker) { + return migrate(); } - return migrate(); - } catch (cause) { - if (cause instanceof OpenClawStateOwnershipError) { - throw cause; + try { + // Check before canonical DDL could recreate the missing publication owner. + if (!tableExists(db, "config_machine_state")) { + throw new Error("Shared state schema publication requires config_machine_state."); + } + return migrate(); + } catch (cause) { + if (cause instanceof OpenClawStateOwnershipError) { + throw cause; + } + throw new UpdateSchemaRefusalError( + [ + { + kind: "state", + path: pathname, + foundVersion: publishedVersion, + supportedVersion: OPENCLAW_STATE_SCHEMA_VERSION, + }, + ], + blocker.updaterVersion, + { targetVersion: VERSION, cause }, + ); } - throw new UpdateSchemaRefusalError( - [ - { - kind: "state", - path: pathname, - foundVersion: publishedVersion, - supportedVersion: OPENCLAW_STATE_SCHEMA_VERSION, - }, - ], - blocker.updaterVersion, - { targetVersion: VERSION, cause }, - ); - } - }, - transactionOptions, - ); + }, + transactionOptions, + ); + } finally { + if (foreignKeysWereEnabled && db.isOpen) { + db.exec("PRAGMA foreign_keys = ON;"); + } + } } export function writeCurrentStateSchemaMetadata(db: DatabaseSync, now: number): void { diff --git a/src/state/openclaw-state-db-repair.ts b/src/state/openclaw-state-db-repair.ts index 887689361777..ae180435f619 100644 --- a/src/state/openclaw-state-db-repair.ts +++ b/src/state/openclaw-state-db-repair.ts @@ -9,6 +9,7 @@ import { assertSqliteIntegrity, assertSqliteTableIntegrity } from "../infra/sqli import { assertSqliteSchemaTablesPresent } from "../infra/sqlite-schema-contract.js"; import { migrateSqliteSchemaToStrictInTransaction } from "../infra/sqlite-strict.js"; import { runSqliteImmediateTransactionSync } from "../infra/sqlite-transaction.js"; +import { migrateLegacyCronRunLogsToTaskRuns } from "../infra/state-migrations.cron-run-logs.js"; import { clearOpenClawDatabaseQuarantine } from "./openclaw-quarantine-store.js"; import { repairAuditEventsSchema } from "./openclaw-state-db-audit-migration.js"; import { clearOpenClawStateDatabaseOpenFailure } from "./openclaw-state-db-cache.js"; @@ -43,9 +44,9 @@ import { repairAgentDatabasesCompositePrimaryKey, repairLegacyGatewayRestartHandoffsForStrictMigration, } from "./openclaw-state-db-schema-repair.js"; +import { ensureOpenClawStateRuntimeSchema } from "./openclaw-state-db-schema-runtime.js"; import { migrateSingletonStateFoldInV12 } from "./openclaw-state-db-schema-v12-foldin.js"; import { - assertSupportedStateSchemaVersion, readStateSchemaContentVersion, readStateSchemaMigrationVersion, } from "./openclaw-state-db-schema-version.js"; @@ -53,10 +54,7 @@ import * as sessionWatchMigration from "./openclaw-state-db-session-watch-migrat import * as retirements from "./openclaw-state-db-table-retirements.js"; import { recoverOrphanTaskDeliveryRows } from "./openclaw-state-db-task-delivery-recovery.js"; import { describeAgentPathMigration } from "./openclaw-state-db.paths.js"; -import { - assertOpenClawStateWriteAllowed, - OpenClawStateOwnershipError, -} from "./openclaw-state-ownership.js"; +import { OpenClawStateOwnershipError } from "./openclaw-state-ownership.js"; import { getOpenClawStateRuntimeSchema } from "./openclaw-state-schema-compatibility.js"; import { OPENCLAW_STATE_SCHEMA_SQL } from "./openclaw-state-schema.js"; import { UpdateSchemaRefusalError } from "./openclaw-update-schema-refusal.js"; @@ -76,11 +74,11 @@ export function repairStateSchema( let ownershipRefused = false; try { setSqliteBusyTimeout(db, OPENCLAW_SQLITE_BUSY_TIMEOUT_MS); - const repairAdmittedSchema = - scope === "automatic" ? undefined : prepareStateDatabaseSchemaRepair(db, pathname, env); - if (!repairAdmittedSchema) { - assertSupportedStateSchemaVersion(db, pathname); - } else if (scope === "readability") { + if (scope === "automatic") { + return { changes: ensureOpenClawStateRuntimeSchema(db, pathname, env), warnings: [] }; + } + const repairAdmittedSchema = prepareStateDatabaseSchemaRepair(db, pathname, env); + if (scope === "readability") { return { changes: runSqliteImmediateTransactionSync( db, @@ -106,9 +104,6 @@ export function repairStateSchema( db, pathname, () => { - if (!repairAdmittedSchema) { - assertOpenClawStateWriteAllowed({ database: db, databasePath: pathname, env }); - } applied.push(...recoverOrphanTaskDeliveryRows(db, pathname)); const previousVersion = readStateSchemaMigrationVersion(db); const preAuditSchema = previousVersion === 1 && !tableExists(db, "audit_events"); @@ -165,7 +160,7 @@ export function repairStateSchema( // Recognized schema-1 stores predate audit; Doctor must finish their schema // before its later read-only workspace and agent readers can consume it. if (preAuditSchema || tableExists(db, "audit_events")) { - ensureAdditiveStateColumns(db); + ensureAdditiveStateColumns(db, "repair"); for (const migration of versionedStateMigrations) { if (migration.migrate(db, previousVersion)) { applied.push(migration.applied); @@ -174,6 +169,7 @@ export function repairStateSchema( executeCanonicalStateSchema(db, { includeVersionLazyAdditiveTables: previousVersion !== OPENCLAW_STATE_SCHEMA_VERSION, }); + migrateLegacyCronRunLogsToTaskRuns(db); if (previousVersion < OPENCLAW_STATE_STRICT_SCHEMA_VERSION) { repairLegacyGatewayRestartHandoffsForStrictMigration(db); ensureFirstUseAdditiveStateColumnsForStrictMigration(db); @@ -213,7 +209,7 @@ export function repairStateSchema( operationLabel: "state.schema.repair", }, () => { - applied.push(...(repairAdmittedSchema?.() ?? [])); + applied.push(...repairAdmittedSchema()); }, ); const quarantineCleared = clearOpenClawDatabaseQuarantine(pathname, { env }); @@ -236,10 +232,13 @@ export function repairStateSchema( } // Reaching this catch inside doctor means repair itself refused or failed, // so the runtime asserts' "run openclaw doctor --fix" advice is circular here. - const reason = String(err).replace( - /has a legacy ([a-z ]+) schema; run openclaw doctor --fix to migrate it\./u, - "has a legacy $1 schema; automatic repair refused the unrecognized schema shape.", - ); + const reason = + scope === "automatic" + ? String(err) + : String(err).replace( + /has a legacy ([a-z ]+) schema; run openclaw doctor --fix to migrate it\./u, + "has a legacy $1 schema; automatic repair refused the unrecognized schema shape.", + ); return { changes: [], warnings: [`Failed migrating shared state database schema at ${pathname}: ${reason}`], diff --git a/src/state/openclaw-state-db-schema-additive.ts b/src/state/openclaw-state-db-schema-additive.ts index 3b6f852229bb..12a5691375d9 100644 --- a/src/state/openclaw-state-db-schema-additive.ts +++ b/src/state/openclaw-state-db-schema-additive.ts @@ -282,13 +282,20 @@ export function ensureFirstUseAdditiveStateColumnsForStrictMigration(db: Databas function ensureColumns( db: DatabaseSync, definitions: readonly (readonly [string, string])[], -): void { - for (const definition of definitions) { - ensureColumn(db, ...definition); +): Array<{ tableName: string; columnName: string }> { + const added: Array<{ tableName: string; columnName: string }> = []; + for (const [tableName, definition] of definitions) { + const columnName = definition.trim().split(/\s+/, 1)[0]; + if (columnName && ensureColumn(db, tableName, definition)) { + added.push({ tableName, columnName }); + } } + return added; } -export function ensureAdditiveStateColumns(db: DatabaseSync): void { +/** Runtime pairs new columns with their transforms; full historical repair stays explicit. */ +export function ensureAdditiveStateColumns(db: DatabaseSync, scope: "runtime" | "repair"): void { + const repairHistoricalRows = scope === "repair"; ensureWorkerSessionToolStateSchema(db); for (const { columnName, @@ -323,15 +330,35 @@ export function ensureAdditiveStateColumns(db: DatabaseSync): void { ); `); } - db.exec("DROP INDEX IF EXISTS idx_diagnostic_events_scope_created;"); - ensureColumns(db, columns.cronRunLogs); - backfillCronRunLogEntryJson(db); - ensureColumns(db, columns.acpReplay); - backfillAcpReplayEstimatedBytes(db); - ensureColumns(db, columns.cronJobs); - backfillCronJobsFromJobJson(db); - ensureColumns(db, columns.deliveryQueue); - backfillDeliveryQueueEntriesFromEntryJson(db); + if (addedDiagnosticEventSequence || repairHistoricalRows) { + db.exec("DROP INDEX IF EXISTS idx_diagnostic_events_scope_created;"); + } + const addedCronLogColumns = ensureColumns(db, columns.cronRunLogs); + if ( + repairHistoricalRows || + addedCronLogColumns.some(({ tableName }) => tableName === "cron_run_logs") + ) { + backfillCronRunLogEntryJson(db); + } + if (ensureColumns(db, columns.acpReplay).length > 0 || repairHistoricalRows) { + backfillAcpReplayEstimatedBytes(db); + } + const addedCronJobColumns = ensureColumns(db, columns.cronJobs); + if ( + repairHistoricalRows || + addedCronJobColumns.some(({ columnName }) => + ["name", "enabled", "agent_id", "payload_kind", "runtime_updated_at_ms"].includes(columnName), + ) + ) { + backfillCronJobsFromJobJson(db); + } + const addedDeliveryColumns = ensureColumns(db, columns.deliveryQueue); + if ( + repairHistoricalRows || + addedDeliveryColumns.some(({ tableName }) => tableName === "delivery_queue_entries") + ) { + backfillDeliveryQueueEntriesFromEntryJson(db); + } // The shipped JSON runtime predeclared this table but never populated it. // The transitional default makes ADD COLUMN portable; schema-v2 tables are // rebuilt from canonical STRICT SQL immediately afterward, removing it. @@ -344,12 +371,18 @@ export function ensureAdditiveStateColumns(db: DatabaseSync): void { if (addedTaskRequesterAgentId) { repairLegacyTaskAgentAttribution(db); } - repairLegacyTaskDeliveryStatuses(db); + if (repairHistoricalRows) { + repairLegacyTaskDeliveryStatuses(db); + } ensureColumns(db, columns.taskRunDetails); - repairLegacySubagentSuspensionReasons(db); - repairLegacySubagentExecutionPayloads(db); - repairLegacySubagentTaskBindings(db); - repairLegacySubagentRetainedResults(db); + if (repairHistoricalRows) { + repairLegacySubagentSuspensionReasons(db); + repairLegacySubagentExecutionPayloads(db); + repairLegacySubagentTaskBindings(db); + repairLegacySubagentRetainedResults(db); + } ensureColumns(db, columns.workerEnvironments); - ensureOperatorApprovalResolutionRefs(db); + if (repairHistoricalRows || !tableHasColumn(db, "operator_approvals", "resolution_ref")) { + ensureOperatorApprovalResolutionRefs(db); + } } diff --git a/src/state/openclaw-state-db-schema-migration-required.ts b/src/state/openclaw-state-db-schema-migration-required.ts index 3024413c0ce4..5ad7b3397319 100644 --- a/src/state/openclaw-state-db-schema-migration-required.ts +++ b/src/state/openclaw-state-db-schema-migration-required.ts @@ -6,6 +6,7 @@ export const LEGACY_SKILL_WORKSHOP_COLLECTION_REVIEWS_INDEX = type OpenClawStateDatabaseSchemaMigrationRequiredKind = | "agent-databases-composite-primary-key" | "audit-events-v2" + | "legacy-cron-run-logs" | "legacy-workshop-review-index"; export class OpenClawStateDatabaseSchemaMigrationRequiredError extends StartupMaintenanceRequiredError { diff --git a/src/state/openclaw-state-db-schema-runtime.ts b/src/state/openclaw-state-db-schema-runtime.ts new file mode 100644 index 000000000000..faa5d1ff1b64 --- /dev/null +++ b/src/state/openclaw-state-db-schema-runtime.ts @@ -0,0 +1,161 @@ +import type { DatabaseSync } from "node:sqlite"; +import { + repairCanonicalSqliteIndexes, + verifyAndRepairCanonicalSqliteIndexes, +} from "../infra/sqlite-index-schema.js"; +import { assertSqliteIntegrity } from "../infra/sqlite-integrity.js"; +import { migrateSqliteSchemaToStrictInTransaction } from "../infra/sqlite-strict.js"; +import { StartupMaintenanceRequiredError } from "../infra/startup-maintenance-required.js"; +import { withStateSchemaFence } from "../infra/state-database-coordinator.js"; +import { migrateLegacyCronRunLogsToTaskRuns } from "../infra/state-migrations.cron-run-logs.js"; +import { createSubsystemLogger } from "../logging/subsystem.js"; +import { + OPENCLAW_SQLITE_BUSY_TIMEOUT_MS, + OPENCLAW_STATE_SCHEMA_VERSION, + OPENCLAW_STATE_STRICT_SCHEMA_VERSION, +} from "./openclaw-state-db-contract.js"; +import { + assertCurrentStateRuntimeSchema, + assertNoLegacyStateRuntimeRepair, + isOpenClawStateSchemaFastPathEligible, +} from "./openclaw-state-db-fast-path.js"; +import { + assertOpenClawStateDatabaseForMaintenance, + executeCanonicalStateSchema, + openClawStateMigrationAssertions, + runStateSchemaMigrationTransaction, + versionedStateMigrations, + writeCurrentStateSchemaMetadata, +} from "./openclaw-state-db-maintenance.js"; +import { + ensureAdditiveStateColumns, + ensureFirstUseAdditiveStateColumnsForStrictMigration, +} from "./openclaw-state-db-schema-additive.js"; +import { + assertCanonicalStateSchemaShape, + dropLegacyStateTables, + migrateAgentDatabaseRelativePaths, + migrateWorkerPlacementExecutionModeSchema, + repairLegacyGatewayRestartHandoffsForStrictMigration, +} from "./openclaw-state-db-schema-repair.js"; +import { migrateSingletonStateFoldInV12 } from "./openclaw-state-db-schema-v12-foldin.js"; +import { + assertSupportedStateSchemaVersion, + readStateSchemaMigrationVersion, +} from "./openclaw-state-db-schema-version.js"; +import { migrateSessionWatchCursorProvenance } from "./openclaw-state-db-session-watch-migration.js"; +import { isUninitializedNativeStartupDatabase } from "./openclaw-state-db-startup-checkpoint.js"; +import * as retirements from "./openclaw-state-db-table-retirements.js"; +import { describeAgentPathMigration, warnAgentPathMigration } from "./openclaw-state-db.paths.js"; +import { assertOpenClawStateWriteAllowed } from "./openclaw-state-ownership.js"; +import { getOpenClawStateRuntimeSchema } from "./openclaw-state-schema-compatibility.js"; +import { OPENCLAW_STATE_SCHEMA_SQL } from "./openclaw-state-schema.js"; + +const stateDbLog = createSubsystemLogger("state/db"); + +/** Runtime converges schema; historical row repair belongs to explicit Doctor maintenance. */ +export function ensureOpenClawStateRuntimeSchema( + db: DatabaseSync, + pathname: string, + env: NodeJS.ProcessEnv, + busyTimeoutMs = OPENCLAW_SQLITE_BUSY_TIMEOUT_MS, + initializeNativeOnly = false, +): string[] { + try { + if (isOpenClawStateSchemaFastPathEligible(db, pathname)) { + // A claim made during validation must not retain a writable handle. + assertOpenClawStateWriteAllowed({ database: db, databasePath: pathname, env }); + return []; + } + } catch (error) { + if (!db.isOpen || error instanceof StartupMaintenanceRequiredError) { + throw error; + } + // Preserve transactional schema convergence and its diagnostics after a clean rollback. + } + + return withStateSchemaFence({ databasePath: pathname }, () => { + const now = Date.now(); + const retiredTableChanges: string[] = []; + const applied = runStateSchemaMigrationTransaction( + db, + pathname, + () => { + assertOpenClawStateWriteAllowed({ database: db, databasePath: pathname, env }); + assertSupportedStateSchemaVersion(db, pathname); + if (initializeNativeOnly && !isUninitializedNativeStartupDatabase(db)) { + return []; + } + const previousVersion = readStateSchemaMigrationVersion(db); + if (previousVersion === OPENCLAW_STATE_SCHEMA_VERSION) { + assertNoLegacyStateRuntimeRepair(db, pathname); + const indexes = verifyAndRepairCanonicalSqliteIndexes( + db, + pathname, + OPENCLAW_STATE_SCHEMA_SQL, + { + allowMissingColumns: true, + validateAfterRepair: () => assertCurrentStateRuntimeSchema(db, pathname), + }, + ); + ensureAdditiveStateColumns(db, "runtime"); + assertCurrentStateRuntimeSchema(db, pathname); + writeCurrentStateSchemaMetadata(db, now); + return indexes.length > 0 + ? [`Rebuilt canonical shared-state SQLite indexes (${indexes.length})`] + : []; + } + + // Older schemas still need atomic content transforms before retiring their columns. + openClawStateMigrationAssertions.get(previousVersion)?.(db, { pathname }); + // Automatic preparation enters without the physical opener's integrity preflight. + assertSqliteIntegrity(db, pathname); + dropLegacyStateTables(db); + const changes = retirements.runRetiredStateTableMigrations(db, previousVersion); + retiredTableChanges.push(...changes); + if (migrateSingletonStateFoldInV12(db, previousVersion)) { + changes.push("Folded singleton state tables into config_machine_state (v12)"); + } + if (migrateWorkerPlacementExecutionModeSchema(db, previousVersion)) { + changes.push("Migrated cloud worker placements to execution modes"); + } + const pathMigration = migrateAgentDatabaseRelativePaths(db, previousVersion, pathname); + changes.push(...describeAgentPathMigration(pathMigration)); + ensureAdditiveStateColumns(db, "repair"); + for (const migration of versionedStateMigrations) { + if (migration.migrate(db, previousVersion)) { + changes.push(migration.applied); + } + } + migrateSessionWatchCursorProvenance(db); + assertCanonicalStateSchemaShape(db, pathname); + executeCanonicalStateSchema(db, { includeVersionLazyAdditiveTables: true }); + migrateLegacyCronRunLogsToTaskRuns(db); + if (previousVersion < OPENCLAW_STATE_STRICT_SCHEMA_VERSION) { + repairLegacyGatewayRestartHandoffsForStrictMigration(db); + ensureFirstUseAdditiveStateColumnsForStrictMigration(db); + const strict = migrateSqliteSchemaToStrictInTransaction( + db, + getOpenClawStateRuntimeSchema({ includeVersionLazyAdditiveTables: true }), + { databaseLabel: pathname }, + ); + if (strict.migratedTables.length > 0) { + changes.push( + `Migrated shared state tables to SQLite STRICT typing (${strict.migratedTables.length})`, + ); + } + } + repairCanonicalSqliteIndexes(db, pathname, OPENCLAW_STATE_SCHEMA_SQL, { + verifyPhysicalIntegrity: false, + }); + writeCurrentStateSchemaMetadata(db, now); + assertOpenClawStateDatabaseForMaintenance(db, { pathname }); + warnAgentPathMigration(stateDbLog, pathMigration, pathname); + return changes; + }, + { busyTimeoutMs, databaseLabel: pathname, operationLabel: "state.schema.ensure" }, + ); + retiredTableChanges.forEach(retirements.logRetiredStateTableMigration); + return applied; + }); +} diff --git a/src/state/openclaw-state-db.test.ts b/src/state/openclaw-state-db.test.ts index 628fdad8b59e..e8a72acbdda8 100644 --- a/src/state/openclaw-state-db.test.ts +++ b/src/state/openclaw-state-db.test.ts @@ -4421,26 +4421,6 @@ describe("openclaw state database", () => { } }); - it("drops unreleased transient verification history on open", () => { - const stateDir = createTempStateDir(); - const options = { env: { OPENCLAW_STATE_DIR: stateDir } }; - const databasePath = materializeCurrentStateDatabase(stateDir); - - const transientHistoryTable = ["database", "verifications"].join("_"); - const { DatabaseSync } = requireNodeSqlite(); - const legacy = new DatabaseSync(databasePath); - legacy.exec(`CREATE TABLE ${transientHistoryTable} (path TEXT PRIMARY KEY) STRICT;`); - markStateDatabaseAsPreviousAppVersion(legacy); - legacy.close(); - - const reopened = openOpenClawStateDatabase(options); - expect( - reopened.db - .prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?") - .get(transientHistoryTable), - ).toBeUndefined(); - }); - it("adopts a canonical device identity seed database without losing the identity", () => { const stateDir = createTempStateDir(); const databasePath = path.join(stateDir, "state", "openclaw.sqlite"); @@ -7647,7 +7627,7 @@ INSERT INTO macos_port_guardian_records VALUES (4242, 18789, '/usr/bin/ssh', 're }); }); - it("normalizes obsolete task delivery statuses in existing state databases", async () => { + it("leaves obsolete task delivery statuses unchanged until Doctor repairs them", async () => { await withOpenClawTestState( { layout: "state-only", prefix: "openclaw-state-task-delivery-status-" }, async ({ stateDir }) => { @@ -7689,6 +7669,15 @@ INSERT INTO macos_port_guardian_records VALUES (4242, 18789, '/usr/bin/ssh', 're { task_id: "pending", delivery_status: "pending" }, ]; + expect(readStatuses()).toEqual([ + { task_id: "canonical", delivery_status: "not_applicable" }, + { task_id: "obsolete", delivery_status: "not-requested" }, + { task_id: "pending", delivery_status: "pending" }, + ]); + closeOpenClawStateDatabaseForTest(); + expect( + repairOpenClawStateDatabaseSchema({ env: { OPENCLAW_STATE_DIR: stateDir } }).warnings, + ).toEqual([]); expect(readStatuses()).toEqual(expectedStatuses); expect( [...loadTaskRegistryStateFromSqlite().tasks.values()].map((task) => ({ @@ -8548,31 +8537,30 @@ INSERT INTO macos_port_guardian_records VALUES (4242, 18789, '/usr/bin/ssh', 're }); }); - it("repairs null schema metadata once before using the current-schema fast path", () => { + it("leaves optional writer metadata unchanged when the current schema needs no migration", () => { const stateDir = createTempStateDir(); const options = { env: { OPENCLAW_STATE_DIR: stateDir } }; const databasePath = openOpenClawStateDatabase(options).path; closeOpenClawStateDatabaseForTest(); const { DatabaseSync } = requireNodeSqlite(); - const corrupt = new DatabaseSync(databasePath); - corrupt + const fixture = new DatabaseSync(databasePath); + fixture .prepare( "UPDATE schema_meta SET app_version = NULL, updated_at = 1 WHERE meta_key = 'primary'", ) .run(); - corrupt.close(); + fixture.close(); openOpenClawStateDatabase(options); closeOpenClawStateDatabaseForTest(); - const afterRepair = new DatabaseSync(databasePath, { readOnly: true }); - const repaired = afterRepair + const afterOpen = new DatabaseSync(databasePath, { readOnly: true }); + const metadata = afterOpen .prepare("SELECT app_version, updated_at FROM schema_meta WHERE meta_key = 'primary'") - .get() as { app_version: string; updated_at: number }; - afterRepair.close(); - expect(repaired.app_version).toBe(VERSION); - expect(repaired.updated_at).toBeGreaterThan(1); + .get(); + afterOpen.close(); + expect(metadata).toEqual({ app_version: null, updated_at: 1 }); openOpenClawStateDatabase(options); closeOpenClawStateDatabaseForTest(); @@ -8583,7 +8571,7 @@ INSERT INTO macos_port_guardian_records VALUES (4242, 18789, '/usr/bin/ssh', 're afterReopen .prepare("SELECT app_version, updated_at FROM schema_meta WHERE meta_key = 'primary'") .get(), - ).toEqual(repaired); + ).toEqual(metadata); } finally { afterReopen.close(); } diff --git a/src/state/openclaw-state-db.ts b/src/state/openclaw-state-db.ts index ea2ba293783a..5bf379f8d8ed 100644 --- a/src/state/openclaw-state-db.ts +++ b/src/state/openclaw-state-db.ts @@ -8,21 +8,14 @@ import { type SqliteLockFailureReporting, } from "../infra/sqlite-busy-timeout.js"; import { createSqliteLifecycleAggregateError } from "../infra/sqlite-coordinator.js"; -import { - repairCanonicalSqliteIndexes, - verifyAndRepairCanonicalSqliteIndexes, -} from "../infra/sqlite-index-schema.js"; import { assertSqliteIntegrity } from "../infra/sqlite-integrity.js"; import { prepareSqliteReadOnlyLocation } from "../infra/sqlite-snapshot-source.js"; -import { migrateSqliteSchemaToStrictInTransaction } from "../infra/sqlite-strict.js"; import type { SqliteTransactionOptions } from "../infra/sqlite-transaction.js"; import { readSqliteUserVersion } from "../infra/sqlite-user-version.js"; import { StateSchemaMutationConflictError, withStateSchemaFence, } from "../infra/state-database-coordinator.js"; -import { migrateLegacyCronRunLogsToTaskRuns } from "../infra/state-migrations.cron-run-logs.js"; -import { createSubsystemLogger } from "../logging/subsystem.js"; import { getOpenClawDatabaseMaintenanceScope, observeOpenClawDatabaseMaintenanceResource, @@ -35,72 +28,43 @@ import { OPENCLAW_DATABASE_SCHEMA_DOCS_URL, OPENCLAW_SQLITE_BUSY_TIMEOUT_MS, OPENCLAW_STATE_SCHEMA_VERSION, - OPENCLAW_STATE_STRICT_SCHEMA_VERSION, type OpenClawStateDatabase, type OpenClawStateDatabaseOptions, } from "./openclaw-state-db-contract.js"; import { openDoctorStateSchemaReadAdmission } from "./openclaw-state-db-doctor-schema.js"; -import { - assertCurrentStateRuntimeSchema, - isOpenClawStateSchemaFastPathEligible, - needsOpenClawStateDatabaseSchemaRepair, -} from "./openclaw-state-db-fast-path.js"; +import { needsOpenClawStateDatabaseSchemaRepair } from "./openclaw-state-db-fast-path.js"; import { assertOpenClawStateDatabaseForMaintenance, markCurrentStateSchemaVersion, - openClawStateMigrationAssertions, resolveDatabasePath, - versionedStateMigrations, - runStateSchemaMigrationTransaction, - writeCurrentStateSchemaMetadata, - executeCanonicalStateSchema, } from "./openclaw-state-db-maintenance.js"; import { openUnpublishedStateDatabase } from "./openclaw-state-db-open.js"; import { ensureOpenClawStatePermissions } from "./openclaw-state-db-permissions.js"; import { openOpenClawStateReadConnection } from "./openclaw-state-db-read-connection.js"; import { withExistingOpenClawStateDatabaseReadOnly } from "./openclaw-state-db-readonly.js"; import { repairStateSchema } from "./openclaw-state-db-repair.js"; -import { - ensureAdditiveStateColumns, - ensureFirstUseAdditiveStateColumnsForStrictMigration, -} from "./openclaw-state-db-schema-additive.js"; -import { - type AgentDatabasePathMigrationSummary as AgentPathSummary, - assertCanonicalStateSchemaShape, - dropLegacyStateTables, - migrateAgentDatabaseRelativePaths as migrateAgentPaths, - migrateWorkerPlacementExecutionModeSchema, - repairLegacyGatewayRestartHandoffsForStrictMigration, -} from "./openclaw-state-db-schema-repair.js"; -import { migrateSingletonStateFoldInV12 } from "./openclaw-state-db-schema-v12-foldin.js"; +import { ensureOpenClawStateRuntimeSchema as ensureSchema } from "./openclaw-state-db-schema-runtime.js"; import { assertSupportedStateSchemaVersion, readStateSchemaContentVersion, - readStateSchemaMigrationVersion, } from "./openclaw-state-db-schema-version.js"; -import * as sessionWatchMigration from "./openclaw-state-db-session-watch-migration.js"; import { initializeNativeOpenClawStateConnection, - isUninitializedNativeStartupDatabase, withOpenClawStateStartupCheckpointConnection, } from "./openclaw-state-db-startup-checkpoint.js"; -import * as retirements from "./openclaw-state-db-table-retirements.js"; import { runCoordinatedStateTransaction, withSharedStateWriteCoordinator, } from "./openclaw-state-db-write-coordination.js"; -import { warnAgentPathMigration } from "./openclaw-state-db.paths.js"; import { assertOpenClawStateWriteAllowed, isOpenClawStateWriteContentionError, runWithOpenClawStateWriteAccess, } from "./openclaw-state-ownership.js"; -import { getOpenClawStateRuntimeSchema } from "./openclaw-state-schema-compatibility.js"; import { readStateSchemaPublicationBlocker, type StateSchemaPublicationBlocker, } from "./openclaw-state-schema-publication.js"; -import { OPENCLAW_STATE_SCHEMA_SQL } from "./openclaw-state-schema.js"; export { registerOpenClawStateDatabaseLifecycleListener } from "./openclaw-state-db-cache.js"; @@ -122,7 +86,6 @@ function assertOpenClawStateDatabaseFreshOpenAllowed( stateDbCache.assertOpenClawStateDatabaseFreshOpenAllowedAtPath(resolveDatabasePath(options), env); } -const stateDbLog = createSubsystemLogger("state/db"); const deferredStateDatabases = new WeakSet(); export function repairOpenClawStateDatabaseSchema(options: OpenClawStateDatabaseOptions = {}): { @@ -173,7 +136,7 @@ export function repairOpenClawStateDatabaseReadabilityForDoctor( ); } -/** Skip the exclusive doctor repair when automatic migration sees a canonical current schema. */ +/** Automatic preparation shares runtime schema convergence; historical repair stays with Doctor. */ export function repairOpenClawStateDatabaseSchemaIfNeeded( options: OpenClawStateDatabaseOptions = {}, ): { @@ -198,104 +161,6 @@ export function repairOpenClawStateDatabaseSchemaIfNeeded( ); } -function ensureSchema( - db: DatabaseSync, - pathname: string, - env: NodeJS.ProcessEnv, - busyTimeoutMs = OPENCLAW_SQLITE_BUSY_TIMEOUT_MS, - initializeNativeOnly = false, -): void { - try { - if (isOpenClawStateSchemaFastPathEligible(db, pathname)) { - // Recheck ownership so a claim made during validation cannot retain a writable handle. - assertOpenClawStateWriteAllowed({ database: db, databasePath: pathname, env }); - return; - } - } catch (error) { - if (!db.isOpen) { - throw error; - } - // Preserve the existing transactional repair and its diagnostics for drift or corruption. - } - - withStateSchemaFence({ databasePath: pathname }, () => { - const now = Date.now(); - db.exec("PRAGMA foreign_keys = OFF;"); // Rebuilding referenced tables requires this before BEGIN. - try { - runStateSchemaMigrationTransaction( - db, - pathname, - () => { - // Recheck ownership after BEGIN IMMEDIATE to exclude a concurrent external claim. - assertOpenClawStateWriteAllowed({ database: db, databasePath: pathname, env }); - assertSupportedStateSchemaVersion(db, pathname); - // Native bootstrap admission is advisory until this transaction owns the - // write. Never migrate state initialized or occupied by a concurrent owner. - if (initializeNativeOnly && !isUninitializedNativeStartupDatabase(db)) { - return []; - } - const previousVersion = readStateSchemaMigrationVersion(db); - if (previousVersion === OPENCLAW_STATE_SCHEMA_VERSION) { - verifyAndRepairCanonicalSqliteIndexes(db, pathname, OPENCLAW_STATE_SCHEMA_SQL, { - allowMissingColumns: true, - validateAfterRepair: () => assertCurrentStateRuntimeSchema(db, pathname), - }); - ensureAdditiveStateColumns(db); - assertCurrentStateRuntimeSchema(db, pathname); - } else { - openClawStateMigrationAssertions.get(previousVersion)?.(db, { pathname }); - } - dropLegacyStateTables(db); - const retirementMessages = retirements.runRetiredStateTableMigrations( - db, - previousVersion, - ); - migrateSingletonStateFoldInV12(db, previousVersion); - migrateWorkerPlacementExecutionModeSchema(db, previousVersion); - const pathMigration: AgentPathSummary = migrateAgentPaths(db, previousVersion, pathname); - ensureAdditiveStateColumns(db); - for (const migration of versionedStateMigrations) { - migration.migrate(db, previousVersion); - } - sessionWatchMigration.migrateSessionWatchCursorProvenance(db); - assertCanonicalStateSchemaShape(db, pathname); - executeCanonicalStateSchema(db, { - includeVersionLazyAdditiveTables: previousVersion !== OPENCLAW_STATE_SCHEMA_VERSION, - }); - migrateLegacyCronRunLogsToTaskRuns(db); - if (previousVersion < OPENCLAW_STATE_STRICT_SCHEMA_VERSION) { - repairLegacyGatewayRestartHandoffsForStrictMigration(db); - ensureFirstUseAdditiveStateColumnsForStrictMigration(db); - migrateSqliteSchemaToStrictInTransaction( - db, - getOpenClawStateRuntimeSchema({ - includeVersionLazyAdditiveTables: previousVersion !== OPENCLAW_STATE_SCHEMA_VERSION, - }), - { databaseLabel: pathname }, - ); - } - repairCanonicalSqliteIndexes(db, pathname, OPENCLAW_STATE_SCHEMA_SQL, { - verifyPhysicalIntegrity: false, - }); - writeCurrentStateSchemaMetadata(db, now); - assertOpenClawStateDatabaseForMaintenance(db, { pathname }); - warnAgentPathMigration(stateDbLog, pathMigration, pathname); - return retirementMessages; - }, - { - busyTimeoutMs, - databaseLabel: pathname, - operationLabel: "state.schema.ensure", - }, - ).forEach(retirements.logRetiredStateTableMigration); - } finally { - if (db.isOpen) { - db.exec("PRAGMA foreign_keys = ON;"); - } - } - }); -} - /** Bootstrap fresh/native-only state canonically before startup checkpoint access. */ export function withOpenClawStateStartupMigrationCheckpointDatabase( callback: (db: DatabaseSync) => T, diff --git a/src/state/openclaw-state-ownership.test.ts b/src/state/openclaw-state-ownership.test.ts index b6fc3c54d3e6..659896a36ab1 100644 --- a/src/state/openclaw-state-ownership.test.ts +++ b/src/state/openclaw-state-ownership.test.ts @@ -824,25 +824,27 @@ describe("external shared-state ownership", () => { it("fences a claim made during a canonical current-schema cold open", () => { const env = createEnv(); - const databasePath = openOpenClawStateDatabase({ env }).path; + const { path: databasePath, db: seeded } = openOpenClawStateDatabase({ env }); + const databaseLocation = seeded.location(); closeOpenClawStateDatabaseForTest(); const { DatabaseSync } = requireNodeSqlite(); - const originalPrepare = Object.getOwnPropertyDescriptor(DatabaseSync.prototype, "prepare") - ?.value as - | (( - this: import("node:sqlite").DatabaseSync, - sql: string, - ) => import("node:sqlite").StatementSync) + const originalExec = Object.getOwnPropertyDescriptor(DatabaseSync.prototype, "exec")?.value as + | ((this: import("node:sqlite").DatabaseSync, sql: string) => void) | undefined; - if (!originalPrepare) { - throw new Error("DatabaseSync.prepare descriptor is unavailable"); + if (!originalExec) { + throw new Error("DatabaseSync.exec descriptor is unavailable"); } let claimInjected = false; - const prepare = vi.spyOn(DatabaseSync.prototype, "prepare").mockImplementation(function ( + const validating = new Set(); + const exec = vi.spyOn(DatabaseSync.prototype, "exec").mockImplementation(function ( this: import("node:sqlite").DatabaseSync, sql: string, ) { - if (!claimInjected && sql.includes("SELECT app_version FROM schema_meta")) { + if (!validating.size && sql === "BEGIN" && this.location() === databaseLocation) { + validating.add(this); + } + originalExec.call(this, sql); + if (!claimInjected && validating.has(this) && sql === "COMMIT") { claimInjected = true; const claimant = new DatabaseSync(databasePath); try { @@ -865,13 +867,12 @@ describe("external shared-state ownership", () => { claimant.close(); } } - return originalPrepare.call(this, sql); }); try { expect(() => openOpenClawStateDatabase({ env })).toThrow(OpenClawStateOwnershipError); } finally { - prepare.mockRestore(); + exec.mockRestore(); } expect(claimInjected).toBe(true); }); diff --git a/src/state/openclaw-state-worker-error.test.ts b/src/state/openclaw-state-worker-error.test.ts index b980268c72d3..c8de10b1eff3 100644 --- a/src/state/openclaw-state-worker-error.test.ts +++ b/src/state/openclaw-state-worker-error.test.ts @@ -303,6 +303,18 @@ describe("shared-state worker error transport", () => { reason: "state database schema migration", }, }, + { + error: new OpenClawStateDatabaseSchemaMigrationRequiredError( + "legacy-cron-run-logs", + "/fixture/state.sqlite", + ), + constructor: OpenClawStateDatabaseSchemaMigrationRequiredError, + fields: { + kind: "legacy-cron-run-logs", + pathname: "/fixture/state.sqlite", + reason: "cron run history migration", + }, + }, { error: new OpenClawAgentDatabaseMediaMigrationRequiredError("/fixture/agent.sqlite", 11), constructor: OpenClawAgentDatabaseMediaMigrationRequiredError, diff --git a/src/state/openclaw-state-worker-error.ts b/src/state/openclaw-state-worker-error.ts index 2988f942bb3d..992872fd8080 100644 --- a/src/state/openclaw-state-worker-error.ts +++ b/src/state/openclaw-state-worker-error.ts @@ -154,6 +154,7 @@ function isMaintenanceKind(kind: unknown): kind is MaintenanceKind { kind === "agent-media" || kind === "agent-databases-composite-primary-key" || kind === "audit-events-v2" || + kind === "legacy-cron-run-logs" || kind === "legacy-workshop-review-index" || kind === "legacy-workspace" || kind === "legacy-session-store" @@ -185,6 +186,7 @@ function parseIdentity(node: Record): ErrorIdentity | undefined case "state-migration": return (node.kind === "agent-databases-composite-primary-key" || node.kind === "audit-events-v2" || + node.kind === "legacy-cron-run-logs" || node.kind === "legacy-workshop-review-index") && typeof node.pathname === "string" ? { type: node.type, kind: node.kind, pathname: node.pathname } diff --git a/test/scripts/upgrade-survivor-cron-history.test.ts b/test/scripts/upgrade-survivor-cron-history.test.ts new file mode 100644 index 000000000000..b1d640c95a45 --- /dev/null +++ b/test/scripts/upgrade-survivor-cron-history.test.ts @@ -0,0 +1,278 @@ +import { execFileSync, spawnSync } from "node:child_process"; +import { mkdirSync, readdirSync, readFileSync, writeFileSync } from "node:fs"; +import path from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import { pathToFileURL } from "node:url"; +import { afterEach, expect, it } from "vitest"; +import { publishDiagnostics } from "../../scripts/e2e/lib/upgrade-survivor/diagnostics.mjs"; +import { + assertCronHistory, + seedCronHistory, +} from "../../scripts/e2e/lib/upgrade-survivor/legacy-operator-cron-history.mjs"; +import { migrateLegacyCronRunLogsToTaskRuns } from "../../src/infra/state-migrations.cron-run-logs.js"; +import { OPENCLAW_STATE_SCHEMA_SQL } from "../../src/state/openclaw-state-schema.js"; +import { useAutoCleanupTempDirTracker } from "../helpers/temp-dir.js"; + +const tempDirs = useAutoCleanupTempDirTracker(afterEach); +const helper = path.resolve("scripts/e2e/lib/upgrade-survivor/legacy-operator-cron-history.mjs"); +const diagnostics = path.resolve("scripts/e2e/lib/upgrade-survivor/diagnostics.mjs"); + +function prepare(version = "2026.9.4") { + const root = tempDirs.make("survivor-cron-history-"); + const stateDir = path.join(root, "state"); + const artifacts = path.join(root, "artifacts"); + const observations = path.join(artifacts, "observation"); + const baseline = path.join(root, "baseline"); + const candidate = path.join(root, "package"); + mkdirSync(path.join(stateDir, "state"), { recursive: true }); + mkdirSync(observations, { recursive: true }); + const schema = version === "2026.9.4" ? 17 : 16; + for (const [directory, packageVersion, stateSchema] of [ + [baseline, version, schema], + [candidate, "2026.9.4", 17], + ] as const) { + mkdirSync(path.join(directory, "dist"), { recursive: true }); + writeFileSync( + path.join(directory, "package.json"), + JSON.stringify({ + name: "openclaw", + version: packageVersion, + type: "module", + openclaw: { schemaVersions: { state: stateSchema } }, + }), + ); + writeFileSync( + path.join(directory, "dist/build-info.json"), + JSON.stringify({ fixture: directory }), + ); + } + const databasePath = path.join(stateDir, "state/openclaw.sqlite"); + const db = new DatabaseSync(databasePath); + db.exec(OPENCLAW_STATE_SCHEMA_SQL); + db.exec(`PRAGMA user_version = ${schema}`); + db.close(); + writeFileSync( + path.join(artifacts, "legacy-operator-baseline.json"), + JSON.stringify({ + jobs: [{ id: "retained-main" }, { id: "retained-ops" }], + }), + ); + const tarball = path.join(root, "candidate.tgz"); + execFileSync("tar", ["-czf", tarball, "-C", root, "package"]); + seedCronHistory(stateDir, artifacts, baseline, tarball); + const entry = path.join(candidate, "openclaw.mjs"); + // This process qualifies the observer, not an installed Doctor. The real + // published updater cell must exercise Doctor's admission and full ordering. + writeFileSync( + entry, + ` + import { DatabaseSync } from 'node:sqlite'; + import { migrateLegacyCronRunLogsToTaskRuns } from ${JSON.stringify(pathToFileURL(path.resolve("src/infra/state-migrations.cron-run-logs.ts")).href)}; + const db = new DatabaseSync(${JSON.stringify(databasePath)}); + db.exec('BEGIN IMMEDIATE'); + if (process.env.FIXTURE_IMPORT === '1') { + migrateLegacyCronRunLogsToTaskRuns(db); + } + db.exec('PRAGMA user_version = 17; COMMIT'); + db.close(); + `, + ); + const preloads = [ + "--import", + path.resolve("scripts/tsx.mjs"), + "--import", + diagnostics, + "--import", + helper, + ]; + const updater = path.join(baseline, "openclaw.mjs"); + writeFileSync( + updater, + ` + import { execFileSync } from 'node:child_process'; + import { DatabaseSync } from 'node:sqlite'; + import { migrateLegacyCronRunLogsToTaskRuns } from ${JSON.stringify(pathToFileURL(path.resolve("src/infra/state-migrations.cron-run-logs.ts")).href)}; + const db = new DatabaseSync(${JSON.stringify(databasePath)}); + db.exec('BEGIN IMMEDIATE'); + migrateLegacyCronRunLogsToTaskRuns(db); + if (process.env.FIXTURE_AFTER_IMPORT_SQL) { + db.exec(process.env.FIXTURE_AFTER_IMPORT_SQL); + } + db.exec('COMMIT'); + db.close(); + execFileSync(process.execPath, ${JSON.stringify([...preloads, entry, "doctor", "--fix", "--non-interactive"])}, { + env: { ...process.env, FIXTURE_IMPORT: '0', OPENCLAW_UPDATE_IN_PROGRESS: '1' }, + stdio: 'inherit', + }); + `, + ); + const invokeProcess = ( + file: string, + command: string, + importHistory: boolean, + afterImportSql?: string, + ) => + spawnSync( + process.execPath, + [ + ...preloads, + file, + command, + ...(command === "doctor" ? ["--fix", "--non-interactive"] : ["--yes", "--no-restart"]), + ], + { + encoding: "utf8", + env: { + ...process.env, + OPENCLAW_STATE_DIR: stateDir, + OPENCLAW_UPDATE_IN_PROGRESS: "1", + OPENCLAW_UPGRADE_SURVIVOR_ARTIFACT_ROOT: observations, + OPENCLAW_UPGRADE_SURVIVOR_CRON_HISTORY_FIXTURE: path.join( + artifacts, + "legacy-operator-cron-history.json", + ), + FIXTURE_IMPORT: importHistory ? "1" : "0", + FIXTURE_AFTER_IMPORT_SQL: afterImportSql, + }, + }, + ); + return { + artifacts, + observations, + databasePath, + stateDir, + invoke: (importHistory: boolean) => invokeProcess(entry, "doctor", importHistory), + invokeUpdater: (afterImportSql?: string) => + invokeProcess(updater, "update", false, afterImportSql), + }; +} + +it.each(["2026.9.3", "2026.9.4"])( + "proves %s retained-history ownership through the updater and Doctor", + (version) => { + const fixture = prepare(version); + const result = version === "2026.9.3" ? fixture.invokeUpdater() : fixture.invoke(true); + expect(result.status, result.stdout + result.stderr).toBe(0); + assertCronHistory(fixture.artifacts, fixture.observations); + const evidence = JSON.parse( + readFileSync(path.join(fixture.artifacts, "legacy-operator-cron-history-proof.json"), "utf8"), + ); + expect(evidence.currentSchemaAtDoctorEntry).toBe(version === "2026.9.4"); + expect(evidence.contract).toBe( + version === "2026.9.3" ? "published-updater-import-preserved" : "candidate-doctor-import", + ); + expect(evidence.doctor.before.legacyRows).toHaveLength(version === "2026.9.3" ? 0 : 2); + expect(evidence.doctor.after.tasks).toHaveLength(2); + if (version === "2026.9.3") { + expect(evidence.updater.before.legacyRows).toHaveLength(2); + expect(evidence.updater.before.tasks).toEqual([]); + expect(evidence.updater.identity).toEqual(evidence.baseline); + expect(evidence.doctor.before.tasks).toEqual(evidence.doctor.after.tasks); + writeFileSync( + path.join(fixture.artifacts, "summary.json"), + JSON.stringify({ + status: "passed", + baseline: { spec: `openclaw@${version}`, version }, + candidate: { kind: "tarball", version: "2026.9.4" }, + scenario: "legacy-operator-state", + installedVersion: "2026.9.4", + candidateInstallMode: "updater", + updateRestartMode: "manual", + updateOutcome: "success", + phases: [], + }), + ); + const published = path.join(fixture.artifacts, "published"); + publishDiagnostics(fixture.artifacts, published, (text: string) => text, "passed"); + const summary = JSON.parse(readFileSync(path.join(published, "summary.json"), "utf8")); + expect(JSON.parse(summary.logs["legacy-operator-cron-history-proof.json"])).toEqual(evidence); + } + }, +); + +it("rejects a successful Doctor exit which leaves retained history for startup", () => { + const fixture = prepare(); + const result = fixture.invoke(false); + expect(result.status, result.stdout + result.stderr).toBe(0); + expect(() => assertCronHistory(fixture.artifacts, fixture.observations)).toThrow( + "table was not retired", + ); +}); + +it("rejects parent-side import even when canonical task rows and Doctor exit are correct", () => { + const fixture = prepare(); + const db = new DatabaseSync(fixture.databasePath); + migrateLegacyCronRunLogsToTaskRuns(db); + db.close(); + const result = fixture.invoke(false); + expect(result.status, result.stdout + result.stderr).toBe(0); + const receiptName = readdirSync(fixture.observations).find((name) => + name.startsWith("cron-history-doctor-"), + ); + expect(receiptName).toBeDefined(); + const receipt = JSON.parse(readFileSync(path.join(fixture.observations, receiptName!), "utf8")); + for (let index = 1; index < 8; index++) { + writeFileSync( + path.join(fixture.observations, `cron-history-doctor-${receipt.pid + index}.json`), + JSON.stringify({ + ...receipt, + pid: receipt.pid + index, + startedAtMs: receipt.startedAtMs + index, + observationError: "rejected rehearsal observation ".repeat(40), + }), + ); + } + expect(() => assertCronHistory(fixture.artifacts, fixture.observations)).toThrow( + "never received the unchanged retained cron history", + ); + const evidence = JSON.parse( + readFileSync(path.join(fixture.artifacts, "legacy-operator-cron-history-proof.json"), "utf8"), + ); + expect(evidence.status).toBe("failed"); + expect(evidence.observations).toHaveLength(8); + expect(Buffer.byteLength(JSON.stringify(evidence, null, 2))).toBeLessThan(16 * 1024); + expect(evidence.observations[0]).toMatchObject({ + role: "doctor", + exitCode: 0, + before: { legacyRows: 0, tasks: 2 }, + after: { legacyRows: 0, tasks: 2 }, + }); + const captured = spawnSync( + process.execPath, + [diagnostics, "capture", fixture.artifacts, "assert-retained-cron-doctor", "1"], + { encoding: "utf8", env: { ...process.env, OPENCLAW_STATE_DIR: fixture.stateDir } }, + ); + expect(captured.status, captured.stderr).toBe(0); + const published = path.join(fixture.artifacts, "published"); + publishDiagnostics(fixture.artifacts, published, (text: string) => text); + const failure = JSON.parse(readFileSync(path.join(published, "failure.json"), "utf8")); + expect(JSON.parse(failure.logs["legacy-operator-cron-history-proof.json"])).toEqual(evidence); +}); + +it("requires the original seed at the published 9.3 updater entry", () => { + const fixture = prepare("2026.9.3"); + const result = fixture.invoke(true); + expect(result.status, result.stdout + result.stderr).toBe(0); + expect(() => assertCronHistory(fixture.artifacts, fixture.observations)).toThrow( + "published updater never received the unchanged retained cron history", + ); +}); + +it.each([ + ["task count", "DELETE FROM task_runs WHERE source_id = 'retained-ops'", "task count changed"], + [ + "task content", + "UPDATE task_runs SET terminal_summary = 'changed' WHERE source_id = 'retained-main'", + "retained cron task changed: terminal_summary", + ], + [ + "import report", + `UPDATE migration_runs SET report_json = '{"imported":1,"alreadyMirrored":0,"malformed":0,"skipped":false}' WHERE id = 'state:cron-run-logs-to-task-runs:v1'`, + "Expected values to be strictly deep-equal", + ], +])("rejects damaged %s from the 9.3 updater before Doctor", (_, sql, message) => { + const fixture = prepare("2026.9.3"); + const result = fixture.invokeUpdater(sql); + expect(result.status, result.stdout + result.stderr).toBe(0); + expect(() => assertCronHistory(fixture.artifacts, fixture.observations)).toThrow(message); +});