Fix SQLite concurrency and queue visibility

backend/src/chat/openClawChatSnapshotStore.ts

@@ -531,7 +531,7 @@ export class SqliteOpenClawChatSnapshotStore implements OpenClawChatSnapshotStor
                 .prepare("DELETE FROM chat_runtime_snapshots WHERE gateway_scope = ?")
                 .run(this.#gatewayScope);
         });
-        clearScope();
+        clearScope.immediate();
     }
 
     delete(sessionKey: string): void {
@@ -544,7 +544,7 @@ export class SqliteOpenClawChatSnapshotStore implements OpenClawChatSnapshotStor
                 )
                 .run(this.#gatewayScope, normalizedKey);
         });
-        deleteSnapshot();
+        deleteSnapshot.immediate();
     }
 
     keys(): string[] {
@@ -688,7 +688,7 @@ export class SqliteOpenClawChatSnapshotStore implements OpenClawChatSnapshotStor
             }
             this.#pruneSnapshots();
         });
-        persist();
+        persist.immediate();
     }
 
     save(sessionKey: string, snapshot: OpenClawRuntimeSnapshot): void {
@@ -699,6 +699,6 @@ export class SqliteOpenClawChatSnapshotStore implements OpenClawChatSnapshotStor
             this.#persistSnapshot(normalizedKey, snapshot, this.#now());
             this.#pruneSnapshots();
         });
-        persist();
+        persist.immediate();
     }
 }

backend/src/database.ts

@@ -442,6 +442,35 @@ function runSchemaSql(databaseConnection: DatabaseSync, schemaSql: string): void
     }
 }
 
