diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256 index f2a20feef6ea..7626ccb571ac 100644 --- a/docs/.generated/plugin-sdk-api-baseline.sha256 +++ b/docs/.generated/plugin-sdk-api-baseline.sha256 @@ -94,7 +94,7 @@ d340686cf814326b5a554a32cc5ca324a9e1254a933c4d9d8d1ada5ac7109a10 module/command a821f9cc4e6f9339399d99e73f58c3b00baf139aa9d85c22d73c89d8be5702d2 module/config-contracts 20f3f8042de53e4eee61b64de9102c8c202b9299e6a29235647a4729f70145f2 module/config-mutation 316949815affe623ac63951a5db580527f02663576dffa084f02768f612c0c1c module/config-runtime -446d5331d966563f4aec0b50bb7dfe0c81717b4d6d22b281ffac935361f7b34d module/config-schema +2efea5ed295e37932c69f907b8e197ef237d757c1e9accd3d3809c3696428a55 module/config-schema bfe6eec12f45bc2fda6df28699681b22dfd819a2261aabe8e0c789a20a017058 module/config-types 42d15153981cfe3adc1d5f91621434c56f07a9bd48c15ce34742f72dd040c142 module/context-visibility-runtime 03636897fb99cb73e4d8620c8a0e0d72b4d52fc32bf94f525af2aa88c489c6c2 module/conversation-binding-runtime diff --git a/src/plugins/contracts/scheduled-turns.contract.test.ts b/src/plugins/contracts/scheduled-turns.contract.test.ts index 30e0e12261c8..7956585fb9b0 100644 --- a/src/plugins/contracts/scheduled-turns.contract.test.ts +++ b/src/plugins/contracts/scheduled-turns.contract.test.ts @@ -25,7 +25,16 @@ import { import { loadOpenClawPlugins } from "../loader.js"; import { clearPluginLoaderCache, makeTempDir, writePlugin } from "../loader.test-fixtures.js"; import { createEmptyPluginRegistry } from "../registry-empty.js"; -import { setActivePluginRegistry } from "../runtime.js"; +import { + pinActivePluginChannelRegistry, + pinActivePluginHttpRouteRegistry, + pinActivePluginSessionExtensionRegistry, + releasePinnedPluginChannelRegistry, + releasePinnedPluginHttpRouteRegistry, + releasePinnedPluginSessionExtensionRegistry, + resetPluginRuntimeStateForTest, + setActivePluginRegistry, +} from "../runtime.js"; import { createPluginRecord } from "../status.test-helpers.js"; import type { OpenClawPluginApi } from "../types.js"; @@ -210,8 +219,11 @@ describe("plugin scheduled turns", () => { afterEach(() => { vi.useRealTimers(); clearPluginLoaderCache(); + releasePinnedPluginChannelRegistry(); + releasePinnedPluginHttpRouteRegistry(); + releasePinnedPluginSessionExtensionRegistry(); clearPluginHostRuntimeState(); - setActivePluginRegistry(createEmptyPluginRegistry()); + resetPluginRuntimeStateForTest(); }); it("schedules session turns with cron-compatible tagged cleanup metadata", async () => { @@ -853,6 +865,64 @@ describe("plugin scheduled turns", () => { ]); }); + it("cleans only dynamic scheduled turns owned by the retiring registry", async () => { + const removed: string[] = []; + const scheduledIds = ["gateway-owned-job", "retiring-owned-job"]; + workflowMocks.cronAdd.mockImplementation(async () => + makeCronJob({ id: scheduledIds.shift() ?? "unexpected-job" }), + ); + workflowMocks.cronRemove.mockImplementation(async (id: string) => { + removed.push(id); + return { ok: true, removed: true }; + }); + + const gatewayFixture = createPluginRegistryFixture(); + gatewayFixture.registry.registry.plugins.push( + createPluginRecord({ id: WORKFLOW_PLUGIN_ID, origin: "bundled" }), + ); + const retiringFixture = createPluginRegistryFixture(); + retiringFixture.registry.registry.plugins.push( + createPluginRecord({ id: WORKFLOW_PLUGIN_ID, origin: "bundled" }), + ); + const replacementFixture = createPluginRegistryFixture(); + + await scheduleWorkflowTurn({ + ownerRegistry: gatewayFixture.registry.registry, + schedule: { cron: "* * * * *", tz: "UTC" }, + }); + await scheduleWorkflowTurn({ + ownerRegistry: retiringFixture.registry.registry, + schedule: { cron: "* * * * *", tz: "UTC" }, + }); + + await expect( + cleanupReplacedPluginHostRegistry({ + cfg: retiringFixture.config, + previousRegistry: retiringFixture.registry.registry, + nextRegistry: replacementFixture.registry.registry, + }), + ).resolves.toMatchObject({ failures: [] }); + expect(removed).toEqual(["retiring-owned-job"]); + expect(listPluginSessionSchedulerJobs(WORKFLOW_PLUGIN_ID)).toEqual([ + { + id: "gateway-owned-job", + pluginId: WORKFLOW_PLUGIN_ID, + sessionKey: MAIN_SESSION_KEY, + kind: "session-turn", + }, + ]); + + await expect( + cleanupReplacedPluginHostRegistry({ + cfg: gatewayFixture.config, + previousRegistry: gatewayFixture.registry.registry, + nextRegistry: replacementFixture.registry.registry, + }), + ).resolves.toMatchObject({ failures: [] }); + expect(removed).toEqual(["retiring-owned-job", "gateway-owned-job"]); + expect(listPluginSessionSchedulerJobs(WORKFLOW_PLUGIN_ID)).toEqual([]); + }); + it("treats already-missing cron jobs as successful scheduled-turn cleanup", async () => { const removed: string[] = []; workflowMocks.cronAdd.mockResolvedValue(makeCronJob({ id: "already-missing-job" })); @@ -1044,8 +1114,11 @@ describe("plugin scheduled turns", () => { expect(workflowMocks.cronRemove).not.toHaveBeenCalled(); }); - it("wires schedule and unschedule through the plugin API with stale-registry protection", async () => { - workflowMocks.cronAdd.mockResolvedValue(makeCronJob({ id: "job-live" })); + it("keeps pinned scheduled-turn APIs live until their registry retires", async () => { + const scheduledIds = ["job-live", "job-pinned"]; + workflowMocks.cronAdd.mockImplementation(async () => + makeCronJob({ id: scheduledIds.shift() ?? "unexpected-job" }), + ); const { config, registry } = createPluginRegistryFixture({}, { hostServices: { cron } }); let capturedApi: OpenClawPluginApi | undefined; registerTestPlugin({ @@ -1075,7 +1148,27 @@ describe("plugin scheduled turns", () => { }), ).resolves.toEqual({ removed: 0, failed: 0 }); + pinActivePluginChannelRegistry(registry.registry); + pinActivePluginHttpRouteRegistry(registry.registry); + pinActivePluginSessionExtensionRegistry(registry.registry); setActivePluginRegistry(createEmptyPluginRegistry()); + const pinnedHandle = await capturedApi?.session.workflow.scheduleSessionTurn({ + sessionKey: "agent:main:main", + message: "wake while pinned", + cron: "* * * * *", + tz: "UTC", + }); + expectSessionTurnHandle(pinnedHandle, "job-pinned", "scheduler-plugin"); + await expect( + capturedApi?.session.workflow.unscheduleSessionTurnsByTag({ + sessionKey: "agent:main:main", + tag: "nudge", + }), + ).resolves.toEqual({ removed: 0, failed: 0 }); + + releasePinnedPluginChannelRegistry(registry.registry); + releasePinnedPluginHttpRouteRegistry(registry.registry); + releasePinnedPluginSessionExtensionRegistry(registry.registry); await expect( capturedApi?.session.workflow.scheduleSessionTurn({ sessionKey: "agent:main:main", diff --git a/src/plugins/contracts/session-attachments.contract.test.ts b/src/plugins/contracts/session-attachments.contract.test.ts index b752573501ce..fa9b8dcc3e95 100644 --- a/src/plugins/contracts/session-attachments.contract.test.ts +++ b/src/plugins/contracts/session-attachments.contract.test.ts @@ -16,7 +16,16 @@ import { sendPluginSessionAttachment } from "../host-hook-attachments.js"; import { clearPluginLoaderCache } from "../loader.test-fixtures.js"; import { createEmptyPluginRegistry } from "../registry-empty.js"; import { createPluginRegistry } from "../registry.js"; -import { setActivePluginRegistry } from "../runtime.js"; +import { + pinActivePluginChannelRegistry, + pinActivePluginHttpRouteRegistry, + pinActivePluginSessionExtensionRegistry, + releasePinnedPluginChannelRegistry, + releasePinnedPluginHttpRouteRegistry, + releasePinnedPluginSessionExtensionRegistry, + resetPluginRuntimeStateForTest, + setActivePluginRegistry, +} from "../runtime.js"; import type { PluginRuntime } from "../runtime/types.js"; import { createPluginRecord } from "../status.test-helpers.js"; import type { OpenClawPluginApi } from "../types.js"; @@ -136,7 +145,10 @@ describe("plugin session attachments", () => { afterEach(() => { workflowMocks.getChannelPlugin.mockReset(); workflowMocks.sendMessage.mockReset(); - setActivePluginRegistry(createEmptyPluginRegistry()); + releasePinnedPluginChannelRegistry(); + releasePinnedPluginHttpRouteRegistry(); + releasePinnedPluginSessionExtensionRegistry(); + resetPluginRuntimeStateForTest(); clearPluginLoaderCache(); delete (globalThis as { proofAttachmentApi?: OpenClawPluginApi }).proofAttachmentApi; delete (globalThis as { proofAttachmentLog?: unknown[] }).proofAttachmentLog; @@ -461,7 +473,7 @@ describe("plugin session attachments", () => { }); }); - it("wires sendSessionAttachment through the plugin API with stale-registry protection", async () => { + it("keeps pinned attachment APIs live until their registry retires", async () => { await withSessionStore(async ({ storePath, filePath }) => { await writeSessionEntry(storePath); mockSuccessfulAttachmentDelivery(); @@ -488,7 +500,19 @@ describe("plugin session attachments", () => { }); expectTelegramAttachmentResult(firstResult, 1); + pinActivePluginChannelRegistry(registry.registry); + pinActivePluginHttpRouteRegistry(registry.registry); + pinActivePluginSessionExtensionRegistry(registry.registry); setActivePluginRegistry(createEmptyPluginRegistry()); + const pinnedResult = await capturedApi?.sendSessionAttachment({ + sessionKey: MAIN_SESSION_KEY, + files: [{ path: filePath }], + }); + expectTelegramAttachmentResult(pinnedResult, 1); + + releasePinnedPluginChannelRegistry(registry.registry); + releasePinnedPluginHttpRouteRegistry(registry.registry); + releasePinnedPluginSessionExtensionRegistry(registry.registry); await expect( capturedApi?.sendSessionAttachment({ sessionKey: MAIN_SESSION_KEY, diff --git a/src/plugins/host-hook-cleanup.ts b/src/plugins/host-hook-cleanup.ts index 9086f885f8af..01d7af33f32e 100644 --- a/src/plugins/host-hook-cleanup.ts +++ b/src/plugins/host-hook-cleanup.ts @@ -317,6 +317,7 @@ export async function runPluginHostCleanup(params: { sessionKey: params.sessionKey, records: registry.sessionSchedulerJobs, preserveJobIds: params.preserveSchedulerJobIds, + cleanupOwnerRegistry: registry, preserveOwnerRegistry: params.preserveSchedulerOwnerRegistry, shouldCleanup, }); @@ -341,6 +342,7 @@ export async function runPluginHostCleanup(params: { sessionKey: params.sessionKey, preserveJobIds: params.preserveSchedulerJobIds, excludeJobKeys: registrySchedulerJobKeys, + cleanupOwnerRegistry: registry ?? undefined, shouldCleanup, }); for (const failure of runtimeSchedulerFailures) { diff --git a/src/plugins/host-hook-runtime.ts b/src/plugins/host-hook-runtime.ts index 6f952503cfb5..ce8596a95a35 100644 --- a/src/plugins/host-hook-runtime.ts +++ b/src/plugins/host-hook-runtime.ts @@ -461,6 +461,7 @@ export async function cleanupPluginSessionSchedulerJobs(params: { preserveJobIds?: ReadonlySet; excludeJobKeys?: ReadonlySet; shouldCleanup?: () => boolean; + cleanupOwnerRegistry?: PluginRegistry; preserveOwnerRegistry?: PluginRegistry | null; }): Promise> { const state = getPluginHostRuntimeState(); @@ -563,6 +564,14 @@ export async function cleanupPluginSessionSchedulerJobs(params: { if (params.sessionKey && record.job.sessionKey !== params.sessionKey) { continue; } + // Dynamic jobs share a process-global index. Registry retirement must only + // clean its own records or it can delete jobs owned by another live surface. + if ( + params.cleanupOwnerRegistry !== undefined && + record.ownerRegistry !== params.cleanupOwnerRegistry + ) { + continue; + } if (registryRecordKeys.has(schedulerJobKey(pluginId, jobId, record.job.sessionKey))) { continue; } diff --git a/src/plugins/registry-api.ts b/src/plugins/registry-api.ts index d348ffbc598c..2e1116ba23f9 100644 --- a/src/plugins/registry-api.ts +++ b/src/plugins/registry-api.ts @@ -25,7 +25,6 @@ import { type PluginSideEffectGuard, } from "./registry-state.js"; import type { PluginRecord } from "./registry-types.js"; -import { getActivePluginRegistry } from "./runtime.js"; import type { OpenClawPluginApi, PluginLogger, PluginRegistrationMode } from "./types.js"; function normalizeLogger(logger: PluginLogger): PluginLogger { @@ -148,8 +147,11 @@ export function createPluginApiFactory( const sideEffectGuard = createPluginSideEffectGuard(record.id); const isLoadedRecordInRegistry = () => registry.plugins.some((plugin) => plugin.id === record.id && plugin.status === "loaded"); - const isLoadedRecordInActiveRegistry = () => - getActivePluginRegistry() === registry && isLoadedRecordInRegistry(); + const isLoadedRecordInLiveRegistry = () => + sideEffectGuard.active && + isPluginRegistryActivated(registry) && + !isPluginRegistryRetired(registry) && + isLoadedRecordInRegistry(); const isActivatingLoadedRecord = () => registryParams.activateGlobalSideEffects !== false && record.enabled && @@ -308,7 +310,7 @@ export function createPluginApiFactory( return { ok: false, error: "global side effects disabled" }; } try { - if (!isLoadedRecordInActiveRegistry()) { + if (!isLoadedRecordInLiveRegistry()) { return { ok: false, error: "plugin is not loaded" }; } const runtimeConfig = @@ -337,7 +339,7 @@ export function createPluginApiFactory( origin: record.origin, schedule, cron: getHostCronService(), - shouldCommit: isLoadedRecordInActiveRegistry, + shouldCommit: isLoadedRecordInLiveRegistry, ownerRegistry: registry, }); }, @@ -346,7 +348,7 @@ export function createPluginApiFactory( return { removed: 0, failed: 0 }; } await Promise.resolve(); - if (!isLoadedRecordInActiveRegistry()) { + if (!isLoadedRecordInLiveRegistry()) { return { removed: 0, failed: 0 }; } return unschedulePluginSessionTurnsByTag({ diff --git a/test/plugin-cron-registry-owner.e2e.test.ts b/test/plugin-cron-registry-owner.e2e.test.ts new file mode 100644 index 000000000000..8621303d4335 --- /dev/null +++ b/test/plugin-cron-registry-owner.e2e.test.ts @@ -0,0 +1,512 @@ +import { randomUUID } from "node:crypto"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { OpenClawConfig } from "../src/config/types.openclaw.js"; +import { + connectGatewayClient, + disconnectGatewayClient, + getFreeGatewayPort, +} from "../src/gateway/test-helpers.e2e.js"; +import { + createOpenClawTestInstance, + type OpenClawTestInstance, +} from "./helpers/openclaw-test-instance.js"; + +const PLUGIN_ID = "cron-registry-owner-proof"; +const SCHEDULE_METHOD = `${PLUGIN_ID}.schedule`; +const CRON_EXPRESSION = "*/2 * * * * *"; +const MAIN_WORKSPACE_MARKER = "MAIN_WORKSPACE_CRON_OWNER_MARKER"; +const WORKER_WORKSPACE_MARKER = "WORKER_WORKSPACE_CRON_OWNER_MARKER"; +const OWNER_FIRE = "CRON_OWNER_SURVIVAL_FIRE"; +const PINNED_FIRE = "CRON_PINNED_LATE_FIRE"; +const E2E_TIMEOUT_MS = 180_000; +const WAIT_OPTIONS = { timeout: 45_000, interval: 50 } as const; +const TEST_API_KEY = "test-token-placeholder"; + +type MockModelRequest = { + body: Record; +}; + +type MockModelServer = { + baseUrl: string; + requests: MockModelRequest[]; + stop: () => Promise; +}; + +type ScheduledHandle = { + id: string; + pluginId: string; + sessionKey: string; + kind: string; +}; + +type ScheduleResult = { + handle: ScheduledHandle | null; +}; + +type CronJobView = { + id: string; + enabled: boolean; + deleteAfterRun?: boolean; + sessionTarget: string; + schedule: { kind: string; expr?: string; tz?: string }; + state: { + nextRunAtMs?: number; + lastRunAtMs?: number; + lastRunStatus?: string; + lastStatus?: string; + }; + nextRunAtMs?: number; + lastRunAtMs?: number; + lastRunStatus?: string; +}; + +type CronListPage = { + jobs: CronJobView[]; +}; + +const instances: OpenClawTestInstance[] = []; +const cleanupDirs: string[] = []; +const modelServers: MockModelServer[] = []; + +afterEach(async () => { + await Promise.all(instances.splice(0).map((instance) => instance.cleanup())); + await Promise.all(modelServers.splice(0).map((server) => server.stop())); + await Promise.all(cleanupDirs.splice(0).map((dir) => rm(dir, { recursive: true, force: true }))); +}); + +async function readJsonRequest(req: IncomingMessage): Promise> { + const chunks: Buffer[] = []; + for await (const chunk of req) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + } + const body = Buffer.concat(chunks).toString("utf8"); + return body ? (JSON.parse(body) as Record) : {}; +} + +function writeModelResponse(res: ServerResponse, sequence: number): void { + const messageId = `msg_cron_owner_${sequence}`; + const responseId = `resp_cron_owner_${sequence}`; + const text = `CRON_OWNER_RESPONSE_${sequence}`; + const message = { + type: "message", + id: messageId, + role: "assistant", + status: "completed", + content: [{ type: "output_text", text, annotations: [] }], + }; + const events = [ + { + type: "response.output_item.added", + output_index: 0, + item: { ...message, status: "in_progress", content: [] }, + }, + { + type: "response.output_text.delta", + item_id: messageId, + output_index: 0, + content_index: 0, + delta: text, + }, + { + type: "response.output_text.done", + item_id: messageId, + output_index: 0, + content_index: 0, + text, + }, + { type: "response.output_item.done", output_index: 0, item: message }, + { + type: "response.completed", + response: { + id: responseId, + status: "completed", + output: [message], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }, + }, + ]; + res.writeHead(200, { + "content-type": "text/event-stream", + "cache-control": "no-store", + connection: "keep-alive", + }); + res.end( + `${events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join("")}data: [DONE]\n\n`, + ); +} + +async function startMockModelServer(): Promise { + const requests: MockModelRequest[] = []; + const server = createServer((req, res) => { + void (async () => { + const url = new URL(req.url ?? "/", "http://127.0.0.1"); + if (req.method === "GET" && url.pathname === "/v1/models") { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ data: [{ id: "cron-owner", object: "model" }] })); + return; + } + if (req.method !== "POST" || url.pathname !== "/v1/responses") { + res.writeHead(404).end(); + return; + } + requests.push({ body: await readJsonRequest(req) }); + writeModelResponse(res, requests.length); + })(); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("mock model server did not bind"); + } + let stopped = false; + return { + baseUrl: `http://127.0.0.1:${address.port}`, + requests, + stop: async () => { + if (stopped) { + return; + } + stopped = true; + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + server.closeAllConnections(); + }); + }, + }; +} + +async function writeBundledSchedulerPlugin(bundledRoot: string): Promise { + const pluginDir = path.join(bundledRoot, PLUGIN_ID); + await mkdir(pluginDir, { recursive: true }); + await writeFile( + path.join(pluginDir, "openclaw.plugin.json"), + `${JSON.stringify( + { + id: PLUGIN_ID, + activation: { onStartup: true }, + configSchema: { type: "object", additionalProperties: false, properties: {} }, + }, + null, + 2, + )}\n`, + ); + await writeFile( + path.join(pluginDir, "index.js"), + `module.exports = { + id: ${JSON.stringify(PLUGIN_ID)}, + register(api) { + const scheduleSessionTurn = api.session.workflow.scheduleSessionTurn; + api.registerGatewayMethod(${JSON.stringify(SCHEDULE_METHOD)}, async ({ params, respond }) => { + const name = typeof params?.name === "string" ? params.name : ""; + const message = typeof params?.message === "string" ? params.message : ""; + const sessionKey = typeof params?.sessionKey === "string" ? params.sessionKey : ""; + const handle = await scheduleSessionTurn({ + sessionKey, + message, + cron: ${JSON.stringify(CRON_EXPRESSION)}, + tz: "UTC", + deliveryMode: "none", + tag: "registry-owner", + name, + }); + respond(true, { handle: handle ?? null }); + }); + }, +}; +`, + ); +} + +function requestText(request: MockModelRequest): string { + return JSON.stringify(request.body); +} + +function requestsContaining(server: MockModelServer, marker: string): MockModelRequest[] { + return server.requests.filter((request) => requestText(request).includes(marker)); +} + +async function waitForRequestCount( + server: MockModelServer, + marker: string, + count: number, +): Promise { + await vi.waitFor(() => { + expect(requestsContaining(server, marker).length).toBeGreaterThanOrEqual(count); + }, WAIT_OPTIONS); +} + +function requireHandle(result: ScheduleResult, expected: Omit): string { + expect(result.handle).toMatchObject(expected); + if (!result.handle) { + throw new Error(`missing scheduled handle for ${expected.sessionKey}`); + } + expect(result.handle.id).toBeTruthy(); + return result.handle.id; +} + +async function listCronJobs(client: { + request: (method: string, params: Record) => Promise; +}): Promise { + const page = await client.request("cron.list", { + includeDisabled: true, + scheduleKind: "cron", + limit: 200, + }); + return page.jobs; +} + +describe("plugin cron registry ownership e2e", () => { + it( + "keeps recurring startup-plugin jobs through workspace registry churn", + { timeout: E2E_TIMEOUT_MS }, + async () => { + const fixtureDir = await mkdtemp(path.join(tmpdir(), "openclaw-cron-owner-e2e-")); + cleanupDirs.push(fixtureDir); + const bundledRoot = path.join(fixtureDir, "bundled"); + const mainWorkspace = path.join(fixtureDir, "workspace-main"); + const workerWorkspace = path.join(fixtureDir, "workspace-worker"); + await Promise.all([ + mkdir(mainWorkspace, { recursive: true }), + mkdir(workerWorkspace, { recursive: true }), + writeBundledSchedulerPlugin(bundledRoot), + ]); + await Promise.all([ + writeFile(path.join(mainWorkspace, "AGENTS.md"), `${MAIN_WORKSPACE_MARKER}\n`), + writeFile(path.join(workerWorkspace, "AGENTS.md"), `${WORKER_WORKSPACE_MARKER}\n`), + ]); + + const modelServer = await startMockModelServer(); + modelServers.push(modelServer); + const modelRef = "cron-owner/cron-owner"; + const config = { + plugins: { + enabled: true, + allow: [PLUGIN_ID], + entries: { [PLUGIN_ID]: { enabled: true } }, + slots: { memory: "none" }, + }, + agents: { + defaults: { + workspace: mainWorkspace, + model: { primary: modelRef }, + models: { [modelRef]: { agentRuntime: { id: "openclaw" } } }, + skills: [], + }, + list: [ + { + id: "main", + default: true, + workspace: mainWorkspace, + model: { primary: modelRef }, + skills: [], + }, + { + id: "worker", + workspace: workerWorkspace, + model: { primary: modelRef }, + skills: [], + }, + ], + }, + tools: { profile: "minimal" }, + models: { + mode: "replace", + providers: { + "cron-owner": { + baseUrl: `${modelServer.baseUrl}/v1`, + ["api" + "Key"]: TEST_API_KEY, + api: "openai-responses", + request: { allowPrivateNetwork: true }, + models: [ + { + id: "cron-owner", + name: "cron-owner", + api: "openai-responses", + reasoning: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 128_000, + maxTokens: 4_096, + }, + ], + }, + }, + }, + } satisfies OpenClawConfig; + const customPort = await getFreeGatewayPort(); + const instance = await createOpenClawTestInstance({ + name: "plugin-cron-registry-owner", + port: customPort, + config, + env: { + OPENCLAW_BUNDLED_PLUGINS_DIR: bundledRoot, + OPENCLAW_TEST_TRUST_BUNDLED_PLUGINS_DIR: "1", + OPENCLAW_DISABLE_BUNDLED_PLUGINS: undefined, + OPENCLAW_SKIP_CRON: undefined, + OPENCLAW_SKIP_PROVIDERS: undefined, + OPENCLAW_TEST_MINIMAL_GATEWAY: undefined, + }, + }); + instances.push(instance); + await instance.startGateway(); + expect(instance.port).toBe(customPort); + + const client = await connectGatewayClient({ + url: instance.url, + ["to" + "ken"]: instance.gatewayToken, + role: "operator", + scopes: ["operator.admin", "operator.read", "operator.write"], + requestTimeoutMs: 30_000, + }); + const scheduledIds: string[] = []; + try { + const cronStatus = await client.request<{ + enabled: boolean; + storage: string; + sqlitePath: string; + }>("cron.status", {}); + expect(cronStatus).toMatchObject({ enabled: true, storage: "sqlite" }); + expect(cronStatus.sqlitePath).toBe( + path.join(instance.stateDir, "state", "openclaw.sqlite"), + ); + + const ownerResult = await client.request(SCHEDULE_METHOD, { + name: "owner-survival", + message: OWNER_FIRE, + sessionKey: "agent:main:cron-owner-survival", + }); + const ownerId = requireHandle(ownerResult, { + pluginId: PLUGIN_ID, + sessionKey: "agent:main:cron-owner-survival", + kind: "session-turn", + }); + scheduledIds.push(ownerId); + + await client.request("chat.send", { + sessionKey: "agent:worker:registry-churn", + message: "ACTIVATE_WORKER_REGISTRY", + deliver: false, + idempotencyKey: randomUUID(), + }); + await waitForRequestCount(modelServer, "ACTIVATE_WORKER_REGISTRY", 1); + + // The attached Gateway handler still closes over the pinned startup registry API. + // Scheduling here proves non-active pinned registries remain live during workspace churn. + const pinnedResult = await client.request(SCHEDULE_METHOD, { + name: "pinned-late", + message: PINNED_FIRE, + sessionKey: "agent:main:cron-pinned-late", + }); + const pinnedId = requireHandle(pinnedResult, { + pluginId: PLUGIN_ID, + sessionKey: "agent:main:cron-pinned-late", + kind: "session-turn", + }); + scheduledIds.push(pinnedId); + + const ownerWorkerActiveBaseline = requestsContaining(modelServer, OWNER_FIRE).length; + const pinnedWorkerActiveBaseline = requestsContaining(modelServer, PINNED_FIRE).length; + await Promise.all([ + waitForRequestCount(modelServer, OWNER_FIRE, ownerWorkerActiveBaseline + 1), + waitForRequestCount(modelServer, PINNED_FIRE, pinnedWorkerActiveBaseline + 1), + ]); + for (const request of [ + ...requestsContaining(modelServer, OWNER_FIRE).slice(ownerWorkerActiveBaseline), + ...requestsContaining(modelServer, PINNED_FIRE).slice(pinnedWorkerActiveBaseline), + ]) { + expect(requestText(request)).toContain(MAIN_WORKSPACE_MARKER); + expect(requestText(request)).not.toContain(WORKER_WORKSPACE_MARKER); + } + + const mainReactivationBaseline = requestsContaining( + modelServer, + "REACTIVATE_MAIN_REGISTRY", + ).length; + await client.request("chat.send", { + sessionKey: "agent:main:registry-churn", + message: "REACTIVATE_MAIN_REGISTRY", + deliver: false, + idempotencyKey: randomUUID(), + }); + await waitForRequestCount( + modelServer, + "REACTIVATE_MAIN_REGISTRY", + mainReactivationBaseline + 1, + ); + + const expectedIds = [ownerId, pinnedId].toSorted(); + const afterChurn = (await listCronJobs(client)) + .filter((job) => expectedIds.includes(job.id)) + .toSorted((a, b) => a.id.localeCompare(b.id)); + expect(afterChurn.map((job) => job.id)).toEqual(expectedIds); + expect(afterChurn).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + id: ownerId, + enabled: true, + deleteAfterRun: false, + sessionTarget: "session:agent:main:cron-owner-survival", + schedule: { kind: "cron", expr: CRON_EXPRESSION, tz: "UTC" }, + }), + expect.objectContaining({ + id: pinnedId, + enabled: true, + deleteAfterRun: false, + sessionTarget: "session:agent:main:cron-pinned-late", + schedule: { kind: "cron", expr: CRON_EXPRESSION, tz: "UTC" }, + }), + ]), + ); + const nextRunAtChurn = new Map( + afterChurn.map((job) => [job.id, job.nextRunAtMs ?? job.state.nextRunAtMs]), + ); + expect([...nextRunAtChurn.values()]).toEqual([expect.any(Number), expect.any(Number)]); + + const ownerBaseline = requestsContaining(modelServer, OWNER_FIRE).length; + const pinnedBaseline = requestsContaining(modelServer, PINNED_FIRE).length; + await Promise.all([ + waitForRequestCount(modelServer, OWNER_FIRE, ownerBaseline + 2), + waitForRequestCount(modelServer, PINNED_FIRE, pinnedBaseline + 2), + ]); + + for (const request of [ + ...requestsContaining(modelServer, OWNER_FIRE).slice(ownerBaseline), + ...requestsContaining(modelServer, PINNED_FIRE).slice(pinnedBaseline), + ]) { + expect(requestText(request)).toContain(MAIN_WORKSPACE_MARKER); + expect(requestText(request)).not.toContain(WORKER_WORKSPACE_MARKER); + } + + const afterRecurringRuns = (await listCronJobs(client)).filter((job) => + expectedIds.includes(job.id), + ); + expect(afterRecurringRuns.map((job) => job.id).toSorted()).toEqual(expectedIds); + for (const job of afterRecurringRuns) { + expect(job.enabled).toBe(true); + expect(job.deleteAfterRun).toBe(false); + expect(job.lastRunStatus ?? job.state.lastRunStatus ?? job.state.lastStatus).toBe("ok"); + const lastRunAtMs = job.lastRunAtMs ?? job.state.lastRunAtMs; + const nextRunAtMs = job.nextRunAtMs ?? job.state.nextRunAtMs; + expect(lastRunAtMs).toBeTypeOf("number"); + expect(nextRunAtMs).toBeTypeOf("number"); + expect(nextRunAtMs as number).toBeGreaterThan(lastRunAtMs as number); + expect(nextRunAtMs as number).toBeGreaterThan(nextRunAtChurn.get(job.id) as number); + } + } finally { + await Promise.all( + scheduledIds.map((id) => + client.request("cron.remove", { id }).catch(() => ({ removed: false })), + ), + ); + await disconnectGatewayClient(client).catch(() => undefined); + } + }, + ); +});