diff --git a/.gitignore b/.gitignore index 713bd89..5d89d6f 100644 --- a/.gitignore +++ b/.gitignore @@ -26,6 +26,7 @@ groups/global/* *.keys.json .env .env.codex +.env.minimax # Temp files .tmp-* diff --git a/package.json b/package.json index e913b09..5f83655 100644 --- a/package.json +++ b/package.json @@ -7,6 +7,7 @@ "scripts": { "build": "tsc", "build:runners": "npm --prefix runners/agent-runner install && npm --prefix runners/agent-runner run build && npm --prefix runners/codex-runner install && npm --prefix runners/codex-runner run build", + "restart:hint": "tsx src/restart-context-cli.ts write", "start": "node dist/index.js", "dev": "tsx src/index.ts", "typecheck": "tsc --noEmit", diff --git a/src/agent-runner.ts b/src/agent-runner.ts index f722ede..6ebe80f 100644 --- a/src/agent-runner.ts +++ b/src/agent-runner.ts @@ -156,6 +156,8 @@ function prepareGroupEnvironment( // Build environment variables for the runner process const envVars = readEnvFile([ 'ANTHROPIC_API_KEY', + 'ANTHROPIC_AUTH_TOKEN', + 'ANTHROPIC_BASE_URL', 'CLAUDE_CODE_OAUTH_TOKEN', 'CLAUDE_MODEL', 'CLAUDE_THINKING', @@ -340,6 +342,8 @@ args = [${JSON.stringify(mementoSseUrl)}, "--header", ${JSON.stringify(`Authoriz // Sanitize secrets: prevent API keys from leaking to codex subprocesses delete env.ANTHROPIC_API_KEY; + delete env.ANTHROPIC_AUTH_TOKEN; + delete env.ANTHROPIC_BASE_URL; delete env.CLAUDE_CODE_OAUTH_TOKEN; env.CODEX_HOME = sessionCodexDir; @@ -349,6 +353,19 @@ args = [${JSON.stringify(mementoSseUrl)}, "--header", ${JSON.stringify(`Authoriz env.ANTHROPIC_API_KEY = envVars.ANTHROPIC_API_KEY || process.env.ANTHROPIC_API_KEY || ''; } + if ( + envVars.ANTHROPIC_AUTH_TOKEN || + process.env.ANTHROPIC_AUTH_TOKEN + ) { + env.ANTHROPIC_AUTH_TOKEN = + envVars.ANTHROPIC_AUTH_TOKEN || + process.env.ANTHROPIC_AUTH_TOKEN || + ''; + } + if (envVars.ANTHROPIC_BASE_URL || process.env.ANTHROPIC_BASE_URL) { + env.ANTHROPIC_BASE_URL = + envVars.ANTHROPIC_BASE_URL || process.env.ANTHROPIC_BASE_URL || ''; + } if ( envVars.CLAUDE_CODE_OAUTH_TOKEN || process.env.CLAUDE_CODE_OAUTH_TOKEN @@ -384,6 +401,7 @@ export async function runAgentProcess( input: AgentInput, onProcess: (proc: ChildProcess, processName: string) => void, onOutput?: (output: AgentOutput) => Promise, + envOverrides?: Record, ): Promise { const startTime = Date.now(); const { env, groupDir, runnerDir } = prepareGroupEnvironment( @@ -391,6 +409,13 @@ export async function runAgentProcess( input.isMain, input.chatJid, ); + + // Apply provider fallback overrides (e.g. Kimi env vars when Claude is in cooldown) + if (envOverrides) { + for (const [key, value] of Object.entries(envOverrides)) { + if (value) env[key] = value; + } + } if (input.runId) { env.NANOCLAW_RUN_ID = input.runId; } diff --git a/src/channels/discord.test.ts b/src/channels/discord.test.ts index 3687772..04b4a9a 100644 --- a/src/channels/discord.test.ts +++ b/src/channels/discord.test.ts @@ -761,7 +761,7 @@ describe('DiscordChannel', () => { expect(currentClient().channels.fetch).toHaveBeenCalledWith('9876543210'); }); - it('handles send failure gracefully', async () => { + it('propagates send failure to the caller', async () => { const opts = createTestOpts(); const channel = new DiscordChannel('test-token', opts); await channel.connect(); @@ -770,10 +770,9 @@ describe('DiscordChannel', () => { new Error('Channel not found'), ); - // Should not throw await expect( channel.sendMessage('dc:1234567890123456', 'Will fail'), - ).resolves.toBeUndefined(); + ).rejects.toThrow('Channel not found'); }); it('does nothing when client is not initialized', async () => { diff --git a/src/channels/discord.ts b/src/channels/discord.ts index ec5f9c1..8a76773 100644 --- a/src/channels/discord.ts +++ b/src/channels/discord.ts @@ -484,6 +484,7 @@ export class DiscordChannel implements Channel { logger.info({ jid, length: text.length }, 'Discord message sent'); } catch (err) { logger.error({ jid, err }, 'Failed to send Discord message'); + throw err; } } @@ -551,7 +552,7 @@ export class DiscordChannel implements Channel { return msg.id; } catch (err) { logger.error({ jid, err }, 'Failed to send tracked Discord message'); - return null; + throw err; } } diff --git a/src/db.test.ts b/src/db.test.ts index e45f9c2..28ff00d 100644 --- a/src/db.test.ts +++ b/src/db.test.ts @@ -3,17 +3,24 @@ import { describe, it, expect, beforeEach } from 'vitest'; import { _initTestDatabase, createTask, + createProducedWorkItem, deleteSession, deleteTask, getAllChats, getAllRegisteredGroups, getDueTasks, + getLatestMessageSeqAtOrBefore, + getMessagesSinceSeq, + getNewMessagesBySeq, + getOpenWorkItem, getRegisteredAgentTypesForJid, getMessagesSince, getNewMessages, isPairedRoomJid, getSession, getTaskById, + markWorkItemDelivered, + markWorkItemDeliveryRetry, setSession, setRegisteredGroup, storeChatMetadata, @@ -412,6 +419,7 @@ describe('task CRUD', () => { id: 'task-claude', group_folder: 'main', chat_jid: 'group@g.us', + agent_type: 'claude-code', prompt: 'claude task', schedule_type: 'once', schedule_value: dueAt, @@ -595,3 +603,81 @@ describe('paired room registration', () => { expect(isPairedRoomJid('dc:solo')).toBe(false); }); }); + +describe('message seq cursors', () => { + beforeEach(() => { + storeChatMetadata('group@g.us', '2024-01-01T00:00:00.000Z'); + store({ + id: 'seq-1', + chat_jid: 'group@g.us', + sender: 'alice', + sender_name: 'Alice', + content: 'first', + timestamp: '2024-01-01T00:00:01.000Z', + }); + store({ + id: 'seq-2', + chat_jid: 'group@g.us', + sender: 'bob', + sender_name: 'Bob', + content: 'second', + timestamp: '2024-01-01T00:00:02.000Z', + }); + store({ + id: 'seq-3', + chat_jid: 'group@g.us', + sender: 'carol', + sender_name: 'Carol', + content: 'third', + timestamp: '2024-01-01T00:00:03.000Z', + }); + }); + + it('assigns monotonic seq values and preserves them on upsert', () => { + const { messages } = getNewMessagesBySeq(['group@g.us'], 0, 'Andy'); + expect(messages.map((m) => m.seq)).toEqual([1, 2, 3]); + + store({ + id: 'seq-2', + chat_jid: 'group@g.us', + sender: 'bob', + sender_name: 'Bob', + content: 'second updated', + timestamp: '2024-01-01T00:00:02.500Z', + }); + + const afterUpdate = getMessagesSinceSeq('group@g.us', 0, 'Andy'); + expect(afterUpdate.map((m) => m.seq)).toEqual([1, 2, 3]); + expect(afterUpdate[1].content).toBe('second updated'); + }); + + it('maps legacy timestamp cursors to the latest seq at or before that time', () => { + expect( + getLatestMessageSeqAtOrBefore('2024-01-01T00:00:02.000Z', 'group@g.us'), + ).toBe(2); + }); +}); + +describe('work items', () => { + it('tracks produced, retry, and delivered states', () => { + const item = createProducedWorkItem({ + group_folder: 'discord_test', + chat_jid: 'dc:123', + agent_type: 'claude-code', + start_seq: 10, + end_seq: 12, + result_payload: 'hello', + }); + + expect(getOpenWorkItem('dc:123', 'claude-code')?.id).toBe(item.id); + + markWorkItemDeliveryRetry(item.id, 'send failed'); + const retried = getOpenWorkItem('dc:123', 'claude-code'); + expect(retried?.status).toBe('delivery_retry'); + expect(retried?.delivery_attempts).toBe(1); + expect(retried?.last_error).toBe('send failed'); + + markWorkItemDelivered(item.id, 'msg-1'); + expect(getOpenWorkItem('dc:123', 'claude-code')).toBeUndefined(); + }); +}); diff --git a/src/db.ts b/src/db.ts index c36fc55..218b5aa 100644 --- a/src/db.ts +++ b/src/db.ts @@ -21,6 +21,61 @@ import { let db: Database.Database; +export interface WorkItem { + id: number; + group_folder: string; + chat_jid: string; + agent_type: AgentType; + status: 'produced' | 'delivery_retry' | 'delivered'; + start_seq: number | null; + end_seq: number | null; + result_payload: string; + delivery_attempts: number; + delivery_message_id: string | null; + last_error: string | null; + created_at: string; + updated_at: string; + delivered_at: string | null; +} + +function backfillMessageSeq(database: Database.Database): void { + const rows = database + .prepare( + `SELECT rowid, seq + FROM messages + ORDER BY CASE WHEN seq IS NULL THEN 1 ELSE 0 END, seq, timestamp, rowid`, + ) + .all() as Array<{ rowid: number; seq: number | null }>; + + if (rows.length === 0) { + return; + } + + let nextSeq = 1; + const assignSeq = database.prepare( + 'UPDATE messages SET seq = ? WHERE rowid = ? AND seq IS NULL', + ); + const tx = database.transaction(() => { + for (const row of rows) { + if (row.seq === null) { + assignSeq.run(nextSeq, row.rowid); + } + nextSeq = Math.max(nextSeq, (row.seq ?? nextSeq) + 1); + } + }); + tx(); + + const maxSeqRow = database + .prepare('SELECT MAX(seq) AS maxSeq FROM messages') + .get() as { maxSeq: number | null }; + const maxSeq = maxSeqRow.maxSeq ?? 0; + if (maxSeq > 0) { + database + .prepare('INSERT OR IGNORE INTO message_sequence (id) VALUES (?)') + .run(maxSeq); + } +} + function createSchema(database: Database.Database): void { database.exec(` CREATE TABLE IF NOT EXISTS chats ( @@ -37,12 +92,39 @@ function createSchema(database: Database.Database): void { sender_name TEXT, content TEXT, timestamp TEXT, + seq INTEGER, is_from_me INTEGER, is_bot_message INTEGER DEFAULT 0, PRIMARY KEY (id, chat_jid), FOREIGN KEY (chat_jid) REFERENCES chats(jid) ); CREATE INDEX IF NOT EXISTS idx_timestamp ON messages(timestamp); + CREATE TABLE IF NOT EXISTS message_sequence ( + id INTEGER PRIMARY KEY AUTOINCREMENT + ); + + CREATE TABLE IF NOT EXISTS work_items ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + group_folder TEXT NOT NULL, + chat_jid TEXT NOT NULL, + agent_type TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'produced', + start_seq INTEGER, + end_seq INTEGER, + result_payload TEXT NOT NULL, + delivery_attempts INTEGER NOT NULL DEFAULT 0, + delivery_message_id TEXT, + last_error TEXT, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + updated_at TEXT NOT NULL DEFAULT (datetime('now')), + delivered_at TEXT, + CHECK (status IN ('produced', 'delivery_retry', 'delivered')) + ); + CREATE INDEX IF NOT EXISTS idx_work_items_status ON work_items(status, updated_at); + CREATE INDEX IF NOT EXISTS idx_work_items_group_agent ON work_items(chat_jid, agent_type, status); + CREATE UNIQUE INDEX IF NOT EXISTS idx_work_items_open + ON work_items(chat_jid, agent_type) + WHERE status IN ('produced', 'delivery_retry'); CREATE TABLE IF NOT EXISTS scheduled_tasks ( id TEXT PRIMARY KEY, @@ -168,6 +250,19 @@ function createSchema(database: Database.Database): void { /* column already exists */ } + try { + database.exec(`ALTER TABLE messages ADD COLUMN seq INTEGER`); + } catch { + /* column already exists */ + } + + backfillMessageSeq(database); + + database.exec(` + CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_seq ON messages(seq); + CREATE INDEX IF NOT EXISTS idx_messages_chat_jid_seq ON messages(chat_jid, seq); + `); + // Migrate registered_groups to composite keys so Claude/Codex can share a jid/folder. const registeredGroupsSql = ( database @@ -412,18 +507,60 @@ export function getAllChats(): ChatInfo[] { * Only call this for registered groups where message history is needed. */ export function storeMessage(msg: NewMessage): void { - db.prepare( - `INSERT OR REPLACE INTO messages (id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, - ).run( - msg.id, - msg.chat_jid, - msg.sender, - msg.sender_name, - msg.content, - msg.timestamp, - msg.is_from_me ? 1 : 0, - msg.is_bot_message ? 1 : 0, - ); + const existing = db + .prepare('SELECT seq FROM messages WHERE id = ? AND chat_jid = ?') + .get(msg.id, msg.chat_jid) as { seq: number | null } | undefined; + + const nextSeq = () => { + const result = db + .prepare('INSERT INTO message_sequence DEFAULT VALUES') + .run() as Database.RunResult; + return Number(result.lastInsertRowid); + }; + + db.transaction(() => { + if (existing?.seq != null) { + db.prepare( + `INSERT INTO messages ( + id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id, chat_jid) DO UPDATE SET + sender = excluded.sender, + sender_name = excluded.sender_name, + content = excluded.content, + timestamp = excluded.timestamp, + is_from_me = excluded.is_from_me, + is_bot_message = excluded.is_bot_message`, + ).run( + msg.id, + msg.chat_jid, + msg.sender, + msg.sender_name, + msg.content, + msg.timestamp, + existing.seq, + msg.is_from_me ? 1 : 0, + msg.is_bot_message ? 1 : 0, + ); + return; + } + + db.prepare( + `INSERT INTO messages ( + id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ).run( + msg.id, + msg.chat_jid, + msg.sender, + msg.sender_name, + msg.content, + msg.timestamp, + nextSeq(), + msg.is_from_me ? 1 : 0, + msg.is_bot_message ? 1 : 0, + ); + })(); } function normalizeMessageRow( @@ -511,6 +648,105 @@ export function getMessagesSince( return rows.map(normalizeMessageRow); } +function normalizeSeqCursor(cursor: string | number | null | undefined): number { + if (typeof cursor === 'number') { + return Number.isFinite(cursor) && cursor > 0 ? cursor : 0; + } + if (!cursor) return 0; + const parsed = Number.parseInt(cursor, 10); + return Number.isFinite(parsed) && parsed > 0 ? parsed : 0; +} + +export function getLatestMessageSeqAtOrBefore( + timestamp: string, + chatJid?: string, +): number { + if (!timestamp) return 0; + const row = (chatJid + ? db + .prepare( + `SELECT COALESCE(MAX(seq), 0) AS maxSeq + FROM messages + WHERE chat_jid = ? AND timestamp <= ?`, + ) + .get(chatJid, timestamp) + : db + .prepare( + `SELECT COALESCE(MAX(seq), 0) AS maxSeq + FROM messages + WHERE timestamp <= ?`, + ) + .get(timestamp)) as { maxSeq: number | null }; + return row.maxSeq ?? 0; +} + +export function getNewMessagesBySeq( + jids: string[], + lastSeqCursor: string | number, + botPrefix: string, + limit: number = 200, +): { messages: NewMessage[]; newSeqCursor: string } { + const sinceSeq = normalizeSeqCursor(lastSeqCursor); + if (jids.length === 0) { + return { messages: [], newSeqCursor: String(sinceSeq) }; + } + + const placeholders = jids.map(() => '?').join(','); + const sql = ` + SELECT id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message + FROM messages + WHERE seq > ? AND chat_jid IN (${placeholders}) + AND content NOT LIKE ? + AND content != '' AND content IS NOT NULL + ORDER BY seq + LIMIT ? + `; + + const rows = db + .prepare(sql) + .all(sinceSeq, ...jids, `${botPrefix}:%`, limit) as Array< + NewMessage & { + seq: number; + is_from_me?: boolean | number; + is_bot_message?: boolean | number; + } + >; + + const lastSeq = rows.length > 0 ? rows[rows.length - 1].seq : sinceSeq; + return { + messages: rows.map(normalizeMessageRow), + newSeqCursor: String(lastSeq), + }; +} + +export function getMessagesSinceSeq( + chatJid: string, + sinceSeqCursor: string | number, + botPrefix: string, + limit: number = 200, +): NewMessage[] { + const sinceSeq = normalizeSeqCursor(sinceSeqCursor); + const sql = ` + SELECT id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message + FROM messages + WHERE chat_jid = ? AND seq > ? + AND content NOT LIKE ? + AND content != '' AND content IS NOT NULL + ORDER BY seq + LIMIT ? + `; + const rows = db + .prepare(sql) + .all(chatJid, sinceSeq, `${botPrefix}:%`, limit) as Array< + NewMessage & { + seq: number; + is_from_me?: boolean | number; + is_bot_message?: boolean | number; + } + >; + return rows.map(normalizeMessageRow); +} + export function getLastHumanMessageTimestamp(chatJid: string): string | null { const row = db .prepare( @@ -523,6 +759,110 @@ export function getLastHumanMessageTimestamp(chatJid: string): string | null { return row?.timestamp ?? null; } +export function hasRecentRestartAnnouncement( + chatJid: string, + sinceTimestamp: string, +): boolean { + const row = db + .prepare( + `SELECT 1 FROM messages + WHERE chat_jid = ? + AND timestamp >= ? + AND is_bot_message = 1 + AND ( + content LIKE '재시작 완료.%' + OR content LIKE '재시작 감지.%' + OR content LIKE '서비스 재시작으로 이전 작업이 중단됐습니다.%' + ) + LIMIT 1`, + ) + .get(chatJid, sinceTimestamp) as { 1: number } | undefined; + return !!row; +} + +export function getOpenWorkItem( + chatJid: string, + agentType: AgentType = SERVICE_AGENT_TYPE, +): WorkItem | undefined { + return db + .prepare( + `SELECT * + FROM work_items + WHERE chat_jid = ? AND agent_type = ? AND status IN ('produced', 'delivery_retry') + ORDER BY id ASC + LIMIT 1`, + ) + .get(chatJid, agentType) as WorkItem | undefined; +} + +export function createProducedWorkItem(input: { + group_folder: string; + chat_jid: string; + agent_type?: AgentType; + start_seq: number | null; + end_seq: number | null; + result_payload: string; +}): WorkItem { + const now = new Date().toISOString(); + const agentType = input.agent_type || SERVICE_AGENT_TYPE; + const result = db + .prepare( + `INSERT INTO work_items ( + group_folder, + chat_jid, + agent_type, + status, + start_seq, + end_seq, + result_payload, + delivery_attempts, + created_at, + updated_at + ) VALUES (?, ?, ?, 'produced', ?, ?, ?, 0, ?, ?)`, + ) + .run( + input.group_folder, + input.chat_jid, + agentType, + input.start_seq, + input.end_seq, + input.result_payload, + now, + now, + ) as Database.RunResult; + + return db + .prepare('SELECT * FROM work_items WHERE id = ?') + .get(Number(result.lastInsertRowid)) as WorkItem; +} + +export function markWorkItemDelivered( + id: number, + deliveryMessageId?: string | null, +): void { + const now = new Date().toISOString(); + db.prepare( + `UPDATE work_items + SET status = 'delivered', + delivered_at = ?, + delivery_message_id = ?, + updated_at = ? + WHERE id = ?`, + ).run(now, deliveryMessageId || null, now, id); +} + +export function markWorkItemDeliveryRetry(id: number, error: string): void { + const now = new Date().toISOString(); + db.prepare( + `UPDATE work_items + SET status = 'delivery_retry', + delivery_attempts = delivery_attempts + 1, + last_error = ?, + updated_at = ? + WHERE id = ?`, + ).run(error, now, id); +} + export function createTask( task: Omit< ScheduledTask, diff --git a/src/group-queue.test.ts b/src/group-queue.test.ts index b914789..0413b5c 100644 --- a/src/group-queue.test.ts +++ b/src/group-queue.test.ts @@ -414,6 +414,30 @@ describe('GroupQueue', () => { await vi.advanceTimersByTimeAsync(10); }); + it('sendMessage touches active run activity after piping follow-up', async () => { + let resolveProcess: () => void; + const touch = vi.fn(); + + const processMessages = vi.fn(async () => { + await new Promise((resolve) => { + resolveProcess = resolve; + }); + return true; + }); + + queue.setProcessMessagesFn(processMessages); + queue.enqueueMessageCheck('group1@g.us'); + await vi.advanceTimersByTimeAsync(10); + queue.registerProcess('group1@g.us', {} as any, 'agent-1', 'test-group'); + queue.setActivityTouch('group1@g.us', touch); + + expect(queue.sendMessage('group1@g.us', 'hello')).toBe(true); + expect(touch).toHaveBeenCalledTimes(1); + + resolveProcess!(); + await vi.advanceTimersByTimeAsync(10); + }); + it('sendMessage returns false for task agents so user messages queue up', async () => { let resolveTask: () => void; diff --git a/src/group-queue.ts b/src/group-queue.ts index 3ac6ed8..042f7a3 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -11,6 +11,12 @@ interface QueuedTask { fn: () => Promise; } +export interface GroupActivityTouchMeta { + source: 'follow-up'; + textLength: number; + filename: string; +} + export interface GroupRunContext { runId: string; reason: 'messages' | 'drain'; @@ -35,6 +41,7 @@ interface GroupState { retryTimer: ReturnType | null; retryScheduledAt: number | null; startedAt: number | null; + activityTouch: ((meta?: GroupActivityTouchMeta) => void) | null; } export interface GroupStatus { @@ -73,6 +80,7 @@ export class GroupQueue { retryTimer: null, retryScheduledAt: null, startedAt: null, + activityTouch: null, }; this.groups.set(groupJid, state); } @@ -85,6 +93,14 @@ export class GroupQueue { this.processMessagesFn = fn; } + setActivityTouch( + groupJid: string, + touch: ((meta?: GroupActivityTouchMeta) => void) | null, + ): void { + const state = this.getGroup(groupJid); + state.activityTouch = touch; + } + private createRunId(): string { return `run-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`; } @@ -262,6 +278,11 @@ export class GroupQueue { const tempPath = `${filepath}.tmp`; fs.writeFileSync(tempPath, JSON.stringify({ type: 'message', text })); fs.renameSync(tempPath, filepath); + state.activityTouch?.({ + source: 'follow-up', + textLength: text.length, + filename, + }); logger.info( { groupJid, @@ -390,6 +411,7 @@ export class GroupQueue { state.processName = null; state.groupFolder = null; state.currentRunId = null; + state.activityTouch = null; this.activeCount--; this.drainGroup(groupJid); } @@ -424,6 +446,7 @@ export class GroupQueue { state.process = null; state.processName = null; state.groupFolder = null; + state.activityTouch = null; this.activeCount--; this.drainGroup(groupJid); } diff --git a/src/index.test.ts b/src/index.test.ts new file mode 100644 index 0000000..cefd5e3 --- /dev/null +++ b/src/index.test.ts @@ -0,0 +1,83 @@ +import { describe, expect, it, vi } from 'vitest'; + +import { + editFormattedTrackedChannelMessage, + sendFormattedChannelMessage, + sendFormattedTrackedChannelMessage, +} from './index.js'; +import { Channel } from './types.js'; + +function makeChannel(overrides: Partial = {}): Channel { + return { + name: 'test', + connect: async () => {}, + sendMessage: async () => {}, + isConnected: () => true, + ownsJid: (jid: string) => jid === 'dc:test', + disconnect: async () => {}, + ...overrides, + }; +} + +describe('index scheduler messaging helpers', () => { + it('sends formatted tracked messages through sendAndTrack', async () => { + const sendAndTrack = vi.fn(async () => 'msg-123'); + const channel = makeChannel({ sendAndTrack }); + + const messageId = await sendFormattedTrackedChannelMessage( + [channel], + 'dc:test', + 'hiddenwatching ci', + ); + + expect(messageId).toBe('msg-123'); + expect(sendAndTrack).toHaveBeenCalledWith('dc:test', 'watching ci'); + }); + + it('edits formatted tracked messages through editMessage', async () => { + const editMessage = vi.fn(async () => {}); + const channel = makeChannel({ editMessage }); + + await editFormattedTrackedChannelMessage( + [channel], + 'dc:test', + 'msg-123', + 'hiddenstill watching', + ); + + expect(editMessage).toHaveBeenCalledWith( + 'dc:test', + 'msg-123', + 'still watching', + ); + }); + + it('skips empty messages after formatting', async () => { + const sendMessage = vi.fn(async () => {}); + const sendAndTrack = vi.fn(async () => 'msg-123'); + const editMessage = vi.fn(async () => {}); + const channel = makeChannel({ sendMessage, sendAndTrack, editMessage }); + + const trackedResult = await sendFormattedTrackedChannelMessage( + [channel], + 'dc:test', + 'only hidden', + ); + await editFormattedTrackedChannelMessage( + [channel], + 'dc:test', + 'msg-123', + 'only hidden', + ); + await sendFormattedChannelMessage( + [channel], + 'dc:test', + 'only hidden', + ); + + expect(trackedResult).toBeNull(); + expect(sendAndTrack).not.toHaveBeenCalled(); + expect(editMessage).not.toHaveBeenCalled(); + expect(sendMessage).not.toHaveBeenCalled(); + }); +}); diff --git a/src/index.ts b/src/index.ts index e860581..f676d4e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -5,6 +5,7 @@ import path from 'path'; import { ASSISTANT_NAME, + DATA_DIR, IDLE_TIMEOUT, POLL_INTERVAL, SERVICE_AGENT_TYPE, @@ -13,6 +14,7 @@ import { STATUS_UPDATE_INTERVAL, TIMEZONE, TRIGGER_PATTERN, + USAGE_DASHBOARD_ENABLED, USAGE_UPDATE_INTERVAL, } from './config.js'; import './channels/index.js'; @@ -31,20 +33,26 @@ import { getAllRegisteredGroups, getAllSessions, getAllTasks, - getLastHumanMessageTimestamp, - getMessagesSince, - getNewMessages, + getLatestMessageSeqAtOrBefore, + getMessagesSinceSeq, + getNewMessagesBySeq, + getOpenWorkItem, + hasRecentRestartAnnouncement, getRegisteredGroup, getRouterState, initDatabase, isPairedRoomJid, + markWorkItemDelivered, + markWorkItemDeliveryRetry, setRegisteredGroup, setRouterState, updateRegisteredGroupName, deleteSession, setSession, storeChatMetadata, + createProducedWorkItem, storeMessage, + WorkItem, } from './db.js'; import { filterProcessableMessages } from './bot-message-filter.js'; import { GroupQueue } from './group-queue.js'; @@ -52,6 +60,13 @@ import { readEnvFile } from './env.js'; import { resolveGroupFolderPath } from './group-folder.js'; import { startIpcWatcher } from './ipc.js'; import { findChannel, formatMessages, formatOutbound } from './router.js'; +import { + buildRestartAnnouncement, + buildInterruptedRestartAnnouncement, + consumeRestartContext, + inferRecentRestartContext, + writeShutdownRestartContext, +} from './restart-context.js'; import { isSenderAllowed, isTriggerAllowed, @@ -63,6 +78,15 @@ import { handleSessionCommand, isSessionCommandAllowed, } from './session-commands.js'; +import { + detectFallbackTrigger, + getActiveProvider, + getFallbackEnvOverrides, + getFallbackProviderName, + hasGroupProviderOverride, + isFallbackEnabled, + markPrimaryCooldown, +} from './provider-fallback.js'; import { shouldResetSessionOnAgentFailure } from './session-recovery.js'; import { readStatusSnapshots, @@ -75,6 +99,51 @@ import { logger } from './logger.js'; // Re-export for backwards compatibility during refactor export { escapeXml, formatMessages } from './router.js'; +export async function sendFormattedChannelMessage( + channels: Channel[], + jid: string, + rawText: string, +): Promise { + const channel = findChannel(channels, jid); + if (!channel) { + logger.warn({ jid }, 'No channel owns JID, cannot send message'); + return; + } + const text = formatOutbound(rawText); + if (text) await channel.sendMessage(jid, text); +} + +export async function sendFormattedTrackedChannelMessage( + channels: Channel[], + jid: string, + rawText: string, +): Promise { + const channel = findChannel(channels, jid); + if (!channel) { + logger.warn({ jid }, 'No channel owns JID, cannot send tracked message'); + return null; + } + const text = formatOutbound(rawText); + if (!text || !channel.sendAndTrack) return null; + return channel.sendAndTrack(jid, text); +} + +export async function editFormattedTrackedChannelMessage( + channels: Channel[], + jid: string, + messageId: string, + rawText: string, +): Promise { + const channel = findChannel(channels, jid); + if (!channel) { + logger.warn({ jid }, 'No channel owns JID, cannot edit tracked message'); + return; + } + const text = formatOutbound(rawText); + if (!text || !channel.editMessage) return; + await channel.editMessage(jid, messageId, text); +} + let lastTimestamp = ''; let sessions: Record = {}; let registeredGroups: Record = {}; @@ -84,8 +153,27 @@ let messageLoopRunning = false; const channels: Channel[] = []; const queue = new GroupQueue(); -function advanceLastAgentCursor(chatJid: string, timestamp: string): void { - lastAgentTimestamp[chatJid] = timestamp; +function normalizeStoredSeqCursor( + cursor: string | undefined, + chatJid?: string, +): string { + if (!cursor) return '0'; + if (/^\d+$/.test(cursor.trim())) return cursor.trim(); + return String(getLatestMessageSeqAtOrBefore(cursor, chatJid)); +} + +function advanceLastAgentCursor( + chatJid: string, + cursorOrTimestamp: string | number, +): void { + if (typeof cursorOrTimestamp === 'number') { + lastAgentTimestamp[chatJid] = String(cursorOrTimestamp); + } else { + lastAgentTimestamp[chatJid] = normalizeStoredSeqCursor( + cursorOrTimestamp, + chatJid, + ); + } saveState(); } @@ -98,13 +186,53 @@ function getProcessableMessages( return filterProcessableMessages(messages, isPairedRoomJid(chatJid), isOwn); } -function loadState(): void { - lastTimestamp = getRouterState('last_timestamp') || ''; - const agentTs = getRouterState('last_agent_timestamp'); +async function deliverOpenWorkItem( + channel: Channel, + item: WorkItem, +): Promise { try { - lastAgentTimestamp = agentTs ? JSON.parse(agentTs) : {}; + await channel.sendMessage(item.chat_jid, item.result_payload); + markWorkItemDelivered(item.id); + logger.info( + { + chatJid: item.chat_jid, + workItemId: item.id, + deliveryAttempts: item.delivery_attempts + 1, + }, + 'Delivered produced work item', + ); + return true; + } catch (err) { + const errorMessage = err instanceof Error ? err.message : String(err); + markWorkItemDeliveryRetry(item.id, errorMessage); + logger.warn( + { + chatJid: item.chat_jid, + workItemId: item.id, + deliveryAttempts: item.delivery_attempts + 1, + err, + }, + 'Failed to deliver produced work item', + ); + return false; + } +} + +function loadState(): void { + lastTimestamp = normalizeStoredSeqCursor( + getRouterState('last_seq') || getRouterState('last_timestamp'), + ); + const agentTs = getRouterState('last_agent_seq') || getRouterState('last_agent_timestamp'); + try { + const parsed = agentTs ? (JSON.parse(agentTs) as Record) : {}; + lastAgentTimestamp = Object.fromEntries( + Object.entries(parsed).map(([chatJid, cursor]) => [ + chatJid, + normalizeStoredSeqCursor(cursor, chatJid), + ]), + ); } catch { - logger.warn('Corrupted last_agent_timestamp in DB, resetting'); + logger.warn('Corrupted last_agent_seq in DB, resetting'); lastAgentTimestamp = {}; } sessions = getAllSessions(); @@ -121,8 +249,8 @@ function loadState(): void { } function saveState(): void { - setRouterState('last_timestamp', lastTimestamp); - setRouterState('last_agent_timestamp', JSON.stringify(lastAgentTimestamp)); + setRouterState('last_seq', lastTimestamp); + setRouterState('last_agent_seq', JSON.stringify(lastAgentTimestamp)); } function clearSession(groupFolder: string): void { @@ -193,12 +321,23 @@ async function processGroupMessages(chatJid: string): Promise { return true; } + const openWorkItem = getOpenWorkItem( + chatJid, + (group.agentType || 'claude-code') as 'claude-code' | 'codex', + ); + if (openWorkItem) { + const delivered = await deliverOpenWorkItem(channel, openWorkItem); + if (!delivered) { + return false; + } + } + const isMainGroup = group.isMain === true; - const sinceTimestamp = lastAgentTimestamp[chatJid] || ''; - const rawMissedMessages = getMessagesSince( + const sinceSeqCursor = lastAgentTimestamp[chatJid] || '0'; + const rawMissedMessages = getMessagesSinceSeq( chatJid, - sinceTimestamp, + sinceSeqCursor, ASSISTANT_NAME, ); const missedMessages = getProcessableMessages( @@ -228,7 +367,10 @@ async function processGroupMessages(chatJid: string): Promise { channel.setTyping?.(chatJid, typing) ?? Promise.resolve(), runAgent: (prompt, onOutput) => runAgent(group, prompt, chatJid, onOutput), - closeStdin: () => queue.closeStdin(chatJid), + closeStdin: () => + queue.closeStdin(chatJid, { + reason: 'session-command', + }), clearSession: () => clearSession(group.folder), advanceCursor: (ts) => { advanceLastAgentCursor(chatJid, ts); @@ -268,11 +410,12 @@ async function processGroupMessages(chatJid: string): Promise { // Advance cursor so the piping path in startMessageLoop won't re-fetch // these messages. Save the old cursor so we can roll back on error. - const previousCursor = lastAgentTimestamp[chatJid] || ''; - advanceLastAgentCursor( - chatJid, - missedMessages[missedMessages.length - 1].timestamp, - ); + const previousCursor = lastAgentTimestamp[chatJid] || '0'; + const startSeq = missedMessages[0].seq ?? null; + const endSeq = missedMessages[missedMessages.length - 1].seq ?? null; + if (endSeq !== null) { + advanceLastAgentCursor(chatJid, endSeq); + } logger.info( { group: group.name, messageCount: missedMessages.length }, @@ -281,14 +424,125 @@ async function processGroupMessages(chatJid: string): Promise { // Track idle timer for closing stdin when agent is idle let idleTimer: ReturnType | null = null; + let producedFinalText: string | null = null; const resetIdleTimer = () => { if (idleTimer) clearTimeout(idleTimer); idleTimer = setTimeout(() => { logger.debug({ group: group.name }, 'Idle timeout, closing agent stdin'); - queue.closeStdin(chatJid); + if (followUpQueuedAt !== null) { + logger.warn( + { + group: group.name, + chatJid, + followUpQueuedAt: new Date(followUpQueuedAt).toISOString(), + followUpQueuedTextLength, + followUpQueuedFilename, + followUpWaitMs: Date.now() - followUpQueuedAt, + }, + 'Idle timeout reached while a queued follow-up still had no agent output', + ); + } + queue.closeStdin(chatJid, { reason: 'idle-timeout' }); }, IDLE_TIMEOUT); }; + let followUpQueuedAt: number | null = null; + let followUpQueuedTextLength: number | null = null; + let followUpQueuedFilename: string | null = null; + let followUpNoOutputWarnTimer: ReturnType | null = null; + const FOLLOW_UP_NO_OUTPUT_WARN_MS = 10_000; + + const clearFollowUpNoOutputWarnTimer = () => { + if (followUpNoOutputWarnTimer) { + clearTimeout(followUpNoOutputWarnTimer); + followUpNoOutputWarnTimer = null; + } + }; + + const clearPendingFollowUpDiagnostics = () => { + clearFollowUpNoOutputWarnTimer(); + followUpQueuedAt = null; + followUpQueuedTextLength = null; + followUpQueuedFilename = null; + }; + + const scheduleFollowUpNoOutputWarning = () => { + clearFollowUpNoOutputWarnTimer(); + followUpNoOutputWarnTimer = setTimeout(() => { + if (followUpQueuedAt === null) return; + logger.warn( + { + group: group.name, + chatJid, + followUpQueuedAt: new Date(followUpQueuedAt).toISOString(), + followUpQueuedTextLength, + followUpQueuedFilename, + elapsedMs: Date.now() - followUpQueuedAt, + }, + 'No agent output observed within 10s of queuing a follow-up message', + ); + }, FOLLOW_UP_NO_OUTPUT_WARN_MS); + }; + + const noteFollowUpQueued = (meta?: { + source: 'follow-up'; + textLength: number; + filename: string; + }) => { + if (meta?.source !== 'follow-up') return; + followUpQueuedAt = Date.now(); + followUpQueuedTextLength = meta.textLength; + followUpQueuedFilename = meta.filename; + scheduleFollowUpNoOutputWarning(); + logger.info( + { + group: group.name, + chatJid, + followUpQueuedTextLength, + followUpQueuedFilename, + }, + 'Registered follow-up activity for active agent', + ); + }; + + const noteAgentOutputObserved = (phase?: string) => { + if (followUpQueuedAt === null) return; + logger.info( + { + group: group.name, + chatJid, + followUpQueuedAt: new Date(followUpQueuedAt).toISOString(), + followUpQueuedTextLength, + followUpQueuedFilename, + followUpWaitMs: Date.now() - followUpQueuedAt, + resultPhase: phase, + }, + 'Agent produced output after a queued follow-up', + ); + clearPendingFollowUpDiagnostics(); + }; + + const warnFollowUpEndedWithoutOutput = (reason: string) => { + if (followUpQueuedAt === null) return; + logger.warn( + { + group: group.name, + chatJid, + followUpQueuedAt: new Date(followUpQueuedAt).toISOString(), + followUpQueuedTextLength, + followUpQueuedFilename, + followUpWaitMs: Date.now() - followUpQueuedAt, + reason, + }, + 'Active agent ended a turn without any output after a queued follow-up', + ); + clearPendingFollowUpDiagnostics(); + }; + + queue.setActivityTouch(chatJid, (meta) => { + noteFollowUpQueued(meta); + resetIdleTimer(); + }); let hadError = false; let latestProgressText: string | null = null; @@ -300,6 +554,8 @@ async function processGroupMessages(chatJid: string): Promise { let progressOutputSentToUser = false; let latestModelProgressTextForFinalFallback: string | null = null; let sawNonProgressOutput = false; + let producedDeliverySucceeded = true; + let isFirstLogicalTurn = true; let poisonedSessionDetected = false; const isClaudeCodeAgent = (group.agentType || 'claude-code') === 'claude-code'; @@ -311,6 +567,47 @@ async function processGroupMessages(chatJid: string): Promise { } }; + const resetTurnOutputState = () => { + finalOutputSentToUser = false; + progressOutputSentToUser = false; + latestModelProgressTextForFinalFallback = null; + sawNonProgressOutput = false; + producedFinalText = null; + }; + + const flushProducedFinalText = async () => { + if (!producedFinalText) { + return; + } + + try { + const workItem = createProducedWorkItem({ + group_folder: group.folder, + chat_jid: chatJid, + agent_type: group.agentType || 'claude-code', + start_seq: isFirstLogicalTurn ? startSeq : null, + end_seq: isFirstLogicalTurn ? endSeq : null, + result_payload: producedFinalText, + }); + const delivered = await deliverOpenWorkItem(channel, workItem); + if (delivered) { + finalOutputSentToUser = true; + } else { + producedDeliverySucceeded = false; + } + } catch (err) { + producedDeliverySucceeded = false; + logger.warn( + { group: group.name, chatJid, err }, + 'Failed to persist produced output for delivery', + ); + } finally { + producedFinalText = null; + latestModelProgressTextForFinalFallback = null; + isFirstLogicalTurn = false; + } + }; + const resetProgressState = () => { clearProgressTicker(); latestProgressText = null; @@ -379,11 +676,12 @@ async function processGroupMessages(chatJid: string): Promise { return; } - latestModelProgressTextForFinalFallback = text; - latestProgressText = text; if (progressStartedAt === null) { + resetTurnOutputState(); progressStartedAt = Date.now(); } + latestModelProgressTextForFinalFallback = text; + latestProgressText = text; const rendered = renderProgressMessage(text); if (progressMessageId && channel.editMessage) { @@ -400,7 +698,15 @@ async function processGroupMessages(chatJid: string): Promise { return; } - progressMessageId = await channel.sendAndTrack(chatJid, rendered); + try { + progressMessageId = await channel.sendAndTrack(chatJid, rendered); + } catch (err) { + logger.warn( + { group: group.name, chatJid, err }, + 'Failed to send tracked progress message', + ); + return; + } if (progressMessageId) { logger.info( { group: group.name, chatJid, progressMessageId, text }, @@ -423,7 +729,9 @@ async function processGroupMessages(chatJid: string): Promise { poisonedSessionDetected = true; hadError = true; clearSession(group.folder); - queue.closeStdin(chatJid); + queue.closeStdin(chatJid, { + reason: 'poisoned-session-detected', + }); logger.warn( { group: group.name, chatJid }, 'Detected poisoned Claude session from streamed output, forcing close', @@ -446,18 +754,31 @@ async function processGroupMessages(chatJid: string): Promise { }, `Agent output: ${raw.slice(0, 200)}`, ); + noteAgentOutputObserved(result.phase); if (result.phase === 'progress') { + if (finalOutputSentToUser || sawNonProgressOutput) { + logger.info( + { group: group.name, chatJid }, + 'New logical turn detected (follow-up), resetting turn state', + ); + resetTurnOutputState(); + resetProgressState(); + isFirstLogicalTurn = false; + await channel.setTyping?.(chatJid, true); + } if (text) { await sendProgressMessage(text); } + if (!poisonedSessionDetected) { + resetIdleTimer(); + } return; } sawNonProgressOutput = true; if (text) { await finalizeProgressMessage(); - await channel.sendMessage(chatJid, text); - finalOutputSentToUser = true; - latestModelProgressTextForFinalFallback = null; + producedFinalText = text; + await flushProducedFinalText(); } else { logger.info( { @@ -473,6 +794,9 @@ async function processGroupMessages(chatJid: string): Promise { latestModelProgressTextForFinalFallback = null; } } else { + if (result.status === 'success') { + warnFollowUpEndedWithoutOutput('success-null-result'); + } await finalizeProgressMessage(); } @@ -500,7 +824,7 @@ async function processGroupMessages(chatJid: string): Promise { if ( output === 'success' && !hadError && - !finalOutputSentToUser && + !producedFinalText && !sawNonProgressOutput && latestModelProgressTextForFinalFallback ) { @@ -508,22 +832,23 @@ async function processGroupMessages(chatJid: string): Promise { { group: group.name, chatJid }, 'Promoting last progress output to final message after agent completion', ); - await channel.sendMessage(chatJid, latestModelProgressTextForFinalFallback); - finalOutputSentToUser = true; - latestModelProgressTextForFinalFallback = null; + producedFinalText = latestModelProgressTextForFinalFallback; + await flushProducedFinalText(); } clearProgressTicker(); if (idleTimer) clearTimeout(idleTimer); + clearPendingFollowUpDiagnostics(); + queue.setActivityTouch(chatJid, null); if (hadError) { - if (finalOutputSentToUser || progressOutputSentToUser) { + if (finalOutputSentToUser || progressOutputSentToUser || producedFinalText) { logger.warn( { group: group.name }, - 'Agent error after output was sent, skipping cursor rollback to prevent duplicates', + 'Agent error after output was produced, skipping cursor rollback to prevent duplicates', ); - return true; + return producedFinalText ? producedDeliverySucceeded : true; } lastAgentTimestamp[chatJid] = previousCursor; saveState(); @@ -534,6 +859,14 @@ async function processGroupMessages(chatJid: string): Promise { return false; } + if (!producedDeliverySucceeded) { + logger.warn( + { group: group.name, chatJid }, + 'Persisted produced output for delivery retry without rerunning agent', + ); + return false; + } + return true; } @@ -571,51 +904,261 @@ async function runAgent( new Set(Object.keys(registeredGroups)), ); - // Wrap onOutput to track session ID from streamed results - const wrappedOnOutput = onOutput - ? async (output: AgentOutput) => { - if (output.newSessionId) { - sessions[group.folder] = output.newSessionId; - setSession(group.folder, output.newSessionId); + const isClaudeCode = (group.agentType || 'claude-code') === 'claude-code'; + const settingsPath = path.join( + DATA_DIR, + 'sessions', + group.folder, + '.claude', + 'settings.json', + ); + const groupHasOverride = hasGroupProviderOverride(settingsPath); + const canFallback = + isClaudeCode && isFallbackEnabled() && !groupHasOverride; + + const agentInput = { + prompt, + sessionId, + groupFolder: group.folder, + chatJid, + isMain, + assistantName: ASSISTANT_NAME, + }; + + const runAttempt = async (provider: string): Promise<{ + output?: AgentOutput; + error?: unknown; + sawOutput: boolean; + sawSuccessNullResultWithoutOutput: boolean; + streamedTriggerReason?: { + reason: string; + retryAfterMs?: number; + }; + }> => { + const persistSessionIds = provider === 'claude'; + let sawOutput = false; + let sawSuccessNullResultWithoutOutput = false; + let streamedTriggerReason: + | { + reason: string; + retryAfterMs?: number; } - await onOutput(output); - } - : undefined; + | undefined; - try { - const output = await runAgentProcess( - group, - { - prompt, - sessionId, - groupFolder: group.folder, - chatJid, - isMain, - assistantName: ASSISTANT_NAME, - }, - (proc, processName) => - queue.registerProcess(chatJid, proc, processName, group.folder), - wrappedOnOutput, - ); + const wrappedOnOutput = onOutput + ? async (output: AgentOutput) => { + if (persistSessionIds && output.newSessionId) { + sessions[group.folder] = output.newSessionId; + setSession(group.folder, output.newSessionId); + } - if (output.newSessionId) { - sessions[group.folder] = output.newSessionId; - setSession(group.folder, output.newSessionId); + if (output.result !== null && output.result !== undefined) { + sawOutput = true; + } else if ( + provider === 'claude' && + output.status === 'success' && + !sawOutput + ) { + sawSuccessNullResultWithoutOutput = true; + } + + if ( + provider === 'claude' && + output.status === 'error' && + !sawOutput && + !streamedTriggerReason + ) { + const trigger = detectFallbackTrigger(output.error); + if (trigger.shouldFallback) { + streamedTriggerReason = { + reason: trigger.reason, + retryAfterMs: trigger.retryAfterMs, + }; + } + } + + await onOutput(output); + } + : undefined; + + if (provider !== 'claude') { + logger.info( + { + group: group.name, + chatJid, + provider, + }, + `Claude provider in cooldown, routing request to ${provider}`, + ); } - if (output.status === 'error') { + logger.info( + { + group: group.name, + chatJid, + provider, + canFallback, + groupHasOverride, + }, + `Using provider: ${provider}`, + ); + + try { + const output = await runAgentProcess( + group, + { + ...agentInput, + sessionId: persistSessionIds ? sessionId : undefined, + }, + (proc, processName) => + queue.registerProcess(chatJid, proc, processName, group.folder), + wrappedOnOutput, + provider === 'claude' ? undefined : getFallbackEnvOverrides(), + ); + + if (persistSessionIds && output.newSessionId) { + sessions[group.folder] = output.newSessionId; + setSession(group.folder, output.newSessionId); + } + + logger.info( + { + group: group.name, + chatJid, + provider, + status: output.status, + sawOutput, + }, + `Provider response completed (provider: ${provider})`, + ); + + return { + output, + sawOutput, + sawSuccessNullResultWithoutOutput, + streamedTriggerReason, + }; + } catch (error) { + return { + error, + sawOutput, + sawSuccessNullResultWithoutOutput, + streamedTriggerReason, + }; + } + }; + + const runFallbackAttempt = async ( + reason: string, + retryAfterMs?: number, + ): Promise<'success' | 'error'> => { + const fallbackName = getFallbackProviderName(); + markPrimaryCooldown(reason, retryAfterMs); + + logger.info( + { + group: group.name, + chatJid, + reason, + retryAfterMs, + fallbackProvider: fallbackName, + }, + `Falling back to provider: ${fallbackName} (reason: ${reason})`, + ); + + const fallbackAttempt = await runAttempt(fallbackName); + if (fallbackAttempt.error) { logger.error( - { group: group.name, error: output.error }, - 'Agent process error', + { + group: group.name, + provider: fallbackName, + err: fallbackAttempt.error, + }, + 'Fallback provider also threw', + ); + return 'error'; + } + + if (fallbackAttempt.output?.status === 'error') { + logger.error( + { + group: group.name, + provider: fallbackName, + error: fallbackAttempt.output.error, + }, + `Fallback provider (${fallbackName}) also failed`, ); return 'error'; } return 'success'; - } catch (err) { - logger.error({ group: group.name, err }, 'Agent error'); + }; + + const provider = canFallback ? getActiveProvider() : 'claude'; + const primaryAttempt = await runAttempt(provider); + + if (primaryAttempt.error) { + if (canFallback && provider === 'claude' && !primaryAttempt.sawOutput) { + const errMsg = + primaryAttempt.error instanceof Error + ? primaryAttempt.error.message + : String(primaryAttempt.error); + const trigger = primaryAttempt.streamedTriggerReason + ? { + shouldFallback: true, + reason: primaryAttempt.streamedTriggerReason.reason, + retryAfterMs: primaryAttempt.streamedTriggerReason.retryAfterMs, + } + : detectFallbackTrigger(errMsg); + if (trigger.shouldFallback) { + return runFallbackAttempt(trigger.reason, trigger.retryAfterMs); + } + } + + logger.error( + { group: group.name, chatJid, provider, err: primaryAttempt.error }, + 'Agent error', + ); return 'error'; } + + const output = primaryAttempt.output; + if (!output) { + logger.error({ group: group.name, chatJid, provider }, 'Agent produced no output object'); + return 'error'; + } + + if ( + canFallback && + provider === 'claude' && + !primaryAttempt.sawOutput && + primaryAttempt.sawSuccessNullResultWithoutOutput + ) { + return runFallbackAttempt('success-null-result'); + } + + if (output.status === 'error') { + if (canFallback && provider === 'claude' && !primaryAttempt.sawOutput) { + const trigger = primaryAttempt.streamedTriggerReason + ? { + shouldFallback: true, + reason: primaryAttempt.streamedTriggerReason.reason, + retryAfterMs: primaryAttempt.streamedTriggerReason.retryAfterMs, + } + : detectFallbackTrigger(output.error); + if (trigger.shouldFallback) { + return runFallbackAttempt(trigger.reason, trigger.retryAfterMs); + } + } + + logger.error( + { group: group.name, chatJid, provider, error: output.error }, + 'Agent process error', + ); + return 'error'; + } + + return 'success'; } // ── Status & Usage Dashboards ─────────────────────────────────── @@ -924,8 +1467,31 @@ interface CodexRateLimit { } let usageApiBackoffUntil = 0; +let usageApi429Streak = 0; +let usageApiPollingDisabled = false; + +function parseRetryAfterMs(retryAfter: string | null): number | null { + if (!retryAfter) return null; + + const seconds = Number(retryAfter); + if (Number.isFinite(seconds) && seconds > 0) { + return seconds * 1000; + } + + const absolute = Date.parse(retryAfter); + if (!Number.isNaN(absolute)) { + return Math.max(0, absolute - Date.now()); + } + + return null; +} async function fetchClaudeUsage(): Promise { + if (usageApiPollingDisabled) { + logger.debug('Skipping usage API call (polling disabled for this process)'); + return null; + } + // Skip if in backoff period (after 429) if (Date.now() < usageApiBackoffUntil) { logger.debug('Skipping usage API call (backoff active)'); @@ -957,15 +1523,27 @@ async function fetchClaudeUsage(): Promise { if (!res.ok) { const body = await res.text().catch(() => ''); if (res.status === 429) { - // Back off for 10 minutes on rate limit - usageApiBackoffUntil = Date.now() + 600_000; + const retryAfter = res.headers.get('retry-after'); + const retryAfterMs = parseRetryAfterMs(retryAfter); + const backoffMs = Math.max(600_000, retryAfterMs ?? 0); + usageApi429Streak += 1; + usageApiBackoffUntil = Date.now() + backoffMs; + if (usageApi429Streak >= 3) { + usageApiPollingDisabled = true; + } logger.warn( { status: 429, - retryAfter: res.headers.get('retry-after'), + retryAfter, + retryAfterMs, + backoffMs, + consecutive429s: usageApi429Streak, + pollingDisabled: usageApiPollingDisabled, body: body.slice(0, 200), }, - 'Usage API rate limited (429), backing off 10min', + usageApiPollingDisabled + ? 'Usage API rate limited repeatedly (429), disabling usage polling for this process' + : 'Usage API rate limited (429), backing off', ); } else { logger.warn( @@ -975,6 +1553,7 @@ async function fetchClaudeUsage(): Promise { } return null; } + usageApi429Streak = 0; return (await res.json()) as ClaudeUsageData; } catch (err) { logger.debug({ err }, 'Usage API fetch failed'); @@ -1082,14 +1661,20 @@ async function buildUsageContent(): Promise { (entry) => entry.status === 'processing' || entry.status === 'waiting', ), ); + const shouldFetchClaudeUsage = USAGE_DASHBOARD_ENABLED; const [liveClaudeUsage, codexUsage] = await Promise.all([ - hasActiveClaudeWork ? Promise.resolve(null) : fetchClaudeUsage(), + shouldFetchClaudeUsage && !hasActiveClaudeWork + ? fetchClaudeUsage() + : Promise.resolve(null), fetchCodexUsage(), ]); - const claudeUsage = liveClaudeUsage || cachedClaudeUsageData; - const claudeUsageIsCached = !liveClaudeUsage && !!cachedClaudeUsageData; - if (liveClaudeUsage) { + const claudeUsage = shouldFetchClaudeUsage + ? liveClaudeUsage || cachedClaudeUsageData + : null; + const claudeUsageIsCached = + shouldFetchClaudeUsage && !liveClaudeUsage && !!cachedClaudeUsageData; + if (shouldFetchClaudeUsage && liveClaudeUsage) { cachedClaudeUsageData = liveClaudeUsage; } @@ -1164,6 +1749,9 @@ async function buildUsageContent(): Promise { } else { lines.push('_조회 불가_'); } + if (shouldFetchClaudeUsage && usageApiPollingDisabled) { + lines.push('_* Claude 사용량 조회는 반복된 429로 이번 프로세스에서 일시 중지_'); + } if (claudeUsageIsCached) { lines.push('_* Claude 사용량은 작업 중일 때는 캐시값 유지_'); } @@ -1314,7 +1902,7 @@ async function startMessageLoop(): Promise { while (true) { try { const jids = Object.keys(registeredGroups); - const { messages, newTimestamp } = getNewMessages( + const { messages, newSeqCursor } = getNewMessagesBySeq( jids, lastTimestamp, ASSISTANT_NAME, @@ -1324,7 +1912,7 @@ async function startMessageLoop(): Promise { logger.info({ count: messages.length }, 'New messages'); // Advance the "seen" cursor for all messages immediately - lastTimestamp = newTimestamp; + lastTimestamp = newSeqCursor; saveState(); // Deduplicate by group @@ -1357,34 +1945,12 @@ async function startMessageLoop(): Promise { if (processableGroupMessages.length === 0) { const lastIgnored = groupMessages[groupMessages.length - 1]; - if (lastIgnored) { - advanceLastAgentCursor(chatJid, lastIgnored.timestamp); + if (lastIgnored?.seq != null) { + advanceLastAgentCursor(chatJid, lastIgnored.seq); } continue; } - // --- Bot-collaboration timeout --- - // If all new messages are from bots, only process if a human - // sent a message within the last 12 hours. - const BOT_COLLAB_TIMEOUT_MS = 12 * 60 * 60 * 1000; - const allFromBots = processableGroupMessages.every( - (m) => m.is_from_me || !!m.is_bot_message, - ); - if (allFromBots) { - const lastHuman = getLastHumanMessageTimestamp(chatJid); - if ( - !lastHuman || - Date.now() - new Date(lastHuman).getTime() > BOT_COLLAB_TIMEOUT_MS - ) { - logger.info( - { chatJid, lastHuman }, - 'Bot-collaboration timeout: no human message within 12h, skipping', - ); - continue; - } - } - // --- End bot-collaboration timeout --- - // --- Session command interception (message loop) --- // Scan ALL messages in the batch for a session command. const loopCmdMsg = processableGroupMessages.find( @@ -1402,7 +1968,9 @@ async function startMessageLoop(): Promise { isSessionCommandSenderAllowed(loopCmdMsg.sender), ) ) { - queue.closeStdin(chatJid); + queue.closeStdin(chatJid, { + reason: 'session-command-detected', + }); } // Enqueue so processGroupMessages handles auth + cursor advancement. // Don't pipe via IPC — slash commands need a fresh agent with @@ -1430,7 +1998,7 @@ async function startMessageLoop(): Promise { // Pull all messages since lastAgentTimestamp so non-trigger // context that accumulated between triggers is included. - const allPending = getMessagesSince( + const allPending = getMessagesSinceSeq( chatJid, lastAgentTimestamp[chatJid] || '', ASSISTANT_NAME, @@ -1451,10 +2019,10 @@ async function startMessageLoop(): Promise { { chatJid, count: messagesToSend.length }, 'Piped messages to active agent', ); - advanceLastAgentCursor( - chatJid, - messagesToSend[messagesToSend.length - 1].timestamp, - ); + const endSeq = messagesToSend[messagesToSend.length - 1].seq; + if (endSeq != null) { + advanceLastAgentCursor(chatJid, endSeq); + } // Show typing indicator while the agent processes the piped message channel .setTyping?.(chatJid, true) @@ -1480,10 +2048,10 @@ async function startMessageLoop(): Promise { */ function recoverPendingMessages(): void { for (const [chatJid, group] of Object.entries(registeredGroups)) { - const sinceTimestamp = lastAgentTimestamp[chatJid] || ''; - const rawPending = getMessagesSince( + const sinceSeqCursor = lastAgentTimestamp[chatJid] || ''; + const rawPending = getMessagesSinceSeq( chatJid, - sinceTimestamp, + sinceSeqCursor, ASSISTANT_NAME, ); const recoveryChannel = findChannel(channels, chatJid); @@ -1499,15 +2067,81 @@ function recoverPendingMessages(): void { ); queue.enqueueMessageCheck(chatJid, group.folder); } else if (rawPending.length > 0) { - advanceLastAgentCursor( - chatJid, - rawPending[rawPending.length - 1].timestamp, - ); + const endSeq = rawPending[rawPending.length - 1].seq; + if (endSeq != null) { + advanceLastAgentCursor(chatJid, endSeq); + } } } } +async function announceRestartRecovery( + processStartedAtMs: number, +): Promise { + const explicitContext = consumeRestartContext(); + const dedupeSince = new Date(processStartedAtMs - 60_000).toISOString(); + if (explicitContext) { + if ( + hasRecentRestartAnnouncement(explicitContext.chatJid, dedupeSince) + ) { + logger.info( + { chatJid: explicitContext.chatJid }, + 'Skipped duplicate restart recovery announcement', + ); + return; + } + + await sendFormattedChannelMessage( + channels, + explicitContext.chatJid, + buildRestartAnnouncement(explicitContext), + ); + logger.info( + { chatJid: explicitContext.chatJid }, + 'Sent explicit restart recovery announcement', + ); + + for (const interrupted of explicitContext.interruptedGroups ?? []) { + if (interrupted.chatJid === explicitContext.chatJid) continue; + if (hasRecentRestartAnnouncement(interrupted.chatJid, dedupeSince)) { + continue; + } + await sendFormattedChannelMessage( + channels, + interrupted.chatJid, + buildInterruptedRestartAnnouncement(interrupted), + ); + } + return; + } + + const inferred = inferRecentRestartContext( + registeredGroups, + processStartedAtMs, + ); + if (!inferred) return; + + if (hasRecentRestartAnnouncement(inferred.chatJid, dedupeSince)) { + logger.info( + { chatJid: inferred.chatJid }, + 'Skipped duplicate inferred restart recovery announcement', + ); + return; + } + + await sendFormattedChannelMessage( + channels, + inferred.chatJid, + inferred.lines.join('\n'), + ); + logger.info( + { chatJid: inferred.chatJid }, + 'Sent inferred restart recovery announcement', + ); +} + async function main(): Promise { + const processStartedAtMs = Date.now(); initDatabase(); logger.info('Database initialized'); loadState(); @@ -1515,6 +2149,34 @@ async function main(): Promise { // Graceful shutdown handlers const shutdown = async (signal: string) => { logger.info({ signal }, 'Shutdown signal received'); + const interruptedGroups = queue + .getStatuses(Object.keys(registeredGroups)) + .filter( + ( + status, + ): status is typeof status & { + status: 'processing' | 'idle' | 'waiting'; + } => status.status !== 'inactive', + ) + .map((status) => ({ + chatJid: status.jid, + groupName: registeredGroups[status.jid]?.name || status.jid, + status: status.status, + elapsedMs: status.elapsedMs, + pendingMessages: status.pendingMessages, + pendingTasks: status.pendingTasks, + })); + const writtenPaths = writeShutdownRestartContext( + registeredGroups, + interruptedGroups, + signal, + ); + if (writtenPaths.length > 0) { + logger.info( + { signal, interruptedGroupCount: interruptedGroups.length, writtenPaths }, + 'Stored shutdown restart context for interrupted groups', + ); + } await queue.shutdown(10000); for (const ch of channels) await ch.disconnect(); process.exit(0); @@ -1581,15 +2243,12 @@ async function main(): Promise { queue, onProcess: (groupJid, proc, processName, groupFolder) => queue.registerProcess(groupJid, proc, processName, groupFolder), - sendMessage: async (jid, rawText) => { - const channel = findChannel(channels, jid); - if (!channel) { - logger.warn({ jid }, 'No channel owns JID, cannot send message'); - return; - } - const text = formatOutbound(rawText); - if (text) await channel.sendMessage(jid, text); - }, + sendMessage: (jid, rawText) => + sendFormattedChannelMessage(channels, jid, rawText), + sendTrackedMessage: (jid, rawText) => + sendFormattedTrackedChannelMessage(channels, jid, rawText), + editTrackedMessage: (jid, messageId, rawText) => + editFormattedTrackedChannelMessage(channels, jid, messageId, rawText), }); startIpcWatcher({ sendMessage: (jid, text) => { @@ -1611,6 +2270,7 @@ async function main(): Promise { }); queue.setProcessMessagesFn(processGroupMessages); recoverPendingMessages(); + await announceRestartRecovery(processStartedAtMs); // Purge old messages in status channel before creating fresh dashboards if (STATUS_CHANNEL_ID && SERVICE_AGENT_TYPE === 'claude-code') { const statusJid = `dc:${STATUS_CHANNEL_ID}`; diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index e06ccff..add652f 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -7,6 +7,7 @@ vi.mock('./agent-runner.js', () => ({ })); vi.mock('./config.js', () => ({ + DATA_DIR: '/tmp/nanoclaw-test-data', isSessionCommandSenderAllowed: vi.fn(() => false), })); @@ -27,6 +28,20 @@ vi.mock('./logger.js', () => ({ }, })); +vi.mock('./provider-fallback.js', () => ({ + detectFallbackTrigger: vi.fn(() => ({ shouldFallback: false, reason: '' })), + getActiveProvider: vi.fn(() => 'claude'), + getFallbackEnvOverrides: vi.fn(() => ({ + ANTHROPIC_BASE_URL: 'https://api.kimi.com/coding/', + ANTHROPIC_AUTH_TOKEN: 'test-kimi-key', + ANTHROPIC_MODEL: 'kimi-k2.5', + })), + getFallbackProviderName: vi.fn(() => 'kimi'), + hasGroupProviderOverride: vi.fn(() => false), + isFallbackEnabled: vi.fn(() => true), + markPrimaryCooldown: vi.fn(), +})); + vi.mock('./sender-allowlist.js', () => ({ isTriggerAllowed: vi.fn(() => true), loadSenderAllowlist: vi.fn(() => ({})), @@ -42,6 +57,7 @@ vi.mock('./session-commands.js', () => ({ import * as agentRunner from './agent-runner.js'; import * as db from './db.js'; import { createMessageRuntime } from './message-runtime.js'; +import * as providerFallback from './provider-fallback.js'; import type { Channel, RegisteredGroup } from './types.js'; function makeGroup(agentType: 'claude-code' | 'codex'): RegisteredGroup { @@ -72,6 +88,21 @@ function makeChannel(chatJid: string): Channel { describe('createMessageRuntime', () => { beforeEach(() => { vi.clearAllMocks(); + vi.mocked(providerFallback.getActiveProvider).mockReturnValue('claude'); + vi.mocked(providerFallback.getFallbackProviderName).mockReturnValue('kimi'); + vi.mocked(providerFallback.getFallbackEnvOverrides).mockReturnValue({ + ANTHROPIC_BASE_URL: 'https://api.kimi.com/coding/', + ANTHROPIC_AUTH_TOKEN: 'test-kimi-key', + ANTHROPIC_MODEL: 'kimi-k2.5', + }); + vi.mocked(providerFallback.hasGroupProviderOverride).mockReturnValue( + false, + ); + vi.mocked(providerFallback.isFallbackEnabled).mockReturnValue(true); + vi.mocked(providerFallback.detectFallbackTrigger).mockReturnValue({ + shouldFallback: false, + reason: '', + }); }); it('clears Claude sessions and closes stdin immediately on poisoned output', async () => { @@ -318,6 +349,114 @@ describe('createMessageRuntime', () => { } }); + it('resets the idle timeout when follow-up Codex progress keeps arriving', async () => { + vi.useFakeTimers(); + const chatJid = 'group@test'; + const group = makeGroup('codex'); + const channel = makeChannel(chatJid); + const closeStdin = vi.fn(); + + vi.mocked(db.getMessagesSince).mockReturnValue([ + { + id: 'msg-1', + chat_jid: chatJid, + sender: 'user@test', + sender_name: 'User', + content: 'hello', + timestamp: '2026-03-19T00:00:00.000Z', + }, + ]); + + vi.mocked(agentRunner.runAgentProcess).mockImplementation( + async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + result: '초기 응답입니다.', + newSessionId: 'session-follow-up', + }); + await vi.advanceTimersByTimeAsync(800); + expect(closeStdin).not.toHaveBeenCalled(); + + await onOutput?.({ + status: 'success', + phase: 'progress', + result: '후속 작업 진행 중입니다.', + newSessionId: 'session-follow-up', + }); + await vi.advanceTimersByTimeAsync(800); + expect(closeStdin).not.toHaveBeenCalled(); + + await onOutput?.({ + status: 'success', + phase: 'progress', + result: '후속 작업 계속 진행 중입니다.', + newSessionId: 'session-follow-up', + }); + await vi.advanceTimersByTimeAsync(800); + expect(closeStdin).not.toHaveBeenCalled(); + + await onOutput?.({ + status: 'success', + result: null, + newSessionId: 'session-follow-up', + }); + + return { + status: 'success', + result: null, + newSessionId: 'session-follow-up', + }; + }, + ); + + const runtime = createMessageRuntime({ + assistantName: 'Andy', + idleTimeout: 1_000, + pollInterval: 1_000, + timezone: 'UTC', + triggerPattern: /^@Andy\b/i, + channels: [channel], + queue: { + registerProcess: vi.fn(), + closeStdin, + notifyIdle: vi.fn(), + } as any, + getRegisteredGroups: () => ({ [chatJid]: group }), + getSessions: () => ({}), + getLastTimestamp: () => '', + setLastTimestamp: vi.fn(), + getLastAgentTimestamps: () => ({}), + saveState: vi.fn(), + persistSession: vi.fn(), + clearSession: vi.fn(), + }); + + try { + const result = await runtime.processGroupMessages(chatJid, { + runId: 'run-follow-up-progress', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(closeStdin).not.toHaveBeenCalled(); + expect(channel.sendMessage).toHaveBeenCalledWith( + chatJid, + '초기 응답입니다.', + ); + expect(channel.sendAndTrack).toHaveBeenCalledWith( + chatJid, + '후속 작업 진행 중입니다.\n\n0초', + ); + expect(channel.editMessage).toHaveBeenCalledWith( + chatJid, + 'progress-1', + '후속 작업 계속 진행 중입니다.\n\n0초', + ); + } finally { + vi.useRealTimers(); + } + }); + it('formats longer Codex progress durations with minutes and hours', async () => { vi.useFakeTimers(); const chatJid = 'group@test'; @@ -1019,4 +1158,172 @@ describe('createMessageRuntime', () => { expect(lastAgentTimestamps[chatJid]).toBe('2026-03-19T00:00:00.000Z'); expect(saveState).toHaveBeenCalled(); }); + + it('retries with the fallback provider when Claude returns a 429 error before any output', async () => { + const chatJid = 'group@test'; + const group = makeGroup('claude-code'); + const channel = makeChannel(chatJid); + + vi.mocked(db.getMessagesSince).mockReturnValue([ + { + id: 'msg-1', + chat_jid: chatJid, + sender: 'user@test', + sender_name: 'User', + content: 'hello', + timestamp: '2026-03-19T00:00:00.000Z', + }, + ]); + + vi.mocked(providerFallback.detectFallbackTrigger).mockReturnValue({ + shouldFallback: true, + reason: '429', + retryAfterMs: 60_000, + }); + + vi.mocked(agentRunner.runAgentProcess) + .mockResolvedValueOnce({ + status: 'error', + result: null, + error: '429 rate limited retry after 60', + }) + .mockImplementationOnce(async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'final', + result: 'fallback 응답입니다.', + }); + return { + status: 'success', + result: null, + }; + }); + + const runtime = createMessageRuntime({ + assistantName: 'Andy', + idleTimeout: 1_000, + pollInterval: 1_000, + timezone: 'UTC', + triggerPattern: /^@Andy\b/i, + channels: [channel], + queue: { + registerProcess: vi.fn(), + closeStdin: vi.fn(), + notifyIdle: vi.fn(), + } as any, + getRegisteredGroups: () => ({ [chatJid]: group }), + getSessions: () => ({}), + getLastTimestamp: () => '', + setLastTimestamp: vi.fn(), + getLastAgentTimestamps: () => ({}), + saveState: vi.fn(), + persistSession: vi.fn(), + clearSession: vi.fn(), + }); + + const result = await runtime.processGroupMessages(chatJid, { + runId: 'run-fallback-429', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(agentRunner.runAgentProcess).toHaveBeenCalledTimes(2); + expect(agentRunner.runAgentProcess).toHaveBeenNthCalledWith( + 2, + expect.anything(), + expect.objectContaining({ sessionId: undefined }), + expect.any(Function), + expect.any(Function), + expect.objectContaining({ + ANTHROPIC_BASE_URL: 'https://api.kimi.com/coding/', + ANTHROPIC_MODEL: 'kimi-k2.5', + }), + ); + expect(providerFallback.markPrimaryCooldown).toHaveBeenCalledWith( + '429', + 60_000, + ); + expect(channel.sendMessage).toHaveBeenCalledWith( + chatJid, + 'fallback 응답입니다.', + ); + }); + + it('retries with the fallback provider when Claude ends with success-null-result before any output', async () => { + const chatJid = 'group@test'; + const group = makeGroup('claude-code'); + const channel = makeChannel(chatJid); + + vi.mocked(db.getMessagesSince).mockReturnValue([ + { + id: 'msg-1', + chat_jid: chatJid, + sender: 'user@test', + sender_name: 'User', + content: 'hello', + timestamp: '2026-03-19T00:00:00.000Z', + }, + ]); + + vi.mocked(agentRunner.runAgentProcess) + .mockImplementationOnce(async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + result: null, + }); + return { + status: 'success', + result: null, + }; + }) + .mockImplementationOnce(async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'final', + result: 'success-null-result 폴백 응답입니다.', + }); + return { + status: 'success', + result: null, + }; + }); + + const runtime = createMessageRuntime({ + assistantName: 'Andy', + idleTimeout: 1_000, + pollInterval: 1_000, + timezone: 'UTC', + triggerPattern: /^@Andy\b/i, + channels: [channel], + queue: { + registerProcess: vi.fn(), + closeStdin: vi.fn(), + notifyIdle: vi.fn(), + } as any, + getRegisteredGroups: () => ({ [chatJid]: group }), + getSessions: () => ({}), + getLastTimestamp: () => '', + setLastTimestamp: vi.fn(), + getLastAgentTimestamps: () => ({}), + saveState: vi.fn(), + persistSession: vi.fn(), + clearSession: vi.fn(), + }); + + const result = await runtime.processGroupMessages(chatJid, { + runId: 'run-fallback-success-null', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(agentRunner.runAgentProcess).toHaveBeenCalledTimes(2); + expect(providerFallback.markPrimaryCooldown).toHaveBeenCalledWith( + 'success-null-result', + undefined, + ); + expect(channel.sendMessage).toHaveBeenCalledWith( + chatJid, + 'success-null-result 폴백 응답입니다.', + ); + }); }); diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 36be661..64773af 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -8,12 +8,20 @@ import { import { getAllChats, getAllTasks, - getLastHumanMessageTimestamp, getMessagesSince, getNewMessages, } from './db.js'; -import { isSessionCommandSenderAllowed } from './config.js'; +import { DATA_DIR, isSessionCommandSenderAllowed } from './config.js'; import { GroupQueue, GroupRunContext } from './group-queue.js'; +import { + detectFallbackTrigger, + getActiveProvider, + getFallbackEnvOverrides, + getFallbackProviderName, + hasGroupProviderOverride, + isFallbackEnabled, + markPrimaryCooldown, +} from './provider-fallback.js'; import { findChannel, formatMessages, formatOutbound } from './router.js'; import { isTriggerAllowed, loadSenderAllowlist } from './sender-allowlist.js'; import { @@ -26,6 +34,7 @@ import { shouldResetSessionOnAgentFailure } from './session-recovery.js'; import { isTaskStatusControlMessage } from './task-scheduler.js'; import { Channel, NewMessage, RegisteredGroup } from './types.js'; import { logger } from './logger.js'; +import path from 'path'; export interface MessageRuntimeDeps { assistantName: string; @@ -121,63 +130,298 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { let resetSessionRequested = false; - const wrappedOnOutput = onOutput - ? async (output: AgentOutput) => { - if (output.newSessionId) { - deps.persistSession(group.folder, output.newSessionId); + const settingsPath = path.join( + DATA_DIR, + 'sessions', + group.folder, + '.claude', + 'settings.json', + ); + const groupHasOverride = hasGroupProviderOverride(settingsPath); + const canFallback = + isClaudeCodeAgent && isFallbackEnabled() && !groupHasOverride; + + const agentInput = { + prompt, + sessionId, + groupFolder: group.folder, + chatJid, + runId, + isMain, + assistantName: deps.assistantName, + }; + + const runAttempt = async (provider: string): Promise<{ + output?: AgentOutput; + error?: unknown; + sawOutput: boolean; + sawSuccessNullResultWithoutOutput: boolean; + streamedTriggerReason?: { + reason: string; + retryAfterMs?: number; + }; + }> => { + const persistSessionIds = provider === 'claude'; + let sawOutput = false; + let sawSuccessNullResultWithoutOutput = false; + let streamedTriggerReason: + | { + reason: string; + retryAfterMs?: number; } - if (isClaudeCodeAgent && shouldResetSessionOnAgentFailure(output)) { - resetSessionRequested = true; + | undefined; + + const wrappedOnOutput = onOutput + ? async (output: AgentOutput) => { + if (persistSessionIds && output.newSessionId) { + deps.persistSession(group.folder, output.newSessionId); + } + if ( + persistSessionIds && + isClaudeCodeAgent && + shouldResetSessionOnAgentFailure(output) + ) { + resetSessionRequested = true; + } + if (output.result !== null && output.result !== undefined) { + sawOutput = true; + } else if ( + provider === 'claude' && + output.status === 'success' && + !sawOutput + ) { + sawSuccessNullResultWithoutOutput = true; + } + if ( + provider === 'claude' && + output.status === 'error' && + !sawOutput && + !streamedTriggerReason + ) { + const trigger = detectFallbackTrigger(output.error); + if (trigger.shouldFallback) { + streamedTriggerReason = { + reason: trigger.reason, + retryAfterMs: trigger.retryAfterMs, + }; + } + } + await onOutput(output); } - await onOutput(output); - } - : undefined; + : undefined; - try { - const output = await runAgentProcess( - group, - { - prompt, - sessionId, - groupFolder: group.folder, - chatJid, - runId, - isMain, - assistantName: deps.assistantName, - }, - (proc, processName) => - deps.queue.registerProcess(chatJid, proc, processName, group.folder), - wrappedOnOutput, - ); - - if (output.newSessionId) { - deps.persistSession(group.folder, output.newSessionId); - } - - if ( - isClaudeCodeAgent && - (resetSessionRequested || shouldResetSessionOnAgentFailure(output)) - ) { - deps.clearSession(group.folder); - logger.warn( - { group: group.name, chatJid, runId }, - 'Cleared poisoned agent session after unrecoverable error', + if (provider !== 'claude') { + logger.info( + { chatJid, group: group.name, groupFolder: group.folder, runId, provider }, + `Claude provider in cooldown, routing request to ${provider}`, ); } - if (output.status === 'error') { + logger.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + provider, + canFallback, + groupHasOverride, + }, + `Using provider: ${provider}`, + ); + + try { + const output = await runAgentProcess( + group, + { + ...agentInput, + sessionId: persistSessionIds ? sessionId : undefined, + }, + (proc, processName) => + deps.queue.registerProcess(chatJid, proc, processName, group.folder), + wrappedOnOutput, + provider === 'claude' ? undefined : getFallbackEnvOverrides(), + ); + + if (persistSessionIds && output.newSessionId) { + deps.persistSession(group.folder, output.newSessionId); + } + + logger.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + provider, + status: output.status, + sawOutput, + }, + `Provider response completed (provider: ${provider})`, + ); + + return { + output, + sawOutput, + sawSuccessNullResultWithoutOutput, + streamedTriggerReason, + }; + } catch (error) { + return { + error, + sawOutput, + sawSuccessNullResultWithoutOutput, + streamedTriggerReason, + }; + } + }; + + const runFallbackAttempt = async ( + reason: string, + retryAfterMs?: number, + ): Promise<'success' | 'error'> => { + const fallbackName = getFallbackProviderName(); + markPrimaryCooldown(reason, retryAfterMs); + + logger.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + reason, + retryAfterMs, + fallbackProvider: fallbackName, + }, + `Falling back to provider: ${fallbackName} (reason: ${reason})`, + ); + + const fallbackAttempt = await runAttempt(fallbackName); + if (fallbackAttempt.error) { logger.error( - { group: group.name, chatJid, runId, error: output.error }, - 'Agent process error', + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + provider: fallbackName, + err: fallbackAttempt.error, + }, + 'Fallback provider also threw', + ); + return 'error'; + } + + if (fallbackAttempt.output?.status === 'error') { + logger.error( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + provider: fallbackName, + error: fallbackAttempt.output.error, + }, + `Fallback provider (${fallbackName}) also failed`, ); return 'error'; } return 'success'; - } catch (err) { - logger.error({ group: group.name, chatJid, runId, err }, 'Agent error'); + }; + + const provider = canFallback ? getActiveProvider() : 'claude'; + const primaryAttempt = await runAttempt(provider); + + if (primaryAttempt.error) { + if (canFallback && provider === 'claude' && !primaryAttempt.sawOutput) { + const errMsg = + primaryAttempt.error instanceof Error + ? primaryAttempt.error.message + : String(primaryAttempt.error); + const trigger = primaryAttempt.streamedTriggerReason + ? { + shouldFallback: true, + reason: primaryAttempt.streamedTriggerReason.reason, + retryAfterMs: primaryAttempt.streamedTriggerReason.retryAfterMs, + } + : detectFallbackTrigger(errMsg); + if (trigger.shouldFallback) { + return runFallbackAttempt(trigger.reason, trigger.retryAfterMs); + } + } + + logger.error( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + provider, + err: primaryAttempt.error, + }, + 'Agent error', + ); return 'error'; } + + const output = primaryAttempt.output; + if (!output) { + logger.error( + { chatJid, group: group.name, groupFolder: group.folder, runId, provider }, + 'Agent produced no output object', + ); + return 'error'; + } + + if ( + canFallback && + provider === 'claude' && + !primaryAttempt.sawOutput && + primaryAttempt.sawSuccessNullResultWithoutOutput + ) { + return runFallbackAttempt('success-null-result'); + } + + if ( + isClaudeCodeAgent && + (resetSessionRequested || shouldResetSessionOnAgentFailure(output)) + ) { + deps.clearSession(group.folder); + logger.warn( + { group: group.name, chatJid, runId }, + 'Cleared poisoned agent session after unrecoverable error', + ); + } + + if (output.status === 'error') { + if (canFallback && provider === 'claude' && !primaryAttempt.sawOutput) { + const trigger = primaryAttempt.streamedTriggerReason + ? { + shouldFallback: true, + reason: primaryAttempt.streamedTriggerReason.reason, + retryAfterMs: primaryAttempt.streamedTriggerReason.retryAfterMs, + } + : detectFallbackTrigger(output.error); + if (trigger.shouldFallback) { + return runFallbackAttempt(trigger.reason, trigger.retryAfterMs); + } + } + + logger.error( + { + group: group.name, + chatJid, + runId, + provider, + error: output.error, + }, + 'Agent process error', + ); + return 'error'; + } + + return 'success'; }; const processGroupMessages = async ( @@ -323,6 +567,21 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { const resetIdleTimer = () => { if (idleTimer) clearTimeout(idleTimer); idleTimer = setTimeout(() => { + if (followUpQueuedAt !== null) { + logger.warn( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + followUpQueuedAt: new Date(followUpQueuedAt).toISOString(), + followUpQueuedTextLength, + followUpQueuedFilename, + followUpWaitMs: Date.now() - followUpQueuedAt, + }, + 'Idle timeout reached while a queued follow-up still had no agent output', + ); + } logger.info( { chatJid, group: group.name, groupFolder: group.folder, runId }, 'Idle timeout reached, closing active agent stdin', @@ -333,6 +592,111 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { }); }, deps.idleTimeout); }; + let followUpQueuedAt: number | null = null; + let followUpQueuedTextLength: number | null = null; + let followUpQueuedFilename: string | null = null; + let followUpNoOutputWarnTimer: ReturnType | null = null; + const FOLLOW_UP_NO_OUTPUT_WARN_MS = 10_000; + + const clearFollowUpNoOutputWarnTimer = () => { + if (followUpNoOutputWarnTimer) { + clearTimeout(followUpNoOutputWarnTimer); + followUpNoOutputWarnTimer = null; + } + }; + + const clearPendingFollowUpDiagnostics = () => { + clearFollowUpNoOutputWarnTimer(); + followUpQueuedAt = null; + followUpQueuedTextLength = null; + followUpQueuedFilename = null; + }; + + const scheduleFollowUpNoOutputWarning = () => { + clearFollowUpNoOutputWarnTimer(); + followUpNoOutputWarnTimer = setTimeout(() => { + if (followUpQueuedAt === null) return; + logger.warn( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + followUpQueuedAt: new Date(followUpQueuedAt).toISOString(), + followUpQueuedTextLength, + followUpQueuedFilename, + elapsedMs: Date.now() - followUpQueuedAt, + }, + 'No agent output observed within 10s of queuing a follow-up message', + ); + }, FOLLOW_UP_NO_OUTPUT_WARN_MS); + }; + + const noteFollowUpQueued = (meta?: { + source: 'follow-up'; + textLength: number; + filename: string; + }) => { + if (meta?.source !== 'follow-up') return; + followUpQueuedAt = Date.now(); + followUpQueuedTextLength = meta.textLength; + followUpQueuedFilename = meta.filename; + scheduleFollowUpNoOutputWarning(); + logger.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + followUpQueuedTextLength, + followUpQueuedFilename, + }, + 'Registered follow-up activity for active agent', + ); + }; + + const noteAgentOutputObserved = (phase?: string) => { + if (followUpQueuedAt === null) return; + logger.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + followUpQueuedAt: new Date(followUpQueuedAt).toISOString(), + followUpQueuedTextLength, + followUpQueuedFilename, + followUpWaitMs: Date.now() - followUpQueuedAt, + resultPhase: phase, + }, + 'Agent produced output after a queued follow-up', + ); + clearPendingFollowUpDiagnostics(); + }; + + const warnFollowUpEndedWithoutOutput = (reason: string) => { + if (followUpQueuedAt === null) return; + logger.warn( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + followUpQueuedAt: new Date(followUpQueuedAt).toISOString(), + followUpQueuedTextLength, + followUpQueuedFilename, + followUpWaitMs: Date.now() - followUpQueuedAt, + reason, + }, + 'Active agent ended a turn without any output after a queued follow-up', + ); + clearPendingFollowUpDiagnostics(); + }; + + deps.queue.setActivityTouch?.(chatJid, (meta) => { + noteFollowUpQueued(meta); + resetIdleTimer(); + }); let hadError = false; let finalOutputSentToUser = false; @@ -463,7 +827,15 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { } if (channel.sendAndTrack) { - progressMessageId = await channel.sendAndTrack(chatJid, rendered); + try { + progressMessageId = await channel.sendAndTrack(chatJid, rendered); + } catch (err) { + logger.warn( + { chatJid, group: group.name, groupFolder: group.folder, runId, err }, + 'Failed to send tracked progress message', + ); + return; + } if (progressMessageId) { logger.info( { @@ -532,6 +904,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { }, `Agent output: ${raw.slice(0, 200)}`, ); + noteAgentOutputObserved(result.phase); if (result.phase === 'progress') { // Detect new logical turn (follow-up IPC): if a final was already // sent in this run, a new progress output means a follow-up turn @@ -557,6 +930,9 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { if (text) { await sendProgressMessage(text); } + if (!poisonedSessionDetected) { + resetIdleTimer(); + } return; } @@ -584,6 +960,9 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { latestModelProgressTextForFinalFallback = null; } } else { + if (result.status === 'success') { + warnFollowUpEndedWithoutOutput('success-null-result'); + } await finalizeProgressMessage(); } @@ -636,6 +1015,8 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { clearProgressTicker(); if (idleTimer) clearTimeout(idleTimer); + clearPendingFollowUpDiagnostics(); + deps.queue.setActivityTouch?.(chatJid, null); if (hadError) { if (finalOutputSentToUser || progressOutputSentToUser) { @@ -723,23 +1104,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { } const isMainGroup = group.isMain === true; - const allFromBots = groupMessages.every( - (msg) => msg.is_from_me || !!msg.is_bot_message, - ); - if (allFromBots) { - const lastHuman = getLastHumanMessageTimestamp(chatJid); - if ( - !lastHuman || - Date.now() - new Date(lastHuman).getTime() > 12 * 60 * 60 * 1000 - ) { - logger.info( - { chatJid, lastHuman }, - 'Bot-collaboration timeout: no human message within 12h, skipping', - ); - continue; - } - } - const loopCmdMsg = groupMessages.find( (msg) => extractSessionCommand(msg.content, deps.triggerPattern) !== diff --git a/src/provider-fallback.ts b/src/provider-fallback.ts new file mode 100644 index 0000000..5274dc7 --- /dev/null +++ b/src/provider-fallback.ts @@ -0,0 +1,297 @@ +/** + * Provider Fallback Module + * + * Manages automatic fallback from the primary provider (Claude) to a + * fallback provider (e.g. Kimi K2.5) when 429/rate-limit or network + * errors are detected. + * + * Cooldown-based recovery: + * Claude 429 → immediate Kimi retry for that turn + * Claude enters cooldown (retry-after header or default 10 min) + * During cooldown → skip Claude, route directly to fallback + * After cooldown → try Claude first again + */ + +import fs from 'fs'; + +import { readEnvFile } from './env.js'; +import { logger } from './logger.js'; + +// ── Types ──────────────────────────────────────────────────────── + +export type ProviderName = 'claude' | string; // fallback name is configurable + +export interface FallbackTriggerResult { + shouldFallback: boolean; + reason: string; + retryAfterMs?: number; +} + +interface CooldownState { + startedAt: number; + expiresAt: number; + reason: string; +} + +interface FallbackConfig { + enabled: boolean; + providerName: string; // e.g. "kimi" + baseUrl: string; + authToken: string; + model: string; + smallModel: string; + defaultCooldownMs: number; +} + +// ── State ──────────────────────────────────────────────────────── + +let cooldown: CooldownState | null = null; + +// ── Config ─────────────────────────────────────────────────────── + +let _config: FallbackConfig | null = null; + +function loadConfig(): FallbackConfig { + if (_config) return _config; + + const env = readEnvFile([ + 'FALLBACK_PROVIDER_NAME', + 'FALLBACK_BASE_URL', + 'FALLBACK_AUTH_TOKEN', + 'FALLBACK_MODEL', + 'FALLBACK_SMALL_MODEL', + 'FALLBACK_COOLDOWN_MS', + ]); + + const baseUrl = + process.env.FALLBACK_BASE_URL || env.FALLBACK_BASE_URL || ''; + const authToken = + process.env.FALLBACK_AUTH_TOKEN || env.FALLBACK_AUTH_TOKEN || ''; + const model = process.env.FALLBACK_MODEL || env.FALLBACK_MODEL || ''; + + _config = { + enabled: Boolean(baseUrl && authToken && model), + providerName: + process.env.FALLBACK_PROVIDER_NAME || + env.FALLBACK_PROVIDER_NAME || + 'kimi', + baseUrl, + authToken, + model, + smallModel: + process.env.FALLBACK_SMALL_MODEL || env.FALLBACK_SMALL_MODEL || model, + defaultCooldownMs: parseInt( + process.env.FALLBACK_COOLDOWN_MS || env.FALLBACK_COOLDOWN_MS || '600000', + 10, + ), + }; + + if (_config.enabled) { + logger.info( + { + provider: _config.providerName, + model: _config.model, + cooldownMs: _config.defaultCooldownMs, + }, + 'Provider fallback configured', + ); + } + + return _config; +} + +/** Force re-read of config (useful after .env changes). */ +export function resetFallbackConfig(): void { + _config = null; +} + +// ── Public API ─────────────────────────────────────────────────── + +/** Check whether the fallback system is configured and available. */ +export function isFallbackEnabled(): boolean { + return loadConfig().enabled; +} + +/** Get the display name of the fallback provider (e.g. "kimi"). */ +export function getFallbackProviderName(): string { + return loadConfig().providerName; +} + +/** + * Determine which provider should be used for the next request. + * Returns 'claude' when Claude is healthy or cooldown has expired, + * or the fallback provider name during an active cooldown. + */ +export function getActiveProvider(): string { + const config = loadConfig(); + if (!config.enabled) return 'claude'; + + if (cooldown) { + if (Date.now() < cooldown.expiresAt) { + return config.providerName; + } + // Cooldown expired — try Claude again + logger.info( + { + provider: 'claude', + cooldownDurationMs: cooldown.expiresAt - cooldown.startedAt, + reason: cooldown.reason, + }, + 'Claude cooldown expired, retrying primary provider', + ); + cooldown = null; + } + + return 'claude'; +} + +/** + * Mark Claude as rate-limited. All subsequent requests will route to + * the fallback provider until the cooldown expires. + */ +export function markPrimaryCooldown( + reason: string, + retryAfterMs?: number, +): void { + const config = loadConfig(); + const durationMs = retryAfterMs || config.defaultCooldownMs; + const now = Date.now(); + + cooldown = { + startedAt: now, + expiresAt: now + durationMs, + reason, + }; + + logger.info( + { + reason, + cooldownMs: durationMs, + expiresAt: new Date(cooldown.expiresAt).toISOString(), + fallbackProvider: config.providerName, + }, + `Falling back to provider: ${config.providerName} (reason: ${reason}, cooldownMs: ${durationMs})`, + ); +} + +/** Manually clear cooldown (e.g. after a successful Claude response). */ +export function clearPrimaryCooldown(): void { + if (cooldown) { + logger.info( + { reason: cooldown.reason }, + 'Claude cooldown cleared manually', + ); + cooldown = null; + } +} + +/** Get current cooldown info (for diagnostics / status dashboard). */ +export function getCooldownInfo(): { + active: boolean; + reason?: string; + expiresAt?: string; + remainingMs?: number; +} { + if (!cooldown || Date.now() >= cooldown.expiresAt) { + return { active: false }; + } + return { + active: true, + reason: cooldown.reason, + expiresAt: new Date(cooldown.expiresAt).toISOString(), + remainingMs: cooldown.expiresAt - Date.now(), + }; +} + +/** + * Build the env-var overrides that make Claude Code SDK talk to + * the fallback provider instead of Claude. + */ +export function getFallbackEnvOverrides(): Record { + const config = loadConfig(); + return { + ANTHROPIC_BASE_URL: config.baseUrl, + ANTHROPIC_AUTH_TOKEN: config.authToken, + ANTHROPIC_MODEL: config.model, + ANTHROPIC_SMALL_FAST_MODEL: config.smallModel, + // Disable non-essential traffic (usage telemetry etc.) on fallback + CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC: '1', + // Generous timeout for third-party APIs + API_TIMEOUT_MS: '3000000', + // Disable tool search (not supported by most fallback providers) + ENABLE_TOOL_SEARCH: 'false', + }; +} + +/** + * Inspect an agent error string and decide whether it warrants + * a provider fallback. + * + * Triggers: + * - 429 / rate limit / too many requests + * - 503 / overloaded (transient provider issue) + * - Network / connection errors + * + * Does NOT trigger for: + * - Poisoned sessions + * - Prompt / tool failures + * - Timeouts (agent took too long, not a provider issue) + */ +export function detectFallbackTrigger( + error?: string | null, +): FallbackTriggerResult { + if (!error) return { shouldFallback: false, reason: '' }; + + const lower = error.toLowerCase(); + + // 429 Rate Limit + if ( + lower.includes('429') || + lower.includes('rate limit') || + lower.includes('too many requests') || + lower.includes('rate_limit') + ) { + // Try to extract retry-after value (seconds → ms) + const retryMatch = error.match(/retry[\s_-]*after[:\s]*(\d+)/i); + const retryAfterMs = retryMatch + ? parseInt(retryMatch[1], 10) * 1000 + : undefined; + return { shouldFallback: true, reason: '429', retryAfterMs }; + } + + // 503 Overloaded + if (lower.includes('503') || lower.includes('overloaded')) { + return { shouldFallback: true, reason: 'overloaded' }; + } + + // Network / connection errors + if ( + lower.includes('econnrefused') || + lower.includes('econnreset') || + lower.includes('etimedout') || + lower.includes('enotfound') || + lower.includes('fetch failed') || + lower.includes('network error') + ) { + return { shouldFallback: true, reason: 'network-error' }; + } + + return { shouldFallback: false, reason: '' }; +} + +/** + * Check whether a per-group settings.json already overrides the + * provider (e.g. the Kimi test channel). If so, we should NOT + * apply fallback env overrides on top — the channel already has + * its own provider configuration. + */ +export function hasGroupProviderOverride(settingsJsonPath: string): boolean { + try { + const raw = fs.readFileSync(settingsJsonPath, 'utf-8'); + const settings = JSON.parse(raw); + const env = settings?.env || {}; + return Boolean(env.ANTHROPIC_BASE_URL || env.ANTHROPIC_MODEL); + } catch { + return false; + } +} diff --git a/src/restart-context-cli.ts b/src/restart-context-cli.ts new file mode 100644 index 0000000..d599e8b --- /dev/null +++ b/src/restart-context-cli.ts @@ -0,0 +1,70 @@ +import { SERVICE_ID } from './config.js'; +import { writeRestartContext } from './restart-context.js'; + +function printUsageAndExit(): never { + console.error( + [ + 'Usage:', + ' tsx src/restart-context-cli.ts write --chat-jid --summary [--verify ...] [--service-id ...]', + ].join('\n'), + ); + process.exit(1); +} + +const [, , command, ...args] = process.argv; +if (command !== 'write') { + printUsageAndExit(); +} + +let chatJid = ''; +let summary = ''; +const verify: string[] = []; +const serviceIds: string[] = []; + +for (let i = 0; i < args.length; i += 1) { + const arg = args[i]; + const value = args[i + 1]; + if (!value) { + printUsageAndExit(); + } + + if (arg === '--chat-jid' || arg === '--jid') { + chatJid = value; + i += 1; + continue; + } + if (arg === '--summary') { + summary = value; + i += 1; + continue; + } + if (arg === '--verify') { + verify.push(value); + i += 1; + continue; + } + if (arg === '--service-id') { + serviceIds.push(value); + i += 1; + continue; + } + + printUsageAndExit(); +} + +if (!chatJid || !summary) { + printUsageAndExit(); +} + +const written = writeRestartContext( + { + chatJid, + summary, + verify, + }, + serviceIds.length > 0 ? serviceIds : [SERVICE_ID], +); + +for (const filePath of written) { + console.log(filePath); +} diff --git a/src/restart-context.ts b/src/restart-context.ts new file mode 100644 index 0000000..7700110 --- /dev/null +++ b/src/restart-context.ts @@ -0,0 +1,258 @@ +import { execSync } from 'child_process'; +import fs from 'fs'; +import path from 'path'; + +import { DATA_DIR, SERVICE_ID, TIMEZONE } from './config.js'; +import { logger } from './logger.js'; +import type { RegisteredGroup } from './types.js'; + +export interface RestartInterruptedGroup { + chatJid: string; + groupName: string; + status: 'processing' | 'idle' | 'waiting'; + elapsedMs: number | null; + pendingMessages: boolean; + pendingTasks: number; +} + +export interface RestartContext { + chatJid: string; + summary: string; + verify: string[]; + writtenAt: string; + source?: 'explicit' | 'shutdown-snapshot'; + signal?: string; + interruptedGroups?: RestartInterruptedGroup[]; +} + +export interface InferredRestartContext { + chatJid: string; + lines: string[]; +} + +const INFER_WINDOW_MS = 3 * 60 * 1000; + +function getRestartContextPath(serviceId: string = SERVICE_ID): string { + return path.join(DATA_DIR, `restart-context.${serviceId}.json`); +} + +function formatKoreanTimestamp(timestamp: string | number | Date): string { + return new Date(timestamp).toLocaleString('ko-KR', { + timeZone: TIMEZONE, + hour12: false, + }); +} + +function getMainGroupJid( + registeredGroups: Record, +): string | null { + const mainEntry = Object.entries(registeredGroups).find( + ([, group]) => group.isMain === true, + ); + return mainEntry?.[0] ?? null; +} + +function readRestartContextFile(filePath: string): RestartContext | null { + if (!fs.existsSync(filePath)) return null; + try { + return JSON.parse(fs.readFileSync(filePath, 'utf8')) as RestartContext; + } catch (err) { + logger.warn( + { err, filePath }, + 'Failed to parse restart context file; ignoring invalid content', + ); + return null; + } +} + +function getLatestDistBuildTime(): number | null { + const distDir = path.join(process.cwd(), 'dist'); + if (!fs.existsSync(distDir)) return null; + + let latestMtime = 0; + for (const entry of fs.readdirSync(distDir)) { + if (!entry.endsWith('.js')) continue; + const stat = fs.statSync(path.join(distDir, entry)); + latestMtime = Math.max(latestMtime, stat.mtimeMs); + } + return latestMtime > 0 ? latestMtime : null; +} + +function getRecentCommit(processStartedAtMs: number): string | null { + try { + const sinceSeconds = Math.max( + 60, + Math.ceil(INFER_WINDOW_MS / 1000) + 30, + ).toString(); + const output = execSync( + `git log --since='${sinceSeconds} seconds ago' -1 --format=%h%x20%s`, + { + cwd: process.cwd(), + stdio: ['ignore', 'pipe', 'ignore'], + encoding: 'utf8', + }, + ).trim(); + if (!output) return null; + + const commitTime = execSync('git log -1 --format=%ct', { + cwd: process.cwd(), + stdio: ['ignore', 'pipe', 'ignore'], + encoding: 'utf8', + }).trim(); + const commitTimeMs = Number(commitTime) * 1000; + if ( + Number.isFinite(commitTimeMs) && + commitTimeMs >= processStartedAtMs - INFER_WINDOW_MS + ) { + return output; + } + } catch { + // Ignore git inference failures on deployed environments without .git. + } + return null; +} + +export function writeRestartContext( + context: Omit, + serviceIds: string[] = [SERVICE_ID], +): string[] { + fs.mkdirSync(DATA_DIR, { recursive: true }); + const payload: RestartContext = { + ...context, + verify: context.verify, + writtenAt: new Date().toISOString(), + }; + + const writtenPaths: string[] = []; + for (const serviceId of serviceIds) { + const filePath = getRestartContextPath(serviceId); + fs.writeFileSync(filePath, JSON.stringify(payload, null, 2), 'utf8'); + writtenPaths.push(filePath); + } + return writtenPaths; +} + +export function writeShutdownRestartContext( + registeredGroups: Record, + interruptedGroups: RestartInterruptedGroup[], + signal: string, + serviceIds: string[] = [SERVICE_ID], +): string[] { + if (interruptedGroups.length === 0) return []; + + fs.mkdirSync(DATA_DIR, { recursive: true }); + const mainChatJid = + getMainGroupJid(registeredGroups) ?? interruptedGroups[0].chatJid; + const writtenPaths: string[] = []; + + for (const serviceId of serviceIds) { + const filePath = getRestartContextPath(serviceId); + const existing = readRestartContextFile(filePath); + const mergedInterrupted = [ + ...(existing?.interruptedGroups ?? []), + ...interruptedGroups, + ].filter( + (group, index, all) => + all.findIndex((candidate) => candidate.chatJid === group.chatJid) === + index, + ); + + const payload: RestartContext = { + chatJid: existing?.chatJid || mainChatJid, + summary: + existing?.summary || '서비스 재시작으로 진행 중인 작업이 중단됨', + verify: existing?.verify ?? [], + writtenAt: new Date().toISOString(), + source: existing?.source || 'shutdown-snapshot', + signal, + interruptedGroups: mergedInterrupted, + }; + + fs.writeFileSync(filePath, JSON.stringify(payload, null, 2), 'utf8'); + writtenPaths.push(filePath); + } + + return writtenPaths; +} + +export function buildInterruptedRestartAnnouncement( + interrupted: RestartInterruptedGroup, +): string { + const lines = ['서비스 재시작으로 이전 작업이 중단됐습니다.']; + lines.push(`- 직전 상태: ${interrupted.status}`); + if (interrupted.elapsedMs !== null) { + lines.push(`- 직전 실행 시간: ${Math.round(interrupted.elapsedMs / 1000)}초`); + } + if (interrupted.pendingMessages) { + lines.push('- 미처리 메시지가 남아 있었음'); + } + if (interrupted.pendingTasks > 0) { + lines.push(`- 대기 태스크: ${interrupted.pendingTasks}개`); + } + lines.push('- 필요하면 이어서 요청해 주세요.'); + return lines.join('\n'); +} + +export function consumeRestartContext(): RestartContext | null { + const filePath = getRestartContextPath(); + if (!fs.existsSync(filePath)) return null; + try { + const parsed = JSON.parse( + fs.readFileSync(filePath, 'utf8'), + ) as RestartContext; + fs.unlinkSync(filePath); + return parsed; + } catch (err) { + logger.warn( + { err, filePath }, + 'Failed to read restart context; removing invalid file', + ); + try { + fs.unlinkSync(filePath); + } catch { + // Ignore cleanup failure. + } + return null; + } +} + +export function buildRestartAnnouncement( + context: RestartContext, +): string { + const lines = ['재시작 완료.', `- 변경: ${context.summary}`]; + if (context.interruptedGroups && context.interruptedGroups.length > 0) { + lines.push(`- 중단 작업 감지: ${context.interruptedGroups.length}개`); + } + if (context.verify.length > 0) { + lines.push(`- 검증: ${context.verify.join(', ')}`); + } + lines.push(`- 기록 시각: ${formatKoreanTimestamp(context.writtenAt)}`); + return lines.join('\n'); +} + +export function inferRecentRestartContext( + registeredGroups: Record, + processStartedAtMs: number, +): InferredRestartContext | null { + const chatJid = getMainGroupJid(registeredGroups); + if (!chatJid) return null; + + const latestBuildTime = getLatestDistBuildTime(); + const recentCommit = getRecentCommit(processStartedAtMs); + const buildLooksRecent = + latestBuildTime !== null && + latestBuildTime >= processStartedAtMs - INFER_WINDOW_MS; + + if (!buildLooksRecent && !recentCommit) return null; + + const lines = ['재시작 감지.', '- 명시적 재시작 힌트는 없어 추론 기반임.']; + if (recentCommit) { + lines.push(`- 최근 커밋: ${recentCommit}`); + } + if (latestBuildTime !== null) { + lines.push(`- 최근 빌드: ${formatKoreanTimestamp(latestBuildTime)}`); + } + lines.push(`- 시작 시각: ${formatKoreanTimestamp(processStartedAtMs)}`); + + return { chatJid, lines }; +} diff --git a/src/task-scheduler.test.ts b/src/task-scheduler.test.ts index 31dc01f..94e9964 100644 --- a/src/task-scheduler.test.ts +++ b/src/task-scheduler.test.ts @@ -61,6 +61,7 @@ describe('task scheduler', () => { id: 'task-claude', group_folder: 'shared-group', chat_jid: 'shared@g.us', + agent_type: 'claude-code', prompt: 'claude task', schedule_type: 'once', schedule_value: dueAt, @@ -126,7 +127,11 @@ Check the run. expect(extractWatchCiTarget(prompt)).toBe('GitHub Actions run 123456'); const rendered = renderWatchCiStatusMessage({ - task: { prompt }, + task: { + prompt, + schedule_type: 'interval', + schedule_value: '60000', + } as any, phase: 'waiting', checkedAt: '2026-03-19T07:02:10.000Z', nextRun: '2026-03-19T07:04:10.000Z', @@ -134,9 +139,11 @@ Check the run. expect(rendered).toContain('CI 감시 중: GitHub Actions run 123456'); expect(rendered).toContain('- 상태: 대기 중'); - expect(rendered).toContain('- 마지막 확인:'); - expect(rendered).toContain('- 다음 확인:'); - expect(rendered).not.toContain('2분 10초'); + expect(rendered).toContain('- 마지막 확인: 16시 02분 10초'); + expect(rendered).toContain('- 확인 간격: 1분'); + expect(rendered).toContain('- 다음 확인: 16시 04분 10초'); + expect(rendered).not.toContain('16:02:10'); + expect(rendered).not.toContain('16:04:10'); }); it('computeNextRun anchors interval tasks to scheduled time to prevent drift', () => { diff --git a/src/task-scheduler.ts b/src/task-scheduler.ts index cb4a85f..e2ed2d0 100644 --- a/src/task-scheduler.ts +++ b/src/task-scheduler.ts @@ -113,11 +113,33 @@ function formatTimeLabel(timestampIso: string): string { second: '2-digit', hour12: false, timeZone: TIMEZONE, - }).format(new Date(timestampIso)); + }) + .format(new Date(timestampIso)) + .replace(/:/g, '시 ') + .replace(/시 (\d{2})$/, '분 $1초'); +} + +function formatWatchIntervalLabel(task: Pick): string | null { + if (task.schedule_type !== 'interval') return null; + const ms = parseInt(task.schedule_value, 10); + if (!Number.isFinite(ms) || ms <= 0) return null; + + const totalSeconds = Math.floor(ms / 1000); + if (totalSeconds < 60) return `${totalSeconds}초`; + + const totalMinutes = Math.floor(totalSeconds / 60); + if (totalMinutes < 60) { + const seconds = totalSeconds % 60; + return seconds > 0 ? `${totalMinutes}분 ${seconds}초` : `${totalMinutes}분`; + } + + const hours = Math.floor(totalMinutes / 60); + const minutes = totalMinutes % 60; + return minutes > 0 ? `${hours}시간 ${minutes}분` : `${hours}시간`; } export function renderWatchCiStatusMessage(args: { - task: Pick; + task: Pick; phase: WatcherStatusPhase; checkedAt: string; nextRun?: string | null; @@ -141,6 +163,10 @@ export function renderWatchCiStatusMessage(args: { `- 상태: ${statusLabel}`, `- 마지막 확인: ${formatTimeLabel(args.checkedAt)}`, ]; + const intervalLabel = formatWatchIntervalLabel(args.task); + if (intervalLabel) { + lines.push(`- 확인 간격: ${intervalLabel}`); + } if (args.nextRun) { lines.push(`- 다음 확인: ${formatTimeLabel(args.nextRun)}`); diff --git a/src/types.ts b/src/types.ts index e464264..6780214 100644 --- a/src/types.ts +++ b/src/types.ts @@ -28,6 +28,7 @@ export interface NewMessage { sender_name: string; content: string; timestamp: string; + seq?: number; is_from_me?: boolean; is_bot_message?: boolean; }