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,25 @@ function runSchemaSql(databaseConnection: DatabaseSync, schemaSql: string): void
     }
 }
 
+export function enableRequiredWalJournalMode(
+    databaseConnection: DatabaseSync,
+    databasePath: string
+): void {
+    const journalModeRow = databaseConnection
+        .query("PRAGMA journal_mode = WAL")
+        .get() as { journal_mode?: unknown } | null;
+    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 +470,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,20 @@ 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 {
     enqueueAndWaitForJobExecution,
     successfulJobExecutionOutput,
+    waitForJobExecution,
 } from "../services/queuedJobExecution.ts";
-import { listScheduledJobs } from "../services/scheduledJobs.ts";
+import { enqueueScheduledJob, 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 +260,36 @@ async function refreshedCacheEntry(key: string, result: Record<string, unknown>)
     return mapCacheRowForResponse(row);
 }
 
+async function enqueueAndWaitForCacheRefresh(
+    key: string,
+    resourceClass: ReturnType<typeof cacheRefreshResourceClass>,
+    signal: AbortSignal
+) {
+    const scheduledJobId = cacheRefreshScheduledJobId(key);
+    if (!scheduledJobId) {
+        return await enqueueAndWaitForJobExecution(
+            {
+                actionKey: "cache.refresh",
+                displayName: `Refresh cache: ${key}`,
+                payload: { key },
+                resourceClass,
+                timeoutMs: CACHE_REFRESH_TIMEOUT_MS,
+            },
+            { signal }
+        );
+    }
+    const run = enqueueScheduledJob(scheduledJobId, "manual");
+    if (!run.executionId) {
+        throw Object.assign(new Error("Scheduled cache refresh was not queued"), {
+            statusCode: 500,
+        });
+    }
+    return await waitForJobExecution(run.executionId, {
+        signal,
+        timeoutMs: CACHE_REFRESH_TIMEOUT_MS,
+    });
+}
+
 type ParametersRequest<T extends string> = Request & { params: Record<T, string> };
 
 export const cacheRoutes = {
@@ -312,15 +348,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,14 @@ const cacheRefreshScheduledJobs = [
     },
 ] as const;
 
+export function cacheRefreshScheduledJobId(key: string): string | undefined {
+    const scheduledKey = MOLTBOOK_CACHE_KEYS.has(key)
+        ? "moltbook"
+        : 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 +2388,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/test/jobExecutionQueue.test.ts

@@ -3,6 +3,7 @@ import { afterEach, describe, expect, it } from "bun:test";
 import { database } from "../src/database.ts";
 import {
     scopedJobProcessCommand,
+    scopedJobProcessEnvironment,
     withJobResourceClass,
 } from "../src/lib/jobResources.ts";
 import {
@@ -172,6 +173,46 @@ describe("persistent job execution queue", () => {
         );
     });
 
+    it("preserves only user-bus variables for scoped children with restricted env", () => {
+        const restrictedEnvironment = { PATH: "/usr/bin" };
+        const inheritedEnvironment = {
+            DBUS_SESSION_BUS_ADDRESS: "unix:path=/run/user/1000/bus",
+            INTERNAL_SECRET: "must-not-leak",
+            XDG_RUNTIME_DIR: "/run/user/1000",
+        };
+
+        const expectedEnvironment = {
+            DBUS_SESSION_BUS_ADDRESS: "unix:path=/run/user/1000/bus",
+            PATH: "/usr/bin",
+            XDG_RUNTIME_DIR: "/run/user/1000",
+        };
+
+        expect(
+            scopedJobProcessEnvironment(
+                "systemd-run",
+                restrictedEnvironment,
+                inheritedEnvironment
+            )
+        ).toEqual(expectedEnvironment);
+        expect(
+            scopedJobProcessEnvironment(
+                "/usr/bin/systemd-run",
+                restrictedEnvironment,
+                inheritedEnvironment
+            )
+        ).toEqual(expectedEnvironment);
+        expect(
+            scopedJobProcessEnvironment(
+                "bash",
+                restrictedEnvironment,
+                inheritedEnvironment
+            )
+        ).toBe(restrictedEnvironment);
+        expect(
+            scopedJobProcessEnvironment("systemd-run", undefined, inheritedEnvironment)
+        ).toBeUndefined();
+    });
+
     it("allows queued cancellation but protects a running mutation", () => {
         const queued = enqueueJobExecution({
             actionKey: `test.protected-${Bun.randomUUIDv7()}`,