fix(upgrade): retain owners in legacy cron stores (#124809)

Amp-Thread-ID: https://ampcode.com/threads/T-01a00a6a-b64e-74a5-8b15-2d3b966a468d

Co-authored-by: Amp <amp@ampcode.com>

src/commands/doctor/shared/default-agent-role-materialization.write.test.ts

@@ -466,6 +466,66 @@ describe("default role materialization authored writes", () => {
     await expect(fs.readFile(configPath, "utf8")).resolves.toBe(firstPersisted);
   });
 
+  it("assigns ownerless jobs in an unmigrated legacy cron file", async () => {
+    const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-legacy-json-cron-owner-"));
+    roots.push(root);
+    const configPath = path.join(root, "openclaw.json");
+    const storePath = path.join(root, "cron", "jobs.json");
+    const env = {
+      HOME: root,
+      OPENCLAW_STATE_DIR: root,
+      OPENCLAW_TEST_FAST: "1",
+    } as NodeJS.ProcessEnv;
+    await fs.mkdir(path.dirname(storePath), { recursive: true });
+    await fs.writeFile(
+      configPath,
+      JSON.stringify({ agents: { list: [{ id: "ops", default: true }, { id: "research" }] } }),
+    );
+    await fs.writeFile(
+      storePath,
+      JSON.stringify({
+        version: 1,
+        jobs: [
+          makeCronJob({ id: "ownerless" }),
+          makeCronJob({ id: "owned", agentId: "research" }),
+          { ...makeCronJob({ id: "session-owned" }), sessionKey: "agent:research:main" },
+        ],
+      }),
+    );
+    writeConfigMachineState("cron.store", storePath, { env });
+    const io = createConfigIO({
+      configPath,
+      env,
+      homedir: () => root,
+      observe: false,
+      logger: { warn: () => {}, error: () => {} },
+    });
+    const snapshot = await io.readConfigFileSnapshot();
+    const nextConfig: OpenClawConfig = {
+      ...snapshot.config,
+      agents: { ...snapshot.config.agents, ownership: "explicit" },
+    };
+
+    await io.writeConfigFile(nextConfig, {
+      baseSnapshot: snapshot,
+      explicitSetPaths: [["agents", "ownership"]],
+      explicitSetValueSource: nextConfig,
+    });
+
+    const persistedConfig = JSON.parse(await fs.readFile(configPath, "utf8"));
+    expect(persistedConfig.agents).toMatchObject({
+      ownership: "explicit",
+      entries: { ops: {}, research: {} },
+    });
+    const persistedStore = JSON.parse(await fs.readFile(storePath, "utf8"));
+    expect(persistedStore.jobs).toMatchObject([
+      { id: "ownerless", agentId: "ops" },
+      { id: "owned", agentId: "research" },
+      { id: "session-owned", sessionKey: "agent:research:main" },
+    ]);
+    expect(persistedStore.jobs[2]).not.toHaveProperty("agentId");
+  });
+
   it("leaves the legacy owner marker intact when a cron row is corrupt", async () => {
     const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-corrupt-cron-owner-write-"));
     roots.push(root);

src/config/io.cron-owner-refusal.test.ts

@@ -31,7 +31,7 @@ const deps = (
     activeGateway ? { ...activeGateway, createdAt: new Date(0).toISOString() } : undefined,
   ),
   loadLegacyCronRepairState: vi.fn(async () => (jobs ? state(jobs) : null)),
-  materializeLegacyDefaultCronJobOwners: vi.fn(() => 0),
+  materializeLegacyDefaultCronJobOwners: vi.fn(async () => 0),
 });
 const cfg = { agents: { entries: { ops: {} } } };
 
@@ -124,7 +124,7 @@ it("materializes a proven retained owner before the commit recheck", async () =>
       ]),
     ),
   );
