mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
refactor: consolidate core stream cleanup, watchdog locality, and approval text duplication (#99901)
* refactor(infra): share guarded body-stream cleanup across fetch consumers * refactor(agents): extract shared runtime model locality for llm watchdogs * refactor(auto-reply): share exec approval route resolution between command handlers * refactor(plugin-sdk): share approval reaction hint text helpers * refactor(plugin-sdk): share planned migration target resolution * refactor(infra): use canonical sleep helper in backup retry * docs(plugins): mention shared migration targets and reaction hint helpers * chore(plugin-sdk): pin surface budgets for reaction hint and migration target helpers
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
eafe2a8d0b
commit
be95bb72d4
@@ -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 |
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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 (
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<Response> => await fetchWithUndici(input instanceof Request ? input.url : input, init);
|
||||
|
||||
const MCP_HTTP_MAX_REDIRECTS = 20;
|
||||
const managedMcpResponseCleanupRegistry = new FinalizationRegistry<{
|
||||
finalize: () => Promise<void>;
|
||||
}>((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<Uint8Array> | 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<Uint8Array>({
|
||||
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,
|
||||
|
||||
@@ -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<void> }>(
|
||||
(held) => {
|
||||
void held.finalize();
|
||||
},
|
||||
);
|
||||
|
||||
function buildManagedResponse(
|
||||
response: Response,
|
||||
release: () => Promise<void>,
|
||||
@@ -569,53 +564,18 @@ function buildManagedResponse(
|
||||
void release().finally(finalizeLocalServiceLease);
|
||||
return response;
|
||||
}
|
||||
const source = response.body;
|
||||
let reader: ReadableStreamDefaultReader<Uint8Array> | 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<Uint8Array>({
|
||||
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,
|
||||
|
||||
@@ -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,
|
||||
});
|
||||
|
||||
@@ -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,
|
||||
});
|
||||
|
||||
@@ -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<ReturnType<typeof getLoadedChannelPlugin>> =>
|
||||
|
||||
@@ -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<void> {
|
||||
return new Promise((resolve) => {
|
||||
setTimeout(resolve, ms);
|
||||
});
|
||||
}
|
||||
|
||||
export type BackupTarRetryLogger = (message: string) => void;
|
||||
|
||||
async function writeTarArchiveWithRetry(params: {
|
||||
|
||||
@@ -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<void> }>(
|
||||
(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<Uint8Array>;
|
||||
cleanup: () => Promise<void> | void;
|
||||
refreshTimeout?: () => void;
|
||||
}): ReadableStream<Uint8Array> {
|
||||
let reader: ReadableStreamDefaultReader<Uint8Array> | 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<Uint8Array>({
|
||||
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;
|
||||
}
|
||||
@@ -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: <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
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user