diff --git a/docs/plugins/sdk-subpaths.md b/docs/plugins/sdk-subpaths.md index 1c5b34c941ad..ae1a4de988f5 100644 --- a/docs/plugins/sdk-subpaths.md +++ b/docs/plugins/sdk-subpaths.md @@ -28,7 +28,7 @@ For the plugin authoring guide, see [Plugin SDK overview](/plugins/sdk-overview) | `plugin-sdk/config-schema` | `OpenClawSchema` | | `plugin-sdk/provider-entry` | `defineSingleProviderPluginEntry` | | `plugin-sdk/migration` | Migration provider item helpers such as `createMigrationItem`, reason constants, item status markers, redaction helpers, and `summarizeMigrationItems` | -| `plugin-sdk/migration-runtime` | Runtime migration helpers such as `copyMigrationFileItem`, `withCachedMigrationConfigRuntime`, and `writeMigrationReport` | +| `plugin-sdk/migration-runtime` | Runtime migration helpers such as `copyMigrationFileItem`, `resolvePlannedMigrationTargets`, `withCachedMigrationConfigRuntime`, and `writeMigrationReport` | | `plugin-sdk/health` | Doctor health-check registration, detection, repair, selection, severity, and finding types for bundled health consumers | ### Deprecated compatibility and test helpers @@ -190,7 +190,7 @@ usage endpoint failed or returned no usable usage data. | `plugin-sdk/approval-handler-adapter-runtime` | Lightweight native approval adapter loading helpers for hot channel entrypoints | | `plugin-sdk/approval-handler-runtime` | Broader approval handler runtime helpers; prefer the narrower adapter/gateway seams when they are enough | | `plugin-sdk/approval-native-runtime` | Native approval target, account-binding, route-gate, forwarding fallback, and local native exec prompt suppression helpers | - | `plugin-sdk/approval-reaction-runtime` | Hardcoded approval reaction bindings, reaction prompt payloads, reaction target stores, and compatibility export for local native exec prompt suppression | + | `plugin-sdk/approval-reaction-runtime` | Hardcoded approval reaction bindings, reaction prompt payloads, reaction target stores, reaction hint text helpers, and compatibility export for local native exec prompt suppression | | `plugin-sdk/approval-reply-runtime` | Exec/plugin approval reply payload helpers | | `plugin-sdk/approval-runtime` | Exec/plugin approval payload helpers, native approval routing/runtime helpers, and structured approval display helpers such as `formatApprovalDisplayPath` | | `plugin-sdk/reply-dedupe` | Narrow inbound reply dedupe reset helpers | @@ -241,7 +241,7 @@ usage endpoint failed or returned no usable usage data. | `plugin-sdk/runtime-config-snapshot` | Current process config snapshot helpers such as `getRuntimeConfig`, `getRuntimeConfigSnapshot`, and test snapshot setters | | `plugin-sdk/telegram-command-config` | Telegram command-name/description normalization and duplicate/conflict checks, even when the bundled Telegram contract surface is unavailable | | `plugin-sdk/text-autolink-runtime` | File-reference autolink detection without the broad text barrel | - | `plugin-sdk/approval-reaction-runtime` | Hardcoded approval reaction bindings, reaction prompt payloads, reaction target stores, and compatibility export for local native exec prompt suppression | + | `plugin-sdk/approval-reaction-runtime` | Hardcoded approval reaction bindings, reaction prompt payloads, reaction target stores, reaction hint text helpers, and compatibility export for local native exec prompt suppression | | `plugin-sdk/approval-runtime` | Exec/plugin approval helpers, approval-capability builders, auth/profile helpers, native routing/runtime helpers, and structured approval display path formatting | | `plugin-sdk/reply-runtime` | Shared inbound/reply runtime helpers, chunking, dispatch, heartbeat, reply planner | | `plugin-sdk/reply-dispatch-runtime` | Narrow reply dispatch/finalize and conversation-label helpers | diff --git a/extensions/imessage/src/approval-reactions.ts b/extensions/imessage/src/approval-reactions.ts index 53d1a5e6db3d..3a031dc2d53f 100644 --- a/extensions/imessage/src/approval-reactions.ts +++ b/extensions/imessage/src/approval-reactions.ts @@ -1,7 +1,9 @@ // Imessage plugin module implements approval reactions behavior. import { + addApprovalReactionHintToText, buildApprovalReactionHint, createApprovalReactionTargetStore, + hasApprovalReactionHintText, listApprovalReactionBindings, resolveApprovalReactionTarget, type ApprovalReactionDecisionBinding, @@ -243,38 +245,15 @@ export function buildIMessageApprovalReactionHint( return buildApprovalReactionHint({ allowedDecisions }); } -function insertIMessageApprovalReactionHintNearHeader(params: { - text: string; - hint: string; -}): string { - const lines = params.text.split(/\r?\n/); - const idLineIndex = lines.findIndex((line) => /^ID:\s*\S+/.test(line.trim())); - if (idLineIndex >= 0) { - const before = lines.slice(0, idLineIndex + 1).join("\n"); - const after = lines - .slice(idLineIndex + 1) - .join("\n") - .replace(/^\n+/, ""); - return after ? `${before}\n\n${params.hint}\n\n${after}` : `${before}\n\n${params.hint}`; - } - return `${params.hint}\n\n${params.text}`; -} - export function addIMessageApprovalReactionHintToText(params: { text: string; allowedDecisions: readonly ExecApprovalReplyDecision[]; }): string { - if (/(^|\n)React with:\s*(\n|$)/i.test(params.text)) { - return params.text; - } - const hint = buildIMessageApprovalReactionHint(params.allowedDecisions); - return hint - ? insertIMessageApprovalReactionHintNearHeader({ text: params.text, hint }) - : params.text; + return addApprovalReactionHintToText(params); } export function appendIMessageApprovalReactionHintForOutboundMessage(text: string): string { - if (/(^|\n)React with:\s*(\n|$)/i.test(text)) { + if (hasApprovalReactionHintText(text)) { return text; } const binding = extractIMessageApprovalPromptBinding(text); diff --git a/extensions/migrate-claude/targets.ts b/extensions/migrate-claude/targets.ts index c23440deb1d6..7390fe78c6c7 100644 --- a/extensions/migrate-claude/targets.ts +++ b/extensions/migrate-claude/targets.ts @@ -1,31 +1,5 @@ -// Migrate Claude plugin module implements targets behavior. -import path from "node:path"; -import { - resolveAgentConfig, - resolveAgentWorkspaceDir, - resolveDefaultAgentId, -} from "openclaw/plugin-sdk/agent-runtime"; -import type { MigrationProviderContext } from "openclaw/plugin-sdk/plugin-entry"; -import { resolveHomePath } from "./helpers.js"; - -export type PlannedTargets = { - workspaceDir: string; - stateDir: string; - agentDir: string; -}; - -export function resolveTargets(ctx: MigrationProviderContext): PlannedTargets { - const cfg = ctx.config; - const agentId = resolveDefaultAgentId(cfg); - const workspaceDir = resolveAgentWorkspaceDir(cfg, agentId); - const configuredAgentDir = resolveAgentConfig(cfg, agentId)?.agentDir?.trim(); - const agentDir = - ctx.runtime?.agent?.resolveAgentDir(cfg, agentId) ?? - (configuredAgentDir ? resolveHomePath(configuredAgentDir) : undefined) ?? - path.join(ctx.stateDir, "agents", agentId, "agent"); - return { - workspaceDir, - stateDir: ctx.stateDir, - agentDir, - }; -} +// Migrate Claude plugin re-exports the shared migration target resolution. +export { + resolvePlannedMigrationTargets as resolveTargets, + type PlannedMigrationTargets as PlannedTargets, +} from "openclaw/plugin-sdk/migration-runtime"; diff --git a/extensions/migrate-hermes/targets.ts b/extensions/migrate-hermes/targets.ts index f2284be1115e..3bb0cf08ab6c 100644 --- a/extensions/migrate-hermes/targets.ts +++ b/extensions/migrate-hermes/targets.ts @@ -1,31 +1,5 @@ -// Migrate Hermes plugin module implements targets behavior. -import path from "node:path"; -import { - resolveAgentConfig, - resolveAgentWorkspaceDir, - resolveDefaultAgentId, -} from "openclaw/plugin-sdk/agent-runtime"; -import type { MigrationProviderContext } from "openclaw/plugin-sdk/plugin-entry"; -import { resolveHomePath } from "./helpers.js"; - -export type PlannedTargets = { - workspaceDir: string; - stateDir: string; - agentDir: string; -}; - -export function resolveTargets(ctx: MigrationProviderContext): PlannedTargets { - const cfg = ctx.config; - const agentId = resolveDefaultAgentId(cfg); - const workspaceDir = resolveAgentWorkspaceDir(cfg, agentId); - const configuredAgentDir = resolveAgentConfig(cfg, agentId)?.agentDir?.trim(); - const agentDir = - ctx.runtime?.agent?.resolveAgentDir(cfg, agentId) ?? - (configuredAgentDir ? resolveHomePath(configuredAgentDir) : undefined) ?? - path.join(ctx.stateDir, "agents", agentId, "agent"); - return { - workspaceDir, - stateDir: ctx.stateDir, - agentDir, - }; -} +// Migrate Hermes plugin re-exports the shared migration target resolution. +export { + resolvePlannedMigrationTargets as resolveTargets, + type PlannedMigrationTargets as PlannedTargets, +} from "openclaw/plugin-sdk/migration-runtime"; diff --git a/extensions/signal/src/approval-reactions.ts b/extensions/signal/src/approval-reactions.ts index 33dd7fb004a7..6cd2e35baae5 100644 --- a/extensions/signal/src/approval-reactions.ts +++ b/extensions/signal/src/approval-reactions.ts @@ -1,8 +1,10 @@ // Signal plugin module implements approval reactions behavior. import { matchesApprovalRequestFilters } from "openclaw/plugin-sdk/approval-client-runtime"; import { + addApprovalReactionHintToText, buildApprovalReactionHint, createApprovalReactionTargetStore, + hasApprovalReactionHintText, listApprovalReactionBindings, resolveApprovalReactionTarget, type ApprovalReactionDecisionBinding, @@ -362,38 +364,11 @@ export function buildSignalApprovalReactionHint( return buildApprovalReactionHint({ allowedDecisions }); } -function insertSignalApprovalReactionHintNearHeader(params: { - text: string; - hint: string; -}): string { - const lines = params.text.split(/\r?\n/); - const idLineIndex = lines.findIndex((line) => /^ID:\s*\S+/.test(line.trim())); - if (idLineIndex >= 0) { - const before = lines.slice(0, idLineIndex + 1).join("\n"); - const after = lines - .slice(idLineIndex + 1) - .join("\n") - .replace(/^\n+/, ""); - return after ? `${before}\n\n${params.hint}\n\n${after}` : `${before}\n\n${params.hint}`; - } - return `${params.hint}\n\n${params.text}`; -} - export function addSignalApprovalReactionHintToText(params: { text: string; allowedDecisions: readonly ExecApprovalReplyDecision[]; }): string { - if (hasSignalApprovalReactionHintText(params.text)) { - return params.text; - } - const hint = buildSignalApprovalReactionHint(params.allowedDecisions); - return hint - ? insertSignalApprovalReactionHintNearHeader({ text: params.text, hint }) - : params.text; -} - -function hasSignalApprovalReactionHintText(text?: string | null): boolean { - return /(^|\n)React with:\s*(\n|$)/i.test(text ?? ""); + return addApprovalReactionHintToText(params); } function buildTargetRoute(params: { @@ -571,8 +546,8 @@ function listDeliveredSignalMessageIdsWithVisibleHint(params: { const ids = candidates .filter((result) => resultsWithVisibleText.length > 0 - ? hasSignalApprovalReactionHintText(readSignalDeliveryVisibleText(result)) - : hasSignalApprovalReactionHintText(params.payload.text), + ? hasApprovalReactionHintText(readSignalDeliveryVisibleText(result)) + : hasApprovalReactionHintText(params.payload.text), ) .map((result) => normalizeOptionalString(result.messageId)) .filter((messageId): messageId is string => Boolean(messageId && messageId !== "unknown")); @@ -595,7 +570,7 @@ export function registerSignalApprovalReactionTargetForDeliveredPayload(params: if (!metadata?.allowedDecisions || metadata.allowedDecisions.length === 0) { return false; } - if (!hasSignalApprovalReactionHintText(params.payload.text)) { + if (!hasApprovalReactionHintText(params.payload.text)) { return false; } if ( diff --git a/scripts/plugin-sdk-surface-report.mjs b/scripts/plugin-sdk-surface-report.mjs index 06a2a3a907c7..4b8d579d1535 100644 --- a/scripts/plugin-sdk-surface-report.mjs +++ b/scripts/plugin-sdk-surface-report.mjs @@ -202,8 +202,8 @@ let publicDeprecatedExportsByEntrypointBudget; try { budgets = { publicEntrypoints: readBudgetEnv("OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_ENTRYPOINTS", 323), - publicExports: readBudgetEnv("OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_EXPORTS", 10416), - publicFunctionExports: readBudgetEnv("OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_FUNCTION_EXPORTS", 5230), + publicExports: readBudgetEnv("OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_EXPORTS", 10421), + publicFunctionExports: readBudgetEnv("OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_FUNCTION_EXPORTS", 5234), publicDeprecatedExports: readBudgetEnv( "OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_DEPRECATED_EXPORTS", 3261, diff --git a/src/agents/embedded-agent-runner/run/llm-idle-timeout.ts b/src/agents/embedded-agent-runner/run/llm-idle-timeout.ts index 4ea8adf5ff63..7dab2db64273 100644 --- a/src/agents/embedded-agent-runner/run/llm-idle-timeout.ts +++ b/src/agents/embedded-agent-runner/run/llm-idle-timeout.ts @@ -198,6 +198,43 @@ function isOllamaCloudModel(model: { id?: string; provider?: string } | undefine return bareModelId.endsWith(":cloud"); } +type RuntimeModelLocality = { + isLocalRuntimeModel: boolean; + isExplicitLocalHostnameRuntimeModel: boolean; + isSelfHostedHostnameRuntimeModel: boolean; +}; + +/** + * Classifies the model endpoint locality shared by the idle and first-event + * watchdogs. Ollama `*:cloud` models stay "cloud" even behind a local proxy. + */ +function resolveRuntimeModelLocality(params?: { + cfg?: OpenClawConfig; + model?: { baseUrl?: string; id?: string; provider?: string }; +}): RuntimeModelLocality { + const baseUrl = params?.model?.baseUrl; + if (typeof baseUrl !== "string" || baseUrl.length === 0) { + return { + isLocalRuntimeModel: false, + isExplicitLocalHostnameRuntimeModel: false, + isSelfHostedHostnameRuntimeModel: false, + }; + } + const notCloudModel = !isOllamaCloudModel(params?.model); + return { + isLocalRuntimeModel: isLocalProviderBaseUrl(baseUrl) && notCloudModel, + isExplicitLocalHostnameRuntimeModel: isExplicitLocalHostnameBaseUrl(baseUrl) && notCloudModel, + isSelfHostedHostnameRuntimeModel: + isBareProviderHostnameBaseUrl(baseUrl) && + (isSelfHostedProviderId(params?.model?.provider) || + hasConfiguredLocalProviderSignal({ + cfg: params?.cfg, + provider: params?.model?.provider, + })) && + notCloudModel, + }; +} + /** * Resolves the stream-idle watchdog timeout for one embedded run. Explicit * provider request timeouts and bounded run/agent timeouts cap the watchdog; @@ -220,22 +257,11 @@ export function resolveLlmIdleTimeoutMs(params?: { const hasExplicitRunTimeout = typeof runTimeoutMs === "number" && Number.isFinite(runTimeoutMs) && runTimeoutMs > 0; const runTimeoutIsNoTimeout = hasExplicitRunTimeout && runTimeoutMs >= MAX_TIMER_TIMEOUT_MS; - const baseUrl = params?.model?.baseUrl; - const isLocalProvider = - typeof baseUrl === "string" && baseUrl.length > 0 && isLocalProviderBaseUrl(baseUrl); - const isLocalRuntimeModel = isLocalProvider && !isOllamaCloudModel(params?.model); - const isExplicitLocalHostnameRuntimeModel = - typeof baseUrl === "string" && - baseUrl.length > 0 && - isExplicitLocalHostnameBaseUrl(baseUrl) && - !isOllamaCloudModel(params?.model); - const isSelfHostedHostnameRuntimeModel = - typeof baseUrl === "string" && - baseUrl.length > 0 && - isBareProviderHostnameBaseUrl(baseUrl) && - (isSelfHostedProviderId(params?.model?.provider) || - hasConfiguredLocalProviderSignal({ cfg: params?.cfg, provider: params?.model?.provider })) && - !isOllamaCloudModel(params?.model); + const { + isLocalRuntimeModel, + isExplicitLocalHostnameRuntimeModel, + isSelfHostedHostnameRuntimeModel, + } = resolveRuntimeModelLocality(params); const timeoutBounds = [ runTimeoutIsNoTimeout ? undefined : runTimeoutMs, hasExplicitRunTimeout ? undefined : agentTimeoutMs, @@ -320,22 +346,11 @@ export function resolveLlmFirstEventTimeoutMs(params?: { const hasExplicitRunTimeout = typeof runTimeoutMs === "number" && Number.isFinite(runTimeoutMs) && runTimeoutMs > 0; const runTimeoutIsBounded = hasExplicitRunTimeout && runTimeoutMs < MAX_TIMER_TIMEOUT_MS; - const baseUrl = params?.model?.baseUrl; - const isLocalProvider = - typeof baseUrl === "string" && baseUrl.length > 0 && isLocalProviderBaseUrl(baseUrl); - const isLocalRuntimeModel = isLocalProvider && !isOllamaCloudModel(params?.model); - const isExplicitLocalHostnameRuntimeModel = - typeof baseUrl === "string" && - baseUrl.length > 0 && - isExplicitLocalHostnameBaseUrl(baseUrl) && - !isOllamaCloudModel(params?.model); - const isSelfHostedHostnameRuntimeModel = - typeof baseUrl === "string" && - baseUrl.length > 0 && - isBareProviderHostnameBaseUrl(baseUrl) && - (isSelfHostedProviderId(params?.model?.provider) || - hasConfiguredLocalProviderSignal({ cfg: params?.cfg, provider: params?.model?.provider })) && - !isOllamaCloudModel(params?.model); + const { + isLocalRuntimeModel, + isExplicitLocalHostnameRuntimeModel, + isSelfHostedHostnameRuntimeModel, + } = resolveRuntimeModelLocality(params); const timeoutBounds = [ runTimeoutIsBounded ? runTimeoutMs : undefined, hasExplicitRunTimeout ? undefined : agentTimeoutMs, diff --git a/src/agents/mcp-http-fetch.ts b/src/agents/mcp-http-fetch.ts index f9909bdefb76..c2ff26f3f6d7 100644 --- a/src/agents/mcp-http-fetch.ts +++ b/src/agents/mcp-http-fetch.ts @@ -6,6 +6,7 @@ import fs from "node:fs"; import type { FetchLike } from "@modelcontextprotocol/sdk/shared/transport.js"; import { fetchWithSsrFGuard } from "../infra/net/fetch-guard.js"; +import { wrapGuardedBodyStream } from "../infra/net/guarded-body-stream.js"; import { ssrfPolicyFromHttpBaseUrlAllowedOrigin, type PinnedDispatcherPolicy, @@ -25,11 +26,6 @@ const fetchWithUndiciGuard = async ( ): Promise => await fetchWithUndici(input instanceof Request ? input.url : input, init); const MCP_HTTP_MAX_REDIRECTS = 20; -const managedMcpResponseCleanupRegistry = new FinalizationRegistry<{ - finalize: () => Promise; -}>((held) => { - void held.finalize(); -}); function resolveFetchRequest(input: RequestInfo | URL, init?: RequestInit) { if (input instanceof Request) { @@ -82,47 +78,11 @@ async function buildManagedMcpResponse( return await ensureGlobalFetchResponse(response); } - const source = response.body; - let reader: ReadableStreamDefaultReader | undefined; - let released = false; - const cleanupRegistrationToken = {}; - const finalize = async () => { - if (released) { - return; - } - released = true; - managedMcpResponseCleanupRegistry.unregister(cleanupRegistrationToken); - await reader?.cancel().catch(() => undefined); - await release().catch(() => undefined); - }; - const wrappedBody = new ReadableStream({ - start() { - reader = source.getReader(); - }, - async pull(controller) { - try { - const chunk = await reader?.read(); - if (!chunk || chunk.done) { - controller.close(); - await finalize(); - return; - } - refreshTimeout?.(); - controller.enqueue(chunk.value); - } catch (error) { - controller.error(error); - await finalize(); - } - }, - async cancel(reason) { - try { - await reader?.cancel(reason); - } finally { - await finalize(); - } - }, + const wrappedBody = wrapGuardedBodyStream({ + body: response.body, + cleanup: release, + refreshTimeout, }); - managedMcpResponseCleanupRegistry.register(wrappedBody, { finalize }, cleanupRegistrationToken); return await ensureGlobalFetchResponse( new Response(wrappedBody, { status: response.status, diff --git a/src/agents/provider-transport-fetch.ts b/src/agents/provider-transport-fetch.ts index 0d0f37b7105a..d24f28150423 100644 --- a/src/agents/provider-transport-fetch.ts +++ b/src/agents/provider-transport-fetch.ts @@ -18,6 +18,7 @@ import { fetchWithSsrFGuard, withTrustedEnvProxyGuardedFetchMode, } from "../infra/net/fetch-guard.js"; +import { wrapGuardedBodyStream } from "../infra/net/guarded-body-stream.js"; import { shouldUseEnvHttpProxyForUrl } from "../infra/net/proxy-env.js"; import { mergeSsrFPolicies, @@ -550,12 +551,6 @@ function shouldBypassLongSdkRetry(response: Response): boolean { return status === 429; } -const managedStreamCleanupRegistry = new FinalizationRegistry<{ finalize: () => Promise }>( - (held) => { - void held.finalize(); - }, -); - function buildManagedResponse( response: Response, release: () => Promise, @@ -569,53 +564,18 @@ function buildManagedResponse( void release().finally(finalizeLocalServiceLease); return response; } - const source = response.body; - let reader: ReadableStreamDefaultReader | undefined; - let released = false; - const cleanupRegistrationToken = {}; - const finalize = async () => { - if (released) { - return; - } - released = true; - managedStreamCleanupRegistry.unregister(cleanupRegistrationToken); - try { - await reader?.cancel().catch(() => undefined); - await release().catch(() => undefined); - } finally { - finalizeLocalServiceLease(); - } - }; - const wrappedBody = new ReadableStream({ - start() { - reader = source.getReader(); - }, - async pull(controller) { + const wrappedBody = wrapGuardedBodyStream({ + body: response.body, + // Lease release must survive a failed guard release so local services do not leak. + cleanup: async () => { try { - const chunk = await reader?.read(); - if (!chunk || chunk.done) { - controller.close(); - await finalize(); - return; - } - refreshTimeout?.(); - controller.enqueue(chunk.value); - } catch (error) { - controller.error(error); - await finalize(); - } - }, - async cancel(reason) { - try { - await reader?.cancel(reason); + await release().catch(() => undefined); } finally { - await finalize(); + finalizeLocalServiceLease(); } }, + refreshTimeout, }); - // Stream consumers should cancel deterministically; this catches abandoned - // wrapper bodies so guarded dispatchers and local-service leases do not leak. - managedStreamCleanupRegistry.register(wrappedBody, { finalize }, cleanupRegistrationToken); return new Response(wrappedBody, { status: response.status, statusText: response.statusText, diff --git a/src/auto-reply/reply/commands-diagnostics.ts b/src/auto-reply/reply/commands-diagnostics.ts index 650252257160..b14d5c1f4727 100644 --- a/src/auto-reply/reply/commands-diagnostics.ts +++ b/src/auto-reply/reply/commands-diagnostics.ts @@ -20,6 +20,7 @@ import { deliverPrivateCommandReply, readCommandDeliveryTarget, readCommandMessageThreadId, + resolveCommandExecApprovalRoute, resolvePrivateCommandApprovalRouteExpiresAtMs, resolvePrivateCommandRouteTargets, type PrivateCommandRouteTarget, @@ -297,7 +298,6 @@ async function requestGatewayDiagnosticsExportApproval( sessionKey: params.sessionKey, config: params.cfg, }); - const messageThreadId = readCommandMessageThreadId(params); const command = buildGatewayDiagnosticsExportJsonCommand(); try { const execTool = deps.createExecTool({ @@ -316,16 +316,10 @@ async function requestGatewayDiagnosticsExportApproval( sessionKey: params.sessionKey, mainKey: params.cfg.session?.mainKey, sessionScope: params.cfg.session?.scope, - messageProvider: options.privateApprovalTarget?.channel ?? params.command.channel, - currentChannelId: options.privateApprovalTarget?.to ?? readCommandDeliveryTarget(params), - currentThreadTs: options.privateApprovalTarget - ? options.privateApprovalTarget.threadId == null - ? undefined - : String(options.privateApprovalTarget.threadId) - : messageThreadId, - accountId: options.privateApprovalTarget - ? (options.privateApprovalTarget.accountId ?? undefined) - : (params.ctx.AccountId ?? undefined), + ...resolveCommandExecApprovalRoute({ + commandParams: params, + privateApprovalTarget: options.privateApprovalTarget, + }), notifyOnExit: params.cfg.tools?.exec?.notifyOnExit, notifyOnExitEmptySuccess: params.cfg.tools?.exec?.notifyOnExitEmptySuccess, }); diff --git a/src/auto-reply/reply/commands-export-trajectory.ts b/src/auto-reply/reply/commands-export-trajectory.ts index 24ba02a132c2..fd4584455ef1 100644 --- a/src/auto-reply/reply/commands-export-trajectory.ts +++ b/src/auto-reply/reply/commands-export-trajectory.ts @@ -15,6 +15,7 @@ import { deliverPrivateCommandReply, readCommandDeliveryTarget, readCommandMessageThreadId, + resolveCommandExecApprovalRoute, resolvePrivateCommandApprovalRouteExpiresAtMs, resolvePrivateCommandRouteTargets, type PrivateCommandRouteTarget, @@ -172,7 +173,6 @@ async function requestTrajectoryExportApproval( sessionKey: params.sessionKey, config: params.cfg, }); - const messageThreadId = readCommandMessageThreadId(params); try { const execTool = deps.createExecTool({ host: "gateway", @@ -190,16 +190,10 @@ async function requestTrajectoryExportApproval( sessionStore: params.cfg.session?.store, mainKey: params.cfg.session?.mainKey, sessionScope: params.cfg.session?.scope, - messageProvider: options.privateApprovalTarget?.channel ?? params.command.channel, - currentChannelId: options.privateApprovalTarget?.to ?? readCommandDeliveryTarget(params), - currentThreadTs: options.privateApprovalTarget - ? options.privateApprovalTarget.threadId == null - ? undefined - : String(options.privateApprovalTarget.threadId) - : messageThreadId, - accountId: options.privateApprovalTarget - ? (options.privateApprovalTarget.accountId ?? undefined) - : (params.ctx.AccountId ?? undefined), + ...resolveCommandExecApprovalRoute({ + commandParams: params, + privateApprovalTarget: options.privateApprovalTarget, + }), notifyOnExit: params.cfg.tools?.exec?.notifyOnExit, notifyOnExitEmptySuccess: params.cfg.tools?.exec?.notifyOnExitEmptySuccess, }); diff --git a/src/auto-reply/reply/commands-private-route.ts b/src/auto-reply/reply/commands-private-route.ts index 793d9482b64a..8248e650a967 100644 --- a/src/auto-reply/reply/commands-private-route.ts +++ b/src/auto-reply/reply/commands-private-route.ts @@ -127,6 +127,36 @@ export function readCommandDeliveryTarget(params: HandleCommandsParams): string ); } +/** + * Resolves where an exec approval prompt for a command should be delivered: + * the private owner-DM target when one was resolved, else the originating + * command surface. Keeps the fallback ternaries in one place so private and + * origin routing cannot drift between command handlers. + */ +export function resolveCommandExecApprovalRoute(params: { + commandParams: HandleCommandsParams; + privateApprovalTarget?: PrivateCommandRouteTarget; +}): { + messageProvider: string; + currentChannelId: string | undefined; + currentThreadTs: string | undefined; + accountId: string | undefined; +} { + const target = params.privateApprovalTarget; + return { + messageProvider: target?.channel ?? params.commandParams.command.channel, + currentChannelId: target?.to ?? readCommandDeliveryTarget(params.commandParams), + currentThreadTs: target + ? target.threadId == null + ? undefined + : String(target.threadId) + : readCommandMessageThreadId(params.commandParams), + accountId: target + ? (target.accountId ?? undefined) + : (params.commandParams.ctx.AccountId ?? undefined), + }; +} + function listPrivateCommandRouteCandidateChannels(originChannel: string) { const plugins = [getLoadedChannelPlugin(originChannel), ...listChannelPlugins()].filter( (plugin): plugin is NonNullable> => diff --git a/src/infra/backup-create.ts b/src/infra/backup-create.ts index e5009802b2e5..3ec63aff0fea 100644 --- a/src/infra/backup-create.ts +++ b/src/infra/backup-create.ts @@ -18,6 +18,7 @@ import { isPathWithin } from "../commands/cleanup-utils.js"; import { createLazyRuntimeModule } from "../shared/lazy-runtime.js"; import { resolveOpenClawStateSqlitePath } from "../state/openclaw-state-db.paths.js"; import { resolveHomeDir, resolveUserPath } from "../utils.js"; +import { sleep } from "../utils/sleep.js"; import { resolveRuntimeServiceVersion } from "../version.js"; import { isVolatileBackupPath } from "./backup-volatile-filter.js"; import { writeJson } from "./json-files.js"; @@ -167,12 +168,6 @@ function isTarEofRaceError(err: unknown): boolean { return /(did not encounter expected|encountered unexpected) EOF|TAR_BAD_ARCHIVE/i.test(message); } -function sleep(ms: number): Promise { - return new Promise((resolve) => { - setTimeout(resolve, ms); - }); -} - export type BackupTarRetryLogger = (message: string) => void; async function writeTarArchiveWithRetry(params: { diff --git a/src/infra/net/guarded-body-stream.ts b/src/infra/net/guarded-body-stream.ts new file mode 100644 index 000000000000..c5dd577d2930 --- /dev/null +++ b/src/infra/net/guarded-body-stream.ts @@ -0,0 +1,70 @@ +/** + * Shared body-stream cleanup for guarded fetch consumers (`fetchWithSsrFGuard` + * callers that re-wrap streaming responses). + */ + +// Catches wrapper bodies abandoned without cancel/consume so guarded dispatchers +// (and caller resources hooked into `cleanup`) do not leak with the stream. +const guardedBodyCleanupRegistry = new FinalizationRegistry<{ finalize: () => Promise }>( + (held) => { + void held.finalize(); + }, +); + +/** + * Wraps a guarded response body so caller cleanup runs exactly once when the + * stream completes, errors, is cancelled, or is garbage-collected unconsumed. + * Cleanup failures are swallowed: releasing guard resources must never break + * response consumption. + */ +export function wrapGuardedBodyStream(params: { + body: ReadableStream; + cleanup: () => Promise | void; + refreshTimeout?: () => void; +}): ReadableStream { + let reader: ReadableStreamDefaultReader | undefined; + let finalized = false; + const cleanupRegistrationToken = {}; + const finalize = async () => { + if (finalized) { + return; + } + finalized = true; + guardedBodyCleanupRegistry.unregister(cleanupRegistrationToken); + await reader?.cancel().catch(() => undefined); + try { + await params.cleanup(); + } catch { + // Best effort: guard cleanup must not surface into stream consumers. + } + }; + const wrappedBody = new ReadableStream({ + start() { + reader = params.body.getReader(); + }, + async pull(controller) { + try { + const chunk = await reader?.read(); + if (!chunk || chunk.done) { + controller.close(); + await finalize(); + return; + } + params.refreshTimeout?.(); + controller.enqueue(chunk.value); + } catch (error) { + controller.error(error); + await finalize(); + } + }, + async cancel(reason) { + try { + await reader?.cancel(reason); + } finally { + await finalize(); + } + }, + }); + guardedBodyCleanupRegistry.register(wrappedBody, { finalize }, cleanupRegistrationToken); + return wrappedBody; +} diff --git a/src/plugin-sdk/approval-reaction-runtime.ts b/src/plugin-sdk/approval-reaction-runtime.ts index e94e8b841904..a3d15b3d2867 100644 --- a/src/plugin-sdk/approval-reaction-runtime.ts +++ b/src/plugin-sdk/approval-reaction-runtime.ts @@ -130,6 +130,43 @@ export function buildApprovalReactionHint(params: { return `React with:\n\n${bindings.map((binding) => `${binding.emoji} ${binding.label}`).join("\n")}`; } +const APPROVAL_REACTION_HINT_PRESENT_RE = /(^|\n)React with:\s*(\n|$)/i; + +/** True when approval prompt text already carries a reaction hint block. */ +export function hasApprovalReactionHintText(text?: string | null): boolean { + return APPROVAL_REACTION_HINT_PRESENT_RE.test(text ?? ""); +} + +/** Inserts a reaction hint after the `ID: ` header line, else prepends it. */ +export function insertApprovalReactionHintNearIdHeader(params: { + text: string; + hint: string; +}): string { + const lines = params.text.split(/\r?\n/); + const idLineIndex = lines.findIndex((line) => /^ID:\s*\S+/.test(line.trim())); + if (idLineIndex >= 0) { + const before = lines.slice(0, idLineIndex + 1).join("\n"); + const after = lines + .slice(idLineIndex + 1) + .join("\n") + .replace(/^\n+/, ""); + return after ? `${before}\n\n${params.hint}\n\n${after}` : `${before}\n\n${params.hint}`; + } + return `${params.hint}\n\n${params.text}`; +} + +/** Adds the canonical reaction hint to approval prompt text unless one is present. */ +export function addApprovalReactionHintToText(params: { + text: string; + allowedDecisions: readonly ExecApprovalReplyDecision[]; +}): string { + if (hasApprovalReactionHintText(params.text)) { + return params.text; + } + const hint = buildApprovalReactionHint({ allowedDecisions: params.allowedDecisions }); + return hint ? insertApprovalReactionHintNearIdHeader({ text: params.text, hint }) : params.text; +} + /** Normalize reaction emoji so skin-tone and text/presentation variants match canonical bindings. */ export function normalizeApprovalReactionEmoji(reactionKey: string): string { const normalized = reactionKey diff --git a/src/plugin-sdk/migration-runtime.ts b/src/plugin-sdk/migration-runtime.ts index cbd3f192c846..233cf96b3568 100644 --- a/src/plugin-sdk/migration-runtime.ts +++ b/src/plugin-sdk/migration-runtime.ts @@ -3,7 +3,10 @@ import crypto from "node:crypto"; import fs from "node:fs/promises"; import path from "node:path"; +import { resolveAgentConfig } from "../agents/agent-scope-config.js"; +import { resolveAgentWorkspaceDir, resolveDefaultAgentId } from "../agents/agent-scope.js"; import { pathExists } from "../infra/fs-safe.js"; +import { resolveHomeRelativePath } from "../infra/home-dir.js"; import type { MigrationApplyResult, MigrationItem, @@ -19,6 +22,37 @@ import { export type { MigrationApplyResult, MigrationItem } from "../plugins/types.js"; +/** Directories a migration provider writes imported agent data into. */ +export type PlannedMigrationTargets = { + workspaceDir: string; + stateDir: string; + agentDir: string; +}; + +/** + * Resolves the default agent's workspace/state/agent directories for a + * migration run. Prefers the runtime resolver, then a configured agentDir + * (home-relative paths honor the effective-home resolution used by the rest + * of the product), then the canonical state-dir layout. + */ +export function resolvePlannedMigrationTargets( + ctx: MigrationProviderContext, +): PlannedMigrationTargets { + const cfg = ctx.config; + const agentId = resolveDefaultAgentId(cfg); + const workspaceDir = resolveAgentWorkspaceDir(cfg, agentId); + const configuredAgentDir = resolveAgentConfig(cfg, agentId)?.agentDir?.trim(); + const agentDir = + ctx.runtime?.agent?.resolveAgentDir(cfg, agentId) ?? + (configuredAgentDir ? resolveHomeRelativePath(configuredAgentDir) : undefined) ?? + path.join(ctx.stateDir, "agents", agentId, "agent"); + return { + workspaceDir, + stateDir: ctx.stateDir, + agentDir, + }; +} + /** Wrap migration runtime config access with a cached mutable snapshot during apply. */ export function withCachedMigrationConfigRuntime( runtime: MigrationProviderContext["runtime"] | undefined,