Ayuriel/openclaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into pr-51359

Browse Source
This commit is contained in:
Val Alexander 2026-03-21 00:50:50 -05:00
commit 877b8f2636
No known key found for this signature in database
22 changed files with 380 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CHANGELOG.md
View File

@ -192,6 +192,9 @@ Docs: https://docs.openclaw.ai
- Web search: align onboarding, configure, and finalize with plugin-owned provider contracts, including disabled-provider recovery, config-aware credential hooks, and runtime-visible summaries. (#50935) Thanks @gumadeiras.
- Agents/replay: sanitize malformed assistant tool-call replay blocks before provider replay so follow-up Anthropic requests do not inherit the downstream `replace` crash. (#50005) Thanks @jalehman.
- Plugins/context engines: retry strict legacy `assemble()` calls without the new `prompt` field when older engines reject it, preserving prompt-aware retrieval compatibility for pre-prompt plugins. (#50848) thanks @danhdoan.
- Agents/embedded transport errors: distinguish common network failures like connection refused, DNS lookup failure, and interrupted sockets from true timeouts in embedded-run user messaging and lifecycle diagnostics. (#51419) Thanks @scoootscooob.
- Discord/startup logging: report client initialization while the gateway is still connecting instead of claiming Discord is logged in before readiness is reached. (#51425) Thanks @scoootscooob.
- Gateway/probe: honor caller `--timeout` for active local loopback probes in `gateway status`, keep inactive remote-mode loopback probes fast, and clamp probe timers to JS-safe bounds so slow local/container gateways stop reporting false timeouts. (#47533) Thanks @MonkeyLeeT.
### Breaking

8
extensions/discord/src/monitor/provider.ts
View File

@ -92,6 +92,7 @@ import { resolveDiscordPresenceUpdate } from "./presence.js";
import { resolveDiscordAllowlistConfig } from "./provider.allowlist.js";
import { runDiscordGatewayLifecycle } from "./provider.lifecycle.js";
import { resolveDiscordRestFetch } from "./rest-fetch.js";
import { formatDiscordStartupStatusMessage } from "./startup-status.js";
import type { DiscordMonitorStatusSink } from "./status.js";
import {
createNoopThreadBindingManager,
@ -972,7 +973,12 @@ export async function monitorDiscordProvider(opts: MonitorDiscordOpts = {}) {
const botIdentity =
botUserId && botUserName ? `${botUserId} (${botUserName})` : (botUserId ?? botUserName ?? "");
runtime.log?.(`logged in to discord${botIdentity ? ` as ${botIdentity}` : ""}`);
runtime.log?.(
formatDiscordStartupStatusMessage({
gatewayReady: lifecycleGateway?.isConnected === true,
botIdentity: botIdentity || undefined,
}),
);
if (lifecycleGateway?.isConnected) {
opts.setStatus?.(createConnectedChannelStatusPatch());
}

30
extensions/discord/src/monitor/startup-status.test.ts Normal file
View File

@ -0,0 +1,30 @@
import { describe, expect, it } from "vitest";
import { formatDiscordStartupStatusMessage } from "./startup-status.js";
describe("formatDiscordStartupStatusMessage", () => {
it("reports logged-in status only after the gateway is ready", () => {
expect(
formatDiscordStartupStatusMessage({
gatewayReady: true,
botIdentity: "bot-1 (Molty)",
}),
).toBe("logged in to discord as bot-1 (Molty)");
});
it("reports client initialization while gateway readiness is still pending", () => {
expect(
formatDiscordStartupStatusMessage({
gatewayReady: false,
botIdentity: "bot-1 (Molty)",
}),
).toBe("discord client initialized as bot-1 (Molty); awaiting gateway readiness");
});
it("handles missing identity without awkward punctuation", () => {
expect(
formatDiscordStartupStatusMessage({
gatewayReady: false,
}),
).toBe("discord client initialized; awaiting gateway readiness");
});
});

10
extensions/discord/src/monitor/startup-status.ts Normal file
View File

@ -0,0 +1,10 @@
export function formatDiscordStartupStatusMessage(params: {
gatewayReady: boolean;
botIdentity?: string;
}): string {
const identitySuffix = params.botIdentity ? ` as ${params.botIdentity}` : "";
if (params.gatewayReady) {
return `logged in to discord${identitySuffix}`;
}
return `discord client initialized${identitySuffix}; awaiting gateway readiness`;
}

58
extensions/feishu/src/thread-bindings.ts
View File

@ -51,16 +51,18 @@ type FeishuThreadBindingsState = {
};
const FEISHU_THREAD_BINDINGS_STATE_KEY = Symbol.for("openclaw.feishuThreadBindingsState");
const state = resolveGlobalSingleton<FeishuThreadBindingsState>(
FEISHU_THREAD_BINDINGS_STATE_KEY,
() => ({
managersByAccountId: new Map(),
bindingsByAccountConversation: new Map(),
}),
);
let state: FeishuThreadBindingsState | undefined;
const MANAGERS_BY_ACCOUNT_ID = state.managersByAccountId;
const BINDINGS_BY_ACCOUNT_CONVERSATION = state.bindingsByAccountConversation;
function getState(): FeishuThreadBindingsState {
state ??= resolveGlobalSingleton<FeishuThreadBindingsState>(
FEISHU_THREAD_BINDINGS_STATE_KEY,
() => ({
managersByAccountId: new Map(),
bindingsByAccountConversation: new Map(),
}),
);
return state;
}
function resolveBindingKey(params: { accountId: string; conversationId: string }): string {
return `${params.accountId}:${params.conversationId}`;
@ -119,7 +121,7 @@ export function createFeishuThreadBindingManager(params: {
cfg: OpenClawConfig;
}): FeishuThreadBindingManager {
const accountId = normalizeAccountId(params.accountId);
const existing = MANAGERS_BY_ACCOUNT_ID.get(accountId);
const existing = getState().managersByAccountId.get(accountId);
if (existing) {
return existing;
}
@ -138,9 +140,11 @@ export function createFeishuThreadBindingManager(params: {
const manager: FeishuThreadBindingManager = {
accountId,
getByConversationId: (conversationId) =>
BINDINGS_BY_ACCOUNT_CONVERSATION.get(resolveBindingKey({ accountId, conversationId })),
getState().bindingsByAccountConversation.get(
resolveBindingKey({ accountId, conversationId }),
),
listBySessionKey: (targetSessionKey) =>
[...BINDINGS_BY_ACCOUNT_CONVERSATION.values()].filter(
[...getState().bindingsByAccountConversation.values()].filter(
(record) => record.accountId === accountId && record.targetSessionKey === targetSessionKey,
),
bindConversation: ({
@ -184,7 +188,7 @@ export function createFeishuThreadBindingManager(params: {
boundAt: now,
lastActivityAt: now,
};
BINDINGS_BY_ACCOUNT_CONVERSATION.set(
getState().bindingsByAccountConversation.set(
resolveBindingKey({ accountId, conversationId: normalizedConversationId }),
record,
);
@ -192,30 +196,30 @@ export function createFeishuThreadBindingManager(params: {
},
touchConversation: (conversationId, at = Date.now()) => {
const key = resolveBindingKey({ accountId, conversationId });
const existingRecord = BINDINGS_BY_ACCOUNT_CONVERSATION.get(key);
const existingRecord = getState().bindingsByAccountConversation.get(key);
if (!existingRecord) {
return null;
}
const updated = { ...existingRecord, lastActivityAt: at };
BINDINGS_BY_ACCOUNT_CONVERSATION.set(key, updated);
getState().bindingsByAccountConversation.set(key, updated);
return updated;
},
unbindConversation: (conversationId) => {
const key = resolveBindingKey({ accountId, conversationId });
const existingRecord = BINDINGS_BY_ACCOUNT_CONVERSATION.get(key);
const existingRecord = getState().bindingsByAccountConversation.get(key);
if (!existingRecord) {
return null;
}
BINDINGS_BY_ACCOUNT_CONVERSATION.delete(key);
getState().bindingsByAccountConversation.delete(key);
return existingRecord;
},
unbindBySessionKey: (targetSessionKey) => {
const removed: FeishuThreadBindingRecord[] = [];
for (const record of [...BINDINGS_BY_ACCOUNT_CONVERSATION.values()]) {
for (const record of [...getState().bindingsByAccountConversation.values()]) {
if (record.accountId !== accountId || record.targetSessionKey !== targetSessionKey) {
continue;
}
BINDINGS_BY_ACCOUNT_CONVERSATION.delete(
getState().bindingsByAccountConversation.delete(
resolveBindingKey({ accountId, conversationId: record.conversationId }),
);
removed.push(record);
@ -223,12 +227,12 @@ export function createFeishuThreadBindingManager(params: {
return removed;
},
stop: () => {
for (const key of [...BINDINGS_BY_ACCOUNT_CONVERSATION.keys()]) {
for (const key of [...getState().bindingsByAccountConversation.keys()]) {
if (key.startsWith(`${accountId}:`)) {
BINDINGS_BY_ACCOUNT_CONVERSATION.delete(key);
getState().bindingsByAccountConversation.delete(key);
}
}
MANAGERS_BY_ACCOUNT_ID.delete(accountId);
getState().managersByAccountId.delete(accountId);
unregisterSessionBindingAdapter({ channel: "feishu", accountId });
},
};
@ -290,22 +294,22 @@ export function createFeishuThreadBindingManager(params: {
},
});
MANAGERS_BY_ACCOUNT_ID.set(accountId, manager);
getState().managersByAccountId.set(accountId, manager);
return manager;
}
export function getFeishuThreadBindingManager(
accountId?: string,
): FeishuThreadBindingManager | null {
return MANAGERS_BY_ACCOUNT_ID.get(normalizeAccountId(accountId)) ?? null;
return getState().managersByAccountId.get(normalizeAccountId(accountId)) ?? null;
}
export const __testing = {
resetFeishuThreadBindingsForTests() {
for (const manager of MANAGERS_BY_ACCOUNT_ID.values()) {
for (const manager of getState().managersByAccountId.values()) {
manager.stop();
}
MANAGERS_BY_ACCOUNT_ID.clear();
BINDINGS_BY_ACCOUNT_CONVERSATION.clear();
getState().managersByAccountId.clear();
getState().bindingsByAccountConversation.clear();
},
};

19
extensions/slack/src/sent-thread-cache.ts
View File

@ -15,7 +15,12 @@ const MAX_ENTRIES = 5000;
*/
const SLACK_THREAD_PARTICIPATION_KEY = Symbol.for("openclaw.slackThreadParticipation");
const threadParticipation = resolveGlobalMap<string, number>(SLACK_THREAD_PARTICIPATION_KEY);
let threadParticipation: Map<string, number> | undefined;
function getThreadParticipation(): Map<string, number> {
threadParticipation ??= resolveGlobalMap<string, number>(SLACK_THREAD_PARTICIPATION_KEY);
return threadParticipation;
}
function makeKey(accountId: string, channelId: string, threadTs: string): string {
return `${accountId}:${channelId}:${threadTs}`;
@ -23,17 +28,17 @@ function makeKey(accountId: string, channelId: string, threadTs: string): string
function evictExpired(): void {
const now = Date.now();
for (const [key, timestamp] of threadParticipation) {
for (const [key, timestamp] of getThreadParticipation()) {
if (now - timestamp > TTL_MS) {
threadParticipation.delete(key);
getThreadParticipation().delete(key);
}
}
}
function evictOldest(): void {
const oldest = threadParticipation.keys().next().value;
const oldest = getThreadParticipation().keys().next().value;
if (oldest) {
threadParticipation.delete(oldest);
getThreadParticipation().delete(oldest);
}
}
@ -45,6 +50,7 @@ export function recordSlackThreadParticipation(
if (!accountId || !channelId || !threadTs) {
return;
}
const threadParticipation = getThreadParticipation();
if (threadParticipation.size >= MAX_ENTRIES) {
evictExpired();
}
@ -63,6 +69,7 @@ export function hasSlackThreadParticipation(
return false;
}
const key = makeKey(accountId, channelId, threadTs);
const threadParticipation = getThreadParticipation();
const timestamp = threadParticipation.get(key);
if (timestamp == null) {
return false;
@ -75,5 +82,5 @@ export function hasSlackThreadParticipation(
}
export function clearSlackThreadParticipationCache(): void {
threadParticipation.clear();
getThreadParticipation().clear();
}

14
extensions/telegram/src/draft-stream.ts
View File

@ -28,11 +28,17 @@ type TelegramSendMessageDraft = (
*/
const TELEGRAM_DRAFT_STREAM_STATE_KEY = Symbol.for("openclaw.telegramDraftStreamState");
const draftStreamState = resolveGlobalSingleton(TELEGRAM_DRAFT_STREAM_STATE_KEY, () => ({
nextDraftId: 0,
}));
let draftStreamState: { nextDraftId: number } | undefined;
function getDraftStreamState(): { nextDraftId: number } {
draftStreamState ??= resolveGlobalSingleton(TELEGRAM_DRAFT_STREAM_STATE_KEY, () => ({
nextDraftId: 0,
}));
return draftStreamState;
}
function allocateTelegramDraftId(): number {
const draftStreamState = getDraftStreamState();
draftStreamState.nextDraftId =
draftStreamState.nextDraftId >= TELEGRAM_DRAFT_ID_MAX ? 1 : draftStreamState.nextDraftId + 1;
return draftStreamState.nextDraftId;
@ -454,6 +460,6 @@ export function createTelegramDraftStream(params: {
export const __testing = {
resetTelegramDraftStreamForTests() {
draftStreamState.nextDraftId = 0;
getDraftStreamState().nextDraftId = 0;
},
};

39
extensions/telegram/src/format.ts
View File

@ -103,17 +103,34 @@ function escapeRegex(str: string): string {
return str.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
}
const FILE_EXTENSIONS_PATTERN = Array.from(FILE_REF_EXTENSIONS_WITH_TLD).map(escapeRegex).join("|");
const AUTO_LINKED_ANCHOR_PATTERN = /<a\s+href="https?:\/\/([^"]+)"[^>]*>\1<\/a>/gi;
const FILE_REFERENCE_PATTERN = new RegExp(
`(^|[^a-zA-Z0-9_\\-/])([a-zA-Z0-9_.\\-./]+\\.(?:${FILE_EXTENSIONS_PATTERN}))(?=$|[^a-zA-Z0-9_\\-/])`,
"gi",
);
const ORPHANED_TLD_PATTERN = new RegExp(
`([^a-zA-Z0-9]|^)([A-Za-z]\\.(?:${FILE_EXTENSIONS_PATTERN}))(?=[^a-zA-Z0-9/]|$)`,
"g",
);
const HTML_TAG_PATTERN = /(<\/?)([a-zA-Z][a-zA-Z0-9-]*)\b[^>]*?>/gi;
let fileReferencePattern: RegExp | undefined;
let orphanedTldPattern: RegExp | undefined;
function getFileReferencePattern(): RegExp {
if (fileReferencePattern) {
return fileReferencePattern;
}
const fileExtensionsPattern = Array.from(FILE_REF_EXTENSIONS_WITH_TLD).map(escapeRegex).join("|");
fileReferencePattern = new RegExp(
`(^|[^a-zA-Z0-9_\\-/])([a-zA-Z0-9_.\\-./]+\\.(?:${fileExtensionsPattern}))(?=$|[^a-zA-Z0-9_\\-/])`,
"gi",
);
return fileReferencePattern;
}
function getOrphanedTldPattern(): RegExp {
if (orphanedTldPattern) {
return orphanedTldPattern;
}
const fileExtensionsPattern = Array.from(FILE_REF_EXTENSIONS_WITH_TLD).map(escapeRegex).join("|");
orphanedTldPattern = new RegExp(
`([^a-zA-Z0-9]|^)([A-Za-z]\\.(?:${fileExtensionsPattern}))(?=[^a-zA-Z0-9/]|$)`,
"g",
);
return orphanedTldPattern;
}
function wrapStandaloneFileRef(match: string, prefix: string, filename: string): string {
if (filename.startsWith("//")) {
@ -134,8 +151,8 @@ function wrapSegmentFileRefs(
if (!text || codeDepth > 0 || preDepth > 0 || anchorDepth > 0) {
return text;
}
const wrappedStandalone = text.replace(FILE_REFERENCE_PATTERN, wrapStandaloneFileRef);
return wrappedStandalone.replace(ORPHANED_TLD_PATTERN, (match, prefix: string, tld: string) =>
const wrappedStandalone = text.replace(getFileReferencePattern(), wrapStandaloneFileRef);
return wrappedStandalone.replace(getOrphanedTldPattern(), (match, prefix: string, tld: string) =>
prefix === ">" ? match : `${prefix}<code>${escapeHtml(tld)}</code>`,
);
}

12
extensions/telegram/src/sent-message-cache.ts
View File

@ -17,7 +17,12 @@ type CacheEntry = {
*/
const TELEGRAM_SENT_MESSAGES_KEY = Symbol.for("openclaw.telegramSentMessages");
const sentMessages = resolveGlobalMap<string, CacheEntry>(TELEGRAM_SENT_MESSAGES_KEY);
let sentMessages: Map<string, CacheEntry> | undefined;
function getSentMessages(): Map<string, CacheEntry> {
sentMessages ??= resolveGlobalMap<string, CacheEntry>(TELEGRAM_SENT_MESSAGES_KEY);
return sentMessages;
}
function getChatKey(chatId: number | string): string {
return String(chatId);
@ -37,6 +42,7 @@ function cleanupExpired(entry: CacheEntry): void {
*/
export function recordSentMessage(chatId: number | string, messageId: number): void {
const key = getChatKey(chatId);
const sentMessages = getSentMessages();
let entry = sentMessages.get(key);
if (!entry) {
entry = { timestamps: new Map() };
@ -54,7 +60,7 @@ export function recordSentMessage(chatId: number | string, messageId: number): v
*/
export function wasSentByBot(chatId: number | string, messageId: number): boolean {
const key = getChatKey(chatId);
const entry = sentMessages.get(key);
const entry = getSentMessages().get(key);
if (!entry) {
return false;
}
@ -67,5 +73,5 @@ export function wasSentByBot(chatId: number | string, messageId: number): boolea
* Clear all cached entries (for testing).
*/
export function clearSentMessageCache(): void {
sentMessages.clear();
getSentMessages().clear();
}

77
extensions/telegram/src/thread-bindings.ts
View File

@ -77,17 +77,19 @@ type TelegramThreadBindingsState = {
*/
const TELEGRAM_THREAD_BINDINGS_STATE_KEY = Symbol.for("openclaw.telegramThreadBindingsState");
const threadBindingsState = resolveGlobalSingleton<TelegramThreadBindingsState>(
TELEGRAM_THREAD_BINDINGS_STATE_KEY,
() => ({
managersByAccountId: new Map<string, TelegramThreadBindingManager>(),
bindingsByAccountConversation: new Map<string, TelegramThreadBindingRecord>(),
persistQueueByAccountId: new Map<string, Promise<void>>(),
}),
);
const MANAGERS_BY_ACCOUNT_ID = threadBindingsState.managersByAccountId;
const BINDINGS_BY_ACCOUNT_CONVERSATION = threadBindingsState.bindingsByAccountConversation;
const PERSIST_QUEUE_BY_ACCOUNT_ID = threadBindingsState.persistQueueByAccountId;
let threadBindingsState: TelegramThreadBindingsState | undefined;
function getThreadBindingsState(): TelegramThreadBindingsState {
threadBindingsState ??= resolveGlobalSingleton<TelegramThreadBindingsState>(
TELEGRAM_THREAD_BINDINGS_STATE_KEY,
() => ({
managersByAccountId: new Map<string, TelegramThreadBindingManager>(),
bindingsByAccountConversation: new Map<string, TelegramThreadBindingRecord>(),
persistQueueByAccountId: new Map<string, Promise<void>>(),
}),
);
return threadBindingsState;
}
function normalizeDurationMs(raw: unknown, fallback: number): number {
if (typeof raw !== "number" || !Number.isFinite(raw)) {
@ -168,7 +170,7 @@ function fromSessionBindingInput(params: {
}): TelegramThreadBindingRecord {
const now = Date.now();
const metadata = params.input.metadata ?? {};
const existing = BINDINGS_BY_ACCOUNT_CONVERSATION.get(
const existing = getThreadBindingsState().bindingsByAccountConversation.get(
resolveBindingKey({
accountId: params.accountId,
conversationId: params.input.conversationId,
@ -310,7 +312,7 @@ async function persistBindingsToDisk(params: {
version: STORE_VERSION,
bindings:
params.bindings ??
[...BINDINGS_BY_ACCOUNT_CONVERSATION.values()].filter(
[...getThreadBindingsState().bindingsByAccountConversation.values()].filter(
(entry) => entry.accountId === params.accountId,
),
};
@ -322,7 +324,7 @@ async function persistBindingsToDisk(params: {
}
function listBindingsForAccount(accountId: string): TelegramThreadBindingRecord[] {
return [...BINDINGS_BY_ACCOUNT_CONVERSATION.values()].filter(
return [...getThreadBindingsState().bindingsByAccountConversation.values()].filter(
(entry) => entry.accountId === accountId,
);
}
@ -335,16 +337,17 @@ function enqueuePersistBindings(params: {
if (!params.persist) {
return Promise.resolve();
}
const previous = PERSIST_QUEUE_BY_ACCOUNT_ID.get(params.accountId) ?? Promise.resolve();
const previous =
getThreadBindingsState().persistQueueByAccountId.get(params.accountId) ?? Promise.resolve();
const next = previous
.catch(() => undefined)
.then(async () => {
await persistBindingsToDisk(params);
});
PERSIST_QUEUE_BY_ACCOUNT_ID.set(params.accountId, next);
getThreadBindingsState().persistQueueByAccountId.set(params.accountId, next);
void next.finally(() => {
if (PERSIST_QUEUE_BY_ACCOUNT_ID.get(params.accountId) === next) {
PERSIST_QUEUE_BY_ACCOUNT_ID.delete(params.accountId);
if (getThreadBindingsState().persistQueueByAccountId.get(params.accountId) === next) {
getThreadBindingsState().persistQueueByAccountId.delete(params.accountId);
}
});
return next;
@ -412,7 +415,7 @@ export function createTelegramThreadBindingManager(
} = {},
): TelegramThreadBindingManager {
const accountId = normalizeAccountId(params.accountId);
const existing = MANAGERS_BY_ACCOUNT_ID.get(accountId);
const existing = getThreadBindingsState().managersByAccountId.get(accountId);
if (existing) {
return existing;
}
@ -430,7 +433,7 @@ export function createTelegramThreadBindingManager(
accountId,
conversationId: entry.conversationId,
});
BINDINGS_BY_ACCOUNT_CONVERSATION.set(key, {
getThreadBindingsState().bindingsByAccountConversation.set(key, {
...entry,
accountId,
});
@ -448,7 +451,7 @@ export function createTelegramThreadBindingManager(
if (!conversationId) {
return undefined;
}
return BINDINGS_BY_ACCOUNT_CONVERSATION.get(
return getThreadBindingsState().bindingsByAccountConversation.get(
resolveBindingKey({
accountId,
conversationId,
@ -471,7 +474,7 @@ export function createTelegramThreadBindingManager(
return null;
}
const key = resolveBindingKey({ accountId, conversationId });
const existing = BINDINGS_BY_ACCOUNT_CONVERSATION.get(key);
const existing = getThreadBindingsState().bindingsByAccountConversation.get(key);
if (!existing) {
return null;
}
@ -479,7 +482,7 @@ export function createTelegramThreadBindingManager(
...existing,
lastActivityAt: normalizeTimestampMs(at ?? Date.now()),
};
BINDINGS_BY_ACCOUNT_CONVERSATION.set(key, nextRecord);
getThreadBindingsState().bindingsByAccountConversation.set(key, nextRecord);
persistBindingsSafely({
accountId,
persist: manager.shouldPersistMutations(),
@ -494,11 +497,11 @@ export function createTelegramThreadBindingManager(
return null;
}
const key = resolveBindingKey({ accountId, conversationId });
const removed = BINDINGS_BY_ACCOUNT_CONVERSATION.get(key) ?? null;
const removed = getThreadBindingsState().bindingsByAccountConversation.get(key) ?? null;
if (!removed) {
return null;
}
BINDINGS_BY_ACCOUNT_CONVERSATION.delete(key);
getThreadBindingsState().bindingsByAccountConversation.delete(key);
persistBindingsSafely({
accountId,
persist: manager.shouldPersistMutations(),
@ -521,7 +524,7 @@ export function createTelegramThreadBindingManager(
accountId,
conversationId: entry.conversationId,
});
BINDINGS_BY_ACCOUNT_CONVERSATION.delete(key);
getThreadBindingsState().bindingsByAccountConversation.delete(key);
removed.push(entry);
}
if (removed.length > 0) {
@ -540,9 +543,9 @@ export function createTelegramThreadBindingManager(
sweepTimer = null;
}
unregisterSessionBindingAdapter({ channel: "telegram", accountId });
const existingManager = MANAGERS_BY_ACCOUNT_ID.get(accountId);
const existingManager = getThreadBindingsState().managersByAccountId.get(accountId);
if (existingManager === manager) {
MANAGERS_BY_ACCOUNT_ID.delete(accountId);
getThreadBindingsState().managersByAccountId.delete(accountId);
}
},
};
@ -574,7 +577,7 @@ export function createTelegramThreadBindingManager(
metadata: input.metadata,
},
});
BINDINGS_BY_ACCOUNT_CONVERSATION.set(
getThreadBindingsState().bindingsByAccountConversation.set(
resolveBindingKey({ accountId, conversationId }),
record,
);
@ -714,14 +717,14 @@ export function createTelegramThreadBindingManager(
sweepTimer.unref?.();
}
MANAGERS_BY_ACCOUNT_ID.set(accountId, manager);
getThreadBindingsState().managersByAccountId.set(accountId, manager);
return manager;
}
export function getTelegramThreadBindingManager(
accountId?: string,
): TelegramThreadBindingManager | null {
return MANAGERS_BY_ACCOUNT_ID.get(normalizeAccountId(accountId)) ?? null;
return getThreadBindingsState().managersByAccountId.get(normalizeAccountId(accountId)) ?? null;
}
function updateTelegramBindingsBySessionKey(params: {
@ -741,7 +744,7 @@ function updateTelegramBindingsBySessionKey(params: {
conversationId: entry.conversationId,
});
const next = params.update(entry, now);
BINDINGS_BY_ACCOUNT_CONVERSATION.set(key, next);
getThreadBindingsState().bindingsByAccountConversation.set(key, next);
updated.push(next);
}
if (updated.length > 0) {
@ -799,12 +802,12 @@ export function setTelegramThreadBindingMaxAgeBySessionKey(params: {
export const __testing = {
async resetTelegramThreadBindingsForTests() {
for (const manager of MANAGERS_BY_ACCOUNT_ID.values()) {
for (const manager of getThreadBindingsState().managersByAccountId.values()) {
manager.stop();
}
await Promise.allSettled(PERSIST_QUEUE_BY_ACCOUNT_ID.values());
PERSIST_QUEUE_BY_ACCOUNT_ID.clear();
MANAGERS_BY_ACCOUNT_ID.clear();
BINDINGS_BY_ACCOUNT_CONVERSATION.clear();
await Promise.allSettled(getThreadBindingsState().persistQueueByAccountId.values());
getThreadBindingsState().persistQueueByAccountId.clear();
getThreadBindingsState().managersByAccountId.clear();
getThreadBindingsState().bindingsByAccountConversation.clear();
},
};

21
src/agents/pi-embedded-helpers.formatassistanterrortext.test.ts
View File

@ -125,6 +125,27 @@ describe("formatAssistantErrorText", () => {
const msg = makeAssistantError("request ended without sending any chunks");
expect(formatAssistantErrorText(msg)).toBe("LLM request timed out.");
});
it("returns a connection-refused message for ECONNREFUSED failures", () => {
const msg = makeAssistantError("connect ECONNREFUSED 127.0.0.1:443 during upstream call");
expect(formatAssistantErrorText(msg)).toBe(
"LLM request failed: connection refused by the provider endpoint.",
);
});
it("returns a DNS-specific message for provider lookup failures", () => {
const msg = makeAssistantError("dial tcp: lookup api.example.com: no such host (ENOTFOUND)");
expect(formatAssistantErrorText(msg)).toBe(
"LLM request failed: DNS lookup for the provider endpoint failed.",
);
});
it("returns an interrupted-connection message for socket hang ups", () => {
const msg = makeAssistantError("socket hang up");
expect(formatAssistantErrorText(msg)).toBe(
"LLM request failed: network connection was interrupted.",
);
});
});
describe("formatRawAssistantErrorForUi", () => {

8
src/agents/pi-embedded-helpers.sanitizeuserfacingtext.test.ts
View File

@ -88,6 +88,14 @@ describe("sanitizeUserFacingText", () => {
);
});
it("returns a transport-specific message for prefixed ECONNREFUSED errors", () => {
expect(
sanitizeUserFacingText("Error: connect ECONNREFUSED 127.0.0.1:443", {
errorContext: true,
}),
).toBe("LLM request failed: connection refused by the provider endpoint.");
});
it.each([
{
input: "Hello there!\n\nHello there!",

60
src/agents/pi-embedded-helpers/errors.ts
View File

@ -65,6 +65,57 @@ function formatRateLimitOrOverloadedErrorCopy(raw: string): string | undefined {
return undefined;
}
function formatTransportErrorCopy(raw: string): string | undefined {
if (!raw) {
return undefined;
}
const lower = raw.toLowerCase();
if (
/\beconnrefused\b/i.test(raw) ||
lower.includes("connection refused") ||
lower.includes("actively refused")
) {
return "LLM request failed: connection refused by the provider endpoint.";
}
if (
/\beconnreset\b|\beconnaborted\b|\benetreset\b|\bepipe\b/i.test(raw) ||
lower.includes("socket hang up") ||
lower.includes("connection reset") ||
lower.includes("connection aborted")
) {
return "LLM request failed: network connection was interrupted.";
}
if (
/\benotfound\b|\beai_again\b/i.test(raw) ||
lower.includes("getaddrinfo") ||
lower.includes("no such host") ||
lower.includes("dns")
) {
return "LLM request failed: DNS lookup for the provider endpoint failed.";
}
if (
/\benetunreach\b|\behostunreach\b|\behostdown\b/i.test(raw) ||
lower.includes("network is unreachable") ||
lower.includes("host is unreachable")
) {
return "LLM request failed: the provider endpoint is unreachable from this host.";
}
if (
lower.includes("fetch failed") ||
lower.includes("connection error") ||
lower.includes("network request failed")
) {
return "LLM request failed: network connection error.";
}
return undefined;
}
function isReasoningConstraintErrorMessage(raw: string): boolean {
if (!raw) {
return false;
@ -566,6 +617,11 @@ export function formatAssistantErrorText(
return transientCopy;
}
const transportCopy = formatTransportErrorCopy(raw);
if (transportCopy) {
return transportCopy;
}
if (isTimeoutErrorMessage(raw)) {
return "LLM request timed out.";
}
@ -626,6 +682,10 @@ export function sanitizeUserFacingText(text: string, opts?: { errorContext?: boo
if (prefixedCopy) {
return prefixedCopy;
}
const transportCopy = formatTransportErrorCopy(trimmed);
if (transportCopy) {
return transportCopy;
}
if (isTimeoutErrorMessage(trimmed)) {
return "LLM request timed out.";
}

10
src/agents/pi-embedded-subscribe.handlers.lifecycle.test.ts
View File

@ -58,14 +58,16 @@ describe("handleAgentEnd", () => {
expect(warn.mock.calls[0]?.[1]).toMatchObject({
event: "embedded_run_agent_end",
runId: "run-1",
error: "connection refused",
error: "LLM request failed: connection refused by the provider endpoint.",
rawErrorPreview: "connection refused",
consoleMessage:
"embedded run agent end: runId=run-1 isError=true model=unknown provider=unknown error=LLM request failed: connection refused by the provider endpoint. rawError=connection refused",
});
expect(onAgentEvent).toHaveBeenCalledWith({
stream: "lifecycle",
data: {
phase: "error",
error: "connection refused",
error: "LLM request failed: connection refused by the provider endpoint.",
},
});
});
@ -92,7 +94,7 @@ describe("handleAgentEnd", () => {
failoverReason: "overloaded",
providerErrorType: "overloaded_error",
consoleMessage:
"embedded run agent end: runId=run-1 isError=true model=claude-test provider=anthropic error=The AI service is temporarily overloaded. Please try again in a moment.",
'embedded run agent end: runId=run-1 isError=true model=claude-test provider=anthropic error=The AI service is temporarily overloaded. Please try again in a moment. rawError={"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}',
});
});
@ -112,7 +114,7 @@ describe("handleAgentEnd", () => {
const meta = warn.mock.calls[0]?.[1];
expect(meta).toMatchObject({
consoleMessage:
"embedded run agent end: runId=run-1 isError=true model=claude sonnet 4 provider=anthropic]8;;https://evil.test error=connection refused",
"embedded run agent end: runId=run-1 isError=true model=claude sonnet 4 provider=anthropic]8;;https://evil.test error=LLM request failed: connection refused by the provider endpoint. rawError=connection refused",
});
expect(meta?.consoleMessage).not.toContain("\n");
expect(meta?.consoleMessage).not.toContain("\r");

4
src/agents/pi-embedded-subscribe.handlers.lifecycle.ts
View File

@ -50,6 +50,8 @@ export function handleAgentEnd(ctx: EmbeddedPiSubscribeContext) {
const safeRunId = sanitizeForConsole(ctx.params.runId) ?? "-";
const safeModel = sanitizeForConsole(lastAssistant.model) ?? "unknown";
const safeProvider = sanitizeForConsole(lastAssistant.provider) ?? "unknown";
const safeRawErrorPreview = sanitizeForConsole(observedError.rawErrorPreview);
const rawErrorConsoleSuffix = safeRawErrorPreview ? ` rawError=${safeRawErrorPreview}` : "";
ctx.log.warn("embedded run agent end", {
event: "embedded_run_agent_end",
tags: ["error_handling", "lifecycle", "agent_end", "assistant_error"],
@ -60,7 +62,7 @@ export function handleAgentEnd(ctx: EmbeddedPiSubscribeContext) {
model: lastAssistant.model,
provider: lastAssistant.provider,
...observedError,
consoleMessage: `embedded run agent end: runId=${safeRunId} isError=true model=${safeModel} provider=${safeProvider} error=${safeErrorText}`,
consoleMessage: `embedded run agent end: runId=${safeRunId} isError=true model=${safeModel} provider=${safeProvider} error=${safeErrorText}${rawErrorConsoleSuffix}`,
});
emitAgentEvent({
runId: ctx.params.runId,

41
src/commands/gateway-status.test.ts
View File

@ -567,6 +567,47 @@ describe("gateway-status command", () => {
expect(targets.some((t) => t.kind === "sshTunnel")).toBe(true);
});
it("passes the full caller timeout through to local loopback probes", async () => {
const { runtime } = createRuntimeCapture();
probeGateway.mockClear();
readBestEffortConfig.mockResolvedValueOnce({
gateway: {
mode: "local",
auth: { mode: "token", token: "ltok" },
},
} as never);
await runGatewayStatus(runtime, { timeout: "15000", json: true });
expect(probeGateway).toHaveBeenCalledWith(
expect.objectContaining({
url: "ws://127.0.0.1:18789",
timeoutMs: 15_000,
}),
);
});
it("keeps inactive local loopback probes on the short timeout in remote mode", async () => {
const { runtime } = createRuntimeCapture();
probeGateway.mockClear();
readBestEffortConfig.mockResolvedValueOnce({
gateway: {
mode: "remote",
auth: { mode: "token", token: "ltok" },
remote: {},
},
} as never);
await runGatewayStatus(runtime, { timeout: "15000", json: true });
expect(probeGateway).toHaveBeenCalledWith(
expect.objectContaining({
url: "ws://127.0.0.1:18789",
timeoutMs: 800,
}),
);
});
it("skips invalid ssh-auto discovery targets", async () => {
const { runtime } = createRuntimeCapture();
await withEnvAsync({ USER: "steipete" }, async () => {

2
src/commands/gateway-status.ts
View File

@ -176,7 +176,7 @@ export async function gatewayStatusCommand(
token: authResolution.token,
password: authResolution.password,
};
const timeoutMs = resolveProbeBudgetMs(overallTimeoutMs, target.kind);
const timeoutMs = resolveProbeBudgetMs(overallTimeoutMs, target);
const probe = await probeGateway({
url: target.url,
auth,

19
src/commands/gateway-status/helpers.test.ts
View File

@ -6,6 +6,7 @@ import {
isScopeLimitedProbeFailure,
renderProbeSummaryLine,
resolveAuthForTarget,
resolveProbeBudgetMs,
} from "./helpers.js";
describe("extractConfigSummary", () => {
@ -273,3 +274,21 @@ describe("probe reachability classification", () => {
expect(renderProbeSummaryLine(probe, false)).toContain("RPC: failed");
});
});
describe("resolveProbeBudgetMs", () => {
it("lets active local loopback probes use the full caller budget", () => {
expect(resolveProbeBudgetMs(15_000, { kind: "localLoopback", active: true })).toBe(15_000);
expect(resolveProbeBudgetMs(3_000, { kind: "localLoopback", active: true })).toBe(3_000);
});
it("keeps inactive local loopback probes on the short cap", () => {
expect(resolveProbeBudgetMs(15_000, { kind: "localLoopback", active: false })).toBe(800);
expect(resolveProbeBudgetMs(500, { kind: "localLoopback", active: false })).toBe(500);
});
it("keeps non-local probe caps unchanged", () => {
expect(resolveProbeBudgetMs(15_000, { kind: "configRemote", active: true })).toBe(1_500);
expect(resolveProbeBudgetMs(15_000, { kind: "explicit", active: true })).toBe(1_500);
expect(resolveProbeBudgetMs(15_000, { kind: "sshTunnel", active: true })).toBe(2_000);
});
});

21
src/commands/gateway-status/helpers.ts
View File

@ -116,14 +116,21 @@ export function resolveTargets(cfg: OpenClawConfig, explicitUrl?: string): Gatew
return targets;
}
export function resolveProbeBudgetMs(overallMs: number, kind: TargetKind): number {
if (kind === "localLoopback") {
return Math.min(800, overallMs);
export function resolveProbeBudgetMs(
overallMs: number,
target: Pick<GatewayStatusTarget, "kind" | "active">,
): number {
switch (target.kind) {
case "localLoopback":
// Active loopback probes should honor the caller budget because local shells/containers
// can legitimately take longer to connect. Inactive loopback probes stay bounded so
// remote-mode status checks do not stall on an expected local miss.
return target.active ? overallMs : Math.min(800, overallMs);
case "sshTunnel":
return Math.min(2_000, overallMs);
default:
return Math.min(1_500, overallMs);
}
if (kind === "sshTunnel") {
return Math.min(2000, overallMs);
}
return Math.min(1500, overallMs);
}
export function sanitizeSshTarget(value: unknown): string | null {

8
src/gateway/probe.test.ts
View File

@ -40,9 +40,15 @@ vi.mock("./client.js", () => ({
GatewayClient: MockGatewayClient,
}));
const { probeGateway } = await import("./probe.js");
const { clampProbeTimeoutMs, probeGateway } = await import("./probe.js");
describe("probeGateway", () => {
it("clamps probe timeout to timer-safe bounds", () => {
expect(clampProbeTimeoutMs(1)).toBe(250);
expect(clampProbeTimeoutMs(2_000)).toBe(2_000);
expect(clampProbeTimeoutMs(3_000_000_000)).toBe(2_147_483_647);
});
it("connects with operator.read scope", async () => {
const result = await probeGateway({
url: "ws://127.0.0.1:18789",

34
src/gateway/probe.ts
View File

@ -29,6 +29,13 @@ export type GatewayProbeResult = {
configSnapshot: unknown;
};
export const MIN_PROBE_TIMEOUT_MS = 250;
export const MAX_TIMER_DELAY_MS = 2_147_483_647;
export function clampProbeTimeoutMs(timeoutMs: number): number {
return Math.min(MAX_TIMER_DELAY_MS, Math.max(MIN_PROBE_TIMEOUT_MS, timeoutMs));
}
export async function probeGateway(opts: {
url: string;
auth?: GatewayProbeAuth;
@ -144,21 +151,18 @@ export async function probeGateway(opts: {
},
});
const timer = setTimeout(
() => {
settle({
ok: false,
connectLatencyMs,
error: connectError ? `connect failed: ${connectError}` : "timeout",
close,
health: null,
status: null,
presence: null,
configSnapshot: null,
});
},
Math.max(250, opts.timeoutMs),
);
const timer = setTimeout(() => {
settle({
ok: false,
connectLatencyMs,
error: connectError ? `connect failed: ${connectError}` : "timeout",
close,
health: null,
status: null,
presence: null,
configSnapshot: null,
});
}, clampProbeTimeoutMs(opts.timeoutMs));
client.start();
});

8
test/fixtures/plugin-extension-import-boundary-inventory.json vendored
View File

@ -31,14 +31,6 @@
"resolvedPath": "extensions/imessage/runtime-api.js",
"reason": "imports extension-owned file from src/plugins"
},
{
"file": "src/plugins/runtime/runtime-matrix.ts",
"line": 4,
"kind": "import",
"specifier": "../../../extensions/matrix/runtime-api.js",
"resolvedPath": "extensions/matrix/runtime-api.js",
"reason": "imports extension-owned file from src/plugins"
},
{
"file": "src/plugins/runtime/runtime-slack-ops.runtime.ts",
"line": 10,