diff --git a/package.json b/package.json index 537a85d..2b5ec7c 100644 --- a/package.json +++ b/package.json @@ -6,7 +6,7 @@ "email": "lobsterai.project@rd.netease.com" }, "license": "MIT", - "version": "2026.3.24", + "version": "2026.3.25", "openclaw": { "version": "v2026.3.2", "repo": "https://github.com/openclaw/openclaw.git", @@ -39,7 +39,7 @@ { "id": "moltbot-popo", "npm": "moltbot-popo", - "version": "1.0.65-beta.21", + "version": "1.0.66", "registry": "https://npm.nie.netease.com", "optional": true }, @@ -86,19 +86,18 @@ "setup:python-runtime": "node scripts/setup-python-runtime.js", "openclaw:ensure": "node scripts/ensure-openclaw-version.cjs", "openclaw:patch": "node scripts/apply-openclaw-patches.cjs", - "openclaw:hotfix": "node scripts/apply-openclaw-runtime-hotfixes.cjs", "openclaw:plugins": "node scripts/ensure-openclaw-plugins.cjs", "openclaw:extensions:local": "node scripts/sync-local-openclaw-extensions.cjs", "openclaw:bundle": "node scripts/bundle-openclaw-gateway.cjs", "openclaw:precompile": "node scripts/precompile-openclaw-extensions.cjs", "openclaw:prune": "node scripts/prune-openclaw-runtime.cjs", "openclaw:runtime:host": "node scripts/openclaw-runtime-host.cjs", - "openclaw:runtime:mac-arm64": "npm run openclaw:ensure && npm run openclaw:patch && node scripts/run-build-openclaw-runtime.cjs mac-arm64 && node scripts/sync-openclaw-runtime-current.cjs mac-arm64 && npm run openclaw:hotfix && npm run openclaw:bundle && npm run openclaw:plugins && npm run openclaw:extensions:local && npm run openclaw:precompile && npm run openclaw:prune", - "openclaw:runtime:mac-x64": "npm run openclaw:ensure && npm run openclaw:patch && node scripts/run-build-openclaw-runtime.cjs mac-x64 && node scripts/sync-openclaw-runtime-current.cjs mac-x64 && npm run openclaw:hotfix && npm run openclaw:bundle && npm run openclaw:plugins && npm run openclaw:extensions:local && npm run openclaw:precompile && npm run openclaw:prune", - "openclaw:runtime:win-x64": "npm run openclaw:ensure && npm run openclaw:patch && node scripts/run-build-openclaw-runtime.cjs win-x64 && node scripts/sync-openclaw-runtime-current.cjs win-x64 && npm run openclaw:hotfix && npm run openclaw:bundle && npm run openclaw:plugins && npm run openclaw:extensions:local && npm run openclaw:precompile && npm run openclaw:prune", - "openclaw:runtime:win-arm64": "npm run openclaw:ensure && npm run openclaw:patch && node scripts/run-build-openclaw-runtime.cjs win-arm64 && node scripts/sync-openclaw-runtime-current.cjs win-arm64 && npm run openclaw:hotfix && npm run openclaw:bundle && npm run openclaw:plugins && npm run openclaw:extensions:local && npm run openclaw:precompile && npm run openclaw:prune", - "openclaw:runtime:linux-x64": "npm run openclaw:ensure && npm run openclaw:patch && node scripts/run-build-openclaw-runtime.cjs linux-x64 && node scripts/sync-openclaw-runtime-current.cjs linux-x64 && npm run openclaw:hotfix && npm run openclaw:bundle && npm run openclaw:plugins && npm run openclaw:extensions:local && npm run openclaw:precompile && npm run openclaw:prune", - "openclaw:runtime:linux-arm64": "npm run openclaw:ensure && npm run openclaw:patch && node scripts/run-build-openclaw-runtime.cjs linux-arm64 && node scripts/sync-openclaw-runtime-current.cjs linux-arm64 && npm run openclaw:hotfix && npm run openclaw:bundle && npm run openclaw:plugins && npm run openclaw:extensions:local && npm run openclaw:precompile && npm run openclaw:prune", + "openclaw:runtime:mac-arm64": "npm run openclaw:ensure && npm run openclaw:patch && node scripts/run-build-openclaw-runtime.cjs mac-arm64 && node scripts/sync-openclaw-runtime-current.cjs mac-arm64 && npm run openclaw:bundle && npm run openclaw:plugins && npm run openclaw:extensions:local && npm run openclaw:precompile && npm run openclaw:prune", + "openclaw:runtime:mac-x64": "npm run openclaw:ensure && npm run openclaw:patch && node scripts/run-build-openclaw-runtime.cjs mac-x64 && node scripts/sync-openclaw-runtime-current.cjs mac-x64 && npm run openclaw:bundle && npm run openclaw:plugins && npm run openclaw:extensions:local && npm run openclaw:precompile && npm run openclaw:prune", + "openclaw:runtime:win-x64": "npm run openclaw:ensure && npm run openclaw:patch && node scripts/run-build-openclaw-runtime.cjs win-x64 && node scripts/sync-openclaw-runtime-current.cjs win-x64 && npm run openclaw:bundle && npm run openclaw:plugins && npm run openclaw:extensions:local && npm run openclaw:precompile && npm run openclaw:prune", + "openclaw:runtime:win-arm64": "npm run openclaw:ensure && npm run openclaw:patch && node scripts/run-build-openclaw-runtime.cjs win-arm64 && node scripts/sync-openclaw-runtime-current.cjs win-arm64 && npm run openclaw:bundle && npm run openclaw:plugins && npm run openclaw:extensions:local && npm run openclaw:precompile && npm run openclaw:prune", + "openclaw:runtime:linux-x64": "npm run openclaw:ensure && npm run openclaw:patch && node scripts/run-build-openclaw-runtime.cjs linux-x64 && node scripts/sync-openclaw-runtime-current.cjs linux-x64 && npm run openclaw:bundle && npm run openclaw:plugins && npm run openclaw:extensions:local && npm run openclaw:precompile && npm run openclaw:prune", + "openclaw:runtime:linux-arm64": "npm run openclaw:ensure && npm run openclaw:patch && node scripts/run-build-openclaw-runtime.cjs linux-arm64 && node scripts/sync-openclaw-runtime-current.cjs linux-arm64 && npm run openclaw:bundle && npm run openclaw:plugins && npm run openclaw:extensions:local && npm run openclaw:precompile && npm run openclaw:prune", "regenerate:icon": "bash scripts/regenerate-mac-icon.sh", "fix:mac-icon": "bash scripts/fix-mac-icon-display.sh" }, diff --git a/scripts/apply-openclaw-patches.cjs b/scripts/apply-openclaw-patches.cjs index 496363a..6e55553 100644 --- a/scripts/apply-openclaw-patches.cjs +++ b/scripts/apply-openclaw-patches.cjs @@ -1,11 +1,11 @@ 'use strict'; /** - * Apply LobsterAI patches to the openclaw source tree. + * Apply version-specific LobsterAI patches to the openclaw source tree. * - * These patches add a dedicated gateway entry point that skips the full CLI - * infrastructure, dramatically reducing startup time inside Electron's - * utilityProcess (~15s instead of ~120s). + * Patches are organised in scripts/patches// directories, where + * matches the "openclaw.version" field in package.json (e.g. + * "v2026.3.2"). Only patches for the currently pinned version are applied. * * Usage: * node scripts/apply-openclaw-patches.cjs [openclaw-src-dir] @@ -25,7 +25,15 @@ const openclawSrc = process.argv[2] ? path.resolve(process.argv[2]) : path.resolve(rootDir, '..', 'openclaw'); -const patchesDir = path.join(rootDir, 'scripts', 'patches'); +// Read pinned openclaw version from package.json. +const pkg = require(path.join(rootDir, 'package.json')); +const openclawVersion = pkg.openclaw && pkg.openclaw.version; +if (!openclawVersion) { + console.error('[apply-openclaw-patches] Missing "openclaw.version" in package.json.'); + process.exit(1); +} + +const patchesDir = path.join(rootDir, 'scripts', 'patches', openclawVersion); if (!fs.existsSync(openclawSrc)) { console.error(`[apply-openclaw-patches] openclaw source not found: ${openclawSrc}`); @@ -37,15 +45,22 @@ if (!fs.existsSync(path.join(openclawSrc, 'package.json'))) { process.exit(1); } +if (!fs.existsSync(patchesDir)) { + console.log(`[apply-openclaw-patches] No patches directory for ${openclawVersion}, nothing to do.`); + process.exit(0); +} + const patchFiles = fs.readdirSync(patchesDir) .filter(f => f.endsWith('.patch')) .sort(); if (patchFiles.length === 0) { - console.log('[apply-openclaw-patches] No patches found, nothing to do.'); + console.log(`[apply-openclaw-patches] No patches found for ${openclawVersion}, nothing to do.`); process.exit(0); } +console.log(`[apply-openclaw-patches] Applying patches for openclaw ${openclawVersion} (${patchFiles.length} file(s))`); + let applied = 0; let skipped = 0; diff --git a/scripts/apply-openclaw-runtime-hotfixes.cjs b/scripts/apply-openclaw-runtime-hotfixes.cjs deleted file mode 100644 index f98be90..0000000 --- a/scripts/apply-openclaw-runtime-hotfixes.cjs +++ /dev/null @@ -1,80 +0,0 @@ -'use strict'; - -/** - * Apply runtime hotfixes to the openclaw dist/ files BEFORE esbuild bundling. - * - * This ensures the gateway-bundle.mjs contains the patched code, eliminating - * the need for the slow runtime hotfix pass (applyRuntimeHotfixes) which scans - * ~1100 JS files via Electron's transparent asar read and takes 250+ seconds - * on Windows. - * - * Usage: - * node scripts/apply-openclaw-runtime-hotfixes.cjs [runtime-dir] - * - * If runtime-dir is not specified, defaults to vendor/openclaw-runtime/current. - */ - -const path = require('path'); - -const rootDir = path.resolve(__dirname, '..'); -const runtimeDir = process.argv[2] - ? path.resolve(process.argv[2]) - : path.join(rootDir, 'vendor', 'openclaw-runtime', 'current'); - -// Import the compiled hotfix module from dist-electron/. -// The hotfix code lives in src/main/libs/openclawRuntimeHotfix.ts and is -// compiled by `npm run compile:electron` to dist-electron/openclawRuntimeHotfix.js. -// However, compile:electron runs AFTER the runtime build in the dist:win chain, -// so we compile just this file on the fly using tsx/ts-node if the compiled -// output is not available. - -let applyBundledOpenClawRuntimeHotfixes; - -const compiledPath = path.join(rootDir, 'dist-electron', 'openclawRuntimeHotfix.js'); -try { - const mod = require(compiledPath); - applyBundledOpenClawRuntimeHotfixes = mod.applyBundledOpenClawRuntimeHotfixes; -} catch { - // Fallback: compile the TypeScript source directly using esbuild's register - // (esbuild is already a devDependency for the bundle step). - try { - const esbuild = require('esbuild'); - const srcPath = path.join(rootDir, 'src', 'main', 'libs', 'openclawRuntimeHotfix.ts'); - const result = esbuild.buildSync({ - entryPoints: [srcPath], - bundle: false, - platform: 'node', - format: 'cjs', - write: false, - }); - const code = result.outputFiles[0].text; - const mod = new module.constructor(); - mod._compile(code, srcPath); - applyBundledOpenClawRuntimeHotfixes = mod.exports.applyBundledOpenClawRuntimeHotfixes; - } catch (err) { - console.error('[apply-openclaw-runtime-hotfixes] Failed to load hotfix module:', err.message || err); - console.error('[apply-openclaw-runtime-hotfixes] Run `npm run compile:electron` first, or ensure esbuild is installed.'); - process.exit(1); - } -} - -if (typeof applyBundledOpenClawRuntimeHotfixes !== 'function') { - console.error('[apply-openclaw-runtime-hotfixes] applyBundledOpenClawRuntimeHotfixes is not a function'); - process.exit(1); -} - -console.log(`[apply-openclaw-runtime-hotfixes] Applying hotfixes to: ${runtimeDir}`); -const result = applyBundledOpenClawRuntimeHotfixes(runtimeDir); - -if (result.changed) { - console.log( - `[apply-openclaw-runtime-hotfixes] Patched ${result.patchedFiles.length} file(s): ` - + result.patchedFiles.map(f => path.relative(runtimeDir, f)).join(', ') - ); -} else { - console.log('[apply-openclaw-runtime-hotfixes] No files needed patching (already up to date).'); -} - -if (result.errors.length > 0) { - console.warn('[apply-openclaw-runtime-hotfixes] Warnings:', result.errors.join(' | ')); -} diff --git a/scripts/build-openclaw-runtime.sh b/scripts/build-openclaw-runtime.sh index 8037cd1..69a79bd 100755 --- a/scripts/build-openclaw-runtime.sh +++ b/scripts/build-openclaw-runtime.sh @@ -91,8 +91,8 @@ try { READVER ) -# Compute a fingerprint of all patch files so the build is invalidated when patches change. -PATCHES_DIR="$ELECTRON_ROOT/scripts/patches" +# Compute a fingerprint of version-specific patch files so the build is invalidated when patches change. +PATCHES_DIR="$ELECTRON_ROOT/scripts/patches/$DESIRED_VERSION" PATCH_HASH="" if [[ -d "$PATCHES_DIR" ]]; then PATCH_HASH=$(cat "$PATCHES_DIR"/*.patch 2>/dev/null | sha256sum | cut -d' ' -f1) diff --git a/scripts/patches/v2026.3.2/openclaw-cron-current-time-suffix.patch b/scripts/patches/v2026.3.2/openclaw-cron-current-time-suffix.patch new file mode 100644 index 0000000..e9cb286 --- /dev/null +++ b/scripts/patches/v2026.3.2/openclaw-cron-current-time-suffix.patch @@ -0,0 +1,13 @@ +diff --git a/src/infra/heartbeat-runner.ts b/src/infra/heartbeat-runner.ts +index 2d0bee48f0..426036bd07 100644 +--- a/src/infra/heartbeat-runner.ts ++++ b/src/infra/heartbeat-runner.ts +@@ -675,7 +675,7 @@ export async function runHeartbeatOnce(opts: { + canRelayToUser, + }); + const ctx = { +- Body: appendCronStyleCurrentTimeLine(prompt, cfg, startedAt), ++ Body: hasCronEvents ? prompt : appendCronStyleCurrentTimeLine(prompt, cfg, startedAt), + From: sender, + To: sender, + OriginatingChannel: delivery.channel !== "none" ? delivery.channel : undefined, diff --git a/scripts/patches/v2026.3.2/openclaw-cron-delivery-inference.patch b/scripts/patches/v2026.3.2/openclaw-cron-delivery-inference.patch new file mode 100644 index 0000000..7f72613 --- /dev/null +++ b/scripts/patches/v2026.3.2/openclaw-cron-delivery-inference.patch @@ -0,0 +1,110 @@ +diff --git a/src/agents/tools/cron-tool.ts b/src/agents/tools/cron-tool.ts +index 14df690102..c36714ece6 100644 +--- a/src/agents/tools/cron-tool.ts ++++ b/src/agents/tools/cron-tool.ts +@@ -1,11 +1,13 @@ + import { Type } from "@sinclair/typebox"; + import { loadConfig } from "../../config/config.js"; ++import { extractDeliveryInfo } from "../../config/sessions/delivery-info.js"; + import { normalizeCronJobCreate, normalizeCronJobPatch } from "../../cron/normalize.js"; + import type { CronDelivery, CronMessageChannel } from "../../cron/types.js"; + import { normalizeHttpWebhookUrl } from "../../cron/webhook-url.js"; + import { parseAgentSessionKey } from "../../sessions/session-key-utils.js"; + import { extractTextFromChatContent } from "../../shared/chat-content.js"; + import { isRecord, truncateUtf16Safe } from "../../utils.js"; ++import { normalizeDeliveryContext } from "../../utils/delivery-context.js"; + import { resolveSessionAgentId } from "../agent-scope.js"; + import { optionalStringEnum, stringEnum } from "../schema/typebox.js"; + import { type AnyAgentTool, jsonResult, readStringParam } from "./common.js"; +@@ -159,6 +161,46 @@ function inferDeliveryFromSessionKey(agentSessionKey?: string): CronDelivery | n + if (!rawSessionKey) { + return null; + } ++ ++ // Try structured delivery info from session store (most reliable). ++ const { deliveryContext, threadId } = extractDeliveryInfo(rawSessionKey); ++ const persistedDelivery = normalizeDeliveryContext(deliveryContext); ++ if (persistedDelivery?.channel && persistedDelivery?.to) { ++ const delivery: CronDelivery = { ++ mode: "announce", ++ channel: persistedDelivery.channel as CronMessageChannel, ++ to: persistedDelivery.to, ++ }; ++ if (persistedDelivery.accountId) delivery.accountId = persistedDelivery.accountId; ++ if (persistedDelivery.threadId != null) delivery.threadId = persistedDelivery.threadId; ++ else if (threadId != null) delivery.threadId = threadId; ++ return delivery; ++ } ++ ++ // Try JSON-encoded delivery context embedded in session key. ++ const jsonIdx = rawSessionKey.indexOf(":{"); ++ if (jsonIdx > 0) { ++ try { ++ const ctx = JSON.parse(rawSessionKey.slice(jsonIdx + 1)); ++ if (ctx && typeof ctx.channel === "string" && ctx.channel.trim()) { ++ const channel = ctx.channel.trim().toLowerCase() as CronMessageChannel; ++ const peerId = (ctx.peerid || ctx.peerId || ctx.conversationId || "").trim(); ++ const chatType = (ctx.chattype || ctx.chatType || "direct").trim().toLowerCase(); ++ const accountId = (ctx.accountid || ctx.accountId || "").trim(); ++ if (peerId) { ++ const to = chatType === "group" ? `group:${peerId}` : `user:${peerId}`; ++ const delivery: CronDelivery = { mode: "announce", channel, to }; ++ if (accountId && accountId !== "__default__") delivery.accountId = accountId; ++ if (threadId != null) delivery.threadId = threadId; ++ return delivery; ++ } ++ } ++ } catch { ++ // fall through to legacy parsing ++ } ++ } ++ ++ // Legacy session-key parsing (channel:marker:peerId patterns). + const parsed = parseAgentSessionKey(stripThreadSuffixFromSessionKey(rawSessionKey)); + if (!parsed || !parsed.rest) { + return null; +@@ -172,6 +214,21 @@ function inferDeliveryFromSessionKey(agentSessionKey?: string): CronDelivery | n + return null; + } + ++ if (head === "dingtalk-connector") { ++ const accountId = parts[1]?.trim(); ++ const senderId = parts[2]?.trim(); ++ if (accountId && senderId) { ++ const delivery: CronDelivery = { ++ mode: "announce", ++ channel: "dingtalk-connector" as CronMessageChannel, ++ to: `user:${senderId}`, ++ accountId, ++ }; ++ if (threadId != null) delivery.threadId = threadId; ++ return delivery; ++ } ++ } ++ + // buildAgentPeerSessionKey encodes peers as: + // - direct: + // - :direct: +@@ -204,6 +261,9 @@ function inferDeliveryFromSessionKey(agentSessionKey?: string): CronDelivery | n + if (channel) { + delivery.channel = channel; + } ++ if (threadId != null) { ++ delivery.threadId = threadId; ++ } + return delivery; + } + +diff --git a/src/cron/types.ts b/src/cron/types.ts +index ef5de924b0..8dde2d0754 100644 +--- a/src/cron/types.ts ++++ b/src/cron/types.ts +@@ -25,6 +25,8 @@ export type CronDelivery = { + to?: string; + /** Explicit channel account id for multi-account setups (e.g. multiple Telegram bots). */ + accountId?: string; ++ /** Thread or topic id for threaded channels (e.g. Telegram forum topics). */ ++ threadId?: string | number; + bestEffort?: boolean; + /** Separate destination for failure notifications. */ + failureDestination?: CronFailureDestination; diff --git a/scripts/patches/v2026.3.2/openclaw-cron-isolated-session-key.patch b/scripts/patches/v2026.3.2/openclaw-cron-isolated-session-key.patch new file mode 100644 index 0000000..3043060 --- /dev/null +++ b/scripts/patches/v2026.3.2/openclaw-cron-isolated-session-key.patch @@ -0,0 +1,13 @@ +diff --git a/src/gateway/server-cron.ts b/src/gateway/server-cron.ts +index 1f1cd1f535..67d4a644c0 100644 +--- a/src/gateway/server-cron.ts ++++ b/src/gateway/server-cron.ts +@@ -291,7 +291,7 @@ export function buildGatewayCronService(params: { + message, + abortSignal, + agentId, +- sessionKey: `cron:${job.id}`, ++ sessionKey: job.sessionKey || `cron:${job.id}`, + lane: "cron", + }); + }, diff --git a/scripts/patches/v2026.3.2/openclaw-cron-owner-fallback.patch b/scripts/patches/v2026.3.2/openclaw-cron-owner-fallback.patch new file mode 100644 index 0000000..cab10ef --- /dev/null +++ b/scripts/patches/v2026.3.2/openclaw-cron-owner-fallback.patch @@ -0,0 +1,13 @@ +diff --git a/src/agents/tool-policy.ts b/src/agents/tool-policy.ts +index 188a9c3361..7d35d4950d 100644 +--- a/src/agents/tool-policy.ts ++++ b/src/agents/tool-policy.ts +@@ -28,7 +28,7 @@ function wrapOwnerOnlyToolExecution(tool: AnyAgentTool, senderIsOwner: boolean): + }; + } + +-const OWNER_ONLY_TOOL_NAME_FALLBACKS = new Set(["whatsapp_login", "cron", "gateway"]); ++const OWNER_ONLY_TOOL_NAME_FALLBACKS = new Set(["whatsapp_login", "gateway"]); + + export function isOwnerOnlyToolName(name: string) { + return OWNER_ONLY_TOOL_NAME_FALLBACKS.has(normalizeToolName(name)); diff --git a/scripts/patches/v2026.3.2/openclaw-cron-reminder-prompt.patch b/scripts/patches/v2026.3.2/openclaw-cron-reminder-prompt.patch new file mode 100644 index 0000000..1156fc4 --- /dev/null +++ b/scripts/patches/v2026.3.2/openclaw-cron-reminder-prompt.patch @@ -0,0 +1,24 @@ +diff --git a/src/infra/heartbeat-events-filter.ts b/src/infra/heartbeat-events-filter.ts +index 1682c3b308..cf1678a3a0 100644 +--- a/src/infra/heartbeat-events-filter.ts ++++ b/src/infra/heartbeat-events-filter.ts +@@ -23,18 +23,7 @@ export function buildCronEventPrompt( + "Reply HEARTBEAT_OK." + ); + } +- if (!deliverToUser) { +- return ( +- "A scheduled reminder has been triggered. The reminder content is:\n\n" + +- eventText + +- "\n\nHandle this reminder internally. Do not relay it to the user unless explicitly requested." +- ); +- } +- return ( +- "A scheduled reminder has been triggered. The reminder content is:\n\n" + +- eventText + +- "\n\nPlease relay this reminder to the user in a helpful and friendly way." +- ); ++ return eventText; + } + + export function buildExecEventPrompt(opts?: { deliverToUser?: boolean }): string { diff --git a/scripts/patches/v2026.3.2/openclaw-cron-tool-owner-only.patch b/scripts/patches/v2026.3.2/openclaw-cron-tool-owner-only.patch new file mode 100644 index 0000000..69df87f --- /dev/null +++ b/scripts/patches/v2026.3.2/openclaw-cron-tool-owner-only.patch @@ -0,0 +1,13 @@ +diff --git a/src/agents/tools/cron-tool.ts b/src/agents/tools/cron-tool.ts +index 14df690102..66bc27d59f 100644 +--- a/src/agents/tools/cron-tool.ts ++++ b/src/agents/tools/cron-tool.ts +@@ -212,7 +212,7 @@ export function createCronTool(opts?: CronToolOptions, deps?: CronToolDeps): Any + return { + label: "Cron", + name: "cron", +- ownerOnly: true, ++ ownerOnly: false, + description: `Manage Gateway cron jobs (status/list/add/update/remove/run/runs) and send wake events. + + ACTIONS: diff --git a/scripts/patches/openclaw-gateway-entry.patch b/scripts/patches/v2026.3.2/openclaw-gateway-entry.patch similarity index 100% rename from scripts/patches/openclaw-gateway-entry.patch rename to scripts/patches/v2026.3.2/openclaw-gateway-entry.patch diff --git a/scripts/patches/v2026.3.2/openclaw-wecom-exec-deny.patch b/scripts/patches/v2026.3.2/openclaw-wecom-exec-deny.patch new file mode 100644 index 0000000..428c244 --- /dev/null +++ b/scripts/patches/v2026.3.2/openclaw-wecom-exec-deny.patch @@ -0,0 +1,12 @@ +diff --git a/src/agents/pi-tools.ts b/src/agents/pi-tools.ts +index 7d6fdf1c14..b622c1adf8 100644 +--- a/src/agents/pi-tools.ts ++++ b/src/agents/pi-tools.ts +@@ -64,6 +64,7 @@ function isOpenAIProvider(provider?: string) { + + const TOOL_DENY_BY_MESSAGE_PROVIDER: Readonly> = { + voice: ["tts"], ++ wecom: ["exec", "process"], + }; + + function normalizeMessageProvider(messageProvider?: string): string | undefined { diff --git a/spec/features/im-conversation-sync/2026-03-19-im-conversation-sync-design.md b/spec/features/im-conversation-sync/2026-03-19-im-conversation-sync-design.md new file mode 100644 index 0000000..2157706 --- /dev/null +++ b/spec/features/im-conversation-sync/2026-03-19-im-conversation-sync-design.md @@ -0,0 +1,575 @@ +# IM 会话同步设计文档 + +## 1. 概述 + +LobsterAI 的 IM 会话通过 OpenClaw 网关实现。用户从各 IM 平台(钉钉、飞书、Telegram、Discord、QQ、企业微信、POPO、云信等)发送消息后,LobsterAI 将消息路由到 OpenClaw 引擎执行,并将 AI 回复同步回 IM。 + +核心挑战在于:OpenClaw 引擎通过 WebSocket 实时推送流式事件,同时也提供 `chat.history` API 查询完整历史。本方案采用 **history-first** 架构,以 `chat.history` 作为唯一数据真相源(single source of truth),流式事件仅用于 UI 实时预览。 + +### 设计原则 + +1. **`chat.history` 是唯一真相源** — 流式事件仅用于实时预览,不作为最终持久化依据 +2. **Turn 完成后全量对账** — 每次 turn 结束时调用 `reconcileWithHistory` 与网关历史对齐 +3. **幂等对账** — 对账操作可重复执行,不产生副作用 +4. **准确性优先于延迟** — 允许 turn 完成后的 HTTP 往返延迟 + +--- + +## 2. 两条 IM 路径 + +### 2.1 Managed IM Sessions(IMCoworkHandler 路径) + +**适用场景**:用户通过 IM 平台直接与 LobsterAI 对话(钉钉、飞书、Telegram 等)。 + +**入口文件**:`src/main/im/imCoworkHandler.ts` + +**流程**: + +``` +IM 消息 (platform:conversationId) + | + v +IMCoworkHandler.processMessage() + | + v +getOrCreateCoworkSession() + |- 查询 im_session_mappings 是否存在映射 + |- 不存在则创建新 cowork session + 持久化映射 + | + v +coworkRuntime.startSession() / continueSession() + | + v +流式事件: + - message -> handleMessage() -> accumulator 收集 + - messageUpdate -> handleMessageUpdate() -> accumulator 更新 + - permissionRequest -> 向 IM 发送确认提示,等待用户回复 + | + v +handleChatFinal() [openclawRuntimeAdapter.ts] + |- await reconcileWithHistory() <- 等待对账完成 + |- emit('complete') + | + v +handleComplete() [imCoworkHandler.ts] + |- 从 coworkStore 读取对账后的消息(而非 accumulator) + |- formatReply() 格式化回复 + |- resolve Promise,回复 IM 用户 +``` + +**关键组件 — MessageAccumulator**: + +```typescript +interface MessageAccumulator { + messages: CoworkMessage[]; // 流式收集的消息 + resolve?: (text: string) => void; // 完成后 resolve + reject?: (error: Error) => void; // 出错后 reject + timeoutId?: NodeJS.Timeout; // 超时保护 + backgroundDelivery?: { // 定时任务后台投递 + conversationId: string; + platform: IMPlatform; + }; +} +``` + +Accumulator 在流式期间收集消息,但 `handleComplete` 最终从 coworkStore 读取对账后的权威消息来构建回复: + +```typescript +const session = this.coworkStore.getSession(sessionId); +const storeMessages = session?.messages ?? []; +const messages = storeMessages.length > 0 ? storeMessages : accumulator.messages; +``` + +### 2.2 Channel Sessions(OpenClaw 轮询路径) + +**适用场景**:Telegram/Discord 等通过 OpenClaw 扩展直接接入的频道会话,LobsterAI 通过轮询发现新会话。 + +**入口文件**:`src/main/libs/agentEngine/openclawRuntimeAdapter.ts` + `src/main/libs/openclawChannelSessionSync.ts` + +**流程**: + +``` +pollChannelSessions() + | + v +遍历 channelSessionKeys: + channelSessionSync.resolveOrCreateSession(key) + |- 解析 sessionKey 获取 platform + conversationId + |- 查询/创建 im_session_mappings + |- 创建 coworkStore session + | + v +gateway 推送 chat 事件 + | + v +Turn 开始: + prefetchChannelUserMessages(sessionId, sessionKey) + |- 调用 chat.history 获取最新消息 + |- syncChannelUserMessages() 将用户消息同步到本地 + |- 回放缓冲的 chat/agent 事件 + | + v +流式: + handleChatDelta -> emit messageUpdate + handleAgentEvent -> emit tool_use/tool_result + | + v +handleChatFinal() + |- await reconcileWithHistory(sessionId, sessionKey) + |- 发送 cowork:sessions:changed 通知渲染进程 +``` + +### 2.3 Session Key 格式 + +| 类型 | 格式 | 示例 | +|------|------|------| +| Managed(LobsterAI 发起) | `agent:main:lobsterai:{sessionId}` | `agent:main:lobsterai:abc123` | +| Channel(平台发起) | `agent:{agentId}:{platform}:{subtype}:{conversationId}` | `agent:bot1:telegram:private:12345` | +| Cron(定时任务) | `cron:{jobId}` 或 `agent:{agentId}:cron:{jobId}` | `cron:task-001` | + +--- + +## 3. 核心对账机制:`reconcileWithHistory` + +### 3.1 方法签名 + +```typescript +// openclawRuntimeAdapter.ts +private async reconcileWithHistory( + sessionId: string, + sessionKey: string, + options?: { isFullSync?: boolean }, +): Promise +``` + +### 3.2 算法步骤 + +``` +1. 调用 chat.history 获取网关权威消息列表 + | + v +2. 同步系统消息(定时提醒等) + - 使用 gatewayHistoryCountBySession 游标跟踪 + - 仅处理游标之后的新系统消息 + | + v +3. 提取权威 user/assistant 条目 + - 过滤 role 为 user 或 assistant 的消息 + - 文本规范化: + - Discord: 移除 <@userId>、<#channelId> 等 mention 标记 + - QQ: 移除机器人系统提示词(【...】块) + - Channel 会话: 追加文件发送路径为 markdown 链接 + | + v +4. 提取本地 user/assistant 条目 + - 从 coworkStore.getSession().messages 中筛选 + | + v +5. 比较 + - 逐条比较 role 和 text + - 如果完全一致 → 跳过,仅更新游标 + | + v +6. 不一致则替换 + - 调用 store.replaceConversationMessages() + - 删除所有本地 user/assistant 消息 + - 按权威列表重新插入 + - 保留 tool_use/tool_result/system 类型消息 + - 通知渲染进程刷新 UI +``` + +### 3.3 调用时机 + +| 调用点 | 触发场景 | 等待方式 | 说明 | +|--------|----------|----------|------| +| `handleChatFinal` | Turn 正常完成 | `await` | 确保 IM 回复使用对账后的数据 | +| `handleChatAborted` | Turn 被中止 | `void` | 同步中止前已交付的消息 | +| `handleChatError` | Turn 出错 | `void` | 同步出错前已交付的消息 | +| `syncFullChannelHistory` | Channel 会话首次发现 | `await` | `isFullSync: true` | +| `incrementalChannelSync` | 轮询增量同步 | `await` | 已知 channel 的增量对账 | + +### 3.4 `replaceConversationMessages` 实现 + +```typescript +// coworkStore.ts +replaceConversationMessages( + sessionId: string, + authoritative: Array<{ role: 'user' | 'assistant'; text: string }>, +): void { + // 1. 删除所有现有 user/assistant 消息(保留 tool/system) + db.run( + "DELETE FROM cowork_messages WHERE session_id = ? AND type IN ('user', 'assistant')", + [sessionId], + ); + + // 2. 获取剩余消息的最大 sequence + let nextSeq = maxExistingSequence + 1; + + // 3. 按权威列表顺序重新插入 + for (const entry of authoritative) { + db.run(`INSERT INTO cowork_messages (...) VALUES (...)`, [ + uuid(), sessionId, entry.role, entry.text, + JSON.stringify({ isStreaming: false, isFinal: true }), + now, nextSeq++, + ]); + } +} +``` + +--- + +## 4. 用户消息预取:`prefetchChannelUserMessages` + +Channel 会话中,当 OpenClaw 开始一个新 turn 时,网关可能已经在推送 assistant 的流式事件。为了保证用户消息在 assistant 回复之前出现在 UI 中,系统会在 turn 开始时预取用户消息。 + +### 4.1 流程 + +1. 标记 `turn.pendingUserSync = true`,缓冲所有到达的流式事件 +2. 调用 `chat.history` 获取最新消息 +3. 通过 `syncChannelUserMessages` 将新的用户消息同步到本地 +4. 如果未发现新用户消息但有缓冲事件,等待 500ms 后重试(最多 2 次) +5. 标记 `turn.pendingUserSync = false` +6. 按 sequence 顺序回放缓冲的 chat + agent 事件 + +### 4.2 事件缓冲与排序 + +```typescript +interface BufferedChatEvent { + payload: unknown; + seq?: number; // 网关分配的序列号 + bufferedAt: number; // 本地缓冲时间 +} +``` + +排序优先级: +1. 有 `seq` 的事件按 `seq` 升序 +2. 有 `seq` 的事件排在无 `seq` 的事件之前 +3. 同类事件按 `bufferedAt` 时间排序 +4. 最后按插入顺序(`idx`)排序 + +### 4.3 与 `reconcileWithHistory` 的关系 + +`prefetchChannelUserMessages` 负责 **turn 开始前** 的快速用户消息同步,确保 UI 显示顺序正确。即使预取不完整,`reconcileWithHistory` 在 **turn 结束后** 会做全量对账,保证最终一致性。因此预取的重试次数从 5 次减少到 2 次。 + +--- + +## 5. 系统消息与定时提醒同步 + +### 5.1 `syncSystemMessagesFromHistory` + +```typescript +private syncSystemMessagesFromHistory( + sessionId: string, + historyMessages: unknown[], + options: { previousCountKnown: boolean; previousCount: number }, +): void +``` + +**游标追踪**:通过 `gatewayHistoryCountBySession` Map 记录上次同步时的历史消息数量。仅处理游标之后的新消息,避免重复添加老的系统消息。 + +**提醒检测**:`extractGatewayHistoryEntries`(`openclawHistory.ts`)会检测用户消息中的定时提醒格式,将其转换为 `role: 'system'` 类型的条目。 + +### 5.2 IM 回复守卫(Reminder Guard) + +**文件**:`src/main/im/imReplyGuard.ts` + +当 assistant 声称创建了定时提醒但实际 `cron.add` 工具调用失败时,守卫会拦截误导性回复: + +- 检测模式:「我会...提醒」、「已...创建...定时任务」、"I'll set a reminder" 等 +- 如果有未成功的 `cron.add` 调用 → 替换为错误提示 +- 后台投递的定时提醒(`backgroundDelivery`)绕过此守卫 + +--- + +## 6. 关键常量与超时 + +### 6.1 消息限制 + +| 常量 | 值 | 用途 | +|------|-----|------| +| `FINAL_HISTORY_SYNC_LIMIT` | 50 | 对账时获取的历史消息数上限 | +| `FULL_HISTORY_SYNC_LIMIT` | 50 | 全量同步时获取的历史消息数上限 | +| `CHANNEL_SESSION_DISCOVERY_LIMIT` | 200 | 轮询时最大发现会话数 | +| `BRIDGE_MAX_MESSAGES` | 20 | agent prompt 中包含的最近消息数 | +| `BRIDGE_MAX_MESSAGE_CHARS` | 1200 | 单条消息最大字符数(截断) | + +### 6.2 超时 + +| 常量 | 值 | 用途 | +|------|-----|------| +| `ACCUMULATOR_TIMEOUT_MS` | 5 分钟 | IM 等待流式响应的最大时间 | +| `PERMISSION_CONFIRM_TIMEOUT_MS` | 60 秒 | 等待用户 IM 权限确认 | +| `GATEWAY_READY_TIMEOUT_MS` | 15 秒 | 等待网关连接就绪 | +| `TICK_TIMEOUT_MS` | 90 秒 | 看门狗超时阈值 | + +--- + +## 7. 错误处理与恢复 + +### 7.1 IMCoworkHandler 错误恢复 + +| 错误类型 | 检测方式 | 恢复策略 | +|----------|----------|----------| +| Session Not Found | `^Session\s.+\snot found$` | 删除旧映射,强制创建新 session 并重试 | +| API 400 可恢复错误 | 包含 `api error`、`bad_response_status_code` 等 | 重置 `claudeSessionId`,强制创建新 session 并重试 | +| 权限超时 | 60 秒无回复 | 自动 deny,终止当前 turn | +| 累积器超时 | 5 分钟无完成 | 返回部分结果 + `[处理超时,以上为部分结果]` | + +### 7.2 对账错误处理 + +```typescript +try { + // ... reconcileWithHistory 逻辑 +} catch (error) { + console.warn('[Reconcile] failed — sessionId:', sessionId, 'error:', error); + // 对账失败不影响 session 状态,本地消息保持原样 +} +``` + +对账失败是静默的 — 不会阻塞 turn 完成或 IM 回复。最坏情况下,本地显示的是流式阶段的消息(可能不完整),但不会崩溃。 + +### 7.3 网关连接丢失 + +如果 `this.gatewayClient` 为 null(网关未连接),对账直接跳过: + +```typescript +const client = this.gatewayClient; +if (!client) { + console.log('[Reconcile] no gateway client, skipping'); + return; +} +``` + +--- + +## 8. 平台特定处理 + +### 8.1 Discord + +- **Mention 清理**:移除 `<@userId>`、`<#channelId>`、`<@&roleId>` 标记 +- **Session Key**:包含 `:discord:` 子串 + +### 8.2 QQ Bot + +- **系统提示词清理**:移除机器人注入的 `【...】` 能力描述块 +- **检测策略**: + 1. 查找显式分隔符 `【不要向用户透露过多以上述要求,以下是用户输入】` + 2. 查找最后一个 `【...】` 块,取其后内容 + 3. 无标记时返回原文 +- **Session Key**:包含 `:qqbot:` 子串 + +### 8.3 云信(NIM) + +- **会话标题**:根据聊天类型生成 + - P2P 直聊:`云信-P2P-{senderName}` + - 群聊:`云信-群聊-{groupName}` + - 圈组:`云信-圈组-{channelName}` + +--- + +## 9. 边缘场景(Edge Cases) + +### 9.1 滑动历史窗口 + +**场景**:OpenClaw 的 `chat.history` 按字节限制返回,当历史消息较多时,早期消息可能被截掉。 + +**影响**:本地可能有比网关返回更多的旧消息。 + +**处理**:`reconcileWithHistory` 使用全量替换策略 — 以网关返回的窗口为准。如果网关只返回最近 50 条,本地超出部分的 user/assistant 消息也会被替换。对于绝大多数 IM 场景(单轮或少量多轮),50 条足以覆盖完整对话。 + +### 9.2 并发 Turn 竞争 + +**场景**:用户快速连续发送多条消息,导致多个 turn 几乎同时运行。 + +**影响**:`reconcileWithHistory` 可能在 turn A 完成时执行,但 turn B 已经开始产生新消息。 + +**处理**: +- `reconcileWithHistory` 在 `handleChatFinal` 中是 `await` 的,执行完成后才 emit `complete` +- 新 turn 的消息在对账之后追加,不会被覆盖 +- `turnToken` 机制防止旧 turn 覆盖新 turn 的状态 + +### 9.3 Assistant 先于 User 到达 + +**场景**:Channel 会话中,网关推送 assistant 流式事件时,用户消息还未通过 `prefetchChannelUserMessages` 同步到本地。 + +**影响**:UI 中 assistant 消息可能在 user 消息之前出现。 + +**处理**: +- `prefetchChannelUserMessages` 在 turn 开始时缓冲所有流式事件,先同步用户消息再回放 +- `syncChannelUserMessages` 使用 `insertMessageBeforeId` 将用户消息插入到已存在的 assistant 消息之前 +- 最终 `reconcileWithHistory` 全量对账修正顺序 + +### 9.4 流式内容与最终内容不一致 + +**场景**:流式期间 `handleChatDelta` 推送的文本片段拼接后,与 `chat.history` 中的最终文本不同(如 token 边界不同、中间有重试等)。 + +**影响**:accumulator 中的消息内容可能不完整或有拼接错误。 + +**处理**: +- `handleComplete` 使用 coworkStore 中对账后的消息,而非 accumulator +- `replaceConversationMessages` 将所有 user/assistant 消息替换为权威版本 + +### 9.5 对账期间网关超时 + +**场景**:`chat.history` 请求在网络不稳定时超时。 + +**影响**:对账失败,本地消息保持流式阶段的状态。 + +**处理**: +- 对账错误被 try-catch 捕获并静默记录日志 +- 不影响 turn 完成状态和 IM 回复 +- 下次 turn 完成时会再次对账,最终收敛到正确状态 + +### 9.6 Session 重建后的历史丢失 + +**场景**:由于 API 400 错误或 session not found,IMCoworkHandler 强制创建新 session(`forceNewSession=true`),新 session 没有历史消息。 + +**影响**:之前的对话上下文丢失。 + +**处理**: +- 这是预期行为 — 错误场景下优先保证可用性 +- 旧 session 的数据仍在 SQLite 中(未删除 cowork_sessions 记录) +- IM 平台映射更新为新 session ID + +### 9.7 Tool 消息的保留 + +**场景**:对账时网关历史仅包含 user/assistant 角色的消息,不包含 tool_use/tool_result。 + +**影响**:如果简单全量替换,tool 相关消息会丢失。 + +**处理**: +- `replaceConversationMessages` 仅删除 `type IN ('user', 'assistant')` 的消息 +- `tool_use`、`tool_result`、`system` 类型的消息不受影响 +- 新插入的 user/assistant 消息的 sequence 从现有最大值 +1 开始 + +### 9.8 空历史响应 + +**场景**:`chat.history` 返回空消息列表(新创建的 session 或网关异常)。 + +**影响**:不应删除本地已有的消息。 + +**处理**: +- 空历史时直接返回,仅将 `channelSyncCursor` 设为 0 +- 不触发 `replaceConversationMessages` + +### 9.9 定时任务后台投递 + +**场景**:Cron 触发的定时提醒需要通过 IM 发送,但此时没有用户主动请求。 + +**影响**:没有活跃的 accumulator 来收集消息。 + +**处理**: +- `ensureBackgroundAccumulator` 创建一个带 `backgroundDelivery` 标记的 accumulator +- `handleComplete` 检测到 `backgroundDelivery` 后使用 `sendAsyncReply` 异步发送 +- 绕过 reminder guard(定时提醒本身就是合法的提醒内容) +- 通过 `isReminderSystemTurn` 验证 turn 确实是定时提醒触发的 + +--- + +## 10. 验证方案 + +### 10.1 自动化测试 + +**测试文件**:`tests/openclawReconcile.test.mjs` + +| 测试用例 | 验证内容 | +|----------|----------| +| already in sync — skips replace | 本地与网关一致时不触发替换 | +| missing assistant message — triggers replace | 本地缺失消息时触发替换 | +| duplicate messages locally — triggers replace | 本地有重复消息时触发替换 | +| content mismatch — triggers replace | 流式内容与最终内容不同时触发替换 | +| preserves tool messages | tool_use/tool_result 消息不被对账影响 | +| empty history — sets cursor to 0 | 空历史不删除本地消息 | +| multi-turn conversation — correct order | 多轮对话顺序正确 | +| gateway error — does not crash | 网关错误不导致崩溃 | + +**运行**: + +```bash +npm run compile:electron && node --test tests/openclawReconcile.test.mjs +``` + +**其他相关测试**: + +```bash +node --test tests/openclawRuntimeAdapter.history.test.mjs # 历史同步测试 +node --test tests/openclawHistory.test.mjs # 历史提取工具测试 +node --test tests/imReplyGuard.test.mjs # IM 回复守卫测试 +``` + +### 10.2 手动测试矩阵 + +#### 基础功能验证 + +| 测试项 | 步骤 | 预期结果 | +|--------|------|----------| +| 单轮对话 | 通过 IM 发送一条消息 | UI 显示用户消息和 AI 回复,内容与 OpenClaw Dashboard 一致 | +| 多轮对话 | 连续发送 3-5 条消息 | 所有消息按正确顺序显示,无丢失 | +| 带 tool 调用 | 发送需要工具调用的任务(如"创建一个文件") | tool_use/tool_result 消息正确显示,最终回复完整 | +| 权限确认 | 触发需要权限的操作 | IM 收到确认提示,回复"允许"后继续执行 | +| 定时提醒 | 发送"5分钟后提醒我开会" | 创建成功,5 分钟后 IM 收到提醒消息 | + +#### 同步准确性验证 + +| 测试项 | 步骤 | 预期结果 | +|--------|------|----------| +| 消息完整性 | 多轮对话后对比 LobsterAI UI 和 OpenClaw Dashboard | 消息数量和内容完全一致 | +| 消息顺序 | 快速连续发送 3 条消息 | user/assistant 交替出现,顺序正确 | +| 无重复 | 同一条消息不应出现两次 | 对比 UI 中无重复消息 | +| 流式→最终 | 观察流式过程和最终结果 | 流式完成后,消息内容与 Dashboard 一致 | + +#### 边缘场景验证 + +| 测试项 | 步骤 | 预期结果 | +|--------|------|----------| +| 网络中断恢复 | 对话过程中断开网络,恢复后继续 | 下次 turn 完成时自动对账修正 | +| 超时处理 | 发送需要长时间处理的任务(超过 5 分钟) | 收到部分结果 + 超时提示 | +| 权限超时 | 触发权限确认但不回复 | 60 秒后自动拒绝,收到提示 | +| 快速连续消息 | 1 秒内连续发送 5 条消息 | 所有消息最终正确同步,无丢失或乱序 | + +#### 多平台验证 + +| 平台 | 特殊验证点 | +|------|-----------| +| Discord | mention 标记 `<@userId>` 不出现在消息内容中 | +| QQ Bot | 系统提示词 `【...】` 被正确移除 | +| 云信(NIM) | 会话标题正确反映聊天类型(P2P/群聊/圈组) | +| Telegram | 基础消息收发正常 | +| 钉钉 | 基础消息收发正常 | + +### 10.3 日志检查 + +对账执行时会输出以下日志,可用于排查问题: + +``` +[Reconcile] already in sync — sessionId: xxx entries: N +[Reconcile] replacing messages — sessionId: xxx local: N → authoritative: M +[Reconcile] empty history — sessionId: xxx +[Reconcile] no user/assistant entries in history — sessionId: xxx +[Reconcile] failed — sessionId: xxx error: ... +[Reconcile] no gateway client, skipping — sessionId: xxx +``` + +预取相关日志: + +``` +[Debug:prefetch] start — sessionId: xxx sessionKey: yyy +[Debug:prefetch] chat.history returned N messages (attempt X) +[Debug:prefetch] synced user messages: N (before: A after: B) +[Debug:prefetch] replay complete, sessionId: xxx +``` + +--- + +## 11. 文件清单 + +| 文件 | 角色 | +|------|------| +| `src/main/libs/agentEngine/openclawRuntimeAdapter.ts` | 核心:`reconcileWithHistory`、`prefetchChannelUserMessages`、`handleChatFinal` | +| `src/main/im/imCoworkHandler.ts` | Managed IM 适配器:`handleComplete` 使用对账后数据 | +| `src/main/coworkStore.ts` | 持久化层:`replaceConversationMessages`、`deleteMessage` | +| `src/main/libs/openclawChannelSessionSync.ts` | Channel 会话发现与映射 | +| `src/main/libs/openclawHistory.ts` | 历史消息提取工具 | +| `src/main/im/imReplyGuard.ts` | IM 回复守卫(防止误导性提醒) | +| `src/main/sqliteStore.ts` | SQLite 数据库初始化与表结构 | +| `tests/openclawReconcile.test.mjs` | 对账算法自动化测试 | diff --git a/src/main/coworkStore.ts b/src/main/coworkStore.ts index b92d42b..d0aaefe 100644 --- a/src/main/coworkStore.ts +++ b/src/main/coworkStore.ts @@ -843,6 +843,67 @@ export class CoworkStore { }; } + /** + * Delete a message from a session. + * Used by reconciliation to remove duplicate or spurious messages. + */ + deleteMessage(sessionId: string, messageId: string): boolean { + this.db.run( + 'DELETE FROM cowork_messages WHERE id = ? AND session_id = ?', + [messageId, sessionId], + ); + const deleted = (this.db.getRowsModified?.() || 0) > 0; + if (deleted) { + this.saveDb(); + } + return deleted; + } + + /** + * Replace all user/assistant messages in a session with the given list. + * Tool messages (tool_use, tool_result, system) are preserved in their existing positions. + * Used by history reconciliation to align local state with the authoritative gateway history. + */ + replaceConversationMessages( + sessionId: string, + authoritative: Array<{ role: 'user' | 'assistant'; text: string }>, + ): void { + const now = Date.now(); + + // Delete all existing user/assistant messages for this session + this.db.run( + "DELETE FROM cowork_messages WHERE session_id = ? AND type IN ('user', 'assistant')", + [sessionId], + ); + + // Re-insert authoritative messages with correct sequence numbers + // First, get the current max sequence from remaining messages (tool_use, tool_result, system) + const seqRow = this.db.exec( + 'SELECT COALESCE(MAX(sequence), 0) as max_seq FROM cowork_messages WHERE session_id = ?', + [sessionId], + ); + let nextSeq = ((seqRow[0]?.values[0]?.[0] as number) || 0) + 1; + + for (const entry of authoritative) { + const id = uuidv4(); + this.db.run(` + INSERT INTO cowork_messages (id, session_id, type, content, metadata, created_at, sequence) + VALUES (?, ?, ?, ?, ?, ?, ?) + `, [ + id, + sessionId, + entry.role, + entry.text, + JSON.stringify({ isStreaming: false, isFinal: true }), + now, + nextSeq++, + ]); + } + + this.db.run('UPDATE cowork_sessions SET updated_at = ? WHERE id = ?', [now, sessionId]); + this.saveDb(); + } + updateMessage(sessionId: string, messageId: string, updates: { content?: string; metadata?: CoworkMessageMetadata }): void { const setClauses: string[] = []; const values: (string | null)[] = []; diff --git a/src/main/im/imCoworkHandler.ts b/src/main/im/imCoworkHandler.ts index f53234f..906960a 100644 --- a/src/main/im/imCoworkHandler.ts +++ b/src/main/im/imCoworkHandler.ts @@ -821,19 +821,27 @@ export class IMCoworkHandler extends EventEmitter { return; } + // Use reconciled messages from the store (authoritative after reconcileWithHistory) + // instead of accumulator messages which may be stale streaming snapshots. + // Fall back to accumulator messages if the store has none (e.g. timeout path). + const session = this.coworkStore.getSession(sessionId); + const storeMessages = session?.messages ?? []; + const messages = storeMessages.length > 0 ? storeMessages : accumulator.messages; + // For cron-triggered background deliveries (scheduled task executions), // skip the reminder guard — the assistant text IS the scheduled reminder // itself, not a promise to create one. const replyText = accumulator.backgroundDelivery - ? this.formatReplyRaw(accumulator.messages) - : this.formatReply(sessionId, accumulator.messages); + ? this.formatReplyRaw(messages) + : this.formatReply(sessionId, messages); console.log(`[IMCoworkHandler] 会话完成:`, JSON.stringify({ sessionId, - messageCount: accumulator.messages.length, + messageCount: messages.length, replyLength: replyText.length, reply: replyText, backgroundDelivery: accumulator.backgroundDelivery ?? null, + usedStoreMessages: storeMessages.length > 0, }, null, 2)); this.cleanupAccumulator(sessionId); @@ -842,7 +850,7 @@ export class IMCoworkHandler extends EventEmitter { if (!this.sendAsyncReply || !replyText || replyText === '处理完成,但没有生成回复。') { return; } - if (!isReminderSystemTurn(accumulator.messages)) { + if (!isReminderSystemTurn(messages)) { return; } void this.sendAsyncReply( diff --git a/src/main/libs/agentEngine/openclawRuntimeAdapter.ts b/src/main/libs/agentEngine/openclawRuntimeAdapter.ts index 5776daa..b88f4c0 100644 --- a/src/main/libs/agentEngine/openclawRuntimeAdapter.ts +++ b/src/main/libs/agentEngine/openclawRuntimeAdapter.ts @@ -1584,7 +1584,6 @@ export class OpenClawRuntimeAdapter extends EventEmitter implements CoworkRuntim } private handleGatewayEvent(event: GatewayEventFrame): void { - if (event.event === 'tick') { this.lastTickTimestamp = Date.now(); return; @@ -1611,6 +1610,10 @@ export class OpenClawRuntimeAdapter extends EventEmitter implements CoworkRuntim if (event.event === 'exec.approval.resolved') { this.handleApprovalResolved(event.payload); } + + if (event.event === 'cron') { + console.debug('[OpenClawRuntime] received cron event:', JSON.stringify(event)); + } } private handleAgentEvent(payload: unknown, seq?: number): void { @@ -2296,7 +2299,7 @@ export class OpenClawRuntimeAdapter extends EventEmitter implements CoworkRuntim } } - private handleChatFinal(sessionId: string, turn: ActiveTurn, payload: ChatEventPayload): void { + private async handleChatFinal(sessionId: string, turn: ActiveTurn, payload: ChatEventPayload): Promise { const previousText = turn.currentText; const previousSegmentText = turn.currentAssistantSegmentText; const finalText = this.resolveFinalTurnText(turn, payload.message); @@ -2363,23 +2366,20 @@ export class OpenClawRuntimeAdapter extends EventEmitter implements CoworkRuntim this.emit('error', sessionId, errorMessage); this.cleanupSessionTurn(sessionId); this.rejectTurn(sessionId, new Error(errorMessage)); - // Sync channel history even on error so the UI shows the messages that - // were already delivered through the IM connector (e.g. DingTalk reply). - this.syncChannelAfterTurn(sessionId, erroredSessionKey); + // Reconcile even on error so the UI shows messages already delivered. + void this.reconcileWithHistory(sessionId, erroredSessionKey); return; } - // Align final assistant text with persisted gateway history to reduce mid-stream drift. - void this.syncFinalAssistantWithHistory(sessionId, turn); + // Reconcile local messages with authoritative gateway history. + // This replaces the old syncFinalAssistantWithHistory + syncChannelAfterTurn flow. + // Awaited so that IM handlers reading from the store see reconciled data. + await this.reconcileWithHistory(sessionId, turn.sessionKey); this.store.updateSession(sessionId, { status: 'completed' }); this.emit('complete', sessionId, payload.runId ?? turn.runId); - const completedSessionKey = turn.sessionKey; this.cleanupSessionTurn(sessionId); this.resolveTurn(sessionId); - // Trigger immediate incremental sync for channel sessions so UI reflects - // the latest messages without waiting for the next polling cycle. - this.syncChannelAfterTurn(sessionId, completedSessionKey); } private handleChatAborted(sessionId: string, turn: ActiveTurn): void { @@ -2398,7 +2398,7 @@ export class OpenClawRuntimeAdapter extends EventEmitter implements CoworkRuntim const abortedSessionKey = turn.sessionKey; this.cleanupSessionTurn(sessionId); this.resolveTurn(sessionId); - this.syncChannelAfterTurn(sessionId, abortedSessionKey); + void this.reconcileWithHistory(sessionId, abortedSessionKey); } private handleChatError(sessionId: string, turn: ActiveTurn, payload: ChatEventPayload): void { @@ -2423,7 +2423,7 @@ export class OpenClawRuntimeAdapter extends EventEmitter implements CoworkRuntim this.emit('error', sessionId, errorMessage); this.cleanupSessionTurn(sessionId); this.rejectTurn(sessionId, new Error(errorMessage)); - this.syncChannelAfterTurn(sessionId, erroredSessionKey); + void this.reconcileWithHistory(sessionId, erroredSessionKey); } private handleApprovalRequested(payload: unknown): void { @@ -2591,6 +2591,148 @@ export class OpenClawRuntimeAdapter extends EventEmitter implements CoworkRuntim this.gatewayHistoryCountBySession.set(sessionId, historyMessages.length); } + /** + * Reconcile local session messages with the authoritative gateway chat.history. + * + * This is the single source-of-truth sync method: after a turn completes, + * it fetches the full conversation from OpenClaw and overwrites local + * user/assistant messages to match exactly. Tool messages (tool_use, + * tool_result, system) are kept as-is because the gateway does not + * expose them in chat.history. + * + * The reconciliation is idempotent — calling it multiple times produces + * the same result. + */ + private async reconcileWithHistory( + sessionId: string, + sessionKey: string, + options?: { isFullSync?: boolean }, + ): Promise { + const client = this.gatewayClient; + if (!client) { + console.log('[Reconcile] no gateway client, skipping — sessionId:', sessionId); + return; + } + + // Skip reconciliation for main-window (managed) sessions — local store is + // the source of truth; only channel/IM sessions need gateway reconciliation. + if (isManagedSessionKey(sessionKey)) { + return; + } + + const limit = options?.isFullSync + ? OpenClawRuntimeAdapter.FULL_HISTORY_SYNC_LIMIT + : FINAL_HISTORY_SYNC_LIMIT; + + try { + const history = await client.request<{ messages?: unknown[] }>('chat.history', { + sessionKey, + limit, + }); + if (!Array.isArray(history?.messages) || history.messages.length === 0) { + console.log('[Reconcile] empty history — sessionId:', sessionId); + this.channelSyncCursor.set(sessionId, 0); + return; + } + + // Update gateway history cursor for system message tracking + this.gatewayHistoryCountBySession.set(sessionId, history.messages.length); + + // Sync system messages (reminders etc.) + const previousHistoryCountKnown = this.gatewayHistoryCountBySession.has(sessionId); + const previousHistoryCount = this.gatewayHistoryCountBySession.get(sessionId) ?? 0; + this.syncSystemMessagesFromHistory(sessionId, history.messages, { + previousCountKnown: previousHistoryCountKnown, + previousCount: previousHistoryCount, + }); + + // Determine if this is a channel session (for Discord/QQ text normalization) + const isChannel = this.channelSessionSync + && !isManagedSessionKey(sessionKey) + && this.channelSessionSync.isChannelSessionKey(sessionKey); + const isDiscord = sessionKey.includes(':discord:'); + const isQQ = sessionKey.includes(':qqbot:'); + + // Extract authoritative user/assistant entries from gateway history + const authoritativeEntries: Array<{ role: 'user' | 'assistant'; text: string }> = []; + for (const message of history.messages) { + if (!isRecord(message)) continue; + const role = typeof message.role === 'string' ? message.role.trim().toLowerCase() : ''; + if (role !== 'user' && role !== 'assistant') continue; + let text = extractMessageText(message).trim(); + if (!text) continue; + if (isDiscord) text = stripDiscordMentions(text); + if (isQQ && role === 'user') text = stripQQBotSystemPrompt(text); + authoritativeEntries.push({ role: role as 'user' | 'assistant', text }); + } + + // For channel sessions, append file paths from "message" tool calls + if (isChannel && authoritativeEntries.length > 0) { + const sentFilePaths = extractSentFilePathsFromHistory(history.messages); + if (sentFilePaths.length > 0) { + const lastAssistantIdx = authoritativeEntries.findLastIndex(e => e.role === 'assistant'); + if (lastAssistantIdx >= 0) { + const fileLinks = sentFilePaths + .map((fp) => `[${path.basename(fp)}](${fp})`) + .join('\n'); + authoritativeEntries[lastAssistantIdx] = { + ...authoritativeEntries[lastAssistantIdx], + text: `${authoritativeEntries[lastAssistantIdx].text}\n\n${fileLinks}`, + }; + } + } + } + + if (authoritativeEntries.length === 0) { + console.log('[Reconcile] no user/assistant entries in history — sessionId:', sessionId); + this.channelSyncCursor.set(sessionId, 0); + return; + } + + // Collect local user/assistant messages for comparison + const session = this.store.getSession(sessionId); + const localEntries: Array<{ role: 'user' | 'assistant'; text: string }> = []; + if (session) { + for (const msg of session.messages) { + if (msg.type !== 'user' && msg.type !== 'assistant') continue; + const text = msg.content.trim(); + if (!text) continue; + localEntries.push({ role: msg.type, text }); + } + } + + // Compare: if already in sync, skip the expensive replace + const isInSync = localEntries.length === authoritativeEntries.length + && localEntries.every((entry, idx) => + entry.role === authoritativeEntries[idx].role + && entry.text === authoritativeEntries[idx].text, + ); + + if (isInSync) { + console.log('[Reconcile] already in sync — sessionId:', sessionId, 'entries:', localEntries.length); + this.channelSyncCursor.set(sessionId, authoritativeEntries.length); + return; + } + + // Replace local messages with authoritative ones + console.log( + '[Reconcile] replacing messages — sessionId:', sessionId, + 'local:', localEntries.length, '→ authoritative:', authoritativeEntries.length, + ); + this.store.replaceConversationMessages(sessionId, authoritativeEntries); + this.channelSyncCursor.set(sessionId, authoritativeEntries.length); + + // Notify renderer to refresh + for (const win of BrowserWindow.getAllWindows()) { + if (!win.isDestroyed()) { + win.webContents.send('cowork:sessions:changed'); + } + } + } catch (error) { + console.warn('[Reconcile] failed — sessionId:', sessionId, 'error:', error); + } + } + private async syncFinalAssistantWithHistory(sessionId: string, turn: ActiveTurn): Promise { console.log('[Debug:syncFinal] start — sessionId:', sessionId, 'sessionKey:', turn.sessionKey); const client = this.gatewayClient; @@ -3036,180 +3178,8 @@ export class OpenClawRuntimeAdapter extends EventEmitter implements CoworkRuntim if (this.fullySyncedSessions.has(sessionId)) return; this.fullySyncedSessions.add(sessionId); - const client = this.gatewayClient; - if (!client) return; - - console.log('[ChannelSync] syncFullChannelHistory: start — sessionId:', sessionId, 'sessionKey:', sessionKey); try { - const history = await client.request<{ messages?: unknown[] }>('chat.history', { - sessionKey, - limit: OpenClawRuntimeAdapter.FULL_HISTORY_SYNC_LIMIT, - }); - if (!Array.isArray(history?.messages) || history.messages.length === 0) { - console.log('[ChannelSync] syncFullChannelHistory: no messages in history'); - return; - } - this.markGatewayHistoryWindowConsumed(sessionId, history.messages); - - const session = this.store.getSession(sessionId); - // Build ordered list of existing local messages for position-based matching - type MsgEntry = { role: 'user' | 'assistant'; text: string }; - const localEntries: MsgEntry[] = []; - if (session) { - for (const msg of session.messages) { - if (msg.type === 'user' || msg.type === 'assistant') { - localEntries.push({ role: msg.type, text: msg.content.trim() }); - } - } - } - - const isDiscord = sessionKey.includes(':discord:'); - const isQQ = sessionKey.includes(':qqbot:'); - // Build history entries - const historyEntries: MsgEntry[] = []; - for (const message of history.messages) { - if (!isRecord(message)) continue; - const role = typeof message.role === 'string' ? message.role.trim().toLowerCase() : ''; - if (role !== 'user' && role !== 'assistant') continue; - let text = extractMessageText(message).trim(); - if (isDiscord) text = stripDiscordMentions(text); - if (isQQ && role === 'user') text = stripQQBotSystemPrompt(text); - if (text) { - historyEntries.push({ role: role as 'user' | 'assistant', text }); - } - } - - // Find where local messages end in the history to determine firstNewIdx. - // Primary strategy: reverse matching from the end of both arrays. - // This handles the sliding window case where oldest local messages may no - // longer appear in history (e.g. local has [m1..m50], history has [m3..m52]). - let firstNewIdx = 0; - if (localEntries.length > 0) { - const lastLocal = localEntries[localEntries.length - 1]; - let matchPos = -1; - for (let i = historyEntries.length - 1; i >= 0; i--) { - if (historyEntries[i].role === lastLocal.role - && historyEntries[i].text === lastLocal.text) { - if (localEntries.length >= 2 && i > 0) { - const secondLastLocal = localEntries[localEntries.length - 2]; - if (historyEntries[i - 1].role === secondLastLocal.role - && historyEntries[i - 1].text === secondLastLocal.text) { - matchPos = i; - break; - } - continue; - } - matchPos = i; - break; - } - } - - // Fallback: if the last local entry is an assistant message and matching failed, - // the stored text may be a segment (from resolveAssistantSegmentText) that differs - // from the full text in gateway history. Retry with the last USER message instead. - let usedUserFallback = false; - if (matchPos < 0 && lastLocal.role === 'assistant') { - let lastLocalUserIdx = -1; - for (let j = localEntries.length - 1; j >= 0; j--) { - if (localEntries[j].role === 'user') { - lastLocalUserIdx = j; - break; - } - } - if (lastLocalUserIdx >= 0) { - const lastLocalUser = localEntries[lastLocalUserIdx]; - // Find preceding user message in local for double-verification - let prevLocalUserText: string | undefined; - for (let j = lastLocalUserIdx - 1; j >= 0; j--) { - if (localEntries[j].role === 'user') { - prevLocalUserText = localEntries[j].text; - break; - } - } - console.log('[ChannelSync] syncFullChannelHistory: assistant match failed, retrying with last user entry'); - for (let i = historyEntries.length - 1; i >= 0; i--) { - if (historyEntries[i].role === 'user' - && historyEntries[i].text === lastLocalUser.text) { - // Double-check: verify the preceding user message also matches - if (prevLocalUserText !== undefined && i > 0) { - let prevHistUserText: string | undefined; - for (let k = i - 1; k >= 0; k--) { - if (historyEntries[k].role === 'user') { - prevHistUserText = historyEntries[k].text; - break; - } - } - if (prevHistUserText !== prevLocalUserText) { - continue; // Double-check failed, keep searching - } - } - matchPos = i; - usedUserFallback = true; - break; - } - } - } - } - - if (matchPos >= 0) { - firstNewIdx = matchPos + 1; - // When user-message fallback was used, the assistant replies immediately following - // the matched user message are already in local (as segments). Skip them to prevent - // duplicate assistant messages. - if (usedUserFallback) { - while (firstNewIdx < historyEntries.length && historyEntries[firstNewIdx].role === 'assistant') { - firstNewIdx++; - } - } - } else { - // Reverse match failed — fall back to forward sequential matching. - // This covers the case where local is a strict prefix of history. - let localIdx = 0; - for (let i = 0; i < historyEntries.length; i++) { - if (localIdx < localEntries.length - && historyEntries[i].role === localEntries[localIdx].role - && historyEntries[i].text === localEntries[localIdx].text) { - localIdx++; - firstNewIdx = i + 1; - } - } - } - } - - let syncedCount = 0; - for (let i = firstNewIdx; i < historyEntries.length; i++) { - const entry = historyEntries[i]; - if (entry.role === 'user') { - const userMsg = this.store.addMessage(sessionId, { - type: 'user', - content: entry.text, - metadata: {}, - }); - this.emit('message', sessionId, userMsg); - } else { - const assistantMsg = this.store.addMessage(sessionId, { - type: 'assistant', - content: entry.text, - metadata: { isStreaming: false, isFinal: true }, - }); - this.emit('message', sessionId, assistantMsg); - } - syncedCount++; - } - - console.log('[ChannelSync] syncFullChannelHistory: synced', syncedCount, 'messages for sessionId:', sessionId); - - // Initialize the sync cursor so incremental syncs know where to start - this.channelSyncCursor.set(sessionId, historyEntries.length); - - // Notify renderer to refresh - if (syncedCount > 0) { - for (const win of BrowserWindow.getAllWindows()) { - if (!win.isDestroyed()) { - win.webContents.send('cowork:sessions:changed'); - } - } - } + await this.reconcileWithHistory(sessionId, sessionKey, { isFullSync: true }); } catch (error) { console.error('[ChannelSync] syncFullChannelHistory: error:', error); // Remove from synced set so retry is possible @@ -3219,31 +3189,10 @@ export class OpenClawRuntimeAdapter extends EventEmitter implements CoworkRuntim /** * Incremental sync for an already-known channel session. - * Fetches recent history and appends any messages not yet in the local store. - * Lightweight: uses position-based matching so only truly new messages are added. + * Delegates to reconcileWithHistory which handles diff and update. */ private async incrementalChannelSync(sessionId: string, sessionKey: string): Promise { - const client = this.gatewayClient; - if (!client) return; - - const history = await client.request<{ messages?: unknown[] }>('chat.history', { - sessionKey, - limit: FINAL_HISTORY_SYNC_LIMIT, - }); - if (!Array.isArray(history?.messages) || history.messages.length === 0) return; - - const beforeCount = this.store.getSession(sessionId)?.messages.length ?? 0; - this.syncChannelUserMessages(sessionId, history.messages, false, sessionKey.includes(':discord:'), sessionKey.includes(':qqbot:')); - const afterCount = this.store.getSession(sessionId)?.messages.length ?? 0; - - if (afterCount > beforeCount) { - console.log('[ChannelSync] incrementalSync: added', afterCount - beforeCount, 'messages for', sessionKey); - for (const win of BrowserWindow.getAllWindows()) { - if (!win.isDestroyed()) { - win.webContents.send('cowork:sessions:changed'); - } - } - } + await this.reconcileWithHistory(sessionId, sessionKey); } /** @@ -3255,7 +3204,7 @@ export class OpenClawRuntimeAdapter extends EventEmitter implements CoworkRuntim if (!this.channelSessionSync.isChannelSessionKey(sessionKey)) return; if (!this.fullySyncedSessions.has(sessionId)) return; - void this.incrementalChannelSync(sessionId, sessionKey).catch((err) => { + void this.reconcileWithHistory(sessionId, sessionKey).catch((err) => { console.warn('[ChannelSync] post-turn incremental sync failed for', sessionKey, err); }); } @@ -3422,8 +3371,9 @@ export class OpenClawRuntimeAdapter extends EventEmitter implements CoworkRuntim private async prefetchChannelUserMessages(sessionId: string, sessionKey: string): Promise { console.log('[Debug:prefetch] start — sessionId:', sessionId, 'sessionKey:', sessionKey); - const MAX_ATTEMPTS = 5; - const BACKOFF_DELAYS = [500, 1000, 1500, 2000]; // exponential-ish backoff + // Best-effort prefetch with 2 attempts. Final correctness is ensured by + // reconcileWithHistory after the turn completes. + const MAX_ATTEMPTS = 2; for (let attempt = 0; attempt < MAX_ATTEMPTS; attempt++) { try { const client = this.gatewayClient; @@ -3449,28 +3399,26 @@ export class OpenClawRuntimeAdapter extends EventEmitter implements CoworkRuntim console.log('[Debug:prefetch] synced user messages:', newUserMessages, '(before:', beforeCount, 'after:', afterCount, ')'); if (newUserMessages > 0) { - break; // Successfully synced new user messages + break; } - // No new user messages but buffered events indicate agent is processing → history may lag - const turn = this.activeTurns.get(sessionId); - if (turn && (turn.bufferedChatPayloads.length > 0 || turn.bufferedAgentPayloads.length > 0)) { - if (attempt < MAX_ATTEMPTS - 1) { - const delay = BACKOFF_DELAYS[Math.min(attempt, BACKOFF_DELAYS.length - 1)]; - console.log('[Debug:prefetch] no new user messages but have buffered events, retrying after', delay, 'ms...'); - await new Promise((resolve) => setTimeout(resolve, delay)); + // Retry once if buffered events suggest history hasn't caught up yet + if (attempt < MAX_ATTEMPTS - 1) { + const turn = this.activeTurns.get(sessionId); + if (turn && (turn.bufferedChatPayloads.length > 0 || turn.bufferedAgentPayloads.length > 0)) { + console.log('[Debug:prefetch] no new user messages but have buffered events, retrying after 500ms...'); + await new Promise((resolve) => setTimeout(resolve, 500)); continue; } } - break; // No buffered events or max attempts reached + break; } else { - // Empty history — session may have just been created - const turn = this.activeTurns.get(sessionId); - if (turn && (turn.bufferedChatPayloads.length > 0 || turn.bufferedAgentPayloads.length > 0)) { - if (attempt < MAX_ATTEMPTS - 1) { - const delay = BACKOFF_DELAYS[Math.min(attempt, BACKOFF_DELAYS.length - 1)]; - console.log('[Debug:prefetch] empty history but have buffered events, retrying after', delay, 'ms...'); - await new Promise((resolve) => setTimeout(resolve, delay)); + // Retry once if buffered events suggest history hasn't caught up yet + if (attempt < MAX_ATTEMPTS - 1) { + const turn = this.activeTurns.get(sessionId); + if (turn && (turn.bufferedChatPayloads.length > 0 || turn.bufferedAgentPayloads.length > 0)) { + console.log('[Debug:prefetch] empty history but have buffered events, retrying after 500ms...'); + await new Promise((resolve) => setTimeout(resolve, 500)); continue; } } @@ -3479,8 +3427,7 @@ export class OpenClawRuntimeAdapter extends EventEmitter implements CoworkRuntim } catch (error) { console.warn('[OpenClawRuntime] prefetchChannelUserMessages attempt', attempt, 'failed:', error); if (attempt < MAX_ATTEMPTS - 1) { - const delay = BACKOFF_DELAYS[Math.min(attempt, BACKOFF_DELAYS.length - 1)]; - await new Promise((resolve) => setTimeout(resolve, delay)); + await new Promise((resolve) => setTimeout(resolve, 500)); } } } diff --git a/src/main/libs/coworkOpenAICompatProxy.ts b/src/main/libs/coworkOpenAICompatProxy.ts index 5167cb1..2546db2 100644 --- a/src/main/libs/coworkOpenAICompatProxy.ts +++ b/src/main/libs/coworkOpenAICompatProxy.ts @@ -75,6 +75,7 @@ let proxyServer: http.Server | null = null; let proxyPort: number | null = null; let upstreamConfig: OpenAICompatUpstreamConfig | null = null; let lastProxyError: string | null = null; +let tokenRefresher: (() => Promise) | null = null; let currentCoworkSessionId: string | null = null; const toolCallExtraContentById = new Map(); @@ -2323,6 +2324,26 @@ async function handleRequest( } if (!upstreamResponse.ok) { + // 401/403 from lobsterai-server likely means the JWT accessToken expired. + // Refresh the token and retry once before falling through to other error handling. + if ((upstreamResponse.status === 401 || upstreamResponse.status === 403) && tokenRefresher) { + console.log(`[CoworkProxy] Got ${upstreamResponse.status}, attempting token refresh and retry...`); + try { + const newToken = await tokenRefresher(); + if (newToken) { + headers.Authorization = `Bearer ${newToken}`; + if (upstreamConfig) { + upstreamConfig.apiKey = newToken; + } + upstreamResponse = await sendUpstreamRequest(upstreamRequest, currentTargetURL); + const retryDuration = Date.now() - fetchStartTime; + console.log(`[CoworkProxy] Token refresh retry: status=${upstreamResponse.status}, ok=${upstreamResponse.ok}, fetchTime=${retryDuration}ms`); + } + } catch (refreshError) { + console.warn('[CoworkProxy] Token refresh retry failed:', refreshError); + } + } + if (upstreamResponse.status === 404 && targetURLs.length > 1) { for (let i = 1; i < targetURLs.length; i += 1) { const retryURL = targetURLs[i]; @@ -2543,6 +2564,14 @@ export function configureCoworkOpenAICompatProxy(config: OpenAICompatUpstreamCon lastProxyError = null; } +/** + * Set a callback that refreshes the auth token and returns the new accessToken. + * Called by the proxy on 401/403 to retry with a fresh token. + */ +export function setProxyTokenRefresher(refresher: () => Promise): void { + tokenRefresher = refresher; +} + export function getCoworkOpenAICompatProxyBaseURL(target: OpenAICompatProxyTarget = 'local'): string | null { if (!proxyServer || !proxyPort) { return null; diff --git a/src/main/libs/mcpServerManager.ts b/src/main/libs/mcpServerManager.ts index 400f37f..fc7eb41 100644 --- a/src/main/libs/mcpServerManager.ts +++ b/src/main/libs/mcpServerManager.ts @@ -83,8 +83,67 @@ interface ResolvedStdioCommand { } /** - * Resolve a stdio MCP server command/args/env for the current platform, - * rewriting node/npx/npm to Electron runtime when packaged. + * Check whether a system-installed Node.js runtime is available on the PATH. + * Caches the result for the lifetime of the process to avoid repeated lookups. + */ +let _systemNodePath: string | false | undefined; + +function findSystemNodePath(): string | null { + if (_systemNodePath !== undefined) { + return _systemNodePath || null; + } + try { + const whichCmd = process.platform === 'win32' ? 'where' : 'which'; + const result = spawnSync(whichCmd, ['node'], { + encoding: 'utf-8', + timeout: 5000, + windowsHide: true, + }); + if (result.status === 0 && result.stdout) { + const resolved = result.stdout.trim().split(/\r?\n/)[0].trim(); + if (resolved) { + _systemNodePath = resolved; + log('INFO', `System Node.js found: ${resolved}`); + return resolved; + } + } + } catch { /* ignore */ } + _systemNodePath = false; + log('INFO', 'System Node.js not found on PATH'); + return null; +} + +/** + * Check if a command is a node/npx/npm variant. + */ +function isNodeCommand(normalized: string): 'node' | 'npx' | 'npm' | null { + if ( + normalized === 'node' || normalized === 'node.exe' + || normalized.endsWith('\\node.cmd') || normalized.endsWith('/node.cmd') + ) { + return 'node'; + } + if ( + normalized === 'npx' || normalized === 'npx.cmd' + || normalized.endsWith('\\npx.cmd') || normalized.endsWith('/npx.cmd') + ) { + return 'npx'; + } + if ( + normalized === 'npm' || normalized === 'npm.cmd' + || normalized.endsWith('\\npm.cmd') || normalized.endsWith('/npm.cmd') + ) { + return 'npm'; + } + return null; +} + +/** + * Resolve a stdio MCP server command/args/env for the current platform. + * + * On packaged builds, node/npx/npm commands are resolved in this order: + * 1. Use system-installed Node.js if available (avoids Electron stdin quirks) + * 2. Fall back to Electron runtime with ELECTRON_RUN_AS_NODE=1 */ async function resolveStdioCommand(server: McpServerRecord): Promise { const stdioCommand = server.command || ''; @@ -100,45 +159,60 @@ async function resolveStdioCommand(server: McpServerRecord): Promise | undefined): Record => ({ - ...(base || {}), - ELECTRON_RUN_AS_NODE: '1', - LOBSTERAI_ELECTRON_PATH: electronNodeRuntimePath, - }); + if (nodeCommandType) { + const systemNode = findSystemNodePath(); + if (systemNode) { + if (nodeCommandType === 'node') { + effectiveCommand = systemNode; + log('INFO', `"${server.name}": using system Node.js "${systemNode}" (preferred over Electron runtime)`); + } else { + const enhancedEnv = await getEnhancedEnv(); + const npmBinDir = enhancedEnv.LOBSTERAI_NPM_BIN_DIR; + const cliJs = nodeCommandType === 'npx' + ? (npmBinDir ? path.join(npmBinDir, 'npx-cli.js') : '') + : (npmBinDir ? path.join(npmBinDir, 'npm-cli.js') : ''); + if (cliJs && fs.existsSync(cliJs)) { + effectiveCommand = systemNode; + effectiveArgs = [cliJs, ...stdioArgs]; + log('INFO', `"${server.name}": using system Node.js "${systemNode}" + ${nodeCommandType}-cli.js (preferred over Electron runtime)`); + } else { + effectiveCommand = stdioCommand; + log('INFO', `"${server.name}": using system "${stdioCommand}" directly`); + } + } + } else { + const enhancedEnv = await getEnhancedEnv(); + const npmBinDir = enhancedEnv.LOBSTERAI_NPM_BIN_DIR; + const npxCliJs = npmBinDir ? path.join(npmBinDir, 'npx-cli.js') : ''; + const npmCliJs = npmBinDir ? path.join(npmBinDir, 'npm-cli.js') : ''; - if ( - normalized === 'node' || normalized === 'node.exe' - || normalized.endsWith('\\node.cmd') || normalized.endsWith('/node.cmd') - ) { - effectiveCommand = electronNodeRuntimePath; - stdioEnv = withElectronNodeEnv(stdioEnv); - shouldInjectWindowsHide = true; - log('INFO', `"${server.name}": rewrote command "${stdioCommand}" → Electron runtime`); - } else if ( - (normalized === 'npx' || normalized === 'npx.cmd' - || normalized.endsWith('\\npx.cmd') || normalized.endsWith('/npx.cmd')) - && npxCliJs && fs.existsSync(npxCliJs) - ) { - effectiveCommand = electronNodeRuntimePath; - effectiveArgs = [npxCliJs, ...stdioArgs]; - stdioEnv = withElectronNodeEnv(stdioEnv); - shouldInjectWindowsHide = true; - log('INFO', `"${server.name}": rewrote command "${stdioCommand}" → Electron + npx-cli.js`); - } else if ( - (normalized === 'npm' || normalized === 'npm.cmd' - || normalized.endsWith('\\npm.cmd') || normalized.endsWith('/npm.cmd')) - && npmCliJs && fs.existsSync(npmCliJs) - ) { - effectiveCommand = electronNodeRuntimePath; - effectiveArgs = [npmCliJs, ...stdioArgs]; - stdioEnv = withElectronNodeEnv(stdioEnv); - shouldInjectWindowsHide = true; - log('INFO', `"${server.name}": rewrote command "${stdioCommand}" → Electron + npm-cli.js`); + const withElectronNodeEnv = (base: Record | undefined): Record => ({ + ...(base || {}), + ELECTRON_RUN_AS_NODE: '1', + LOBSTERAI_ELECTRON_PATH: electronNodeRuntimePath, + }); + + if (nodeCommandType === 'node') { + effectiveCommand = electronNodeRuntimePath; + stdioEnv = withElectronNodeEnv(stdioEnv); + shouldInjectWindowsHide = true; + log('WARN', `"${server.name}": no system Node.js found, falling back to Electron runtime (may cause stdin issues)`); + } else if (nodeCommandType === 'npx' && npxCliJs && fs.existsSync(npxCliJs)) { + effectiveCommand = electronNodeRuntimePath; + effectiveArgs = [npxCliJs, ...stdioArgs]; + stdioEnv = withElectronNodeEnv(stdioEnv); + shouldInjectWindowsHide = true; + log('WARN', `"${server.name}": no system Node.js found, falling back to Electron + npx-cli.js (may cause stdin issues)`); + } else if (nodeCommandType === 'npm' && npmCliJs && fs.existsSync(npmCliJs)) { + effectiveCommand = electronNodeRuntimePath; + effectiveArgs = [npmCliJs, ...stdioArgs]; + stdioEnv = withElectronNodeEnv(stdioEnv); + shouldInjectWindowsHide = true; + log('WARN', `"${server.name}": no system Node.js found, falling back to Electron + npm-cli.js (may cause stdin issues)`); + } + } } } @@ -243,6 +317,17 @@ export class McpServerManager { env: spawnEnv, }); + const stderrChunks: string[] = []; + if (transport.stderr) { + transport.stderr.on('data', (chunk: Buffer) => { + const text = chunk.toString().trim(); + if (text) { + stderrChunks.push(text); + log('WARN', `"${record.name}" stderr: ${text}`); + } + }); + } + const client = new Client( { name: `lobsterai-mcp-bridge`, version: '1.0.0' }, { capabilities: {} }, @@ -252,7 +337,11 @@ export class McpServerManager { await client.connect(transport); log('INFO', `Connected to MCP server "${record.name}"`); } catch (error) { - log('ERROR', `Failed to connect to "${record.name}": ${error instanceof Error ? error.message : String(error)}`); + const errMsg = error instanceof Error ? error.message : String(error); + const stderrSummary = stderrChunks.length > 0 + ? ` | stderr: ${stderrChunks.join(' ').slice(0, 500)}` + : ''; + log('ERROR', `Failed to connect to "${record.name}": ${errMsg}${stderrSummary}`); try { await transport.close(); } catch { /* ignore */ } return null; } diff --git a/src/main/libs/openclawEngineManager.ts b/src/main/libs/openclawEngineManager.ts index 825ce24..d4265ff 100644 --- a/src/main/libs/openclawEngineManager.ts +++ b/src/main/libs/openclawEngineManager.ts @@ -7,7 +7,6 @@ import net from 'net'; import path from 'path'; import { getElectronNodeRuntimePath, ensureElectronNodeShim } from './coworkUtil'; import { syncLocalOpenClawExtensionsIntoRuntime } from './openclawLocalExtensions'; -import { applyBundledOpenClawRuntimeHotfixes } from './openclawRuntimeHotfix'; import { appendPythonRuntimeToEnv } from './pythonRuntime'; import { isSystemProxyEnabled, resolveSystemProxyUrl } from './systemProxy'; @@ -358,19 +357,6 @@ export class OpenClawEngineManager extends EventEmitter { this.ensureBareEntryFiles(runtime.root); console.log(`[OpenClaw] startGateway: ensureBareEntryFiles done (${elapsed()})`); - // Skip hotfixes when gateway-bundle.mjs exists. The hotfixes patch individual - // JS files in dist/, but the bundle is a single esbuild artifact that doesn't - // use those files. Applying hotfixes with bundle present is wasteful (Electron's - // transparent asar read causes walkJsFiles to scan ~1100 files inside gateway.asar) - // and can take 250+ seconds on Windows due to Defender scanning. - const bundlePath = path.join(runtime.root, 'gateway-bundle.mjs'); - if (fs.existsSync(bundlePath)) { - console.log(`[OpenClaw] startGateway: skipping applyRuntimeHotfixes (bundle exists) (${elapsed()})`); - } else { - this.applyRuntimeHotfixes(runtime.root); - console.log(`[OpenClaw] startGateway: applyRuntimeHotfixes done (${elapsed()})`); - } - const openclawEntry = this.resolveOpenClawEntry(runtime.root); console.log(`[OpenClaw] startGateway: resolveOpenClawEntry done (${elapsed()}), entry=${openclawEntry}`); if (!openclawEntry) { @@ -426,6 +412,18 @@ export class OpenClawEngineManager extends EventEmitter { // This keeps plaintext credentials out of the config file on disk. ...this.secretEnvVars, }; + + // Ensure the gateway process uses the host's local timezone for logging. + // macOS does not set TZ in the environment by default (it uses NSTimeZone/ICU), + // so utilityProcess.fork() children may fall back to UTC for date formatting. + if (!env.TZ) { + const hostTimezone = Intl.DateTimeFormat().resolvedOptions().timeZone; + if (hostTimezone) { + env.TZ = hostTimezone; + console.log(`[OpenClaw] injected TZ=${hostTimezone} into gateway env`); + } + } + if (cliShimDir) { // Plain object is case-sensitive: the spread key from process.env on Windows is "Path", // not "PATH". We must read the actual key to avoid creating a PATH with only cliShimDir. @@ -680,20 +678,6 @@ export class OpenClawEngineManager extends EventEmitter { } } - private applyRuntimeHotfixes(runtimeRoot: string): void { - const result = applyBundledOpenClawRuntimeHotfixes(runtimeRoot); - if (result.changed) { - console.log( - `[OpenClaw] Applied runtime hotfixes: ${result.patchedFiles - .map((filePath) => path.relative(runtimeRoot, filePath)) - .join(', ')}`, - ); - } - if (result.errors.length > 0) { - console.warn('[OpenClaw] Runtime hotfix warnings:', result.errors.join(' | ')); - } - } - private ensureBundledCliShims(): string | null { const shimDir = path.join(this.stateDir, 'bin'); const shellWrapper = [ @@ -1222,6 +1206,24 @@ export class OpenClawEngineManager extends EventEmitter { }, 1200); } + // Workaround: Electron utilityProcess V8 isolate reports getTimezoneOffset()=0. + private static rewriteUtcTimestamps(text: string): string { + return text.replace( + /\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z/g, + (utc) => { + const d = new Date(utc); + if (Number.isNaN(d.getTime())) return utc; + const pad = (n: number) => String(n).padStart(2, '0'); + const ms = String(d.getMilliseconds()).padStart(3, '0'); + const offsetMin = -d.getTimezoneOffset(); + const sign = offsetMin >= 0 ? '+' : '-'; + const absH = Math.floor(Math.abs(offsetMin) / 60); + const absM = Math.abs(offsetMin) % 60; + return `${d.getFullYear()}-${pad(d.getMonth() + 1)}-${pad(d.getDate())}T${pad(d.getHours())}:${pad(d.getMinutes())}:${pad(d.getSeconds())}.${ms}${sign}${pad(absH)}:${pad(absM)}`; + }, + ); + } + private attachGatewayProcessLogs(child: GatewayProcess): void { ensureDir(path.dirname(this.gatewayLogPath)); const appendLog = (chunk: Buffer | string, stream: 'stdout' | 'stderr') => { @@ -1234,11 +1236,13 @@ export class OpenClawEngineManager extends EventEmitter { child.stdout?.on('data', (chunk) => { appendLog(chunk, 'stdout'); - console.log(`[OpenClaw stdout] ${typeof chunk === 'string' ? chunk : chunk.toString()}`); + const text = typeof chunk === 'string' ? chunk : chunk.toString(); + console.log(`[OpenClaw stdout] ${OpenClawEngineManager.rewriteUtcTimestamps(text)}`); }); child.stderr?.on('data', (chunk) => { appendLog(chunk, 'stderr'); - console.error(`[OpenClaw stderr] ${typeof chunk === 'string' ? chunk : chunk.toString()}`); + const text = typeof chunk === 'string' ? chunk : chunk.toString(); + console.error(`[OpenClaw stderr] ${OpenClawEngineManager.rewriteUtcTimestamps(text)}`); }); } diff --git a/src/main/libs/openclawRuntimeHotfix.ts b/src/main/libs/openclawRuntimeHotfix.ts deleted file mode 100644 index 70008be..0000000 --- a/src/main/libs/openclawRuntimeHotfix.ts +++ /dev/null @@ -1,288 +0,0 @@ -import fs from 'fs'; -import path from 'path'; - -const CRON_OWNER_ONLY_PATTERN = - /(function createCronTool\([^)]*\)\s*{[\s\S]*?\bname:\s*"cron",\s*\n\s*ownerOnly:\s*)true,/; -const OWNER_ONLY_FALLBACK_PATTERN = - /(const OWNER_ONLY_TOOL_NAME_FALLBACKS = new Set\(\[\s*\n\s*"whatsapp_login",\s*\n\s*)"cron",\s*\n(\s*"gateway"\s*\n\s*\]\);)/; -const WECOM_EXEC_DENY_PATTERN = - /const TOOL_DENY_BY_MESSAGE_PROVIDER = \{ voice: \["tts"\](?:,\s*wecom: \["exec", "process"\])? \};/; -const CRON_DELIVERY_INFERENCE_OLD = `function inferDeliveryFromSessionKey(agentSessionKey) { -\tconst rawSessionKey = agentSessionKey?.trim(); -\tif (!rawSessionKey) return null; -\tconst parsed = parseAgentSessionKey(stripThreadSuffixFromSessionKey(rawSessionKey)); -\tif (!parsed || !parsed.rest) return null; -\tconst parts = parsed.rest.split(":").filter(Boolean); -\tif (parts.length === 0) return null; -\tconst head = parts[0]?.trim().toLowerCase(); -\tif (!head || head === "main" || head === "subagent" || head === "acp") return null; -\tconst markerIndex = parts.findIndex((part) => part === "direct" || part === "dm" || part === "group" || part === "channel"); -\tif (markerIndex === -1) return null; -\tconst peerId = parts.slice(markerIndex + 1).join(":").trim(); -\tif (!peerId) return null; -\tlet channel; -\tif (markerIndex >= 1) channel = parts[0]?.trim().toLowerCase(); -\tconst delivery = { -\t\tmode: "announce", -\t\tto: peerId -\t}; -\tif (channel) delivery.channel = channel; -\treturn delivery; -}`; -const CRON_DELIVERY_INFERENCE_NEW = `function inferDeliveryFromSessionKey(agentSessionKey) { -\tconst rawSessionKey = agentSessionKey?.trim(); -\tif (!rawSessionKey) return null; -\tconst { deliveryContext, threadId } = extractDeliveryInfo(rawSessionKey); -\tconst persistedDelivery = normalizeDeliveryContext(deliveryContext); -\tif (persistedDelivery?.channel && persistedDelivery?.to) { -\t\tconst delivery = { -\t\t\tmode: "announce", -\t\t\tchannel: persistedDelivery.channel, -\t\t\tto: persistedDelivery.to -\t\t}; -\t\tif (persistedDelivery.accountId) delivery.accountId = persistedDelivery.accountId; -\t\tif (persistedDelivery.threadId != null) delivery.threadId = persistedDelivery.threadId; -\t\telse if (threadId != null) delivery.threadId = threadId; -\t\treturn delivery; -\t} -\tconst jsonIdx = rawSessionKey.indexOf(":{"); -\tif (jsonIdx > 0) { -\t\ttry { -\t\t\tconst ctx = JSON.parse(rawSessionKey.slice(jsonIdx + 1)); -\t\t\tif (ctx && typeof ctx.channel === "string" && ctx.channel.trim()) { -\t\t\t\tconst channel = ctx.channel.trim().toLowerCase(); -\t\t\t\tconst peerId = (ctx.peerid || ctx.peerId || ctx.conversationId || "").trim(); -\t\t\t\tconst chatType = (ctx.chattype || ctx.chatType || "direct").trim().toLowerCase(); -\t\t\t\tconst accountId = (ctx.accountid || ctx.accountId || "").trim(); -\t\t\t\tif (peerId) { -\t\t\t\t\tconst to = chatType === "group" ? \`group:\${peerId}\` : \`user:\${peerId}\`; -\t\t\t\t\tconst delivery = { mode: "announce", channel, to }; -\t\t\t\t\tif (accountId && accountId !== "__default__") delivery.accountId = accountId; -\t\t\t\t\tif (threadId != null) delivery.threadId = threadId; -\t\t\t\t\treturn delivery; -\t\t\t\t} -\t\t\t} -\t\t} catch {} -\t} -\tconst parsed = parseAgentSessionKey(stripThreadSuffixFromSessionKey(rawSessionKey)); -\tif (!parsed || !parsed.rest) return null; -\tconst parts = parsed.rest.split(":").filter(Boolean); -\tif (parts.length === 0) return null; -\tconst head = parts[0]?.trim().toLowerCase(); -\tif (!head || head === "main" || head === "subagent" || head === "acp") return null; -\tif (head === "dingtalk-connector") { -\t\tconst accountId = parts[1]?.trim(); -\t\tconst senderId = parts[2]?.trim(); -\t\tif (accountId && senderId) { -\t\t\tconst delivery = { -\t\t\t\tmode: "announce", -\t\t\t\tchannel: "dingtalk-connector", -\t\t\t\tto: \`user:\${senderId}\`, -\t\t\t\taccountId -\t\t\t}; -\t\t\tif (threadId != null) delivery.threadId = threadId; -\t\t\treturn delivery; -\t\t} -\t} -\tconst markerIndex = parts.findIndex((part) => part === "direct" || part === "dm" || part === "group" || part === "channel"); -\tif (markerIndex === -1) return null; -\tconst peerId = parts.slice(markerIndex + 1).join(":").trim(); -\tif (!peerId) return null; -\tlet channel; -\tif (markerIndex >= 1) channel = parts[0]?.trim().toLowerCase(); -\tconst delivery = { -\t\tmode: "announce", -\t\tto: peerId -\t}; -\tif (channel) delivery.channel = channel; -\tif (threadId != null) delivery.threadId = threadId; -\treturn delivery; -}`; -const CRON_REMINDER_PROMPT_OLD = [ - 'function buildCronEventPrompt(pendingEvents, opts) {', - '\tconst deliverToUser = opts?.deliverToUser ?? true;', - '\tconst eventText = pendingEvents.join("\\n").trim();', - '\tif (!eventText) {', - '\t\tif (!deliverToUser) return "A scheduled cron event was triggered, but no event content was found. Handle this internally and reply HEARTBEAT_OK when nothing needs user-facing follow-up.";', - '\t\treturn "A scheduled cron event was triggered, but no event content was found. Reply HEARTBEAT_OK.";', - '\t}', - '\tif (!deliverToUser) return "A scheduled reminder has been triggered. The reminder content is:\\n\\n" + eventText + "\\n\\nHandle this reminder internally. Do not relay it to the user unless explicitly requested.";', - '\treturn "A scheduled reminder has been triggered. The reminder content is:\\n\\n" + eventText + "\\n\\nPlease relay this reminder to the user in a helpful and friendly way.";', - '}', -].join('\n'); -const CRON_REMINDER_PROMPT_NEW = [ - 'function buildCronEventPrompt(pendingEvents, opts) {', - '\tconst deliverToUser = opts?.deliverToUser ?? true;', - '\tconst eventText = pendingEvents.join("\\n").trim();', - '\tif (!eventText) {', - '\t\tif (!deliverToUser) return "A scheduled cron event was triggered, but no event content was found. Handle this internally and reply HEARTBEAT_OK when nothing needs user-facing follow-up.";', - '\t\treturn "A scheduled cron event was triggered, but no event content was found. Reply HEARTBEAT_OK.";', - '\t}', - '\treturn eventText;', - '}', -].join('\n'); -const CRON_REMINDER_CURRENT_TIME_BODY_OLD = 'Body: appendCronStyleCurrentTimeLine(prompt, cfg, startedAt),'; -const CRON_REMINDER_CURRENT_TIME_BODY_NEW = 'Body: hasCronEvents ? prompt : appendCronStyleCurrentTimeLine(prompt, cfg, startedAt),'; - -const walkJsFiles = (dirPath: string, files: string[]): void => { - let entries: fs.Dirent[]; - try { - entries = fs.readdirSync(dirPath, { withFileTypes: true }); - } catch { - return; - } - - for (const entry of entries) { - const fullPath = path.join(dirPath, entry.name); - if (entry.isDirectory()) { - walkJsFiles(fullPath, files); - continue; - } - if (entry.isFile() && fullPath.endsWith('.js')) { - files.push(fullPath); - } - } -}; - -export const patchCronToolOwnerOnly = (source: string): { changed: boolean; content: string } => { - if (!source.includes('function createCronTool(') || !source.includes('name: "cron"')) { - return { changed: false, content: source }; - } - const next = source.replace(CRON_OWNER_ONLY_PATTERN, '$1false,'); - return { - changed: next !== source, - content: next, - }; -}; - -export const patchCronOwnerFallback = (source: string): { changed: boolean; content: string } => { - if (!source.includes('OWNER_ONLY_TOOL_NAME_FALLBACKS') || !source.includes('"cron"')) { - return { changed: false, content: source }; - } - const next = source.replace(OWNER_ONLY_FALLBACK_PATTERN, '$1$2'); - return { - changed: next !== source, - content: next, - }; -}; - -export const patchWecomMessageProviderExecDeny = ( - source: string, -): { changed: boolean; content: string } => { - if (!source.includes('TOOL_DENY_BY_MESSAGE_PROVIDER') || !source.includes('voice: ["tts"]')) { - return { changed: false, content: source }; - } - if (source.includes('wecom: ["exec", "process"]')) { - return { changed: false, content: source }; - } - const next = source.replace( - WECOM_EXEC_DENY_PATTERN, - 'const TOOL_DENY_BY_MESSAGE_PROVIDER = { voice: ["tts"], wecom: ["exec", "process"] };', - ); - return { - changed: next !== source, - content: next, - }; -}; - -export const patchCronSessionDeliveryInference = ( - source: string, -): { changed: boolean; content: string } => { - if ( - !source.includes('function inferDeliveryFromSessionKey(agentSessionKey) {') || - !source.includes('const parsed = parseAgentSessionKey(stripThreadSuffixFromSessionKey(rawSessionKey));') - ) { - return { changed: false, content: source }; - } - const next = source.replace(CRON_DELIVERY_INFERENCE_OLD, CRON_DELIVERY_INFERENCE_NEW); - return { - changed: next !== source, - content: next, - }; -}; - -export const patchCronReminderPromptEnvelope = ( - source: string, -): { changed: boolean; content: string } => { - if ( - !source.includes('function buildCronEventPrompt(') || - !source.includes('A scheduled reminder has been triggered. The reminder content is:') - ) { - return { changed: false, content: source }; - } - const next = source.replace(CRON_REMINDER_PROMPT_OLD, CRON_REMINDER_PROMPT_NEW); - return { - changed: next !== source, - content: next, - }; -}; - -export const patchCronReminderCurrentTimeSuffix = ( - source: string, -): { changed: boolean; content: string } => { - if ( - !source.includes(CRON_REMINDER_CURRENT_TIME_BODY_OLD) || - !source.includes('hasCronEvents') - ) { - return { changed: false, content: source }; - } - const next = source.replace( - CRON_REMINDER_CURRENT_TIME_BODY_OLD, - CRON_REMINDER_CURRENT_TIME_BODY_NEW, - ); - return { - changed: next !== source, - content: next, - }; -}; - -export const applyBundledOpenClawRuntimeHotfixes = ( - runtimeRoot: string, -): { changed: boolean; patchedFiles: string[]; errors: string[] } => { - const distRoot = path.join(runtimeRoot, 'dist'); - const jsFiles: string[] = []; - walkJsFiles(distRoot, jsFiles); - - const patchedFiles: string[] = []; - const errors: string[] = []; - - for (const filePath of jsFiles) { - let source = ''; - try { - source = fs.readFileSync(filePath, 'utf8'); - } catch (error) { - errors.push(`${filePath}: ${error instanceof Error ? error.message : String(error)}`); - continue; - } - - const cronOwnerPatch = patchCronToolOwnerOnly(source); - const fallbackPatch = patchCronOwnerFallback(cronOwnerPatch.content); - const wecomExecPatch = patchWecomMessageProviderExecDeny(fallbackPatch.content); - const cronDeliveryPatch = patchCronSessionDeliveryInference(wecomExecPatch.content); - const cronReminderPromptPatch = patchCronReminderPromptEnvelope(cronDeliveryPatch.content); - const cronReminderTimePatch = patchCronReminderCurrentTimeSuffix(cronReminderPromptPatch.content); - if ( - !cronOwnerPatch.changed && - !fallbackPatch.changed && - !wecomExecPatch.changed && - !cronDeliveryPatch.changed && - !cronReminderPromptPatch.changed && - !cronReminderTimePatch.changed - ) { - continue; - } - - try { - fs.writeFileSync(filePath, cronReminderTimePatch.content, 'utf8'); - patchedFiles.push(filePath); - } catch (error) { - errors.push(`${filePath}: ${error instanceof Error ? error.message : String(error)}`); - } - } - - return { - changed: patchedFiles.length > 0, - patchedFiles, - errors, - }; -}; diff --git a/src/main/main.ts b/src/main/main.ts index cbb4379..bd1bcb8 100644 --- a/src/main/main.ts +++ b/src/main/main.ts @@ -17,7 +17,7 @@ import type { PermissionResult } from '@anthropic-ai/claude-agent-sdk'; import { getCurrentApiConfig, resolveCurrentApiConfig, setStoreGetter, setAuthTokensGetter, setServerBaseUrlGetter } from './libs/claudeSettings'; import { saveCoworkApiConfig } from './libs/coworkConfigStore'; import { generateSessionTitle, probeCoworkModelReadiness } from './libs/coworkUtil'; -import { startCoworkOpenAICompatProxy, stopCoworkOpenAICompatProxy } from './libs/coworkOpenAICompatProxy'; +import { startCoworkOpenAICompatProxy, stopCoworkOpenAICompatProxy, setProxyTokenRefresher } from './libs/coworkOpenAICompatProxy'; import { OpenClawEngineManager, type OpenClawEngineStatus } from './libs/openclawEngineManager'; import { listPairingRequests, @@ -4285,6 +4285,32 @@ if (!gotTheLock) { }); setServerBaseUrlGetter(() => getServerApiBaseUrl()); + // Wire up token refresher for the OpenAI compat proxy so it can retry + // on 401/403 with a fresh accessToken instead of failing immediately. + setProxyTokenRefresher(async () => { + const tokens = getAuthTokens(); + if (!tokens?.refreshToken) return null; + const serverBaseUrl = getServerApiBaseUrl(); + try { + const resp = await net.fetch(`${serverBaseUrl}/api/auth/refresh`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ refreshToken: tokens.refreshToken }), + }); + if (resp.ok) { + const body = await resp.json() as { code: number; data: { accessToken: string; refreshToken?: string } }; + if (body.code === 0 && body.data) { + saveAuthTokens(body.data.accessToken, body.data.refreshToken || tokens.refreshToken); + console.log('[Auth] proxy token refresh succeeded'); + return body.data.accessToken; + } + } + } catch (err) { + console.warn('[Auth] proxy token refresh failed:', err); + } + return null; + }); + bindCoworkRuntimeForwarder(); bindOpenClawStatusForwarder(); diff --git a/src/renderer/components/cowork/CoworkSessionDetail.tsx b/src/renderer/components/cowork/CoworkSessionDetail.tsx index e1b6ad5..9e7b429 100644 --- a/src/renderer/components/cowork/CoworkSessionDetail.tsx +++ b/src/renderer/components/cowork/CoworkSessionDetail.tsx @@ -1306,6 +1306,10 @@ const CoworkSessionDetail: React.FC = ({ const navigatingTimerRef = useRef | null>(null); const turnElsCacheRef = useRef([]); const [isScrollable, setIsScrollable] = useState(false); + const [showTurnIndexPopover, setShowTurnIndexPopover] = useState(false); + const showTurnIndexPopoverRef = useRef(false); + const turnIndexPopoverRef = useRef(null); + const turnIndexButtonRef = useRef(null); // Menu and action states const [menuPosition, setMenuPosition] = useState<{ x: number; y: number } | null>(null); @@ -1359,6 +1363,8 @@ const CoworkSessionDetail: React.FC = ({ turnElsCacheRef.current = []; if (hideNavTimerRef.current) clearTimeout(hideNavTimerRef.current); if (navigatingTimerRef.current) clearTimeout(navigatingTimerRef.current); + setShowTurnIndexPopover(false); + showTurnIndexPopoverRef.current = false; }, [currentSession?.id]); // Close menu on outside click @@ -1678,7 +1684,9 @@ const CoworkSessionDetail: React.FC = ({ setShowTurnNav((prev) => (prev ? prev : true)); if (hideNavTimerRef.current) clearTimeout(hideNavTimerRef.current); - hideNavTimerRef.current = setTimeout(() => setShowTurnNav(false), NAV_HIDE_DELAY); + if (!showTurnIndexPopoverRef.current) { + hideNavTimerRef.current = setTimeout(() => setShowTurnNav(false), NAV_HIDE_DELAY); + } // Skip index recalculation during programmatic navigation if (isNavigatingRef.current) return; @@ -1708,28 +1716,31 @@ const CoworkSessionDetail: React.FC = ({ setCurrentTurnIndex(visibleIndex); }, []); + const navigateToTurnByIndex = useCallback((index: number) => { + const turnEls = turnElsCacheRef.current; + if (index < 0 || index >= turnEls.length) return; + + isNavigatingRef.current = true; + if (navigatingTimerRef.current) clearTimeout(navigatingTimerRef.current); + navigatingTimerRef.current = setTimeout(() => { isNavigatingRef.current = false; }, NAV_SCROLL_LOCK_DURATION); + + turnEls[index].scrollIntoView({ behavior: 'smooth', block: 'start' }); + currentTurnIndexRef.current = index; + setCurrentTurnIndex(index); + + setShowTurnNav(true); + if (hideNavTimerRef.current) clearTimeout(hideNavTimerRef.current); + hideNavTimerRef.current = setTimeout(() => setShowTurnNav(false), NAV_HIDE_DELAY); + }, []); + const navigateToTurn = useCallback((direction: 'prev' | 'next') => { const turnEls = turnElsCacheRef.current; if (turnEls.length === 0) return; const idx = currentTurnIndexRef.current; const targetIndex = direction === 'prev' ? idx - 1 : idx + 1; if (targetIndex < 0 || targetIndex >= turnEls.length) return; - - // Block scroll handler from overriding index during smooth scroll - isNavigatingRef.current = true; - if (navigatingTimerRef.current) clearTimeout(navigatingTimerRef.current); - navigatingTimerRef.current = setTimeout(() => { isNavigatingRef.current = false; }, NAV_SCROLL_LOCK_DURATION); - - turnEls[targetIndex].scrollIntoView({ behavior: 'smooth', block: 'start' }); - currentTurnIndexRef.current = targetIndex; - setCurrentTurnIndex(targetIndex); - // Reset hide timer - setShowTurnNav(true); - if (hideNavTimerRef.current) clearTimeout(hideNavTimerRef.current); - hideNavTimerRef.current = setTimeout(() => setShowTurnNav(false), NAV_HIDE_DELAY); - }, []); - - // Get the last message content for auto-scroll on streaming updates + navigateToTurnByIndex(targetIndex); + }, [navigateToTurnByIndex]); const lastMessage = currentSession?.messages?.[currentSession.messages.length - 1]; const lastMessageContent = lastMessage?.content; @@ -1789,6 +1800,60 @@ const CoworkSessionDetail: React.FC = ({ ); }, [turns]); + // Close turn index popover on click-outside or Escape + useEffect(() => { + if (!showTurnIndexPopover) return; + + const handleClickOutside = (e: MouseEvent) => { + if ( + turnIndexPopoverRef.current && !turnIndexPopoverRef.current.contains(e.target as Node) + && !turnIndexButtonRef.current?.contains(e.target as Node) + ) { + setShowTurnIndexPopover(false); + } + }; + const handleEscape = (e: KeyboardEvent) => { + if (e.key === 'Escape') setShowTurnIndexPopover(false); + }; + + document.addEventListener('mousedown', handleClickOutside); + document.addEventListener('keydown', handleEscape); + return () => { + document.removeEventListener('mousedown', handleClickOutside); + document.removeEventListener('keydown', handleEscape); + }; + }, [showTurnIndexPopover]); + + // Auto-scroll popover to bottom (show latest conversations) when opened + useEffect(() => { + if (showTurnIndexPopover && turnIndexPopoverRef.current) { + turnIndexPopoverRef.current.scrollTop = turnIndexPopoverRef.current.scrollHeight; + } + }, [showTurnIndexPopover]); + + // Sync popover ref and manage hide timer based on popover visibility + useEffect(() => { + showTurnIndexPopoverRef.current = showTurnIndexPopover; + if (showTurnIndexPopover) { + // Popover opened — clear hide timer to keep nav visible + if (hideNavTimerRef.current) { + clearTimeout(hideNavTimerRef.current); + hideNavTimerRef.current = null; + } + } else { + // Popover closed — restart hide timer + if (showTurnNav) { + if (hideNavTimerRef.current) clearTimeout(hideNavTimerRef.current); + hideNavTimerRef.current = setTimeout(() => setShowTurnNav(false), NAV_HIDE_DELAY); + } + } + }, [showTurnIndexPopover, showTurnNav]); + + // Close popover when nav hides + useEffect(() => { + if (!showTurnNav) setShowTurnIndexPopover(false); + }, [showTurnNav]); + // Auto scroll to bottom when new messages arrive or content updates (streaming) useEffect(() => { if (!shouldAutoScroll) { @@ -2043,39 +2108,90 @@ const CoworkSessionDetail: React.FC = ({ {/* Turn Navigation Buttons */} {turns.length > 1 && isScrollable && ( -
+
- -
- -
+ +
+ +
+ +
+ + {/* Turn Index Popover */} + {showTurnIndexPopover && ( +
+ {turns.map((turn, index) => { + const label = turn.userMessage?.content + ? turn.userMessage.content.split('\n')[0].slice(0, 60) + : '...'; + const isActive = index === currentTurnIndex; + return ( + + ); + })} +
+ )} + )}
diff --git a/src/renderer/components/im/IMSettings.tsx b/src/renderer/components/im/IMSettings.tsx index 7e4dd12..82c1757 100644 --- a/src/renderer/components/im/IMSettings.tsx +++ b/src/renderer/components/im/IMSettings.tsx @@ -139,6 +139,7 @@ const IMSettings: React.FC = () => { const [weixinQrStatus, setWeixinQrStatus] = useState<'idle' | 'loading' | 'showing' | 'waiting' | 'success' | 'error'>('idle'); const [weixinQrUrl, setWeixinQrUrl] = useState(''); const [weixinQrError, setWeixinQrError] = useState(''); + const weixinTimerRef = useRef | null>(null); const [localIp, setLocalIp] = useState(''); const isMountedRef = useRef(true); @@ -253,6 +254,7 @@ const IMSettings: React.FC = () => { // Reset weixin QR login state when switching away from weixin useEffect(() => { if (activePlatform !== 'weixin') { + if (weixinTimerRef.current) { clearTimeout(weixinTimerRef.current); weixinTimerRef.current = null; } setWeixinQrStatus('idle'); setWeixinQrUrl(''); setWeixinQrError(''); @@ -497,9 +499,18 @@ const IMSettings: React.FC = () => { setWeixinQrUrl(startResult.qrDataUrl); setWeixinQrStatus('showing'); + // QR expires in ~2 minutes. Show error and let user retry. + if (weixinTimerRef.current) clearTimeout(weixinTimerRef.current); + weixinTimerRef.current = setTimeout(() => { + if (!isMountedRef.current) return; + setWeixinQrStatus('error'); + setWeixinQrError(i18nService.t('imWeixinQrExpired')); + }, 120000); + // Start polling for scan result setWeixinQrStatus('waiting'); const waitResult = await window.electron.im.weixinQrLoginWait(startResult.sessionKey); + if (weixinTimerRef.current) { clearTimeout(weixinTimerRef.current); weixinTimerRef.current = null; } if (!isMountedRef.current) return; if (waitResult.success && waitResult.connected) { @@ -516,6 +527,7 @@ const IMSettings: React.FC = () => { setWeixinQrError(waitResult.message || i18nService.t('imWeixinQrFailed')); } } catch (err) { + if (weixinTimerRef.current) { clearTimeout(weixinTimerRef.current); weixinTimerRef.current = null; } if (!isMountedRef.current) return; setWeixinQrStatus('error'); setWeixinQrError(String(err)); diff --git a/src/renderer/services/auth.ts b/src/renderer/services/auth.ts index 1818f99..7c827c5 100644 --- a/src/renderer/services/auth.ts +++ b/src/renderer/services/auth.ts @@ -6,6 +6,8 @@ import type { Model } from '../store/slices/modelSlice'; class AuthService { private unsubCallback: (() => void) | null = null; private unsubQuotaChanged: (() => void) | null = null; + private unsubWindowState: (() => void) | null = null; + private lastRefreshTime = 0; /** * Initialize: try to restore login state from persisted token. @@ -35,6 +37,19 @@ class AuthService { // Listen for quota changes (e.g. after cowork session using server model) this.unsubQuotaChanged = window.electron.auth.onQuotaChanged(() => { this.refreshQuota(); + this.loadServerModels(); + }); + + // Refresh quota and models when Electron window gains focus — user may have purchased on portal + this.unsubWindowState = window.electron.window.onStateChanged((state) => { + if (state.isFocused && store.getState().auth.isLoggedIn) { + const now = Date.now(); + if (now - this.lastRefreshTime > 30_000) { + this.lastRefreshTime = now; + this.refreshQuota(); + this.loadServerModels(); + } + } }); } @@ -139,6 +154,8 @@ class AuthService { this.unsubCallback = null; this.unsubQuotaChanged?.(); this.unsubQuotaChanged = null; + this.unsubWindowState?.(); + this.unsubWindowState = null; } /** diff --git a/src/renderer/services/i18n.ts b/src/renderer/services/i18n.ts index 981af89..d96b7e7 100644 --- a/src/renderer/services/i18n.ts +++ b/src/renderer/services/i18n.ts @@ -458,6 +458,7 @@ const translations: Record> = { coworkExportImageSuccess: '图片导出成功', coworkExportImageFailed: '导出图片失败', deleteSession: '删除任务', + turnIndex: '对话索引', deleteTaskConfirmTitle: '确认删除任务', deleteTaskConfirmMessage: '此操作无法撤销,任务的所有消息记录将被永久删除。', batchOperations: '批量操作', @@ -1562,6 +1563,7 @@ const translations: Record> = { coworkExportImageSuccess: 'Image exported successfully', coworkExportImageFailed: 'Failed to export image', deleteSession: 'Delete Task', + turnIndex: 'Turn index', deleteTaskConfirmTitle: 'Confirm Deletion', deleteTaskConfirmMessage: 'This action cannot be undone. All messages in this task will be permanently deleted.', batchOperations: 'Batch Operations', diff --git a/tests/openclawReconcile.test.mjs b/tests/openclawReconcile.test.mjs new file mode 100644 index 0000000..da21a6d --- /dev/null +++ b/tests/openclawReconcile.test.mjs @@ -0,0 +1,282 @@ +import assert from 'node:assert/strict'; +import Module from 'node:module'; +import test from 'node:test'; +import { createRequire } from 'node:module'; + +const require = createRequire(import.meta.url); +const originalModuleLoad = Module._load; + +Module._load = function patchedModuleLoad(request, parent, isMain) { + if (request === 'electron') { + return { + app: { + getAppPath: () => process.cwd(), + getPath: () => process.cwd(), + }, + BrowserWindow: { + getAllWindows: () => [], + }, + }; + } + return originalModuleLoad.call(this, request, parent, isMain); +}; + +const { OpenClawRuntimeAdapter } = require('../dist-electron/main/libs/agentEngine/openclawRuntimeAdapter.js'); + +const createStore = (messages) => { + const session = { + id: 'session-1', + title: 'Test Session', + claudeSessionId: null, + status: 'completed', + pinned: false, + cwd: '', + systemPrompt: '', + executionMode: 'local', + activeSkillIds: [], + messages: [...messages], + createdAt: 1, + updatedAt: 1, + }; + let nextId = session.messages.length + 1; + let replaceCallCount = 0; + let lastReplaceArgs = null; + + return { + session, + getReplaceCallCount: () => replaceCallCount, + getLastReplaceArgs: () => lastReplaceArgs, + store: { + getSession: (sessionId) => (sessionId === session.id ? session : null), + addMessage: (sessionId, message) => { + assert.equal(sessionId, session.id); + const created = { + id: `msg-${nextId++}`, + timestamp: nextId, + metadata: {}, + ...message, + }; + session.messages.push(created); + return created; + }, + updateSession: () => {}, + replaceConversationMessages: (sessionId, authoritative) => { + replaceCallCount++; + lastReplaceArgs = { sessionId, authoritative }; + // Simulate: remove old user/assistant, insert new ones + session.messages = session.messages.filter( + (m) => m.type !== 'user' && m.type !== 'assistant', + ); + for (const entry of authoritative) { + session.messages.push({ + id: `msg-${nextId++}`, + type: entry.role, + content: entry.text, + metadata: { isStreaming: false, isFinal: true }, + timestamp: nextId, + }); + } + }, + deleteMessage: () => true, + }, + }; +}; + +test.after(() => { + Module._load = originalModuleLoad; +}); + +test('reconcileWithHistory: already in sync — skips replace', async () => { + const { session, store, getReplaceCallCount } = createStore([ + { id: 'msg-1', type: 'user', content: 'Hello', timestamp: 1, metadata: {} }, + { id: 'msg-2', type: 'assistant', content: 'Hi there', timestamp: 2, metadata: {} }, + ]); + + const adapter = new OpenClawRuntimeAdapter(store, {}); + adapter.gatewayClient = { + start: () => {}, + stop: () => {}, + request: async () => ({ + messages: [ + { role: 'user', content: 'Hello' }, + { role: 'assistant', content: 'Hi there' }, + ], + }), + }; + + await adapter.reconcileWithHistory(session.id, 'managed:session-1'); + + assert.equal(getReplaceCallCount(), 0, 'should not call replaceConversationMessages when in sync'); + assert.equal(session.messages.length, 2); +}); + +test('reconcileWithHistory: missing assistant message — triggers replace', async () => { + const { session, store, getReplaceCallCount, getLastReplaceArgs } = createStore([ + { id: 'msg-1', type: 'user', content: 'Hello', timestamp: 1, metadata: {} }, + // assistant message missing locally + ]); + + const adapter = new OpenClawRuntimeAdapter(store, {}); + adapter.gatewayClient = { + start: () => {}, + stop: () => {}, + request: async () => ({ + messages: [ + { role: 'user', content: 'Hello' }, + { role: 'assistant', content: 'Hi there' }, + ], + }), + }; + + await adapter.reconcileWithHistory(session.id, 'managed:session-1'); + + assert.equal(getReplaceCallCount(), 1, 'should call replaceConversationMessages'); + const args = getLastReplaceArgs(); + assert.equal(args.sessionId, session.id); + assert.deepEqual(args.authoritative, [ + { role: 'user', text: 'Hello' }, + { role: 'assistant', text: 'Hi there' }, + ]); +}); + +test('reconcileWithHistory: duplicate messages locally — triggers replace', async () => { + const { session, store, getReplaceCallCount, getLastReplaceArgs } = createStore([ + { id: 'msg-1', type: 'user', content: 'Hello', timestamp: 1, metadata: {} }, + { id: 'msg-2', type: 'assistant', content: 'Hi there', timestamp: 2, metadata: {} }, + { id: 'msg-3', type: 'assistant', content: 'Hi there', timestamp: 3, metadata: {} }, // duplicate + ]); + + const adapter = new OpenClawRuntimeAdapter(store, {}); + adapter.gatewayClient = { + start: () => {}, + stop: () => {}, + request: async () => ({ + messages: [ + { role: 'user', content: 'Hello' }, + { role: 'assistant', content: 'Hi there' }, + ], + }), + }; + + await adapter.reconcileWithHistory(session.id, 'managed:session-1'); + + assert.equal(getReplaceCallCount(), 1, 'should replace to fix duplicates'); + const args = getLastReplaceArgs(); + assert.equal(args.authoritative.length, 2); +}); + +test('reconcileWithHistory: content mismatch — triggers replace', async () => { + const { session, store, getReplaceCallCount, getLastReplaceArgs } = createStore([ + { id: 'msg-1', type: 'user', content: 'Hello', timestamp: 1, metadata: {} }, + { id: 'msg-2', type: 'assistant', content: 'Streaming partial...', timestamp: 2, metadata: {} }, + ]); + + const adapter = new OpenClawRuntimeAdapter(store, {}); + adapter.gatewayClient = { + start: () => {}, + stop: () => {}, + request: async () => ({ + messages: [ + { role: 'user', content: 'Hello' }, + { role: 'assistant', content: 'Full complete response from the model.' }, + ], + }), + }; + + await adapter.reconcileWithHistory(session.id, 'managed:session-1'); + + assert.equal(getReplaceCallCount(), 1, 'should replace when content mismatches'); + const args = getLastReplaceArgs(); + assert.equal(args.authoritative[1].text, 'Full complete response from the model.'); +}); + +test('reconcileWithHistory: preserves tool messages', async () => { + const { session, store, getReplaceCallCount } = createStore([ + { id: 'msg-1', type: 'user', content: 'Run a command', timestamp: 1, metadata: {} }, + { id: 'msg-2', type: 'tool_use', content: 'Using bash', timestamp: 2, metadata: {} }, + { id: 'msg-3', type: 'tool_result', content: 'OK', timestamp: 3, metadata: {} }, + { id: 'msg-4', type: 'assistant', content: 'Done!', timestamp: 4, metadata: {} }, + ]); + + const adapter = new OpenClawRuntimeAdapter(store, {}); + adapter.gatewayClient = { + start: () => {}, + stop: () => {}, + request: async () => ({ + messages: [ + { role: 'user', content: 'Run a command' }, + { role: 'assistant', content: 'Done!' }, + ], + }), + }; + + await adapter.reconcileWithHistory(session.id, 'managed:session-1'); + + assert.equal(getReplaceCallCount(), 0, 'should be in sync (tool messages ignored in comparison)'); +}); + +test('reconcileWithHistory: empty history — sets cursor to 0', async () => { + const { session, store, getReplaceCallCount } = createStore([ + { id: 'msg-1', type: 'user', content: 'Hello', timestamp: 1, metadata: {} }, + ]); + + const adapter = new OpenClawRuntimeAdapter(store, {}); + adapter.gatewayClient = { + start: () => {}, + stop: () => {}, + request: async () => ({ messages: [] }), + }; + + await adapter.reconcileWithHistory(session.id, 'managed:session-1'); + + assert.equal(getReplaceCallCount(), 0, 'should not replace on empty history'); + assert.equal(adapter.channelSyncCursor.get(session.id), 0); +}); + +test('reconcileWithHistory: multi-turn conversation — correct order', async () => { + const { session, store, getReplaceCallCount, getLastReplaceArgs } = createStore([ + { id: 'msg-1', type: 'user', content: 'First', timestamp: 1, metadata: {} }, + { id: 'msg-2', type: 'assistant', content: 'Reply 1', timestamp: 2, metadata: {} }, + // Missing second turn + ]); + + const adapter = new OpenClawRuntimeAdapter(store, {}); + adapter.gatewayClient = { + start: () => {}, + stop: () => {}, + request: async () => ({ + messages: [ + { role: 'user', content: 'First' }, + { role: 'assistant', content: 'Reply 1' }, + { role: 'user', content: 'Second' }, + { role: 'assistant', content: 'Reply 2' }, + ], + }), + }; + + await adapter.reconcileWithHistory(session.id, 'managed:session-1'); + + assert.equal(getReplaceCallCount(), 1, 'should replace to add missing turn'); + const args = getLastReplaceArgs(); + assert.equal(args.authoritative.length, 4); + assert.equal(args.authoritative[2].text, 'Second'); + assert.equal(args.authoritative[3].text, 'Reply 2'); +}); + +test('reconcileWithHistory: gateway error — does not crash', async () => { + const { session, store, getReplaceCallCount } = createStore([ + { id: 'msg-1', type: 'user', content: 'Hello', timestamp: 1, metadata: {} }, + ]); + + const adapter = new OpenClawRuntimeAdapter(store, {}); + adapter.gatewayClient = { + start: () => {}, + stop: () => {}, + request: async () => { throw new Error('Network timeout'); }, + }; + + // Should not throw + await adapter.reconcileWithHistory(session.id, 'managed:session-1'); + + assert.equal(getReplaceCallCount(), 0, 'should not replace on error'); +}); diff --git a/tests/openclawRuntimeHotfix.test.mjs b/tests/openclawRuntimeHotfix.test.mjs deleted file mode 100644 index d847d2a..0000000 --- a/tests/openclawRuntimeHotfix.test.mjs +++ /dev/null @@ -1,275 +0,0 @@ -import test from 'node:test'; -import assert from 'node:assert/strict'; -import fs from 'node:fs'; -import { createRequire } from 'node:module'; -import os from 'node:os'; -import path from 'node:path'; - -const require = createRequire(import.meta.url); -const { - applyBundledOpenClawRuntimeHotfixes, - patchCronReminderCurrentTimeSuffix, - patchCronReminderPromptEnvelope, - patchCronSessionDeliveryInference, - patchCronToolOwnerOnly, - patchCronOwnerFallback, - patchWecomMessageProviderExecDeny, -} = require('../dist-electron/main/libs/openclawRuntimeHotfix.js'); - -const walkJsFiles = (dirPath, files = []) => { - for (const entry of fs.readdirSync(dirPath, { withFileTypes: true })) { - const fullPath = path.join(dirPath, entry.name); - if (entry.isDirectory()) { - walkJsFiles(fullPath, files); - continue; - } - if (entry.isFile() && fullPath.endsWith('.js')) { - files.push(fullPath); - } - } - return files; -}; - -test('patchCronToolOwnerOnly only flips the cron tool guard', () => { - const source = [ - 'function createCronTool(opts, deps) {', - ' return {', - ' label: "Cron",', - ' name: "cron",', - ' ownerOnly: true,', - ' };', - '}', - 'function createGatewayTool() {', - ' return {', - ' label: "Gateway",', - ' name: "gateway",', - ' ownerOnly: true,', - ' };', - '}', - ].join('\n'); - - const result = patchCronToolOwnerOnly(source); - - assert.equal(result.changed, true); - assert.match(result.content, /name: "cron",\n\s+ownerOnly: false,/); - assert.match(result.content, /name: "gateway",\n\s+ownerOnly: true,/); -}); - -test('patchCronOwnerFallback removes cron from owner-only fallback names only', () => { - const source = [ - 'const OWNER_ONLY_TOOL_NAME_FALLBACKS = new Set([', - ' "whatsapp_login",', - ' "cron",', - ' "gateway"', - ']);', - ].join('\n'); - - const result = patchCronOwnerFallback(source); - - assert.equal(result.changed, true); - assert.doesNotMatch(result.content, /"cron"/); - assert.match(result.content, /"whatsapp_login"/); - assert.match(result.content, /"gateway"/); -}); - -test('patchWecomMessageProviderExecDeny denies exec/process for WeCom native sessions only', () => { - const source = 'const TOOL_DENY_BY_MESSAGE_PROVIDER = { voice: ["tts"] };'; - - const result = patchWecomMessageProviderExecDeny(source); - - assert.equal(result.changed, true); - assert.match( - result.content, - /const TOOL_DENY_BY_MESSAGE_PROVIDER = \{ voice: \["tts"\], wecom: \["exec", "process"\] \};/, - ); -}); - -test('patchCronSessionDeliveryInference restores IM routing for persisted sessions and DingTalk keys', () => { - const source = [ - 'function inferDeliveryFromSessionKey(agentSessionKey) {', - ' const rawSessionKey = agentSessionKey?.trim();', - ' if (!rawSessionKey) return null;', - ' const parsed = parseAgentSessionKey(stripThreadSuffixFromSessionKey(rawSessionKey));', - ' if (!parsed || !parsed.rest) return null;', - ' const parts = parsed.rest.split(":").filter(Boolean);', - ' if (parts.length === 0) return null;', - ' const head = parts[0]?.trim().toLowerCase();', - ' if (!head || head === "main" || head === "subagent" || head === "acp") return null;', - ' const markerIndex = parts.findIndex((part) => part === "direct" || part === "dm" || part === "group" || part === "channel");', - ' if (markerIndex === -1) return null;', - ' const peerId = parts.slice(markerIndex + 1).join(":").trim();', - ' if (!peerId) return null;', - ' let channel;', - ' if (markerIndex >= 1) channel = parts[0]?.trim().toLowerCase();', - ' const delivery = {', - ' mode: "announce",', - ' to: peerId', - ' };', - ' if (channel) delivery.channel = channel;', - ' return delivery;', - '}', - ].join('\n').replaceAll(' ', '\t'); - - const result = patchCronSessionDeliveryInference(source); - - assert.equal(result.changed, true); - assert.match(result.content, /extractDeliveryInfo\(rawSessionKey\)/); - assert.match(result.content, /persistedDelivery\?\.channel && persistedDelivery\?\.to/); - assert.match(result.content, /head === "dingtalk-connector"/); - assert.match(result.content, /to: `user:\$\{senderId\}`/); - assert.match(result.content, /delivery\.accountId = persistedDelivery\.accountId/); -}); - -test('patchCronReminderPromptEnvelope strips scheduled reminder wrapper text', () => { - const source = [ - 'function buildCronEventPrompt(pendingEvents, opts) {', - '\tconst deliverToUser = opts?.deliverToUser ?? true;', - '\tconst eventText = pendingEvents.join("\\n").trim();', - '\tif (!eventText) {', - '\t\tif (!deliverToUser) return "A scheduled cron event was triggered, but no event content was found. Handle this internally and reply HEARTBEAT_OK when nothing needs user-facing follow-up.";', - '\t\treturn "A scheduled cron event was triggered, but no event content was found. Reply HEARTBEAT_OK.";', - '\t}', - '\tif (!deliverToUser) return "A scheduled reminder has been triggered. The reminder content is:\\n\\n" + eventText + "\\n\\nHandle this reminder internally. Do not relay it to the user unless explicitly requested.";', - '\treturn "A scheduled reminder has been triggered. The reminder content is:\\n\\n" + eventText + "\\n\\nPlease relay this reminder to the user in a helpful and friendly way.";', - '}', - ].join('\n'); - - const result = patchCronReminderPromptEnvelope(source); - - assert.equal(result.changed, true); - assert.match(result.content, /return eventText;/); - assert.doesNotMatch(result.content, /A scheduled reminder has been triggered/); -}); - -test('patchCronReminderCurrentTimeSuffix skips appending Current time for cron reminder prompts', () => { - const source = [ - 'const hasCronEvents = true;', - 'const ctx = {', - '\tBody: appendCronStyleCurrentTimeLine(prompt, cfg, startedAt),', - '};', - ].join('\n'); - - const result = patchCronReminderCurrentTimeSuffix(source); - - assert.equal(result.changed, true); - assert.match( - result.content, - /Body: hasCronEvents \? prompt : appendCronStyleCurrentTimeLine\(prompt, cfg, startedAt\),/, - ); -}); - -test('applyBundledOpenClawRuntimeHotfixes patches matching runtime dist files', () => { - const tmpRoot = fs.mkdtempSync(path.join(os.tmpdir(), 'openclaw-runtime-hotfix-')); - const distDir = path.join(tmpRoot, 'dist', 'plugin-sdk'); - fs.mkdirSync(distDir, { recursive: true }); - - const targetFile = path.join(distDir, 'reply.js'); - fs.writeFileSync( - targetFile, - [ - 'function createCronTool(opts, deps) {', - ' return {', - ' label: "Cron",', - ' name: "cron",', - ' ownerOnly: true,', - ' };', - '}', - 'const OWNER_ONLY_TOOL_NAME_FALLBACKS = new Set([', - ' "whatsapp_login",', - ' "cron",', - ' "gateway"', - ']);', - 'const TOOL_DENY_BY_MESSAGE_PROVIDER = { voice: ["tts"] };', - 'function inferDeliveryFromSessionKey(agentSessionKey) {', - '\tconst rawSessionKey = agentSessionKey?.trim();', - '\tif (!rawSessionKey) return null;', - '\tconst parsed = parseAgentSessionKey(stripThreadSuffixFromSessionKey(rawSessionKey));', - '\tif (!parsed || !parsed.rest) return null;', - '\tconst parts = parsed.rest.split(":").filter(Boolean);', - '\tif (parts.length === 0) return null;', - '\tconst head = parts[0]?.trim().toLowerCase();', - '\tif (!head || head === "main" || head === "subagent" || head === "acp") return null;', - '\tconst markerIndex = parts.findIndex((part) => part === "direct" || part === "dm" || part === "group" || part === "channel");', - '\tif (markerIndex === -1) return null;', - '\tconst peerId = parts.slice(markerIndex + 1).join(":").trim();', - '\tif (!peerId) return null;', - '\tlet channel;', - '\tif (markerIndex >= 1) channel = parts[0]?.trim().toLowerCase();', - '\tconst delivery = {', - '\t\tmode: "announce",', - '\t\tto: peerId', - '\t};', - '\tif (channel) delivery.channel = channel;', - '\treturn delivery;', - '}', - 'function buildCronEventPrompt(pendingEvents, opts) {', - '\tconst deliverToUser = opts?.deliverToUser ?? true;', - '\tconst eventText = pendingEvents.join("\\n").trim();', - '\tif (!eventText) {', - '\t\tif (!deliverToUser) return "A scheduled cron event was triggered, but no event content was found. Handle this internally and reply HEARTBEAT_OK when nothing needs user-facing follow-up.";', - '\t\treturn "A scheduled cron event was triggered, but no event content was found. Reply HEARTBEAT_OK.";', - '\t}', - '\tif (!deliverToUser) return "A scheduled reminder has been triggered. The reminder content is:\\n\\n" + eventText + "\\n\\nHandle this reminder internally. Do not relay it to the user unless explicitly requested.";', - '\treturn "A scheduled reminder has been triggered. The reminder content is:\\n\\n" + eventText + "\\n\\nPlease relay this reminder to the user in a helpful and friendly way.";', - '}', - 'const hasCronEvents = true;', - 'const ctx = {', - '\tBody: appendCronStyleCurrentTimeLine(prompt, cfg, startedAt),', - '};', - ].join('\n'), - 'utf8', - ); - - const controlFile = path.join(tmpRoot, 'dist', 'other.js'); - fs.writeFileSync( - controlFile, - [ - 'function noop() {', - ' return { ownerOnly: true };', - '}', - ].join('\n'), - 'utf8', - ); - - const result = applyBundledOpenClawRuntimeHotfixes(tmpRoot); - - assert.equal(result.changed, true); - assert.deepEqual(result.errors, []); - assert.deepEqual(result.patchedFiles, [targetFile]); - assert.match(fs.readFileSync(targetFile, 'utf8'), /ownerOnly: false,/); - assert.doesNotMatch( - fs.readFileSync(targetFile, 'utf8'), - /OWNER_ONLY_TOOL_NAME_FALLBACKS[\s\S]*"cron"/, - ); - assert.match( - fs.readFileSync(targetFile, 'utf8'), - /TOOL_DENY_BY_MESSAGE_PROVIDER = \{ voice: \["tts"\], wecom: \["exec", "process"\] \};/, - ); - assert.match(fs.readFileSync(targetFile, 'utf8'), /extractDeliveryInfo\(rawSessionKey\)/); - assert.match(fs.readFileSync(targetFile, 'utf8'), /head === "dingtalk-connector"/); - assert.match(fs.readFileSync(targetFile, 'utf8'), /return eventText;/); - assert.match( - fs.readFileSync(targetFile, 'utf8'), - /Body: hasCronEvents \? prompt : appendCronStyleCurrentTimeLine\(prompt, cfg, startedAt\),/, - ); - assert.match(fs.readFileSync(controlFile, 'utf8'), /ownerOnly: true/); -}); - -test('bundled OpenClaw runtime exposes cron to non-owner native sessions', () => { - const runtimeDist = path.resolve('vendor/openclaw-runtime/mac-arm64/dist'); - const cronOwnerOffendingFiles = []; - const cronFallbackOffendingFiles = []; - - for (const filePath of walkJsFiles(runtimeDist)) { - const source = fs.readFileSync(filePath, 'utf8'); - if (patchCronToolOwnerOnly(source).changed) { - cronOwnerOffendingFiles.push(path.relative(process.cwd(), filePath)); - } - if (patchCronOwnerFallback(source).changed) { - cronFallbackOffendingFiles.push(path.relative(process.cwd(), filePath)); - } - } - - assert.deepEqual(cronOwnerOffendingFiles, []); - assert.deepEqual(cronFallbackOffendingFiles, []); -});