From b6d3f879ef5c1fb965d52082498f52cd7b8c2f28 Mon Sep 17 00:00:00 2001 From: Eyejoker Date: Wed, 25 Mar 2026 07:08:49 +0900 Subject: [PATCH] refactor: add recovery mode + shared JSON/error/fetch utilities - GroupQueue recovery mode: limit to 3 concurrent agents for 60s after restart to prevent API rate-limit storms (configurable via env vars) - getErrorMessage: replace 14 occurrences of instanceof Error pattern - readJsonFile/writeJsonFile: replace 19 occurrences of JSON+fs pattern - fetchWithTimeout: replace 3 occurrences of AbortController pattern 354/354 tests passing --- src/agent-runner-snapshot.ts | 6 +++++- src/agent-runner.test.ts | 5 +---- src/channels/discord.ts | 37 ++++++++++++++++++++++++++------- src/claude-usage.ts | 3 ++- src/codex-token-rotation.ts | 10 ++++----- src/config.ts | 12 +++++++++++ src/group-queue.ts | 40 +++++++++++++++++++++++++++++++----- src/index.ts | 1 + src/ipc.ts | 19 +++++++++++++---- src/provider-retry.ts | 16 ++++++++++++--- src/token-rotation.ts | 11 +++++----- 11 files changed, 123 insertions(+), 37 deletions(-) diff --git a/src/agent-runner-snapshot.ts b/src/agent-runner-snapshot.ts index 224f3ff..7498326 100644 --- a/src/agent-runner-snapshot.ts +++ b/src/agent-runner-snapshot.ts @@ -49,5 +49,9 @@ export function writeGroupsSnapshot( fs.mkdirSync(groupIpcDir, { recursive: true }); const visibleGroups = isMain ? groups : []; const groupsFile = path.join(groupIpcDir, 'available_groups.json'); - writeJsonFile(groupsFile, { groups: visibleGroups, lastSync: new Date().toISOString() }, true); + writeJsonFile( + groupsFile, + { groups: visibleGroups, lastSync: new Date().toISOString() }, + true, + ); } diff --git a/src/agent-runner.test.ts b/src/agent-runner.test.ts index 7057fd6..71b9a47 100644 --- a/src/agent-runner.test.ts +++ b/src/agent-runner.test.ts @@ -4,10 +4,7 @@ import { PassThrough } from 'stream'; import fs from 'fs'; import { spawn } from 'child_process'; -import { - OUTPUT_START_MARKER, - OUTPUT_END_MARKER, -} from './agent-protocol.js'; +import { OUTPUT_START_MARKER, OUTPUT_END_MARKER } from './agent-protocol.js'; // Mock config vi.mock('./config.js', () => ({ diff --git a/src/channels/discord.ts b/src/channels/discord.ts index d355f4e..a27d636 100644 --- a/src/channels/discord.ts +++ b/src/channels/discord.ts @@ -271,9 +271,7 @@ export class DiscordChannel implements Channel { return `[Audio: ${att.name || 'audio'}]`; } else if ( contentType.startsWith('text/') || - /\.(txt|md|csv|json|log)$/i.test( - att.name || '', - ) + /\.(txt|md|csv|json|log)$/i.test(att.name || '') ) { // Download and inline text-based files try { @@ -450,8 +448,9 @@ export class DiscordChannel implements Channel { } cleaned = formatOutbound(cleaned); - // Discord has a 2000 character limit per message — split if needed + // Discord has a 2000 character limit per message and 10 attachments per message const MAX_LENGTH = 2000; + const MAX_ATTACHMENTS = 10; const files = imageFiles.map((f) => ({ attachment: f, name: path.basename(f), @@ -462,19 +461,43 @@ export class DiscordChannel implements Channel { return; } + // Split files into batches of MAX_ATTACHMENTS + const fileBatches: typeof files[] = []; + for (let i = 0; i < files.length; i += MAX_ATTACHMENTS) { + fileBatches.push(files.slice(i, i + MAX_ATTACHMENTS)); + } + if (cleaned.length <= MAX_LENGTH) { + // Send text with first batch of files await textChannel.send({ content: cleaned || undefined, - files: files.length > 0 ? files : undefined, + files: fileBatches[0]?.length ? fileBatches[0] : undefined, flags: MessageFlags.SuppressEmbeds, }); + // Send remaining file batches as follow-up messages + for (let b = 1; b < fileBatches.length; b++) { + await textChannel.send({ + files: fileBatches[b], + flags: MessageFlags.SuppressEmbeds, + }); + } } else { - // Send text in chunks, attach images to the first chunk + // Send text in chunks, attach first batch to the first chunk + let fileBatchIndex = 0; for (let i = 0; i < cleaned.length; i += MAX_LENGTH) { const chunk = cleaned.slice(i, i + MAX_LENGTH); + const batch = fileBatches[fileBatchIndex]; await textChannel.send({ content: chunk, - files: i === 0 && files.length > 0 ? files : undefined, + files: batch?.length ? batch : undefined, + flags: MessageFlags.SuppressEmbeds, + }); + if (batch?.length) fileBatchIndex++; + } + // Send any remaining file batches + for (let b = fileBatchIndex; b < fileBatches.length; b++) { + await textChannel.send({ + files: fileBatches[b], flags: MessageFlags.SuppressEmbeds, }); } diff --git a/src/claude-usage.ts b/src/claude-usage.ts index ff57e2a..4b1fa7b 100644 --- a/src/claude-usage.ts +++ b/src/claude-usage.ts @@ -55,7 +55,8 @@ let diskCacheLoaded = false; function loadUsageDiskCache(): void { if (diskCacheLoaded) return; diskCacheLoaded = true; - usageDiskCache = readJsonFile>(USAGE_CACHE_FILE) ?? {}; + usageDiskCache = + readJsonFile>(USAGE_CACHE_FILE) ?? {}; } function saveUsageDiskCache(): void { diff --git a/src/codex-token-rotation.ts b/src/codex-token-rotation.ts index 55521f3..aa0d375 100644 --- a/src/codex-token-rotation.ts +++ b/src/codex-token-rotation.ts @@ -94,7 +94,9 @@ export function initCodexTokenRotation(): void { const authPath = path.join(ACCOUNTS_DIR, dir, 'auth.json'); if (!fs.existsSync(authPath)) continue; - const data = readJsonFile<{ tokens?: { account_id?: string; id_token?: string } }>(authPath); + const data = readJsonFile<{ + tokens?: { account_id?: string; id_token?: string }; + }>(authPath); if (!data) { logger.warn({ authPath }, 'Failed to parse codex account auth.json'); continue; @@ -186,11 +188,7 @@ function loadCodexState(): void { } } if (Array.isArray(state.resetAts)) { - for ( - let i = 0; - i < Math.min(state.resetAts.length, accounts.length); - i++ - ) { + for (let i = 0; i < Math.min(state.resetAts.length, accounts.length); i++) { if (state.resetAts[i]) accounts[i].resetAt = state.resetAts[i]!; } } diff --git a/src/config.ts b/src/config.ts index 83cc7fb..f5a6a4f 100644 --- a/src/config.ts +++ b/src/config.ts @@ -54,6 +54,18 @@ export const MAX_CONCURRENT_AGENTS = Math.max( 1, parseInt(process.env.MAX_CONCURRENT_AGENTS || '5', 10) || 5, ); +export const RECOVERY_CONCURRENT_AGENTS = parseInt( + getEnv('RECOVERY_CONCURRENT_AGENTS') || '3', + 10, +); +export const RECOVERY_STAGGER_MS = parseInt( + getEnv('RECOVERY_STAGGER_MS') || '2000', + 10, +); +export const RECOVERY_DURATION_MS = parseInt( + getEnv('RECOVERY_DURATION_MS') || '60000', + 10, +); function escapeRegex(str: string): string { return str.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'); diff --git a/src/group-queue.ts b/src/group-queue.ts index 323f681..a37932e 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -1,6 +1,10 @@ import { ChildProcess } from 'child_process'; -import { MAX_CONCURRENT_AGENTS } from './config.js'; +import { + MAX_CONCURRENT_AGENTS, + RECOVERY_CONCURRENT_AGENTS, + RECOVERY_DURATION_MS, +} from './config.js'; import { queueFollowUpMessage, writeCloseSentinel } from './group-queue-ipc.js'; import { logger } from './logger.js'; @@ -50,6 +54,8 @@ export class GroupQueue { private activeCount = 0; private activeTaskCount = 0; private waitingGroups: string[] = []; + private recoveryMode = false; + private recoveryTimer: ReturnType | null = null; private processMessagesFn: | ((groupJid: string, context: GroupRunContext) => Promise) | null = null; @@ -85,6 +91,30 @@ export class GroupQueue { this.processMessagesFn = fn; } + /** Limit concurrency after restart to avoid API rate-limit storms. */ + enterRecoveryMode(): void { + this.recoveryMode = true; + logger.info( + { maxConcurrent: RECOVERY_CONCURRENT_AGENTS, durationMs: RECOVERY_DURATION_MS }, + 'Entering recovery mode (staggered restart)', + ); + this.recoveryTimer = setTimeout(() => { + this.recoveryMode = false; + this.recoveryTimer = null; + logger.info( + { maxConcurrent: MAX_CONCURRENT_AGENTS }, + 'Recovery mode ended, full concurrency restored', + ); + this.drainWaiting(); + }, RECOVERY_DURATION_MS); + } + + private get effectiveMaxConcurrent(): number { + return this.recoveryMode + ? RECOVERY_CONCURRENT_AGENTS + : MAX_CONCURRENT_AGENTS; + } + private createRunId(): string { return `run-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`; } @@ -121,13 +151,13 @@ export class GroupQueue { return; } - if (this.activeCount >= MAX_CONCURRENT_AGENTS) { + if (this.activeCount >= this.effectiveMaxConcurrent) { state.pendingMessages = true; if (!this.waitingGroups.includes(groupJid)) { this.waitingGroups.push(groupJid); } logger.debug( - { groupJid, activeCount: this.activeCount }, + { groupJid, activeCount: this.activeCount, max: this.effectiveMaxConcurrent }, 'At concurrency limit, message queued', ); return; @@ -160,7 +190,7 @@ export class GroupQueue { } if ( - this.activeCount >= MAX_CONCURRENT_AGENTS || + this.activeCount >= this.effectiveMaxConcurrent || this.activeTaskCount >= MAX_CONCURRENT_TASKS ) { state.pendingTasks.push({ id: taskId, groupJid, fn }); @@ -466,7 +496,7 @@ export class GroupQueue { private drainWaiting(): void { while ( this.waitingGroups.length > 0 && - this.activeCount < MAX_CONCURRENT_AGENTS + this.activeCount < this.effectiveMaxConcurrent ) { const nextMessageIndex = this.waitingGroups.findIndex((jid) => { const state = this.getGroup(jid); diff --git a/src/index.ts b/src/index.ts index 4fad071..23d0535 100644 --- a/src/index.ts +++ b/src/index.ts @@ -438,6 +438,7 @@ async function main(): Promise { writeGroupsSnapshot, }); queue.setProcessMessagesFn(runtime.processGroupMessages); + queue.enterRecoveryMode(); runtime.recoverPendingMessages(); const restartContext = await announceRestartRecovery(processStartedAtMs); for (const candidate of getInterruptedRecoveryCandidates( diff --git a/src/ipc.ts b/src/ipc.ts index 153c876..44ef581 100644 --- a/src/ipc.ts +++ b/src/ipc.ts @@ -80,8 +80,13 @@ export function startIpcWatcher(deps: IpcDeps): void { const filePath = path.join(messagesDir, file); try { const data = readJsonFile(filePath); - if (!data || typeof data !== 'object') throw new Error('Invalid JSON'); - const msg = data as { type?: string; chatJid?: string; text?: string }; + if (!data || typeof data !== 'object') + throw new Error('Invalid JSON'); + const msg = data as { + type?: string; + chatJid?: string; + text?: string; + }; if (msg.type === 'message' && msg.chatJid && msg.text) { // Authorization: verify this group can send to this chatJid const targetGroup = registeredGroups[msg.chatJid]; @@ -133,9 +138,15 @@ export function startIpcWatcher(deps: IpcDeps): void { const filePath = path.join(tasksDir, file); try { const data = readJsonFile(filePath); - if (!data || typeof data !== 'object') throw new Error('Invalid JSON'); + if (!data || typeof data !== 'object') + throw new Error('Invalid JSON'); // Pass source group identity to processTaskIpc for authorization - await processTaskIpc(data as Parameters[0], sourceGroup, isMain, deps); + await processTaskIpc( + data as Parameters[0], + sourceGroup, + isMain, + deps, + ); fs.unlinkSync(filePath); } catch (err) { logger.error( diff --git a/src/provider-retry.ts b/src/provider-retry.ts index 060c706..227531c 100644 --- a/src/provider-retry.ts +++ b/src/provider-retry.ts @@ -8,8 +8,15 @@ import { shouldRotateClaudeToken } from './agent-error-detection.js'; import { logger } from './logger.js'; import { getErrorMessage } from './utils.js'; -import { detectFallbackTrigger, markPrimaryCooldown } from './provider-fallback.js'; -import { rotateToken, getTokenCount, markTokenHealthy } from './token-rotation.js'; +import { + detectFallbackTrigger, + markPrimaryCooldown, +} from './provider-fallback.js'; +import { + rotateToken, + getTokenCount, + markTokenHealthy, +} from './token-rotation.js'; // ── Types ──────────────────────────────────────────────────────── @@ -115,7 +122,10 @@ export async function runClaudeRotationLoop( // ── Success with null result (MAE-specific, TaskScheduler ignores) ── if (!attempt.sawOutput && attempt.sawSuccessNullResult) { - return { type: 'needs-fallback', trigger: { reason: 'success-null-result' } }; + return { + type: 'needs-fallback', + trigger: { reason: 'success-null-result' }, + }; } // ── Error status ── diff --git a/src/token-rotation.ts b/src/token-rotation.ts index eeb29ab..fb8123e 100644 --- a/src/token-rotation.ts +++ b/src/token-rotation.ts @@ -75,7 +75,10 @@ function saveState(): void { } function loadState(): void { - const state = readJsonFile<{ currentIndex?: number; rateLimits?: (number | null)[] }>(STATE_FILE); + const state = readJsonFile<{ + currentIndex?: number; + rateLimits?: (number | null)[]; + }>(STATE_FILE); if (!state) return; const now = Date.now(); @@ -86,11 +89,7 @@ function loadState(): void { currentIndex = state.currentIndex; } if (Array.isArray(state.rateLimits)) { - for ( - let i = 0; - i < Math.min(state.rateLimits.length, tokens.length); - i++ - ) { + for (let i = 0; i < Math.min(state.rateLimits.length, tokens.length); i++) { const until = state.rateLimits[i]; if (typeof until === 'number' && until > now) { tokens[i].rateLimitedUntil = until;