mirror of
https://github.com/openclaw/openclaw.git
synced 2026-10-03 09:39:25 +00:00
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
This commit is contained in:
Peter Steinberger
GitHub
parent
bd3633bdf8
commit
d314668148
32 changed files with 1467 additions and 360 deletions
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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]);
|
||||
}
|
||||
}
|
||||
|
|
@ -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 \
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}>;
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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 = ?")
|
||||
|
|
|
|||
|
|
@ -3037,7 +3037,7 @@ export async function prepareLegacyStateDatabaseSchema(
|
|||
createStateSchemaMigrationStep({
|
||||
stateDir: resolveStateDir(env),
|
||||
env,
|
||||
mode: "automatic",
|
||||
mode: "doctor",
|
||||
requiredness: "conditional",
|
||||
}),
|
||||
]);
|
||||
|
|
|
|||
|
|
@ -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<typeof listPluginDoctorStateMigrationEntries>,
|
||||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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];
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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<T>(pathname: string, operation: () => T): T {
|
||||
const open = nodeSqlite.openNodeSqliteDatabase;
|
||||
const opened = new Set<DatabaseSync>();
|
||||
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);
|
||||
}
|
||||
});
|
||||
},
|
||||
);
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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<DatabaseSync>();
|
||||
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<DatabaseSync>();
|
||||
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);
|
||||
},
|
||||
);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
});
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 = ?")
|
||||
|
|
|
|||
|
|
@ -543,45 +543,57 @@ export function runStateSchemaMigrationTransaction<T>(
|
|||
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 {
|
||||
|
|
|
|||
|
|
@ -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}`],
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
161
src/state/openclaw-state-db-schema-runtime.ts
Normal file
161
src/state/openclaw-state-db-schema-runtime.ts
Normal file
|
|
@ -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;
|
||||
});
|
||||
}
|
||||
|
|
@ -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();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<DatabaseSync>();
|
||||
|
||||
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<T>(
|
||||
callback: (db: DatabaseSync) => T,
|
||||
|
|
|
|||
|
|
@ -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<import("node:sqlite").DatabaseSync>();
|
||||
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);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>): 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 }
|
||||
|
|
|
|||
278
test/scripts/upgrade-survivor-cron-history.test.ts
Normal file
278
test/scripts/upgrade-survivor-cron-history.test.ts
Normal file
|
|
@ -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);
|
||||
});
|
||||
Loading…
Add table
Add a link
Reference in a new issue