Skip to content

Commit bd3ad34

Browse files
mbelinkymbelinky
authored
tasks: add detached runtime plugin registration contract (#68915)
* tasks: register detached runtime plugins * tasks: harden detached runtime ownership * tasks: extract detached runtime contract types * changelog: note detached runtime contract * changelog: attribute detached runtime contract
1 parent c67a9c5 commit bd3ad34

19 files changed

Lines changed: 759 additions & 49 deletions

‎CHANGELOG.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,8 @@ Docs: https://docs.openclaw.ai
66

77
### Changes
88

9+
- Plugins/tasks: add a detached runtime registration contract so plugin executors can own detached task lifecycle and cancellation without reaching into core task internals. (#68915) Thanks @mbelinky.
10+
911
### Fixes
1012

1113
- Models/Kimi: default bundled Kimi thinking to off and normalize Anthropic-compatible `thinking` payloads so stale session `/think` state no longer silently re-enables reasoning on Kimi runs. (#68907) Thanks @frankekn.

‎src/commands/tasks.ts‎

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,7 @@
11
import type { RuntimeEnv } from "../runtime.js";
22
import { normalizeOptionalString } from "../shared/string-coerce.js";
3-
import {
4-
cancelTaskById,
5-
getTaskById,
6-
updateTaskNotifyPolicyById,
7-
} from "../tasks/runtime-internal.js";
3+
import { getTaskById, updateTaskNotifyPolicyById } from "../tasks/runtime-internal.js";
4+
import { cancelDetachedTaskRunById } from "../tasks/task-executor.js";
85
import {
96
listTaskFlowAuditFindings,
107
summarizeTaskFlowAuditFindings,
@@ -391,7 +388,7 @@ export async function tasksCancelCommand(opts: { lookup: string }, runtime: Runt
391388
runtime.exit(1);
392389
return;
393390
}
394-
const result = await cancelTaskById({
391+
const result = await cancelDetachedTaskRunById({
395392
cfg: await loadTaskCancelConfig(),
396393
taskId: task.taskId,
397394
});

‎src/plugin-sdk/index.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@ export type {
6464
export type {
6565
BoundTaskFlowsRuntime,
6666
BoundTaskRunsRuntime,
67+
DetachedTaskLifecycleRuntime,
6768
PluginRuntimeTaskFlows,
6869
PluginRuntimeTaskRuns,
6970
PluginRuntimeTasks,

‎src/plugins/api-builder.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@ export type BuildPluginApiParams = {
4848
| "registerContextEngine"
4949
| "registerCompactionProvider"
5050
| "registerAgentHarness"
51+
| "registerDetachedTaskRuntime"
5152
| "registerMemoryCapability"
5253
| "registerMemoryPromptSection"
5354
| "registerMemoryPromptSupplement"
@@ -98,6 +99,7 @@ const noopRegisterCommand: OpenClawPluginApi["registerCommand"] = () => {};
9899
const noopRegisterContextEngine: OpenClawPluginApi["registerContextEngine"] = () => {};
99100
const noopRegisterCompactionProvider: OpenClawPluginApi["registerCompactionProvider"] = () => {};
100101
const noopRegisterAgentHarness: OpenClawPluginApi["registerAgentHarness"] = () => {};
102+
const noopRegisterDetachedTaskRuntime: OpenClawPluginApi["registerDetachedTaskRuntime"] = () => {};
101103
const noopRegisterMemoryCapability: OpenClawPluginApi["registerMemoryCapability"] = () => {};
102104
const noopRegisterMemoryPromptSection: OpenClawPluginApi["registerMemoryPromptSection"] = () => {};
103105
const noopRegisterMemoryPromptSupplement: OpenClawPluginApi["registerMemoryPromptSupplement"] =
@@ -164,6 +166,8 @@ export function buildPluginApi(params: BuildPluginApiParams): OpenClawPluginApi
164166
registerCompactionProvider:
165167
handlers.registerCompactionProvider ?? noopRegisterCompactionProvider,
166168
registerAgentHarness: handlers.registerAgentHarness ?? noopRegisterAgentHarness,
169+
registerDetachedTaskRuntime:
170+
handlers.registerDetachedTaskRuntime ?? noopRegisterDetachedTaskRuntime,
167171
registerMemoryCapability: handlers.registerMemoryCapability ?? noopRegisterMemoryCapability,
168172
registerMemoryPromptSection:
169173
handlers.registerMemoryPromptSection ?? noopRegisterMemoryPromptSection,

‎src/plugins/loader.test.ts‎

Lines changed: 175 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,12 @@ import {
1111
triggerInternalHook,
1212
} from "../hooks/internal-hooks.js";
1313
import { emitDiagnosticEvent } from "../infra/diagnostic-events.js";
14+
import {
15+
clearDetachedTaskLifecycleRuntimeRegistration,
16+
getDetachedTaskLifecycleRuntimeRegistration,
17+
registerDetachedTaskLifecycleRuntime,
18+
type DetachedTaskLifecycleRuntime,
19+
} from "../tasks/detached-task-runtime-state.js";
1420
import { withEnv } from "../test-utils/env.js";
1521
import { clearPluginCommands, getPluginCommandSpecs } from "./command-registry-state.js";
1622
import { getGlobalHookRunner, resetGlobalHookRunner } from "./hook-runner-global.js";
@@ -65,6 +71,26 @@ import {
6571
import type { PluginSdkResolutionPreference } from "./sdk-alias.js";
6672
let cachedBundledTelegramDir = "";
6773
let cachedBundledMemoryDir = "";
74+
75+
function createDetachedTaskRuntimeStub(id: string): DetachedTaskLifecycleRuntime {
76+
const fail = (name: string): never => {
77+
throw new Error(`detached runtime ${id} should not execute ${name} in this test`);
78+
};
79+
return {
80+
createQueuedTaskRun: () => fail("createQueuedTaskRun"),
81+
createRunningTaskRun: () => fail("createRunningTaskRun"),
82+
startTaskRunByRunId: () => fail("startTaskRunByRunId"),
83+
recordTaskRunProgressByRunId: () => fail("recordTaskRunProgressByRunId"),
84+
completeTaskRunByRunId: () => fail("completeTaskRunByRunId"),
85+
failTaskRunByRunId: () => fail("failTaskRunByRunId"),
86+
setDetachedTaskDeliveryStatusByRunId: () => fail("setDetachedTaskDeliveryStatusByRunId"),
87+
cancelDetachedTaskRunById: async () => ({
88+
found: true,
89+
cancelled: true,
90+
}),
91+
};
92+
}
93+
6894
const BUNDLED_TELEGRAM_PLUGIN_BODY = `module.exports = {
6995
id: "telegram",
7096
register(api) {
@@ -2025,6 +2051,155 @@ module.exports = { id: "throws-after-import", register() {} };`,
20252051
expect(listMemoryEmbeddingProviders()).toEqual([]);
20262052
});
20272053

2054+
it("does not replace the active detached task runtime during non-activating loads", () => {
2055+
useNoBundledPlugins();
2056+
const activeRuntime = createDetachedTaskRuntimeStub("active");
2057+
registerDetachedTaskLifecycleRuntime("active-runtime", activeRuntime);
2058+
2059+
const plugin = writePlugin({
2060+
id: "snapshot-detached-runtime",
2061+
filename: "snapshot-detached-runtime.cjs",
2062+
body: `module.exports = {
2063+
id: "snapshot-detached-runtime",
2064+
register(api) {
2065+
api.registerDetachedTaskRuntime({
2066+
createQueuedTaskRun() { throw new Error("snapshot createQueuedTaskRun should not run"); },
2067+
createRunningTaskRun() { throw new Error("snapshot createRunningTaskRun should not run"); },
2068+
startTaskRunByRunId() { throw new Error("snapshot startTaskRunByRunId should not run"); },
2069+
recordTaskRunProgressByRunId() { throw new Error("snapshot recordTaskRunProgressByRunId should not run"); },
2070+
completeTaskRunByRunId() { throw new Error("snapshot completeTaskRunByRunId should not run"); },
2071+
failTaskRunByRunId() { throw new Error("snapshot failTaskRunByRunId should not run"); },
2072+
setDetachedTaskDeliveryStatusByRunId() { throw new Error("snapshot setDetachedTaskDeliveryStatusByRunId should not run"); },
2073+
async cancelDetachedTaskRunById() { return { found: true, cancelled: true }; },
2074+
});
2075+
},
2076+
};`,
2077+
});
2078+
2079+
const scoped = loadOpenClawPlugins({
2080+
cache: false,
2081+
activate: false,
2082+
workspaceDir: plugin.dir,
2083+
config: {
2084+
plugins: {
2085+
load: { paths: [plugin.file] },
2086+
allow: ["snapshot-detached-runtime"],
2087+
},
2088+
},
2089+
onlyPluginIds: ["snapshot-detached-runtime"],
2090+
});
2091+
2092+
expect(scoped.plugins.find((entry) => entry.id === "snapshot-detached-runtime")?.status).toBe(
2093+
"loaded",
2094+
);
2095+
expect(getDetachedTaskLifecycleRuntimeRegistration()).toMatchObject({
2096+
pluginId: "active-runtime",
2097+
runtime: activeRuntime,
2098+
});
2099+
});
2100+
2101+
it("clears newly-registered detached task runtimes when plugin register fails", () => {
2102+
useNoBundledPlugins();
2103+
const plugin = writePlugin({
2104+
id: "failing-detached-runtime",
2105+
filename: "failing-detached-runtime.cjs",
2106+
body: `module.exports = {
2107+
id: "failing-detached-runtime",
2108+
register(api) {
2109+
api.registerDetachedTaskRuntime({
2110+
createQueuedTaskRun() { throw new Error("failing createQueuedTaskRun should not run"); },
2111+
createRunningTaskRun() { throw new Error("failing createRunningTaskRun should not run"); },
2112+
startTaskRunByRunId() { throw new Error("failing startTaskRunByRunId should not run"); },
2113+
recordTaskRunProgressByRunId() { throw new Error("failing recordTaskRunProgressByRunId should not run"); },
2114+
completeTaskRunByRunId() { throw new Error("failing completeTaskRunByRunId should not run"); },
2115+
failTaskRunByRunId() { throw new Error("failing failTaskRunByRunId should not run"); },
2116+
setDetachedTaskDeliveryStatusByRunId() { throw new Error("failing setDetachedTaskDeliveryStatusByRunId should not run"); },
2117+
async cancelDetachedTaskRunById() { return { found: true, cancelled: true }; },
2118+
});
2119+
throw new Error("detached runtime register failed");
2120+
},
2121+
};`,
2122+
});
2123+
2124+
const registry = loadOpenClawPlugins({
2125+
cache: false,
2126+
workspaceDir: plugin.dir,
2127+
config: {
2128+
plugins: {
2129+
load: { paths: [plugin.file] },
2130+
allow: ["failing-detached-runtime"],
2131+
},
2132+
},
2133+
onlyPluginIds: ["failing-detached-runtime"],
2134+
});
2135+
2136+
expect(registry.plugins.find((entry) => entry.id === "failing-detached-runtime")?.status).toBe(
2137+
"error",
2138+
);
2139+
expect(getDetachedTaskLifecycleRuntimeRegistration()).toBeUndefined();
2140+
});
2141+
2142+
it("restores cached detached task runtime registrations on cache hits", () => {
2143+
useNoBundledPlugins();
2144+
const plugin = writePlugin({
2145+
id: "cached-detached-runtime",
2146+
filename: "cached-detached-runtime.cjs",
2147+
body: `module.exports = {
2148+
id: "cached-detached-runtime",
2149+
register(api) {
2150+
api.registerDetachedTaskRuntime({
2151+
createQueuedTaskRun() { throw new Error("cached createQueuedTaskRun should not run"); },
2152+
createRunningTaskRun() { throw new Error("cached createRunningTaskRun should not run"); },
2153+
startTaskRunByRunId() { throw new Error("cached startTaskRunByRunId should not run"); },
2154+
recordTaskRunProgressByRunId() { throw new Error("cached recordTaskRunProgressByRunId should not run"); },
2155+
completeTaskRunByRunId() { throw new Error("cached completeTaskRunByRunId should not run"); },
2156+
failTaskRunByRunId() { throw new Error("cached failTaskRunByRunId should not run"); },
2157+
setDetachedTaskDeliveryStatusByRunId() { throw new Error("cached setDetachedTaskDeliveryStatusByRunId should not run"); },
2158+
async cancelDetachedTaskRunById() { return { found: true, cancelled: true }; },
2159+
});
2160+
},
2161+
};`,
2162+
});
2163+
2164+
const loadOptions = {
2165+
workspaceDir: plugin.dir,
2166+
config: {
2167+
plugins: {
2168+
load: { paths: [plugin.file] },
2169+
allow: ["cached-detached-runtime"],
2170+
},
2171+
},
2172+
onlyPluginIds: ["cached-detached-runtime"],
2173+
} satisfies Parameters<typeof loadOpenClawPlugins>[0];
2174+
2175+
loadOpenClawPlugins(loadOptions);
2176+
expect(getDetachedTaskLifecycleRuntimeRegistration()?.pluginId).toBe("cached-detached-runtime");
2177+
2178+
clearDetachedTaskLifecycleRuntimeRegistration();
2179+
expect(getDetachedTaskLifecycleRuntimeRegistration()).toBeUndefined();
2180+
2181+
loadOpenClawPlugins(loadOptions);
2182+
2183+
expect(getDetachedTaskLifecycleRuntimeRegistration()?.pluginId).toBe("cached-detached-runtime");
2184+
});
2185+
2186+
it("clears stale detached task runtime registrations on active reloads when no plugin re-registers one", () => {
2187+
useNoBundledPlugins();
2188+
registerDetachedTaskLifecycleRuntime("stale-runtime", createDetachedTaskRuntimeStub("stale"));
2189+
2190+
loadOpenClawPlugins({
2191+
cache: false,
2192+
config: {
2193+
plugins: {
2194+
load: { paths: [] },
2195+
allow: [],
2196+
},
2197+
},
2198+
});
2199+
2200+
expect(getDetachedTaskLifecycleRuntimeRegistration()).toBeUndefined();
2201+
});
2202+
20282203
it("restores cached memory capability public artifacts on cache hits", async () => {
20292204
useNoBundledPlugins();
20302205
const workspaceDir = makeTempDir();

‎src/plugins/loader.ts‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,11 @@ import {
2222
normalizeLowercaseStringOrEmpty,
2323
normalizeOptionalString,
2424
} from "../shared/string-coerce.js";
25+
import {
26+
clearDetachedTaskLifecycleRuntimeRegistration,
27+
getDetachedTaskLifecycleRuntimeRegistration,
28+
restoreDetachedTaskLifecycleRuntimeRegistration,
29+
} from "../tasks/detached-task-runtime-state.js";
2530
import { resolveUserPath } from "../utils.js";
2631
import { buildPluginApi } from "./api-builder.js";
2732
import { inspectBundleMcpRuntimeSupport } from "./bundle-mcp.js";
@@ -187,6 +192,7 @@ export class PluginLoadReentryError extends Error {
187192

188193
type CachedPluginState = {
189194
registry: PluginRegistry;
195+
detachedTaskRuntimeRegistration: ReturnType<typeof getDetachedTaskLifecycleRuntimeRegistration>;
190196
memoryCapability: ReturnType<typeof getMemoryCapabilityRegistration>;
191197
memoryCorpusSupplements: ReturnType<typeof listMemoryCorpusSupplements>;
192198
agentHarnesses: ReturnType<typeof listRegisteredAgentHarnesses>;
@@ -225,6 +231,7 @@ export function clearPluginLoaderCache(): void {
225231
openAllowlistWarningCache.clear();
226232
clearAgentHarnesses();
227233
clearCompactionProviders();
234+
clearDetachedTaskLifecycleRuntimeRegistration();
228235
clearMemoryEmbeddingProviders();
229236
clearMemoryPluginState();
230237
}
@@ -1441,6 +1448,7 @@ export function loadOpenClawPlugins(options: PluginLoadOptions = {}): PluginRegi
14411448
if (cached) {
14421449
restoreRegisteredAgentHarnesses(cached.agentHarnesses);
14431450
restoreRegisteredCompactionProviders(cached.compactionProviders);
1451+
restoreDetachedTaskLifecycleRuntimeRegistration(cached.detachedTaskRuntimeRegistration);
14441452
restoreRegisteredMemoryEmbeddingProviders(cached.memoryEmbeddingProviders);
14451453
restoreMemoryPluginState({
14461454
capability: cached.memoryCapability,
@@ -1472,6 +1480,7 @@ export function loadOpenClawPlugins(options: PluginLoadOptions = {}): PluginRegi
14721480
clearAgentHarnesses();
14731481
clearPluginCommands();
14741482
clearPluginInteractiveHandlers();
1483+
clearDetachedTaskLifecycleRuntimeRegistration();
14751484
clearMemoryPluginState();
14761485
}
14771486

@@ -2212,6 +2221,7 @@ export function loadOpenClawPlugins(options: PluginLoadOptions = {}): PluginRegi
22122221
const registrySnapshot = snapshotPluginRegistry(registry);
22132222
const previousAgentHarnesses = listRegisteredAgentHarnesses();
22142223
const previousCompactionProviders = listRegisteredCompactionProviders();
2224+
const previousDetachedTaskRuntimeRegistration = getDetachedTaskLifecycleRuntimeRegistration();
22152225
const previousMemoryEmbeddingProviders = listRegisteredMemoryEmbeddingProviders();
22162226
const previousMemoryFlushPlanResolver = getMemoryFlushPlanResolver();
22172227
const previousMemoryPromptBuilder = getMemoryPromptSectionBuilder();
@@ -2225,6 +2235,7 @@ export function loadOpenClawPlugins(options: PluginLoadOptions = {}): PluginRegi
22252235
if (!shouldActivate) {
22262236
restoreRegisteredAgentHarnesses(previousAgentHarnesses);
22272237
restoreRegisteredCompactionProviders(previousCompactionProviders);
2238+
restoreDetachedTaskLifecycleRuntimeRegistration(previousDetachedTaskRuntimeRegistration);
22282239
restoreRegisteredMemoryEmbeddingProviders(previousMemoryEmbeddingProviders);
22292240
restoreMemoryPluginState({
22302241
corpusSupplements: previousMemoryCorpusSupplements,
@@ -2241,6 +2252,7 @@ export function loadOpenClawPlugins(options: PluginLoadOptions = {}): PluginRegi
22412252
restorePluginRegistry(registry, registrySnapshot);
22422253
restoreRegisteredAgentHarnesses(previousAgentHarnesses);
22432254
restoreRegisteredCompactionProviders(previousCompactionProviders);
2255+
restoreDetachedTaskLifecycleRuntimeRegistration(previousDetachedTaskRuntimeRegistration);
22442256
restoreRegisteredMemoryEmbeddingProviders(previousMemoryEmbeddingProviders);
22452257
restoreMemoryPluginState({
22462258
corpusSupplements: previousMemoryCorpusSupplements,
@@ -2297,6 +2309,7 @@ export function loadOpenClawPlugins(options: PluginLoadOptions = {}): PluginRegi
22972309

22982310
if (cacheEnabled) {
22992311
setCachedPluginRegistry(cacheKey, {
2312+
detachedTaskRuntimeRegistration: getDetachedTaskLifecycleRuntimeRegistration(),
23002313
memoryCapability: getMemoryCapabilityRegistration(),
23012314
memoryCorpusSupplements: listMemoryCorpusSupplements(),
23022315
registry,

‎src/plugins/registry.ts‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,10 @@ import {
2222
import { normalizePluginGatewayMethodScope } from "../shared/gateway-method-policy.js";
2323
import { resolveGlobalSingleton } from "../shared/global-singleton.js";
2424
import { normalizeOptionalString } from "../shared/string-coerce.js";
25+
import {
26+
getDetachedTaskLifecycleRuntimeRegistration,
27+
registerDetachedTaskLifecycleRuntime,
28+
} from "../tasks/detached-task-runtime-state.js";
2529
import { resolveUserPath } from "../utils.js";
2630
import { buildPluginApi } from "./api-builder.js";
2731
import { normalizeRegisteredChannelPlugin } from "./channel-validation.js";
@@ -1169,6 +1173,19 @@ export function createPluginRegistry(registryParams: PluginRegistryParams) {
11691173
registerHttpRoute: (routeParams) => registerHttpRoute(record, routeParams),
11701174
registerProvider: (provider) => registerProvider(record, provider),
11711175
registerAgentHarness: (harness) => registerAgentHarness(record, harness),
1176+
registerDetachedTaskRuntime: (runtime) => {
1177+
const existing = getDetachedTaskLifecycleRuntimeRegistration();
1178+
if (existing && existing.pluginId !== record.id) {
1179+
pushDiagnostic({
1180+
level: "error",
1181+
pluginId: record.id,
1182+
source: record.source,
1183+
message: `detached task runtime already registered by ${existing.pluginId}`,
1184+
});
1185+
return;
1186+
}
1187+
registerDetachedTaskLifecycleRuntime(record.id, runtime);
1188+
},
11721189
registerSpeechProvider: (provider) => registerSpeechProvider(record, provider),
11731190
registerRealtimeTranscriptionProvider: (provider) =>
11741191
registerRealtimeTranscriptionProvider(record, provider),

‎src/plugins/runtime/runtime-task-test-harness.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import { vi } from "vitest";
2+
import { resetDetachedTaskLifecycleRuntimeForTests } from "../../tasks/detached-task-runtime.js";
23
import {
34
resetTaskRegistryControlRuntimeForTests,
45
resetTaskRegistryDeliveryRuntimeForTests,
@@ -33,6 +34,7 @@ export function installRuntimeTaskDeliveryMock(): void {
3334
export function resetRuntimeTaskTestState(
3435
taskRegistryOptions?: Parameters<typeof resetTaskRegistryForTests>[0],
3536
): void {
37+
resetDetachedTaskLifecycleRuntimeForTests();
3638
resetTaskRegistryControlRuntimeForTests();
3739
resetTaskRegistryDeliveryRuntimeForTests();
3840
resetTaskRegistryForTests(taskRegistryOptions);

0 commit comments

Comments
 (0)