mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-27 04:51:15 +00:00
fix(plugins): recurring plugin-scheduled jobs silently deleted after first post-boot agent turn (#107752)
* fix(agents): skip runtime plugin reload when a registry is already active
ensureRuntimePluginsLoaded rebuilt and activated a replacement plugin
registry on every call site invocation (agent turns, cron isolated-agent
runs, the gateway prewarm sidecar). Activating a replacement retires the
active registry and runs its plugin host cleanup, which cron.remove()s
persistent plugin-scheduled session turns before the replacement's async
re-registration catches up — silently deleting the cron row with only a
low-level 'cron service unavailable' WARN. Plugins using
api.session.workflow.scheduleSessionTurn for recurring jobs lose their
schedule on the next agent turn after boot.
Guard with getActiveRuntimePluginRegistry(): when a registry is already
active every caller's request is satisfied, so return before building a
replacement.
* fix(agents): make active-registry reuse compatibility-aware, not presence-only
The original guard (1de432a9) returned early whenever ANY registry was
active, without checking whether it was actually compatible with this
call's workspace, plugin scope, config, or runtime mode. A later call
requesting a different workspace could silently keep an unrelated
workspace's registry active instead of loading its own — a correctness
and workspace-isolation issue (caught in upstream review, PR #107752).
Fix: compute workspaceDir/startupPluginIds/loadOptions first, then check
getLoadedRuntimePluginRegistry({loadOptions, workspaceDir, requiredPluginIds})
before deciding to skip — reusing the existing loader cache-key
compatibility rules (workspace, config, runtime subagent mode, required
plugin IDs) that already govern normal cache-hit reuse elsewhere. Only
skips the reload when an active registry is genuinely compatible; still
loads a fresh one otherwise.
Tests: replaced the presence-only regression test with three cases —
compatible-registry reuse (skip), incompatible/different-workspace (still
loads), and startup-scoped plugin IDs passed through to the compatibility
check. All 5 acceptance criteria from the review verified locally:
runtime-plugins.test.ts, active-runtime-registry.test.ts,
standalone-runtime-registry-loader.test.ts, and the full cron lane
(33 files / 460 tests, including run.runtime-plugins.test.ts) all green;
pnpm build completes clean (5m14s, no errors).
* fix(plugins): preserve live scheduled turns
Scope dynamic scheduler cleanup to the retiring registry owner and keep runtime workflow APIs callable while their registry remains pinned and live.\n\nCo-authored-by: Dan Scarafoni <dan@poweredbyzon.com>
* test(plugins): prove recurring cron ownership
* chore(plugin-sdk): refresh api baseline
---------
Co-authored-by: Peter Steinberger <steipete@gmail.com>
This commit is contained in:
Dan Scarafoni
@@ -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
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -461,6 +461,7 @@ export async function cleanupPluginSessionSchedulerJobs(params: {
|
||||
preserveJobIds?: ReadonlySet<string>;
|
||||
excludeJobKeys?: ReadonlySet<string>;
|
||||
shouldCleanup?: () => boolean;
|
||||
cleanupOwnerRegistry?: PluginRegistry;
|
||||
preserveOwnerRegistry?: PluginRegistry | null;
|
||||
}): Promise<Array<{ pluginId: string; hookId: string; error: unknown }>> {
|
||||
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;
|
||||
}
|
||||
|
||||
@@ -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({
|
||||
|
||||
512
test/plugin-cron-registry-owner.e2e.test.ts
Normal file
512
test/plugin-cron-registry-owner.e2e.test.ts
Normal file
@@ -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<string, unknown>;
|
||||
};
|
||||
|
||||
type MockModelServer = {
|
||||
baseUrl: string;
|
||||
requests: MockModelRequest[];
|
||||
stop: () => Promise<void>;
|
||||
};
|
||||
|
||||
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<Record<string, unknown>> {
|
||||
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<string, unknown>) : {};
|
||||
}
|
||||
|
||||
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<MockModelServer> {
|
||||
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<void>((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<void>((resolve, reject) => {
|
||||
server.close((error) => (error ? reject(error) : resolve()));
|
||||
server.closeAllConnections();
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function writeBundledSchedulerPlugin(bundledRoot: string): Promise<void> {
|
||||
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<void> {
|
||||
await vi.waitFor(() => {
|
||||
expect(requestsContaining(server, marker).length).toBeGreaterThanOrEqual(count);
|
||||
}, WAIT_OPTIONS);
|
||||
}
|
||||
|
||||
function requireHandle(result: ScheduleResult, expected: Omit<ScheduledHandle, "id">): 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: <T>(method: string, params: Record<string, unknown>) => Promise<T>;
|
||||
}): Promise<CronJobView[]> {
|
||||
const page = await client.request<CronListPage>("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<ScheduleResult>(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<ScheduleResult>(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);
|
||||
}
|
||||
},
|
||||
);
|
||||
});
|
||||
Reference in New Issue
Block a user