diff --git a/README.md b/README.md index 820abaa..6adab48 100644 --- a/README.md +++ b/README.md @@ -20,6 +20,7 @@ - **工具权限审批** — WeChat 弹权限卡片，支持 `once` / `always` / `reject` 三选一；`/auto-permission` 可切换自动接收模式；30 分钟软超时自动 reject - **静默模式** — 启用 `/silent`（别名 `/sl`）后，一轮 turn 中只发送最终文本回复；推理、工具摘要、增量文本在 turn 结束前隐藏。Question / Permission 请求不受影响；设置独立且跨重启持久化 - **跨会话通知** — 其他 session 的 question/permission/error/completion 事件推送到微信；切换到有 pending 的会话自动弹出卡片 +- **卡住 / 异常自动提示** — Agent 5 分钟无响应（卡住）自动结束本轮并微信提示；SSE 事件连接中断超过 1 分钟、微信消息通道连续失败或登录过期、以及恢复时均自动微信通知，无需打开终端观察 - **二维码登录** — 终端渲染二维码，扫码登录微信 - **OpenCode Server** — 基于 HTTP API，不再需要 ACP 子进程 @@ -217,9 +218,13 @@ export WECHAT_OPENCODE_STARTUP_TIMEOUT_MS=600000 # 10 分钟（极慢网络） ### 消息计数（`/next`） +微信限制 bot 连续发送 10 条消息，超出后剩余消息会缓存，不再继续发送。 + | 命令 | 说明 | |------|------| -| `/next` | 微信限制连续发送 10 条消息，超出后需用户回复才能继续。发送 `/next` 重置计数，不转发给 Agent | +| `/next` | 立即发送缓存的剩余消息并重置计数，不转发给 Agent | + +**自动续发（无需手动 `/next`）**：默认开启。超出 10 条限制后，缓存的剩余消息会在约 60 秒后自动发送，无需用户任何操作；若一次缓存超过 10 条，会自动分轮续发直到发完。通过配置 `wechat.autoFlushMs`（毫秒）调整间隔，设为 `0` 关闭（恢复纯手动 `/next` 行为）。 ### LLM 问答（`/reject-question`） diff --git a/src/__tests__/test-auto-flush.mjs b/src/__tests__/test-auto-flush.mjs new file mode 100644 index 0000000..51ae359 --- /dev/null +++ b/src/__tests__/test-auto-flush.mjs @@ -0,0 +1,166 @@ +/** + * Tests for the auto-flush ("auto /next") feature. + * + * When the bridge hits WeChat's 10-consecutive-message send limit, the + * remaining outbound segments are cached in `pendingOutbound`. Without + * auto-flush they are only delivered on the next user message or an + * explicit `/next`. With `wechat.autoFlushMs` > 0, the bridge arms a + * timer on the first cache hit and flushes the cached messages + * automatically once the delay elapses; if the limit still holds + * (more than 10 segments cached), the timer re-arms for another round. + * + * The methods under test are private — TS `private` is compile-time + * only, so the compiled dist/ methods are reachable at runtime. We + * drive the real instance with fake timers and stub the outbound send + * path (`flushPending` / `sendReplyImpl`) so no network is touched. + * + * Run via `npm test` (requires `npm run build` first to produce dist/). + */ +import { describe, test, expect, vi } from "vitest"; +import { WeChatOpencodeBridge } from "../../dist/src/bridge.js"; +import { defaultConfig } from "../../dist/src/config.js"; + +function makeBridge(overrides = {}) { + const config = defaultConfig(); + Object.assign(config.wechat, overrides); + const bridge = new WeChatOpencodeBridge(config, () => {}); + return bridge; +} + +describe("auto-flush (auto /next)", () => { + test("does not arm a timer when autoFlushMs is 0 (disabled)", async () => { + vi.useFakeTimers(); + try { + const bridge = makeBridge({ autoFlushMs: 0 }); + bridge.pendingOutbound = [{ kind: "text", text: "cached", contextToken: "ctx-1" }]; + + const flushSpy = vi.spyOn(bridge, "flushPending").mockResolvedValue(undefined); + bridge.scheduleAutoFlush("ctx-1"); + await vi.advanceTimersByTimeAsync(60_000); + + expect(flushSpy).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + test("does not arm a timer when nothing is cached", async () => { + vi.useFakeTimers(); + try { + const bridge = makeBridge({ autoFlushMs: 60_000 }); + const flushSpy = vi.spyOn(bridge, "flushPending").mockResolvedValue(undefined); + + bridge.scheduleAutoFlush("ctx-1"); + await vi.advanceTimersByTimeAsync(60_000); + + expect(flushSpy).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + test("flushes cached messages once the delay elapses", async () => { + vi.useFakeTimers(); + try { + const bridge = makeBridge({ autoFlushMs: 60_000 }); + bridge.pendingOutbound = [{ kind: "text", text: "cached", contextToken: "ctx-1" }]; + const flushSpy = vi.spyOn(bridge, "flushPending").mockResolvedValue(undefined); + + bridge.scheduleAutoFlush("ctx-1"); + expect(bridge.autoFlushTimer).not.toBeNull(); + + await vi.advanceTimersByTimeAsync(59_000); + expect(flushSpy).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(2_000); + expect(flushSpy).toHaveBeenCalledTimes(1); + expect(flushSpy).toHaveBeenCalledWith("ctx-1"); + } finally { + vi.useRealTimers(); + } + }); + + test("re-arms for another round when messages remain cached after a flush", async () => { + vi.useFakeTimers(); + try { + const bridge = makeBridge({ autoFlushMs: 60_000 }); + bridge.pendingOutbound = [{ kind: "text", text: "cached", contextToken: "ctx-1" }]; + vi.spyOn(bridge, "flushPending").mockImplementation(async () => { + // Simulate the 10-msg limit still holding: messages stay cached. + bridge.pendingOutbound = [{ kind: "text", text: "still-cached", contextToken: "ctx-1" }]; + }); + + bridge.scheduleAutoFlush("ctx-1"); + await vi.advanceTimersByTimeAsync(60_000); + expect(bridge.autoFlushTimer).not.toBeNull(); + + await vi.advanceTimersByTimeAsync(60_000); + expect(bridge.pendingOutbound).toHaveLength(1); + } finally { + vi.useRealTimers(); + } + }); + + test("does not flush while an in-flight outbound send is still running", async () => { + vi.useFakeTimers(); + try { + const bridge = makeBridge({ autoFlushMs: 60_000 }); + bridge.pendingOutbound = [{ kind: "text", text: "cached", contextToken: "ctx-1" }]; + const flushSpy = vi.spyOn(bridge, "flushPending").mockResolvedValue(undefined); + + // Simulate an agent reply in flight on the same contextToken. + let releaseInflight = () => {}; + const inflight = new Promise((resolve) => { + releaseInflight = () => resolve(undefined); + }); + bridge.outboundQueue.set("ctx-1", inflight); + + bridge.scheduleAutoFlush("ctx-1"); + await vi.advanceTimersByTimeAsync(60_000); + expect(flushSpy).not.toHaveBeenCalled(); + + releaseInflight(); + await vi.advanceTimersByTimeAsync(0); + expect(flushSpy).toHaveBeenCalledTimes(1); + } finally { + vi.useRealTimers(); + } + }); + + test("cache hits after the first timer is pending do not double-arm", async () => { + vi.useFakeTimers(); + try { + const bridge = makeBridge({ autoFlushMs: 60_000 }); + bridge.pendingOutbound = [{ kind: "text", text: "cached", contextToken: "ctx-1" }]; + const flushSpy = vi.spyOn(bridge, "flushPending").mockResolvedValue(undefined); + + bridge.scheduleAutoFlush("ctx-1"); + const timer = bridge.autoFlushTimer; + bridge.scheduleAutoFlush("ctx-1"); // second cache hit + expect(bridge.autoFlushTimer).toBe(timer); + + await vi.advanceTimersByTimeAsync(60_000); + expect(flushSpy).toHaveBeenCalledTimes(1); + } finally { + vi.useRealTimers(); + } + }); + + test("stop() clears a pending auto-flush timer", async () => { + vi.useFakeTimers(); + try { + const bridge = makeBridge({ autoFlushMs: 60_000 }); + bridge.pendingOutbound = [{ kind: "text", text: "cached", contextToken: "ctx-1" }]; + const flushSpy = vi.spyOn(bridge, "flushPending").mockResolvedValue(undefined); + + bridge.scheduleAutoFlush("ctx-1"); + expect(bridge.autoFlushTimer).not.toBeNull(); + + await bridge.stop(); + await vi.advanceTimersByTimeAsync(60_000); + expect(flushSpy).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); +}); diff --git a/src/__tests__/test-stuck-notices.mjs b/src/__tests__/test-stuck-notices.mjs new file mode 100644 index 0000000..4beab78 --- /dev/null +++ b/src/__tests__/test-stuck-notices.mjs @@ -0,0 +1,272 @@ +/** + * Tests for user-visible system notices ("stuck"/outage alerts). + * + * Two sources of notices: + * 1. `SessionManager` — the 5-minute stuck-turn timeout (force-finalize + * now also tells the user the agent was unresponsive) and SSE event + * pipeline outages (reconnecting > 60s → warning, recovery → OK). + * 2. `startMonitor` — WeChat-channel session expiry, repeated poll + * failures (backoff), and recovery. + * + * The methods under test are private — TS `private` is compile-time + * only, so the compiled dist/ methods are reachable at runtime. The + * SessionManager is driven through a fake EventPipeline injected via + * the `useEventStream: false` path is NOT used here; instead we call + * the pipeline-status handler directly with crafted status values. + * + * Run via `npm test` (requires `npm run build` first to produce dist/). + */ +import { describe, test, expect, vi } from "vitest"; +import { SessionManager } from "../../dist/src/server/session.js"; +import { defaultConfig } from "../../dist/src/config.js"; + +function makeManager(onNotify) { + const config = defaultConfig(); + const manager = new SessionManager({ + serverUrl: config.server.url, + cwd: config.agent.cwd, + log: () => {}, + onReply: async () => {}, + onMediaReply: async () => {}, + sendTyping: async () => {}, + onNotify, + }); + return manager; +} + +describe("SessionManager stuck-turn notice", () => { + test("stuck timeout notifies the user before force-finalizing", async () => { + vi.useFakeTimers(); + try { + const notified = vi.fn().mockResolvedValue(undefined); + const manager = makeManager(notified); + const finalizeSpy = vi.spyOn(manager, "finalizeTurn").mockImplementation(() => { + manager.currentTurn = null; + }); + + // Begin a turn, then let the 5-minute stuck timer elapse. + manager.currentTurn = { + contextToken: "ctx-1", + status: "accumulating", + }; + manager.scheduleStuckTimeout(); + await vi.advanceTimersByTimeAsync(5 * 60_000); + + expect(notified).toHaveBeenCalledTimes(1); + const [ctx, text] = notified.mock.calls[0]; + expect(ctx).toBe("ctx-1"); + expect(text).toContain("无响应"); + expect(finalizeSpy).toHaveBeenCalledWith("finalized"); + } finally { + vi.useRealTimers(); + } + }); + + test("stuck notice is not sent when onNotify is omitted", async () => { + vi.useFakeTimers(); + try { + const manager = makeManager(undefined); + const finalizeSpy = vi.spyOn(manager, "finalizeTurn").mockImplementation(() => { + manager.currentTurn = null; + }); + manager.currentTurn = { contextToken: "ctx-1", status: "accumulating" }; + manager.scheduleStuckTimeout(); + await vi.advanceTimersByTimeAsync(5 * 60_000); + expect(finalizeSpy).toHaveBeenCalledWith("finalized"); // force-finalize still ran + } finally { + vi.useRealTimers(); + } + }); +}); + +describe("SessionManager pipeline-outage notice", () => { + test("reconnecting for >60s notifies once; recovery notifies again", async () => { + vi.useFakeTimers(); + try { + const notified = vi.fn().mockResolvedValue(undefined); + const manager = makeManager(notified); + manager.lastEnqueuedContextToken = "ctx-9"; + + manager.handlePipelineStatus("reconnecting"); + await vi.advanceTimersByTimeAsync(59_000); + manager.handlePipelineStatus("reconnecting"); + expect(notified).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(2_000); + manager.handlePipelineStatus("reconnecting"); + expect(notified).toHaveBeenCalledTimes(1); + expect(notified.mock.calls[0][0]).toBe("ctx-9"); + expect(notified.mock.calls[0][1]).toContain("重连"); + + // Recovery: connected clears the outage and sends an OK notice. + manager.handlePipelineStatus("connected"); + expect(notified).toHaveBeenCalledTimes(2); + expect(notified.mock.calls[1][1]).toContain("已恢复"); + + // A later reconnect must NOT re-notify instantly (fresh outage). + manager.handlePipelineStatus("reconnecting"); + expect(notified).toHaveBeenCalledTimes(2); + } finally { + vi.useRealTimers(); + } + }); + + test("brief reconnects under the threshold never notify", async () => { + vi.useFakeTimers(); + try { + const notified = vi.fn().mockResolvedValue(undefined); + const manager = makeManager(notified); + + manager.handlePipelineStatus("reconnecting"); + await vi.advanceTimersByTimeAsync(10_000); + manager.handlePipelineStatus("connected"); + manager.handlePipelineStatus("reconnecting"); + await vi.advanceTimersByTimeAsync(10_000); + manager.handlePipelineStatus("connected"); + + expect(notified).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); +}); + +describe("SessionManager retry-failure notice", () => { + test("zero-output turn after retry tells the user the request failed", async () => { + vi.useFakeTimers(); + try { + const replies = []; + const config = defaultConfig(); + const manager = new SessionManager({ + serverUrl: config.server.url, + cwd: config.agent.cwd, + log: () => {}, + onReply: async (contextToken, text) => { + replies.push({ contextToken, text }); + }, + onMediaReply: async () => {}, + sendTyping: async () => {}, + }); + manager.currentTurn = { + contextToken: "ctx-retry", + status: "accumulating", + retried: true, + }; + const finalizeSpy = vi.spyOn(manager, "finalizeTurn").mockImplementation(() => { + manager.currentTurn = null; + }); + + // The retry flag is set when session.status=retry arrives mid-turn. + manager.handleSessionStatus({ + type: "session.status", + properties: { sessionID: "ses-x", status: { type: "retry" } }, + }); + expect(manager.currentTurn.retried).toBe(true); + + // Force-finalize (what the debounce does); currentTurn is cleared by + // the spy so finalizeTurn's fallback runs against a snapshot... it + // doesn't — finalizeTurn reads this.currentTurn. Simulate directly: + finalizeSpy.mockRestore(); + // Rebuild a turn and call finalizeTurn for real. + manager.currentTurn = { + sessionId: "ses-x", + userMessageId: "msg-1", + assistantMessageId: null, + parts: new Map(), + textBuffer: "", + finalText: "", + toolCalls: new Map(), + hasBackgroundTasks: false, + contextToken: "ctx-retry", + hint: null, + status: "accumulating", + startedAt: Date.now(), + lastEventAt: Date.now(), + retried: true, + sentTextPartIds: new Set(), + pendingTextParts: [], + pendingReasoningParts: [], + showThoughtsSnapshot: false, + showToolsSnapshot: false, + immersiveSnapshot: false, + immersiveLastText: "", + reasoningCharCount: 0, + reasoningStartMs: null, + reasoningEndMs: null, + sentReasoningPartIds: new Set(), + reasoningPartTimestamps: new Map(), + currentPartType: null, + currentPartID: null, + currentReasoningText: "", + currentReasoningStartMs: null, + currentReasoningEndMs: null, + currentText: "", + currentToolKey: null, + }; + manager.finalizeTurn("finalized"); + + expect(replies).toHaveLength(1); + expect(replies[0].contextToken).toBe("ctx-retry"); + expect(replies[0].text).toContain("❌ Agent 请求失败"); + } finally { + vi.useRealTimers(); + } + }); + + test("zero-output turn WITHOUT retry stays silent (no false positives)", async () => { + vi.useFakeTimers(); + try { + const replies = []; + const config = defaultConfig(); + const manager = new SessionManager({ + serverUrl: config.server.url, + cwd: config.agent.cwd, + log: () => {}, + onReply: async (contextToken, text) => { + replies.push({ contextToken, text }); + }, + onMediaReply: async () => {}, + sendTyping: async () => {}, + }); + manager.currentTurn = { + sessionId: "ses-y", + userMessageId: "msg-2", + assistantMessageId: null, + parts: new Map(), + textBuffer: "", + finalText: "", + toolCalls: new Map(), + hasBackgroundTasks: false, + contextToken: "ctx-silent", + hint: null, + status: "accumulating", + startedAt: Date.now(), + lastEventAt: Date.now(), + retried: false, + sentTextPartIds: new Set(), + pendingTextParts: [], + pendingReasoningParts: [], + showThoughtsSnapshot: false, + showToolsSnapshot: false, + immersiveSnapshot: false, + immersiveLastText: "", + reasoningCharCount: 0, + reasoningStartMs: null, + reasoningEndMs: null, + sentReasoningPartIds: new Set(), + reasoningPartTimestamps: new Map(), + currentPartType: null, + currentPartID: null, + currentReasoningText: "", + currentReasoningStartMs: null, + currentReasoningEndMs: null, + currentText: "", + currentToolKey: null, + }; + manager.finalizeTurn("interrupted"); + expect(replies).toHaveLength(0); + } finally { + vi.useRealTimers(); + } + }); +}); diff --git a/src/bridge.ts b/src/bridge.ts index 4d73af1..a5c5c8e 100644 --- a/src/bridge.ts +++ b/src/bridge.ts @@ -229,6 +229,13 @@ export class WeChatOpencodeBridge { private outboundQueue = new Map>(); private static readonly MSG_LIMIT_WARN = 7; private static readonly MSG_LIMIT_MAX = 10; + /** + * Auto-flush ("auto /next") timer. Armed on the first cache hit while + * `pendingOutbound` is non-empty; fires once and re-arms if the 10-msg + * limit left messages cached. `null` when disabled (`autoFlushMs <= 0`) + * or while a timer is pending. + */ + private autoFlushTimer: ReturnType | null = null; private log: (msg: string) => void; private restartServer?: () => Promise; @@ -329,6 +336,16 @@ export class WeChatOpencodeBridge { // further escalation available when the WeChat gateway itself // is unreachable). onSendWarning: (contextToken, text) => this.sendWarningReply(contextToken, text), + // System-level notices (stuck-turn detection, SSE pipeline + // outage/recovery). Delivered through the normal outbound queue so + // they can't reorder with an agent reply; best-effort on failure. + // An empty contextToken (e.g. WeChat-channel notices that don't + // carry one) falls back to the most recent user message's token. + onNotify: (contextToken, text) => { + const ctx = contextToken || this.currentContextToken; + if (!ctx) return Promise.resolve(); + return this.sendReply(ctx, text); + }, onMediaReply: (contextToken, blocks) => this.sendMediaReply(contextToken, blocks), sendTyping: (contextToken) => this.sendTypingIndicator(contextToken), cancelTyping: (contextToken) => this.cancelTypingIndicator(contextToken), @@ -445,6 +462,7 @@ export class WeChatOpencodeBridge { // notify text is semantically the same as cached agent text // for delivery purposes. this.pendingOutbound.push({ kind: "text", text, contextToken: "" }); + this.scheduleAutoFlush(""); return Promise.resolve(); }, log: (msg) => this.log(msg), @@ -548,11 +566,19 @@ export class WeChatOpencodeBridge { abortSignal: this.abortController.signal, log: this.log, onMessage: (msg) => this.handleMessage(msg), + // WeChat-channel condition notices (session expiry, poll failures, + // recovery). No contextToken of their own — the bridge falls back + // to the most recent user message's token for delivery. + onNotify: (text) => this.sendReply(this.currentContextToken ?? "", text), }); } async stop(): Promise { this.log("Stopping bridge..."); + if (this.autoFlushTimer) { + clearTimeout(this.autoFlushTimer); + this.autoFlushTimer = null; + } this.abortController.abort(); // Stop the SSE event pipeline before tearing down the session manager. if (this.sessionManager) { @@ -721,12 +747,13 @@ export class WeChatOpencodeBridge { const remaining = segments.slice(segments.indexOf(segment)); const cached: PendingMessage[] = remaining.map((s) => ({ kind: "tool_text", text: s, contextToken })); this.pendingOutbound = [...this.pendingOutbound, ...cached]; + this.scheduleAutoFlush(contextToken); break; } this.wechatMsgCount++; let payload = segment; if (this.wechatMsgCount > WeChatOpencodeBridge.MSG_LIMIT_WARN) { - payload += `\n\n⚠️ 微信限制连续发送消息数量10条（已发 ${this.wechatMsgCount} 条），发送 /next 可重置`; + payload += `\n\n⚠️ 微信限制连续发送消息数量10条（已发 ${this.wechatMsgCount} 条）${this.msgLimitNotice()}`; } await sendTextMessage(targetUserId, payload, { baseUrl: this.tokenData!.baseUrl, @@ -744,6 +771,7 @@ export class WeChatOpencodeBridge { this.log(`[tool-api] WeChat 10-msg limit reached (sent=${this.wechatMsgCount}), caching file`); const fName = path.basename(filePath); this.pendingOutbound.push({ kind: "tool_file", filePath, fileName: fName, mimeType, contextToken }); + this.scheduleAutoFlush(contextToken); } else { this.wechatMsgCount++; const fileBuffer = await fs.promises.readFile(filePath); @@ -2650,6 +2678,43 @@ const apCmd = parseAutoPermissionCommand(trimmed); await this.sessionManager!.enqueue(parts, contextToken); } + /** + * Arm the auto-flush ("auto /next") timer, if enabled and not already + * pending. Called whenever messages are cached into `pendingOutbound` + * (outbound send limit hit). When the timer fires, `runAutoFlush` + * delivers the cached messages without requiring a user message or a + * manual `/next`. + */ + private scheduleAutoFlush(contextToken: string): void { + const delayMs = this.config.wechat.autoFlushMs; + if (!delayMs || delayMs <= 0 || this.autoFlushTimer) return; + this.autoFlushTimer = setTimeout(() => { + this.autoFlushTimer = null; + this.runAutoFlush(contextToken).catch((err: unknown) => { + this.log(`auto-flush error: ${String(err)}`); + }); + }, delayMs); + } + + /** + * Auto-flush body. Waits for any in-flight outbound send on the same + * contextToken to settle (so the flush doesn't interleave with an agent + * reply and reorder messages on WeChat), then runs the same flush the + * manual `/next` command uses. If the 10-msg limit still left messages + * cached, re-arms the timer for another round instead of waiting for the + * user. + */ + private async runAutoFlush(contextToken: string): Promise { + if (this.pendingOutbound.length === 0) return; + const inflight = this.outboundQueue.get(contextToken); + if (inflight) await inflight.catch(() => {}); + if (this.pendingOutbound.length === 0) return; + await this.flushPending(contextToken); + if (this.pendingOutbound.length > 0) { + this.scheduleAutoFlush(contextToken); + } + } + private async flushPending(contextToken: string): Promise { if (this.pendingOutbound.length === 0) { await this.sendReply(contextToken, "✅ No cached messages"); @@ -2673,7 +2738,7 @@ const apCmd = parseAutoPermissionCommand(trimmed); const text = msg.kind === "text" || msg.kind === "tool_text" ? msg.text : ""; const payload = text && this.wechatMsgCount > WeChatOpencodeBridge.MSG_LIMIT_WARN - ? text + `\n\n⚠️ 微信限制连续发送消息数量10条（已发 ${this.wechatMsgCount} 条），发送 /next 可重置` + ? text + `\n\n⚠️ 微信限制连续发送消息数量10条（已发 ${this.wechatMsgCount} 条）${this.msgLimitNotice()}` : text; switch (msg.kind) { @@ -2732,6 +2797,7 @@ const apCmd = parseAutoPermissionCommand(trimmed); if (remaining.length > 0) { this.pendingOutbound = remaining; await this.sendReply(contextToken, `✅ Sent ${sent}, ${remaining.length} cached, /next to continue`); + this.scheduleAutoFlush(contextToken); } else { this.pendingOutbound = []; await this.sendReply(contextToken, `✅ All ${sent} cached messages sent`); @@ -2742,6 +2808,18 @@ const apCmd = parseAutoPermissionCommand(trimmed); return this.enqueueOutbound(contextToken, () => this.sendReplyImpl(contextToken, text)); } + /** + * Suffix for the WeChat 10-msg-limit warning. When auto-flush is enabled + * the user doesn't need to do anything — remaining messages are delivered + * automatically — so we say so instead of asking for a manual `/next`. + */ + private msgLimitNotice(): string { + const autoFlushMs = this.config.wechat.autoFlushMs; + return autoFlushMs && autoFlushMs > 0 + ? "（其余消息将自动续发）" + : "（发送 /next 可重置）"; + } + private async sendReplyImpl(contextToken: string, text: string): Promise { const formatted = formatForWeChat(text); const segments = splitText(formatted, TEXT_CHUNK_LIMIT); @@ -2752,13 +2830,14 @@ const apCmd = parseAutoPermissionCommand(trimmed); const remaining = segments.slice(segments.indexOf(segment)); const cached: PendingMessage[] = remaining.map((s) => ({ kind: "text", text: s, contextToken })); this.pendingOutbound = [...this.pendingOutbound, ...cached]; + this.scheduleAutoFlush(contextToken); break; } this.wechatMsgCount++; let payload = segment; if (this.wechatMsgCount > WeChatOpencodeBridge.MSG_LIMIT_WARN) { - payload += `\n\n⚠️ 微信限制连续发送消息数量10条（已发 ${this.wechatMsgCount} 条），发送 /next 可重置`; + payload += `\n\n⚠️ 微信限制连续发送消息数量10条（已发 ${this.wechatMsgCount} 条）${this.msgLimitNotice()}`; } await sendTextMessage(this.userState?.userId ?? "", payload, { @@ -2810,6 +2889,7 @@ const apCmd = parseAutoPermissionCommand(trimmed); const remaining = blocks.slice(blocks.indexOf(block)); const cached: PendingMessage[] = remaining.map((b) => ({ kind: "media", block: b, contextToken })); this.pendingOutbound = [...this.pendingOutbound, ...cached]; + this.scheduleAutoFlush(contextToken); break; } diff --git a/src/config.ts b/src/config.ts index 6f6aa3c..041595c 100644 --- a/src/config.ts +++ b/src/config.ts @@ -58,6 +58,21 @@ export interface WeChatOpencodeConfig { baseUrl: string; cdnBaseUrl: string; botType: string; + /** + * Auto-flush ("auto /next") delay in milliseconds. + * + * When the bridge hits WeChat's 10-consecutive-message send limit, + * remaining outbound segments are cached in `pendingOutbound` and + * normally only delivered on the next user message or an explicit + * `/next`. With `autoFlushMs` > 0, the bridge additionally starts a + * timer after the first cache hit and flushes the cached messages + * automatically once the delay elapses — no user message required. + * If the 10-msg limit still holds after a flush (more than 10 + * segments were cached), the timer is re-armed for another round. + * + * Set to 0 to disable and keep the manual `/next`-only behavior. + */ + autoFlushMs: number; }; /** OpenCode Server connection config. */ server: ServerConfig; @@ -165,6 +180,7 @@ export function defaultConfig(): WeChatOpencodeConfig { baseUrl: "https://ilinkai.weixin.qq.com", cdnBaseUrl: "https://novac2c.cdn.weixin.qq.com/c2c", botType: "3", + autoFlushMs: 60_000, }, server: { url: "http://localhost:4096", diff --git a/src/server/session.ts b/src/server/session.ts index 2e61446..6ea0d72 100644 --- a/src/server/session.ts +++ b/src/server/session.ts @@ -51,6 +51,7 @@ import type { TextPart, ToolPart, TrackedTool, + EventPipelineStatus, } from "../types/events.js"; import type { PendingQuestion, @@ -68,8 +69,14 @@ import { DEFAULT_AUTO_PERMISSION_MODE } from "../types/permission.js"; /** Idle debounce: wait this long after the last delta before considering the turn final. */ const TURN_FINALIZE_DEBOUNCE_MS = 500; -/** Hard ceiling: if no event for this long, force-finalize the turn. */ const TURN_STUCK_TIMEOUT_MS = 5 * 60_000; +/** + * How long the SSE event pipeline may sit in `connecting`/`reconnecting` + * before the user is told the agent's event stream is down (via + * `onNotify`). The pipeline retries with exponential backoff on its own; + * this is purely a user-visible warning, not a recovery mechanism. + */ +const PIPELINE_ISSUE_NOTIFY_MS = 60_000; /** * Soft timeout for unanswered `question.asked` events. After this many ms * with no user reply, SessionManager auto-rejects the question (POST @@ -236,6 +243,20 @@ export interface SessionManagerOpts { * (the pre-retry behavior). */ onSendWarning?: (contextToken: string, text: string) => Promise; + /** + * Invoked when the SessionManager needs to surface a system-level + * condition to the WeChat user that is neither a reply nor a send + * failure, e.g.: + * - the 5-minute stuck timeout fired (the agent produced no events + * for too long and the turn was force-finalized — the user was + * probably staring at a conversation that silently went nowhere) + * - the SSE event pipeline lost its connection and has been + * reconnecting for a while, then recovered + * + * The bridge should push `text` to the user's WeChat chat + * (best-effort; failures are logged, not thrown). + */ + onNotify?: (contextToken: string, text: string) => Promise | void; } interface QueueItem { @@ -441,6 +462,14 @@ export class SessionManager { private finalizeTimer: ReturnType | null = null; /** Hard timeout timer in case events stop arriving entirely. */ private stuckTimer: ReturnType | null = null; + /** + * Timestamp (ms) when the SSE pipeline entered `connecting` / + * `reconnecting`. `null` while the pipeline is connected (or stopped). + * Powers the "event stream down" warning via `onNotify`. + */ + private pipelineIssueAt: number | null = null; + /** True once the pipeline-down warning has been sent for the current outage. */ + private pipelineIssueNotified = false; /** * Session-level dedup map for the tool summary: `callID` → last status * that was emitted to WeChat for that tool. Survives turn finalization @@ -540,6 +569,8 @@ export class SessionManager { * sees `⚠️ 上一条