From 898c77fcfc501733b6951a4f6b080761250d8ecb Mon Sep 17 00:00:00 2001 From: Eyejoker Date: Mon, 23 Mar 2026 06:39:26 +0900 Subject: [PATCH] refactor: consolidate runtime helpers --- src/available-groups.ts | 19 ++ src/config.ts | 4 + src/dashboard-render.ts | 90 ++++++++ src/dashboard.test.ts | 28 ++- src/dashboard.ts | 51 +++-- src/group-queue-ipc.ts | 21 ++ src/group-queue.ts | 20 +- src/index.test.ts | 19 ++ src/index.ts | 101 ++++----- src/message-agent-executor.ts | 381 ++++++++++++++++++++++++++++++++ src/message-cursor.ts | 10 + src/message-runtime-rules.ts | 11 +- src/message-runtime.ts | 397 ++-------------------------------- src/task-scheduler.ts | 136 ++---------- src/task-watch-status.ts | 128 +++++++++++ 15 files changed, 808 insertions(+), 608 deletions(-) create mode 100644 src/available-groups.ts create mode 100644 src/dashboard-render.ts create mode 100644 src/group-queue-ipc.ts create mode 100644 src/message-agent-executor.ts create mode 100644 src/message-cursor.ts create mode 100644 src/task-watch-status.ts diff --git a/src/available-groups.ts b/src/available-groups.ts new file mode 100644 index 0000000..8e7ae66 --- /dev/null +++ b/src/available-groups.ts @@ -0,0 +1,19 @@ +import type { AvailableGroup } from './agent-runner.js'; +import { getAllChats } from './db.js'; +import type { RegisteredGroup } from './types.js'; + +export function listAvailableGroups( + registeredGroups: Record, +): AvailableGroup[] { + const chats = getAllChats(); + const registeredJids = new Set(Object.keys(registeredGroups)); + + return chats + .filter((chat) => chat.jid !== '__group_sync__' && chat.is_group) + .map((chat) => ({ + jid: chat.jid, + name: chat.name, + lastActivity: chat.last_message_time, + isRegistered: registeredJids.has(chat.jid), + })); +} diff --git a/src/config.ts b/src/config.ts index 0fed572..7f7bb57 100644 --- a/src/config.ts +++ b/src/config.ts @@ -11,6 +11,7 @@ const envConfig = readEnvFile([ 'SERVICE_AGENT_TYPE', 'SESSION_COMMAND_ALLOWED_SENDERS', 'SESSION_COMMAND_USER_IDS', + 'STATUS_SHOW_ROOMS', 'USAGE_DASHBOARD', ]); @@ -82,6 +83,9 @@ export const TRIGGER_PATTERN = new RegExp( export const STATUS_CHANNEL_ID = process.env.STATUS_CHANNEL_ID || ''; export const STATUS_UPDATE_INTERVAL = 10000; // 10s export const USAGE_UPDATE_INTERVAL = 300000; // 5 minutes +export const STATUS_SHOW_ROOMS = + (process.env.STATUS_SHOW_ROOMS || envConfig.STATUS_SHOW_ROOMS || 'true') !== + 'false'; export const USAGE_DASHBOARD_ENABLED = (process.env.USAGE_DASHBOARD || envConfig.USAGE_DASHBOARD) === 'true'; diff --git a/src/dashboard-render.ts b/src/dashboard-render.ts new file mode 100644 index 0000000..e942abd --- /dev/null +++ b/src/dashboard-render.ts @@ -0,0 +1,90 @@ +import { STATUS_SHOW_ROOMS } from './config.js'; +import type { GroupStatus } from './group-queue.js'; + +export function formatElapsed(ms: number): string { + const s = Math.floor(ms / 1000); + if (s < 60) return `${s}s`; + if (s < 3600) { + const m = Math.floor(s / 60); + const rem = s % 60; + return `${m}m${rem.toString().padStart(2, '0')}s`; + } + const h = Math.floor(s / 3600); + const m = Math.floor((s % 3600) / 60); + if (h < 24) return `${h}h${m.toString().padStart(2, '0')}m`; + const d = Math.floor(h / 24); + const remH = h % 24; + return `${d}d${remH}h`; +} + +export function getStatusLabel(status: { + status: GroupStatus['status']; + elapsedMs: number | null; + pendingTasks: number; +}): string { + if (status.status === 'processing') { + return `처리 중 (${formatElapsed(status.elapsedMs || 0)})`; + } + if (status.status === 'waiting') { + return status.pendingTasks > 0 + ? `큐 대기 (태스크 ${status.pendingTasks}개)` + : '큐 대기 (메시지)'; + } + return '비활성'; +} + +export interface DashboardRoomLine { + category: string; + categoryPosition: number; + position: number; + line: string; +} + +export function renderCategorizedRoomSections(args: { + lines: DashboardRoomLine[]; + showCategoryHeaders: boolean; +}): string { + const { lines, showCategoryHeaders } = args; + const categoryMap = new Map(); + + for (const line of lines) { + if (!categoryMap.has(line.category)) { + categoryMap.set(line.category, []); + } + categoryMap.get(line.category)!.push(line); + } + + const sortedCategories = [...categoryMap.entries()].sort( + ([, a], [, b]) => + (a[0]?.categoryPosition ?? Number.MAX_SAFE_INTEGER) - + (b[0]?.categoryPosition ?? Number.MAX_SAFE_INTEGER), + ); + + return sortedCategories + .map(([category, entries]) => { + entries.sort((a, b) => a.position - b.position); + const content = entries.map((entry) => entry.line).join('\n'); + if (!showCategoryHeaders || category === '기타') { + return content; + } + return `📁 **${category}**\n${content}`; + }) + .join('\n\n'); +} + +export function composeDashboardContent( + sections: string[], + now = new Date(), +): string { + const parts = sections.map((section) => section.trim()).filter(Boolean); + if (parts.length === 0) return ''; + + parts.push( + `_${now.toLocaleTimeString('ko-KR', { hour: '2-digit', minute: '2-digit' })}_`, + ); + return parts.join('\n\n'); +} + +export function shouldShowRoomStatus(): boolean { + return STATUS_SHOW_ROOMS; +} diff --git a/src/dashboard.test.ts b/src/dashboard.test.ts index 0cb2120..9cf6f70 100644 --- a/src/dashboard.test.ts +++ b/src/dashboard.test.ts @@ -1,6 +1,6 @@ -import { describe, expect, it } from 'vitest'; +import { afterEach, describe, expect, it, vi } from 'vitest'; -import { buildStatusContent, type DashboardOptions } from './dashboard.js'; +import type { DashboardOptions } from './dashboard.js'; function makeOptions(sessionId?: string): DashboardOptions { const sessions: Record = sessionId @@ -37,13 +37,33 @@ function makeOptions(sessionId?: string): DashboardOptions { } describe('buildStatusContent', () => { - it('shows cleared sessions as empty', () => { + async function loadBuildStatusContent( + statusShowRooms = 'true', + ): Promise<(opts: DashboardOptions) => string> { + vi.resetModules(); + process.env.STATUS_SHOW_ROOMS = statusShowRooms; + const mod = await import('./dashboard.js'); + return mod.buildStatusContent; + } + + afterEach(() => { + delete process.env.STATUS_SHOW_ROOMS; + }); + + it('shows cleared sessions as empty', async () => { + const buildStatusContent = await loadBuildStatusContent(); const content = buildStatusContent(makeOptions()); expect(content).toContain('세션 없음'); }); - it('shows a shortened session id when a session exists', () => { + it('shows a shortened session id when a session exists', async () => { + const buildStatusContent = await loadBuildStatusContent(); const content = buildStatusContent(makeOptions('session-1234567890')); expect(content).toContain('세션 34567890'); }); + + it('returns an empty string when room status display is disabled', async () => { + const buildStatusContent = await loadBuildStatusContent('false'); + expect(buildStatusContent(makeOptions())).toBe(''); + }); }); diff --git a/src/dashboard.ts b/src/dashboard.ts index 4f7134c..e969190 100644 --- a/src/dashboard.ts +++ b/src/dashboard.ts @@ -7,7 +7,13 @@ import { fetchClaudeUsageViaCli, type ClaudeUsageData, } from './claude-usage.js'; -import { USAGE_DASHBOARD_ENABLED } from './config.js'; +import { STATUS_SHOW_ROOMS, USAGE_DASHBOARD_ENABLED } from './config.js'; +import { + composeDashboardContent, + type DashboardRoomLine, + getStatusLabel as formatDashboardStatusLabel, + renderCategorizedRoomSections, +} from './dashboard-render.js'; import { readEnvFile } from './env.js'; import { GroupQueue, GroupStatus } from './group-queue.js'; import { logger } from './logger.js'; @@ -103,15 +109,7 @@ async function refreshChannelMeta(opts: DashboardOptions): Promise { } function getStatusLabel(status: GroupStatus): string { - if (status.status === 'processing') { - return `처리 중 (${formatElapsed(status.elapsedMs || 0)})`; - } - if (status.status === 'waiting') { - return status.pendingTasks > 0 - ? `큐 대기 (태스크 ${status.pendingTasks}개)` - : '큐 대기 (메시지)'; - } - return '비활성'; + return formatDashboardStatusLabel(status); } function getSessionLabel(sessionId: string | undefined): string { @@ -122,6 +120,8 @@ function getSessionLabel(sessionId: string | undefined): string { /** @internal - exported for testing */ export function buildStatusContent(opts: DashboardOptions): string { + if (!STATUS_SHOW_ROOMS) return ''; + const registeredGroups = opts.registeredGroups(); const sessions = opts.getSessions(); const jids = Object.keys(registeredGroups); @@ -148,7 +148,7 @@ export function buildStatusContent(opts: DashboardOptions): string { return posA - posB; }); - const sections: string[] = []; + const roomLines: DashboardRoomLine[] = []; let totalActive = 0; let totalWaiting = 0; let total = 0; @@ -158,20 +158,19 @@ export function buildStatusContent(opts: DashboardOptions): string { (a, b) => (a.meta?.position ?? 999) - (b.meta?.position ?? 999), ); - const lines = categoryEntries.map((entry) => { + categoryEntries.forEach((entry) => { const icon = STATUS_ICONS[entry.status.status] || '⚪'; const label = getStatusLabel(entry.status); const sessionLabel = getSessionLabel(sessions[entry.group.folder]); const name = entry.meta?.name ? `#${entry.meta.name}` : entry.group.name; - return ` ${icon} **${name}** — ${label} · ${sessionLabel}`; + roomLines.push({ + category: categoryName, + categoryPosition: entry.meta?.categoryPosition ?? 999, + position: entry.meta?.position ?? 999, + line: ` ${icon} **${name}** — ${label} · ${sessionLabel}`, + }); }); - if (channelMetaCache.size > 0 && categoryName !== '기타') { - sections.push(`📁 **${categoryName}**\n${lines.join('\n')}`); - } else { - sections.push(lines.join('\n')); - } - totalActive += categoryEntries.filter( (entry) => entry.status.status === 'processing', ).length; @@ -182,7 +181,15 @@ export function buildStatusContent(opts: DashboardOptions): string { } const header = `**에이전트 상태** (${opts.assistantName}) — 활성 ${totalActive} | 큐대기 ${totalWaiting} | 전체 ${total}`; - return `${header}\n\n${sections.join('\n\n')}\n\n_${new Date().toLocaleTimeString('ko-KR')}_`; + return composeDashboardContent( + [ + `${header}\n\n${renderCategorizedRoomSections({ + lines: roomLines, + showCategoryHeaders: channelMetaCache.size > 0, + })}`, + ], + new Date(), + ); } async function fetchClaudeUsage(): Promise { @@ -473,6 +480,10 @@ export async function startStatusDashboard( try { await refreshChannelMeta(opts); const content = buildStatusContent(opts); + if (!content) { + statusMessageId = null; + return; + } if (statusMessageId && ch.editMessage) { await ch.editMessage(statusJid, statusMessageId, content); diff --git a/src/group-queue-ipc.ts b/src/group-queue-ipc.ts new file mode 100644 index 0000000..eafc288 --- /dev/null +++ b/src/group-queue-ipc.ts @@ -0,0 +1,21 @@ +import fs from 'fs'; +import path from 'path'; + +import { DATA_DIR } from './config.js'; + +export function queueFollowUpMessage(groupFolder: string, text: string): string { + const inputDir = path.join(DATA_DIR, 'ipc', groupFolder, 'input'); + fs.mkdirSync(inputDir, { recursive: true }); + const filename = `${Date.now()}-${Math.random().toString(36).slice(2, 6)}.json`; + const filepath = path.join(inputDir, filename); + const tempPath = `${filepath}.tmp`; + fs.writeFileSync(tempPath, JSON.stringify({ type: 'message', text })); + fs.renameSync(tempPath, filepath); + return filename; +} + +export function writeCloseSentinel(groupFolder: string): void { + const inputDir = path.join(DATA_DIR, 'ipc', groupFolder, 'input'); + fs.mkdirSync(inputDir, { recursive: true }); + fs.writeFileSync(path.join(inputDir, '_close'), ''); +} diff --git a/src/group-queue.ts b/src/group-queue.ts index a10c732..88cea37 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -1,8 +1,10 @@ import { ChildProcess } from 'child_process'; -import fs from 'fs'; -import path from 'path'; -import { DATA_DIR, MAX_CONCURRENT_AGENTS } from './config.js'; +import { MAX_CONCURRENT_AGENTS } from './config.js'; +import { + queueFollowUpMessage, + writeCloseSentinel, +} from './group-queue-ipc.js'; import { logger } from './logger.js'; interface QueuedTask { @@ -213,14 +215,8 @@ export class GroupQueue { return false; } - const inputDir = path.join(DATA_DIR, 'ipc', state.groupFolder, 'input'); try { - fs.mkdirSync(inputDir, { recursive: true }); - const filename = `${Date.now()}-${Math.random().toString(36).slice(2, 6)}.json`; - const filepath = path.join(inputDir, filename); - const tempPath = `${filepath}.tmp`; - fs.writeFileSync(tempPath, JSON.stringify({ type: 'message', text })); - fs.renameSync(tempPath, filepath); + const filename = queueFollowUpMessage(state.groupFolder, text); logger.info( { groupJid, @@ -257,10 +253,8 @@ export class GroupQueue { if (!state.active || !state.groupFolder) return; state.closingStdin = true; - const inputDir = path.join(DATA_DIR, 'ipc', state.groupFolder, 'input'); try { - fs.mkdirSync(inputDir, { recursive: true }); - fs.writeFileSync(path.join(inputDir, '_close'), ''); + writeCloseSentinel(state.groupFolder); logger.info( { groupJid, diff --git a/src/index.test.ts b/src/index.test.ts index cefd5e3..fa19cb7 100644 --- a/src/index.test.ts +++ b/src/index.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it, vi } from 'vitest'; import { + composeDashboardContent, editFormattedTrackedChannelMessage, sendFormattedChannelMessage, sendFormattedTrackedChannelMessage, @@ -81,3 +82,21 @@ describe('index scheduler messaging helpers', () => { expect(sendMessage).not.toHaveBeenCalled(); }); }); + +describe('composeDashboardContent', () => { + it('returns an empty string when all dashboard sections are disabled', () => { + expect( + composeDashboardContent([], new Date('2026-03-23T04:00:00+09:00')), + ).toBe(''); + }); + + it('keeps non-status sections when room status is hidden', () => { + const content = composeDashboardContent( + ['**사용량**\nCodex OK'], + new Date('2026-03-23T04:00:00+09:00'), + ); + + expect(content).toContain('**사용량**'); + expect(content).not.toContain('**📊 에이전트 상태**'); + }); +}); diff --git a/src/index.ts b/src/index.ts index 35a2854..c527bb0 100644 --- a/src/index.ts +++ b/src/index.ts @@ -11,6 +11,7 @@ import { SERVICE_AGENT_TYPE, isSessionCommandSenderAllowed, STATUS_CHANNEL_ID, + STATUS_SHOW_ROOMS, STATUS_UPDATE_INTERVAL, TIMEZONE, TRIGGER_PATTERN, @@ -23,6 +24,7 @@ import { getRegisteredChannelNames, } from './channels/registry.js'; import { writeGroupsSnapshot } from './agent-runner.js'; +import { listAvailableGroups } from './available-groups.js'; import { getAllChats, getAllRegisteredGroups, @@ -41,6 +43,12 @@ import { storeChatMetadata, storeMessage, } from './db.js'; +import { + composeDashboardContent, + formatElapsed, + getStatusLabel as formatDashboardStatusLabel, + renderCategorizedRoomSections, +} from './dashboard-render.js'; import { GroupQueue } from './group-queue.js'; import { readEnvFile } from './env.js'; import { resolveGroupFolderPath } from './group-folder.js'; @@ -69,9 +77,11 @@ import { import { startSchedulerLoop } from './task-scheduler.js'; import { Channel, ChannelMeta, NewMessage, RegisteredGroup } from './types.js'; import { logger } from './logger.js'; +import { normalizeStoredSeqCursor } from './message-cursor.js'; // Re-export for backwards compatibility during refactor export { escapeXml, formatMessages } from './router.js'; +export { composeDashboardContent } from './dashboard-render.js'; export async function sendFormattedChannelMessage( channels: Channel[], @@ -148,15 +158,6 @@ const runtime = createMessageRuntime({ clearSession, }); -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 loadState(): void { lastTimestamp = normalizeStoredSeqCursor( getRouterState('last_seq') || getRouterState('last_timestamp'), @@ -229,17 +230,7 @@ function registerGroup(jid: string, group: RegisteredGroup): void { * Returns groups ordered by most recent activity. */ export function getAvailableGroups(): import('./agent-runner.js').AvailableGroup[] { - const chats = getAllChats(); - const registeredJids = new Set(Object.keys(registeredGroups)); - - return chats - .filter((c) => c.jid !== '__group_sync__' && c.is_group) - .map((c) => ({ - jid: c.jid, - name: c.name, - lastActivity: c.last_message_time, - isRegistered: registeredJids.has(c.jid), - })); + return listAvailableGroups(registeredGroups); } /** @internal - exported for testing */ @@ -251,22 +242,6 @@ export function _setRegisteredGroups( // ── Status & Usage Dashboards ─────────────────────────────────── -function formatElapsed(ms: number): string { - const s = Math.floor(ms / 1000); - if (s < 60) return `${s}s`; - if (s < 3600) { - const m = Math.floor(s / 60); - const rem = s % 60; - return `${m}m${rem.toString().padStart(2, '0')}s`; - } - const h = Math.floor(s / 3600); - const m = Math.floor((s % 3600) / 60); - if (h < 24) return `${h}h${m.toString().padStart(2, '0')}m`; - const d = Math.floor(h / 24); - const remH = h % 24; - return `${d}d${remH}h`; -} - function formatBytes(bytes: number): string { if (bytes >= 1073741824) return `${(bytes / 1073741824).toFixed(1)}GB`; if (bytes >= 1048576) return `${(bytes / 1048576).toFixed(0)}MB`; @@ -360,13 +335,11 @@ function getStatusLabel(s: { pendingMessages: boolean; pendingTasks: number; }): string { - if (s.status === 'processing') - return `처리 중 (${formatElapsed(s.elapsedMs || 0)})`; - if (s.status === 'waiting') - return s.pendingTasks > 0 - ? `큐 대기 (태스크 ${s.pendingTasks}개)` - : '큐 대기 (메시지)'; - return '비활성'; + return formatDashboardStatusLabel({ + status: s.status, + elapsedMs: s.elapsedMs, + pendingTasks: s.pendingTasks, + }); } function getAgentDisplayName(agentType: 'claude-code' | 'codex'): string { @@ -434,6 +407,8 @@ function writeLocalStatusSnapshot(): void { } function buildStatusContent(): string { + if (!STATUS_SHOW_ROOMS) return ''; + const snapshots = readStatusSnapshots(STATUS_SNAPSHOT_MAX_AGE_MS); const chatNameByJid = new Map( getAllChats().map((chat) => [chat.jid, chat.name]), @@ -504,11 +479,9 @@ function buildStatusContent(): string { return posA - posB; }); - const sections: string[] = []; + const roomLines = []; for (const [catName, rooms] of sortedCategories) { rooms.sort((a, b) => (a.meta?.position ?? 999) - (b.meta?.position ?? 999)); - - const roomLines: string[] = []; for (const room of rooms) { // Sort agents: claude-code first, codex second room.agents.sort((a, b) => @@ -525,18 +498,21 @@ function buildStatusContent(): string { const tag = getAgentDisplayName(a.agentType); return `${tag} ${icon} ${label}`; }); - roomLines.push(` **${room.name}** — ${agentParts.join(' | ')}`); - } - - if (channelMetaCache.size > 0 && catName !== '기타') { - sections.push(`📁 **${catName}**\n${roomLines.join('\n')}`); - } else { - sections.push(roomLines.join('\n')); + roomLines.push({ + category: catName, + categoryPosition: room.meta?.categoryPosition ?? 999, + position: room.meta?.position ?? 999, + line: ` **${room.name}** — ${agentParts.join(' | ')}`, + }); } } const header = `**📊 에이전트 상태** — 활성 ${totalActive} / ${totalRooms}`; - return `${header}\n\n${sections.join('\n\n')}`; + const sections = renderCategorizedRoomSections({ + lines: roomLines, + showCategoryHeaders: channelMetaCache.size > 0, + }); + return `${header}\n\n${sections}`; } // ── API Usage Fetchers ────────────────────────────────────────── @@ -907,17 +883,14 @@ async function refreshUsageCache(): Promise { } function buildUnifiedDashboard(): string { - const status = buildStatusContent(); - const parts = [status]; - + const parts: string[] = []; + if (STATUS_SHOW_ROOMS) { + parts.push(buildStatusContent()); + } if (cachedUsageContent) { parts.push(cachedUsageContent); } - - parts.push( - `_${new Date().toLocaleTimeString('ko-KR', { hour: '2-digit', minute: '2-digit' })}_`, - ); - return parts.join('\n\n'); + return composeDashboardContent(parts); } async function startStatusDashboard(): Promise { @@ -944,6 +917,10 @@ async function startStatusDashboard(): Promise { try { await refreshChannelMeta(); const content = buildUnifiedDashboard(); + if (!content) { + statusMessageId = null; + return; + } if (statusMessageId && ch.editMessage) { await ch.editMessage(statusJid, statusMessageId, content); diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts new file mode 100644 index 0000000..3debef4 --- /dev/null +++ b/src/message-agent-executor.ts @@ -0,0 +1,381 @@ +import path from 'path'; + +import { + AgentOutput, + runAgentProcess, + writeGroupsSnapshot, + writeTasksSnapshot, +} from './agent-runner.js'; +import { listAvailableGroups } from './available-groups.js'; +import { DATA_DIR } from './config.js'; +import { getAllTasks } from './db.js'; +import { GroupQueue } from './group-queue.js'; +import { logger } from './logger.js'; +import { + detectFallbackTrigger, + getActiveProvider, + getFallbackEnvOverrides, + getFallbackProviderName, + hasGroupProviderOverride, + isFallbackEnabled, + markPrimaryCooldown, +} from './provider-fallback.js'; +import { shouldResetSessionOnAgentFailure } from './session-recovery.js'; +import type { RegisteredGroup } from './types.js'; + +export interface MessageAgentExecutorDeps { + assistantName: string; + queue: Pick; + getRegisteredGroups: () => Record; + getSessions: () => Record; + persistSession: (groupFolder: string, sessionId: string) => void; + clearSession: (groupFolder: string) => void; +} + +export async function runAgentForGroup( + deps: MessageAgentExecutorDeps, + args: { + group: RegisteredGroup; + prompt: string; + chatJid: string; + runId: string; + onOutput?: (output: AgentOutput) => Promise; + }, +): Promise<'success' | 'error'> { + const { group, prompt, chatJid, runId, onOutput } = args; + const isMain = group.isMain === true; + const isClaudeCodeAgent = + (group.agentType || 'claude-code') === 'claude-code'; + const sessions = deps.getSessions(); + const sessionId = sessions[group.folder]; + + const tasks = getAllTasks(group.agentType || 'claude-code'); + writeTasksSnapshot( + group.folder, + isMain, + tasks.map((task) => ({ + id: task.id, + groupFolder: task.group_folder, + prompt: task.prompt, + schedule_type: task.schedule_type, + schedule_value: task.schedule_value, + status: task.status, + next_run: task.next_run, + })), + ); + + writeGroupsSnapshot( + group.folder, + isMain, + listAvailableGroups(deps.getRegisteredGroups()), + ); + + let resetSessionRequested = false; + + 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; + } + | 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); + } + : undefined; + + if (provider !== 'claude') { + logger.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + provider, + }, + `Claude provider in cooldown, routing request to ${provider}`, + ); + } + + 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( + { + 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'; + }; + + 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'; +} diff --git a/src/message-cursor.ts b/src/message-cursor.ts new file mode 100644 index 0000000..dbe0bb7 --- /dev/null +++ b/src/message-cursor.ts @@ -0,0 +1,10 @@ +import { getLatestMessageSeqAtOrBefore } from './db.js'; + +export 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)); +} diff --git a/src/message-runtime-rules.ts b/src/message-runtime-rules.ts index 55f7a56..3af80fb 100644 --- a/src/message-runtime-rules.ts +++ b/src/message-runtime-rules.ts @@ -1,9 +1,9 @@ import { - getLatestMessageSeqAtOrBefore, getLastHumanMessageTimestamp, isPairedRoomJid, } from './db.js'; import { filterProcessableMessages } from './bot-message-filter.js'; +import { normalizeStoredSeqCursor } from './message-cursor.js'; import { isTriggerAllowed, loadSenderAllowlist } from './sender-allowlist.js'; import { type Channel, @@ -13,15 +13,6 @@ import { const BOT_COLLABORATION_WINDOW_MS = 12 * 60 * 60 * 1000; -export 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)); -} - export function advanceLastAgentCursor( lastAgentTimestamps: Record, saveState: () => void, diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 3f07b7d..8c00d50 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -1,13 +1,5 @@ +import { AgentOutput } from './agent-runner.js'; import { - AgentOutput, - AvailableGroup, - runAgentProcess, - writeGroupsSnapshot, - writeTasksSnapshot, -} from './agent-runner.js'; -import { - getAllChats, - getAllTasks, getMessagesSinceSeq, getNewMessagesBySeq, getOpenWorkItem, @@ -16,17 +8,8 @@ import { markWorkItemDeliveryRetry, type WorkItem, } from './db.js'; -import { DATA_DIR, isSessionCommandSenderAllowed } from './config.js'; +import { 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 } from './router.js'; import { isTriggerAllowed, loadSenderAllowlist } from './sender-allowlist.js'; import { @@ -37,6 +20,7 @@ import { hasAllowedTrigger, shouldSkipBotOnlyCollaboration, } from './message-runtime-rules.js'; +import { runAgentForGroup } from './message-agent-executor.js'; import { MessageTurnController } from './message-turn-controller.js'; import { extractSessionCommand, @@ -44,10 +28,8 @@ import { isSessionCommandAllowed, isSessionCommandControlMessage, } from './session-commands.js'; -import { shouldResetSessionOnAgentFailure } from './session-recovery.js'; import { Channel, NewMessage, RegisteredGroup } from './types.js'; import { logger } from './logger.js'; -import path from 'path'; export interface MessageRuntimeDeps { assistantName: string; @@ -67,22 +49,6 @@ export interface MessageRuntimeDeps { clearSession: (groupFolder: string) => void; } -export function getAvailableGroups( - registeredGroups: Record, -): AvailableGroup[] { - const chats = getAllChats(); - const registeredJids = new Set(Object.keys(registeredGroups)); - - return chats - .filter((chat) => chat.jid !== '__group_sync__' && chat.is_group) - .map((chat) => ({ - jid: chat.jid, - name: chat.name, - lastActivity: chat.last_message_time, - isRegistered: registeredJids.has(chat.jid), - })); -} - export function createMessageRuntime(deps: MessageRuntimeDeps): { processGroupMessages: ( chatJid: string, @@ -97,9 +63,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { deps.idleTimeout, ); - const getCurrentAvailableGroups = (): AvailableGroup[] => - getAvailableGroups(deps.getRegisteredGroups()); - const deliverOpenWorkItem = async ( channel: Channel, item: WorkItem, @@ -176,345 +139,25 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { chatJid: string, runId: string, onOutput?: (output: AgentOutput) => Promise, - ): Promise<'success' | 'error'> => { - const isMain = group.isMain === true; - const isClaudeCodeAgent = - (group.agentType || 'claude-code') === 'claude-code'; - const sessions = deps.getSessions(); - const sessionId = sessions[group.folder]; - - const tasks = getAllTasks(group.agentType || 'claude-code'); - writeTasksSnapshot( - group.folder, - isMain, - tasks.map((task) => ({ - id: task.id, - groupFolder: task.group_folder, - prompt: task.prompt, - schedule_type: task.schedule_type, - schedule_value: task.schedule_value, - status: task.status, - next_run: task.next_run, - })), + ): Promise<'success' | 'error'> => + runAgentForGroup( + { + assistantName: deps.assistantName, + queue: deps.queue, + getRegisteredGroups: deps.getRegisteredGroups, + getSessions: deps.getSessions, + persistSession: deps.persistSession, + clearSession: deps.clearSession, + }, + { + group, + prompt, + chatJid, + runId, + onOutput, + }, ); - writeGroupsSnapshot(group.folder, isMain, getCurrentAvailableGroups()); - - let resetSessionRequested = false; - - 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; - } - | 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); - } - : undefined; - - if (provider !== 'claude') { - logger.info( - { - chatJid, - group: group.name, - groupFolder: group.folder, - runId, - provider, - }, - `Claude provider in cooldown, routing request to ${provider}`, - ); - } - - 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( - { - 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'; - }; - - 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 ( chatJid: string, context: GroupRunContext, diff --git a/src/task-scheduler.ts b/src/task-scheduler.ts index fb0a56d..00d8792 100644 --- a/src/task-scheduler.ts +++ b/src/task-scheduler.ts @@ -25,7 +25,21 @@ import { import { GroupQueue } from './group-queue.js'; import { resolveGroupFolderPath } from './group-folder.js'; import { logger } from './logger.js'; +import { + getTaskQueueJid, + isWatchCiTask, + renderWatchCiStatusMessage, + TASK_STATUS_MESSAGE_PREFIX, + type WatcherStatusPhase, +} from './task-watch-status.js'; import { AgentType, RegisteredGroup, ScheduledTask } from './types.js'; +export { + extractWatchCiTarget, + getTaskQueueJid, + isTaskStatusControlMessage, + isWatchCiTask, + renderWatchCiStatusMessage, +} from './task-watch-status.js'; /** * Compute the next run time for a recurring task, anchored to the @@ -88,128 +102,6 @@ export interface SchedulerDependencies { ) => Promise; } -type WatcherStatusPhase = 'checking' | 'waiting' | 'retrying' | 'completed'; - -const WATCH_CI_PREFIX = '[BACKGROUND CI WATCH]'; -const TASK_STATUS_MESSAGE_PREFIX = '\u2063\u2063\u2063'; - -export function isWatchCiTask(task: Pick): boolean { - return task.prompt.startsWith(WATCH_CI_PREFIX); -} - -export function isTaskStatusControlMessage(content: string): boolean { - return content.startsWith(TASK_STATUS_MESSAGE_PREFIX); -} - -export function extractWatchCiTarget(prompt: string): string | null { - const match = prompt.match(/Watch target:\n([\s\S]*?)\n\nTask ID:/); - return match?.[1]?.trim() || null; -} - -function formatTimeLabel(timestampIso: string): string { - return new Intl.DateTimeFormat('ko-KR', { - hour: '2-digit', - minute: '2-digit', - second: '2-digit', - hour12: false, - timeZone: TIMEZONE, - }) - .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}시간`; -} - -function formatElapsedLabel( - startedAtIso: string, - checkedAtIso: string, -): string { - const elapsedMs = Math.max( - 0, - new Date(checkedAtIso).getTime() - new Date(startedAtIso).getTime(), - ); - const totalSeconds = Math.floor(elapsedMs / 1000); - const hours = Math.floor(totalSeconds / 3600); - const minutes = Math.floor((totalSeconds % 3600) / 60); - const seconds = totalSeconds % 60; - - const parts: string[] = []; - if (hours > 0) parts.push(`${hours}시간`); - if (minutes > 0) parts.push(`${minutes}분`); - parts.push(`${seconds}초`); - return parts.join(' '); -} - -export function renderWatchCiStatusMessage(args: { - task: Pick; - phase: WatcherStatusPhase; - checkedAt: string; - statusStartedAt?: string | null; - nextRun?: string | null; -}): string { - const target = extractWatchCiTarget(args.task.prompt) || 'CI watcher'; - const title = - args.phase === 'completed' - ? `CI 감시 종료: ${target}` - : `CI 감시 중: ${target}`; - const statusLabel = - args.phase === 'checking' - ? '확인 중' - : args.phase === 'retrying' - ? '재시도 대기' - : args.phase === 'completed' - ? '완료' - : '대기 중'; - - const lines = [ - title, - `- 상태: ${statusLabel}`, - `- 마지막 확인: ${formatTimeLabel(args.checkedAt)}`, - ]; - if (args.statusStartedAt) { - lines.push( - `- 경과 시간: ${formatElapsedLabel(args.statusStartedAt, args.checkedAt)}`, - ); - } - const intervalLabel = formatWatchIntervalLabel(args.task); - if (intervalLabel) { - lines.push(`- 확인 간격: ${intervalLabel}`); - } - - if (args.nextRun) { - lines.push(`- 다음 확인: ${formatTimeLabel(args.nextRun)}`); - } - return lines.join('\n'); -} - -function getTaskQueueJid( - task: Pick, -): string { - return task.context_mode === 'isolated' || isWatchCiTask(task) - ? `${task.chat_jid}::task:${task.id}` - : task.chat_jid; -} - async function runTask( task: ScheduledTask, deps: SchedulerDependencies, diff --git a/src/task-watch-status.ts b/src/task-watch-status.ts new file mode 100644 index 0000000..9abab5d --- /dev/null +++ b/src/task-watch-status.ts @@ -0,0 +1,128 @@ +import { TIMEZONE } from './config.js'; +import type { ScheduledTask } from './types.js'; + +export type WatcherStatusPhase = + | 'checking' + | 'waiting' + | 'retrying' + | 'completed'; + +export const WATCH_CI_PREFIX = '[BACKGROUND CI WATCH]'; +export const TASK_STATUS_MESSAGE_PREFIX = '\u2063\u2063\u2063'; + +export function isWatchCiTask(task: Pick): boolean { + return task.prompt.startsWith(WATCH_CI_PREFIX); +} + +export function isTaskStatusControlMessage(content: string): boolean { + return content.startsWith(TASK_STATUS_MESSAGE_PREFIX); +} + +export function extractWatchCiTarget(prompt: string): string | null { + const match = prompt.match(/Watch target:\n([\s\S]*?)\n\nTask ID:/); + return match?.[1]?.trim() || null; +} + +function formatTimeLabel(timestampIso: string): string { + return new Intl.DateTimeFormat('ko-KR', { + hour: '2-digit', + minute: '2-digit', + second: '2-digit', + hour12: false, + timeZone: TIMEZONE, + }) + .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}시간`; +} + +function formatElapsedLabel( + startedAtIso: string, + checkedAtIso: string, +): string { + const elapsedMs = Math.max( + 0, + new Date(checkedAtIso).getTime() - new Date(startedAtIso).getTime(), + ); + const totalSeconds = Math.floor(elapsedMs / 1000); + const hours = Math.floor(totalSeconds / 3600); + const minutes = Math.floor((totalSeconds % 3600) / 60); + const seconds = totalSeconds % 60; + + const parts: string[] = []; + if (hours > 0) parts.push(`${hours}시간`); + if (minutes > 0) parts.push(`${minutes}분`); + parts.push(`${seconds}초`); + return parts.join(' '); +} + +export function renderWatchCiStatusMessage(args: { + task: Pick; + phase: WatcherStatusPhase; + checkedAt: string; + statusStartedAt?: string | null; + nextRun?: string | null; +}): string { + const target = extractWatchCiTarget(args.task.prompt) || 'CI watcher'; + const title = + args.phase === 'completed' + ? `CI 감시 종료: ${target}` + : `CI 감시 중: ${target}`; + const statusLabel = + args.phase === 'checking' + ? '확인 중' + : args.phase === 'retrying' + ? '재시도 대기' + : args.phase === 'completed' + ? '완료' + : '대기 중'; + + const lines = [ + title, + `- 상태: ${statusLabel}`, + `- 마지막 확인: ${formatTimeLabel(args.checkedAt)}`, + ]; + if (args.statusStartedAt) { + lines.push( + `- 경과 시간: ${formatElapsedLabel(args.statusStartedAt, args.checkedAt)}`, + ); + } + const intervalLabel = formatWatchIntervalLabel(args.task); + if (intervalLabel) { + lines.push(`- 확인 간격: ${intervalLabel}`); + } + + if (args.nextRun) { + lines.push(`- 다음 확인: ${formatTimeLabel(args.nextRun)}`); + } + return lines.join('\n'); +} + +export function getTaskQueueJid( + task: Pick, +): string { + return task.context_mode === 'isolated' || isWatchCiTask(task) + ? `${task.chat_jid}::task:${task.id}` + : task.chat_jid; +}