+export function enableRequiredWalJournalMode(
+    databaseConnection: DatabaseSync,
+    databasePath: string
+): void {
+    let journalModeRow: { journal_mode?: unknown } | null;
+    try {
+        journalModeRow = databaseConnection.query("PRAGMA journal_mode = WAL").get() as {
+            journal_mode?: unknown;
+        } | null;
+    } catch (error) {
+        try {
+            databaseConnection.close();
+        } catch {
+            // Preserve the original SQLite error.
+        }
+        throw error;
+    }
+    const journalMode =
+        typeof journalModeRow?.journal_mode === "string"
+            ? journalModeRow.journal_mode
+            : undefined;
+    if (journalMode?.toLowerCase() !== "wal") {
+        databaseConnection.close();
+        throw new Error(
+            `SQLite WAL journal mode is required for ${databasePath}; got ${journalMode ?? "unknown"}`
+        );
+    }
+}
+
 function initializeDatabase(databasePath: string): DatabaseSync {
     const { configuredDatabasePath } = resolveDatabasePath();
     assertTestDatabasePath(databasePath, configuredDatabasePath);
@@ -451,6 +480,7 @@ function initializeDatabase(databasePath: string): DatabaseSync {
     const initializedDatabase = new Database(databasePath);
     initializedDatabase.run("PRAGMA foreign_keys = ON");
     initializedDatabase.run("PRAGMA busy_timeout = 5000");
+    enableRequiredWalJournalMode(initializedDatabase, databasePath);
     runSchemaSql(initializedDatabase, SCHEMA_SQL);
 
     return initializedDatabase;

backend/src/lib/jobResources.ts

@@ -98,6 +98,31 @@ export function withJobResourceClass<T>(
     return resourceContext.run({ resourceClass }, operation);
 }
 
+function isSystemdRunExecutable(executable: string): boolean {
+    return executable === "systemd-run" || executable.endsWith("/systemd-run");
+}
+
+/** Keeps the user-bus variables needed by a scoped launcher without widening child env. */
+export function scopedJobProcessEnvironment(
+    executable: string,
+    environment: Record<string, string | undefined> | undefined,
+    inheritedEnvironment: Record<string, string | undefined> = process.env
+): Record<string, string | undefined> | undefined {
+    if (environment === undefined || !isSystemdRunExecutable(executable)) {
+        return environment;
+    }
+    const busAddress =
+        environment.DBUS_SESSION_BUS_ADDRESS ||
+        inheritedEnvironment.DBUS_SESSION_BUS_ADDRESS;
+    const runtimeDirectory =
+        environment.XDG_RUNTIME_DIR || inheritedEnvironment.XDG_RUNTIME_DIR;
+    return {
+        ...environment,
+        ...(busAddress && { DBUS_SESSION_BUS_ADDRESS: busAddress }),
+        ...(runtimeDirectory && { XDG_RUNTIME_DIR: runtimeDirectory }),
+    };
+}
+
 function scopeOwnerProperties(environment: Record<string, string | undefined>): string[] {
     const owner = environment.MIRA_DASHBOARD_JOB_SCOPE_OWNER?.trim();
     if (!owner || !/^[A-Za-z0-9_.@-]+\.service$/u.test(owner)) return [];
@@ -113,9 +138,8 @@ export function scopedJobProcessCommand(
     const context = resourceContext.getStore();
     if (
         !context ||
-        executable === "systemd-run" ||
-        environment.MIRA_DASHBOARD_ENABLE_JOB_SCOPES !== "1" ||
-        executable.endsWith("/systemd-run")
+        isSystemdRunExecutable(executable) ||
+        environment.MIRA_DASHBOARD_ENABLE_JOB_SCOPES !== "1"
     ) {
         return { arguments: [...arguments_], executable };
     }

backend/src/lib/processes.ts

@@ -1,4 +1,4 @@
-import { scopedJobProcessCommand } from "./jobResources.ts";
+import { scopedJobProcessCommand, scopedJobProcessEnvironment } from "./jobResources.ts";
 
 export interface RunProcessOptions {
     cwd?: string;
@@ -63,7 +63,7 @@ export function spawnProcess(
         cmd: [command.executable, ...command.arguments],
         cwd: options.cwd,
         detached: options.detached ?? true,
-        env: options.env,
+        env: scopedJobProcessEnvironment(command.executable, options.env),
         stderr: "pipe",
         stdin: "ignore",
         stdout: "pipe",

backend/src/routes/cacheRoutes.ts

@@ -8,14 +8,25 @@ import {
 } from "../lib/cacheStore.ts";
 import { errorMessage, httpStatusCode } from "../lib/errors.ts";
 import { stringFallback } from "../lib/values.ts";
-import { cacheRefreshResourceClass } from "../services/cacheRefresh.ts";
+import {
+    cacheRefreshResourceClass,
+    cacheRefreshScheduledJobId,
+} from "../services/cacheRefresh.ts";
+import { getLatestScheduledJobExecution } from "../services/jobExecutionQueue.ts";
 import {
     enqueueAndWaitForJobExecution,
     successfulJobExecutionOutput,
+    waitForJobExecution,
 } from "../services/queuedJobExecution.ts";
-import { listScheduledJobs } from "../services/scheduledJobs.ts";
+import {
+    enqueueScheduledJob,
+    getScheduledJob,
+    listScheduledJobs,
+} from "../services/scheduledJobs.ts";
 import { getHeartbeatAutomationSnapshot } from "../services/taskAutomation.ts";
 
+const CACHE_REFRESH_TIMEOUT_MS = 5 * 60 * 1000;
+
 function parseJsonFieldOrValue(value: string) {
     const parsed = parseJsonField<unknown>(value);
     return parsed ?? value;
@@ -254,6 +265,71 @@ async function refreshedCacheEntry(key: string, result: Record<string, unknown>)
     return mapCacheRowForResponse(row);
 }
 
+async function enqueueAndWaitForCacheRefresh(
+    key: string,
+    resourceClass: ReturnType<typeof cacheRefreshResourceClass>,
+    signal: AbortSignal
+) {
+    const enqueueUnscheduledRefresh = async () =>
+        await enqueueAndWaitForJobExecution(
+            {
+                actionKey: "cache.refresh",
+                displayName: `Refresh cache: ${key}`,
+                payload: { key },
+                resourceClass,
+                timeoutMs: CACHE_REFRESH_TIMEOUT_MS,
+            },
+            { signal }
+        );
+    const scheduledJobId = cacheRefreshScheduledJobId(key);
+    if (!scheduledJobId) {
+        return await enqueueUnscheduledRefresh();
+    }
+    let shouldCancelQueuedOnTimeout = true;
+    let executionId: string | undefined;
+    try {
+        executionId = enqueueScheduledJob(scheduledJobId, "manual").executionId;
+    } catch (error) {
+        const statusCode = httpStatusCode(error);
+        if (statusCode === 404) {
+            return await enqueueUnscheduledRefresh();
+        }
+        if (statusCode !== 409) throw error;
+        const existingExecution = getLatestScheduledJobExecution(scheduledJobId);
+        if (!existingExecution) throw error;
+        const scheduledJob = getScheduledJob(scheduledJobId);
+        const canReuseExecution =
+            existingExecution.status === "running" ||
+            (existingExecution.status === "queued" &&
+                scheduledJob !== undefined &&
+                (existingExecution.triggerType === "manual" || scheduledJob.enabled));
+        if (!canReuseExecution) {
+            return await enqueueUnscheduledRefresh();
+        }
+        shouldCancelQueuedOnTimeout = false;
+        executionId = existingExecution.id;
+    }
+    if (!executionId) {
+        throw Object.assign(new Error("Scheduled cache refresh was not queued"), {
+            statusCode: 500,
+        });
+    }
+    const execution = await waitForJobExecution(executionId, {
+        cancelQueuedOnTimeout: shouldCancelQueuedOnTimeout,
+        signal,
+        timeoutMs: CACHE_REFRESH_TIMEOUT_MS,
+    });
+    if (
+        !shouldCancelQueuedOnTimeout &&
+        execution.status === "cancelled" &&
+        (execution.message === "Scheduled job was disabled before execution" ||
+            execution.message === "Scheduled job was removed before execution")
+    ) {
+        return await enqueueUnscheduledRefresh();
+    }
+    return execution;
+}
+
 type ParametersRequest<T extends string> = Request & { params: Record<T, string> };
 
 export const cacheRoutes = {
@@ -312,15 +388,10 @@ export const cacheRoutes = {
             if (!key) return json({ error: "Missing cache key" }, { status: 400 });
             try {
                 const resourceClass = cacheRefreshResourceClass(key);
-                const execution = await enqueueAndWaitForJobExecution(
-                    {
-                        actionKey: "cache.refresh",
-                        displayName: `Refresh cache: ${key}`,
-                        payload: { key },
-                        resourceClass,
-                        timeoutMs: 5 * 60 * 1000,
-                    },
-                    { signal: request.signal }
+                const execution = await enqueueAndWaitForCacheRefresh(
+                    key,
+                    resourceClass,
+                    request.signal
                 );
                 const entry = await refreshedCacheEntry(
                     key,

backend/src/services/backups.ts

@@ -1092,7 +1092,7 @@ export function registerBackupScheduledJobs(): void {
             return { backup: await clearBackupAttention(type, backupExecutionId) };
         }
     );
-    database.run("BEGIN");
+    database.run("BEGIN IMMEDIATE");
     try {
         removeScheduledJobsNotInAction(
             "backup.run",

backend/src/services/cacheRefresh.ts

@@ -2270,6 +2270,12 @@ const cacheRefreshScheduledJobs = [
     },
 ] as const;
 
+export function cacheRefreshScheduledJobId(key: string): string | undefined {
+    const scheduledKey = cacheRefreshScopeKey(key);
+    return cacheRefreshScheduledJobs.find((job) => job.actionPayload.key === scheduledKey)
+        ?.id;
+}
+
 function getScheduledCacheKey(job: ScheduledJob): string {
     const key = job.actionPayload.key;
     if (typeof key !== "string" || key.trim() === "") {
@@ -2380,7 +2386,7 @@ export function registerCacheRefreshScheduledJobs(
         return { key, ...result };
     });
     const seedJobs: Array<{ id: string; key: string }> = [];
-    database.run("BEGIN");
+    database.run("BEGIN IMMEDIATE");
     try {
         removeScheduledJobsNotInAction(
             "cache.refresh",

backend/src/services/dockerUpdater.ts

@@ -2066,7 +2066,7 @@ export async function registerDockerUpdaterServices(
     let isTxnStarted = false;
     try {
         signal?.throwIfAborted();
-        database.run("BEGIN");
+        database.run("BEGIN IMMEDIATE");
         isTxnStarted = true;
         const discoveredAppSlugs = new Set(
             successfulOrPartialDiscoveries.map((item) => item.appSlug)
@@ -2803,7 +2803,7 @@ export function registerDockerUpdaterScheduledJobs(): void {
         },
         { timeoutMs: 30 * 60 * 1000 }
     );
-    database.run("BEGIN");
+    database.run("BEGIN IMMEDIATE");
     try {
         removeScheduledJobsNotInAction("docker.updater", [job.id]);
         const existing = getScheduledJob(job.id);

backend/src/services/gitHygiene.ts

@@ -507,7 +507,7 @@ export function registerGitHygieneScheduledJobs(): void {
         },
         { timeoutMs: GIT_WORKSPACE_SYNC_TIMEOUT_MS }
     );
-    database.run("BEGIN");
+    database.run("BEGIN IMMEDIATE");
     try {
         removeScheduledJobsNotInAction("git.openclaw.workspace-sync", [job.id]);
         const existing = getScheduledJob(job.id);

backend/src/services/logRotation.ts

@@ -1984,7 +1984,7 @@ export function registerLogRotationScheduledJobs(): void {
         }
         return { logRotation };
     });
-    database.run("BEGIN");
+    database.run("BEGIN IMMEDIATE");
     try {
         removeScheduledJobsNotInAction(LOG_ROTATION_JOB_ID, [LOG_ROTATION_JOB_ID]);
         const existing = getScheduledJob(LOG_ROTATION_JOB_ID);

backend/src/services/queuedJobExecution.ts

@@ -11,6 +11,7 @@ const DEFAULT_POLL_INTERVAL_MS = 100;
 const DEFAULT_WAIT_TIMEOUT_MS = 30 * 60 * 1000;
 
 export interface WaitForJobExecutionOptions {
+    cancelQueuedOnTimeout?: boolean;
     pollIntervalMs?: number;
     signal?: AbortSignal;
     timeoutMs?: number;
@@ -55,7 +56,11 @@ export async function waitForJobExecution(
         }
         if (isTerminalJobExecution(execution)) return execution;
         if (Date.now() - startedAt >= timeoutMs) {
-            if (execution.status === "queued" && execution.cancellable) {
+            if (
+                options.cancelQueuedOnTimeout !== false &&
+                execution.status === "queued" &&
+                execution.cancellable
+            ) {
                 try {
                     cancelJobExecution(id);
                 } catch (error) {