-  injected.materializeLegacyDefaultCronJobOwners.mockImplementationOnce(() => {
+  injected.materializeLegacyDefaultCronJobOwners.mockImplementationOnce(async () => {
     injected.loadLegacyCronRepairState.mockResolvedValue(
       state([
         { id: "ownerless", agentId: "ops" },
@@ -161,7 +161,7 @@ it("keeps ambiguous and failed owner handoffs as typed refusals", async () => {
   expect(ambiguous.materializeLegacyDefaultCronJobOwners).not.toHaveBeenCalled();
 
   const failed = deps(undefined, [{ id: "ownerless" }]);
-  failed.materializeLegacyDefaultCronJobOwners.mockImplementationOnce(() => {
+  failed.materializeLegacyDefaultCronJobOwners.mockImplementationOnce(async () => {
     throw new Error("database is temporarily read-only");
   });
   const failure = prepareCronOwnerWriteRefusal(

src/config/io.cron-owner-refusal.ts

@@ -104,7 +104,7 @@ async function assertSafe(
   }
   if ((unresolved > 0 || projectedDynamicDefaults > 0) && provenOwnerAgentId) {
     try {
-      deps.materializeLegacyDefaultCronJobOwners({
+      await deps.materializeLegacyDefaultCronJobOwners({
         storePath,
         legacyDefaultAgentId: provenOwnerAgentId,
         env,

src/cron/legacy-default-agent-owner-migration.test.ts

@@ -33,13 +33,13 @@ function fixture(label: string) {
   return { env, storePath, storeKey, database };
 }
 
-it("preserves undecodable JSON and bumps the epoch once", () => {
+it("preserves undecodable JSON and bumps the epoch once", async () => {
   const { env, storePath, storeKey, database } = fixture("openclaw-cron-owner-");
   database
     .prepare("UPDATE cron_jobs SET agent_id = ' ', job_json = ? WHERE store_key = ?")
     .run("{malformed", storeKey);
 
-  expect(migrate(storePath, env)).toBe(1);
+  expect(await migrate(storePath, env)).toBe(1);
   expect(loadCronRows(database, storeKey)[0]).toMatchObject({
     agent_id: "ops",
     job_json: "{malformed",
@@ -53,7 +53,7 @@ it("preserves undecodable JSON and bumps the epoch once", () => {
   ).toBe(1);
 });
 
-it("preserves a session-scoped owner stored only in job JSON", () => {
+it("preserves a session-scoped owner stored only in job JSON", async () => {
   const { env, storePath, storeKey, database } = fixture("openclaw-cron-json-owner-");
   const row = loadCronRows(database, storeKey)[0];
   const jobJson = JSON.parse(row?.job_json ?? "{}") as Record<string, unknown>;
@@ -65,7 +65,7 @@ it("preserves a session-scoped owner stored only in job JSON", () => {
     )
     .run(JSON.stringify(jobJson), storeKey);
 
-  expect(migrate(storePath, env)).toBe(0);
+  expect(await migrate(storePath, env)).toBe(0);
   const preserved = loadCronRows(database, storeKey)[0];
   const preservedJobJson = JSON.parse(preserved?.job_json ?? "{}") as Record<string, unknown>;
   expect(preserved?.agent_id).toBeNull();
@@ -75,13 +75,13 @@ it("preserves a session-scoped owner stored only in job JSON", () => {
   expect(preservedJobJson).not.toHaveProperty("agentId");
 });
 
-it("rolls back the row when the epoch bump fails", () => {
+it("rolls back the row when the epoch bump fails", async () => {
   const { env, storePath, storeKey, database } = fixture("openclaw-cron-atomic-");
   ensureCronStoreEpochSchema(database);
   database.exec(`CREATE TRIGGER fail_epoch BEFORE UPDATE OF store_epoch ON cron_store_epochs
       BEGIN SELECT RAISE(ABORT, 'synthetic epoch failure'); END`);
 
-  expect(() => migrate(storePath, env)).toThrow("synthetic epoch failure");
+  await expect(migrate(storePath, env)).rejects.toThrow("synthetic epoch failure");
   expect(loadCronRows(database, storeKey)[0]?.agent_id).toBeNull();
 });
 

src/cron/legacy-default-agent-owner-migration.ts

@@ -1,19 +1,70 @@
+import fs from "node:fs/promises";
 import path from "node:path";
-import { normalizeAgentId } from "../routing/session-key.js";
+import { isRecord } from "@openclaw/normalization-core/record-coerce";
+import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
+import { isMissingPathError } from "../infra/errors.js";
+import { writeTextAtomic } from "../infra/json-files.js";
+import { normalizeAgentId, parseAgentSessionKey } from "../routing/session-key.js";
 import { runOpenClawStateWriteTransaction } from "../state/openclaw-state-db.js";
+import { parseJsonWithJson5Fallback } from "../utils/parse-json-compat.js";
 import { cronStoreKey } from "./store/key.js";
 import { materializeCronRowAgentOwners } from "./store/row-codec.js";
 
-export function materializeLegacyDefaultCronJobOwners(params: {
+async function materializeLegacyJsonOwners(storePath: string, agentId: string): Promise<number> {
+  let raw: string;
+  try {
+    raw = await fs.readFile(storePath, "utf8");
+  } catch (error) {
+    if (isMissingPathError(error)) {
+      return 0;
+    }
+    throw error;
+  }
+  const parsed = parseJsonWithJson5Fallback(raw);
+  const jobs = Array.isArray(parsed)
+    ? parsed
+    : isRecord(parsed) && Array.isArray(parsed.jobs)
+      ? parsed.jobs
+      : [];
+  let rewritten = 0;
+  for (const job of jobs) {
+    if (
+      !isRecord(job) ||
+      normalizeOptionalString(job.agentId) ||
+      parseAgentSessionKey(normalizeOptionalString(job.sessionKey))?.agentId
+    ) {
+      continue;
+    }
+    job.agentId = agentId;
+    rewritten += 1;
+  }
+  if (rewritten === 0) {
+    return 0;
+  }
+  await writeTextAtomic(storePath, JSON.stringify(parsed, null, 2), {
+    mode: 0o600,
+    tempPrefix: path.basename(storePath),
+    trailingNewline: true,
+    beforeRename: async () => {
+      if ((await fs.readFile(storePath, "utf8")) !== raw) {
+        throw new Error("legacy cron source changed while assigning its retained owner");
+      }
+    },
+  });
+  return rewritten;
+}
+
+export async function materializeLegacyDefaultCronJobOwners(params: {
   storePath: string;
   legacyDefaultAgentId: string;
   env?: NodeJS.ProcessEnv;
-}): number {
+}): Promise<number> {
   const agentId = normalizeAgentId(params.legacyDefaultAgentId);
-  return runOpenClawStateWriteTransaction(
-    ({ db }) =>
-      materializeCronRowAgentOwners(db, cronStoreKey(path.resolve(params.storePath)), agentId),
+  const storePath = path.resolve(params.storePath);
+  const sqliteCount = runOpenClawStateWriteTransaction(
+    ({ db }) => materializeCronRowAgentOwners(db, cronStoreKey(storePath), agentId),
     { env: params.env },
     { operationLabel: "cron.legacy-default-owner" },
   );
+  return sqliteCount + (await materializeLegacyJsonOwners(storePath, agentId));
 }

src/cron/service/ops-lifecycle.ts

@@ -134,7 +134,7 @@ export async function start(state: CronServiceState): Promise<void> {
       return;
     }
     if (state.deps.legacyDefaultAgentId) {
-      const rewritten = materializeLegacyDefaultCronJobOwners({
+      const rewritten = await materializeLegacyDefaultCronJobOwners({
         storePath: state.deps.storePath,
         legacyDefaultAgentId: state.deps.legacyDefaultAgentId,
       });