diff --git a/runners/agent-runner/src/index.ts b/runners/agent-runner/src/index.ts index 69702e2..0c55883 100644 --- a/runners/agent-runner/src/index.ts +++ b/runners/agent-runner/src/index.ts @@ -32,6 +32,7 @@ interface ContainerInput { interface ContainerOutput { status: 'success' | 'error'; + phase?: 'progress' | 'final'; result: string | null; newSessionId?: string; error?: string; @@ -59,6 +60,11 @@ interface SDKUserMessage { session_id: string; } +interface AssistantContentBlock { + type?: string; + text?: string; +} + // Paths configurable via env vars. const GROUP_DIR = process.env.EJCLAW_GROUP_DIR || '/workspace/group'; const IPC_DIR = process.env.EJCLAW_IPC_DIR || '/workspace/ipc'; @@ -174,6 +180,25 @@ function log(message: string): void { console.error(`[agent-runner] ${message}`); } +function extractAssistantText(message: unknown): string | null { + const assistant = message as { + message?: { + content?: AssistantContentBlock[]; + }; + }; + const blocks = assistant.message?.content; + if (!Array.isArray(blocks)) return null; + + const text = blocks + .filter((block) => block?.type === 'text' && typeof block.text === 'string') + .map((block) => block.text!.trim()) + .filter(Boolean) + .join('\n\n') + .trim(); + + return text || null; +} + function getSessionSummary(sessionId: string, transcriptPath: string): string | null { const projectDir = path.dirname(transcriptPath); const indexPath = path.join(projectDir, 'sessions-index.json'); @@ -525,6 +550,32 @@ async function runQuery( log(`Task notification: task=${tn.task_id} status=${tn.status} summary=${tn.summary}`); } + if (message.type === 'tool_progress') { + const tp = message as { + tool_name: string; + elapsed_time_seconds: number; + }; + const label = `${tp.tool_name} (${Math.round(tp.elapsed_time_seconds)}s)`; + log(`Tool progress: ${label}`); + writeOutput({ + status: 'success', + phase: 'progress', + result: label, + newSessionId, + }); + } + + if (message.type === 'tool_use_summary') { + const ts = message as { summary: string }; + log(`Tool use summary: ${ts.summary.slice(0, 200)}`); + writeOutput({ + status: 'success', + phase: 'progress', + result: ts.summary, + newSessionId, + }); + } + if (message.type === 'result') { resultCount++; const textResult = 'result' in message ? (message as { result?: string }).result : null; @@ -569,6 +620,26 @@ async function runQuery( log('Terminal result observed, ending query stream'); break; } + + if (message.type === 'assistant') { + const stopReason = (message as { stop_reason?: string }).stop_reason; + const textResult = extractAssistantText(message); + if (stopReason === 'end_turn' && textResult) { + resultCount++; + log( + `Terminal assistant turn observed without result event (${textResult.length} chars), ending query stream`, + ); + writeOutput({ + status: 'success', + result: textResult, + newSessionId, + }); + terminalResultObserved = true; + ipcPolling = false; + stream.end(); + break; + } + } } ipcPolling = false; diff --git a/src/db.ts b/src/db.ts index dc60ec4..0f2dc05 100644 --- a/src/db.ts +++ b/src/db.ts @@ -214,6 +214,14 @@ function createSchema(database: Database.Database): void { /* column already exists */ } + try { + database.exec( + `ALTER TABLE scheduled_tasks ADD COLUMN suspended_until TEXT`, + ); + } catch { + /* column already exists */ + } + database.exec(` UPDATE scheduled_tasks SET agent_type = COALESCE( @@ -928,7 +936,12 @@ export function updateTask( updates: Partial< Pick< ScheduledTask, - 'prompt' | 'schedule_type' | 'schedule_value' | 'next_run' | 'status' + | 'prompt' + | 'schedule_type' + | 'schedule_value' + | 'next_run' + | 'status' + | 'suspended_until' > >, ): void { @@ -955,6 +968,10 @@ export function updateTask( fields.push('status = ?'); values.push(updates.status); } + if (updates.suspended_until !== undefined) { + fields.push('suspended_until = ?'); + values.push(updates.suspended_until); + } if (fields.length === 0) return; @@ -1005,10 +1022,11 @@ export function getDueTasks( ` SELECT * FROM scheduled_tasks WHERE status = 'active' AND agent_type = ? AND next_run IS NOT NULL AND next_run <= ? + AND (suspended_until IS NULL OR suspended_until <= ?) ORDER BY next_run `, ) - .all(agentType, now) as ScheduledTask[]; + .all(agentType, now, now) as ScheduledTask[]; } export function updateTaskAfterRun( @@ -1042,6 +1060,25 @@ export function logTaskRun(log: TaskRunLog): void { ); } +export function getRecentConsecutiveErrors( + taskId: string, + limit: number = 5, +): string[] { + const rows = db + .prepare( + `SELECT status, error FROM task_run_logs + WHERE task_id = ? ORDER BY run_at DESC LIMIT ?`, + ) + .all(taskId, limit) as Array<{ status: string; error: string | null }>; + + const errors: string[] = []; + for (const row of rows) { + if (row.status !== 'error' || !row.error) break; + errors.push(row.error); + } + return errors; +} + // --- Router state accessors --- export function getRouterState(key: string): string | undefined { diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index 3dda543..c1b4d8f 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -167,17 +167,19 @@ export async function runAgentForGroup( ); } + const agentType = group.agentType || 'claude-code'; + const providerLabel = canFallback ? provider : agentType; logger.info( { chatJid, group: group.name, groupFolder: group.folder, runId, - provider, + provider: providerLabel, canFallback, groupHasOverride, }, - `Using provider: ${provider}`, + `Using provider: ${providerLabel}`, ); try { diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index dcfd96c..06f4dd5 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -1,5 +1,10 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { TASK_STATUS_MESSAGE_PREFIX } from './task-watch-status.js'; + +/** Prefix helper for progress message assertions */ +const P = (text: string) => `${TASK_STATUS_MESSAGE_PREFIX}${text}`; + vi.mock('./agent-runner.js', () => ({ runAgentProcess: vi.fn(), writeGroupsSnapshot: vi.fn(), @@ -686,17 +691,17 @@ describe('createMessageRuntime', () => { expect(result).toBe(true); expect(channel.sendAndTrack).toHaveBeenCalledWith( chatJid, - 'CI 상태 확인 중입니다.\n\n0초', + P('CI 상태 확인 중입니다.\n\n0초'), ); expect(channel.editMessage).toHaveBeenCalledWith( chatJid, 'progress-1', - 'CI 상태 확인 중입니다.\n\n10초', + P('CI 상태 확인 중입니다.\n\n10초'), ); expect(channel.editMessage).toHaveBeenCalledWith( chatJid, 'progress-1', - 'CI 상태 확인 중입니다.\n\n10초', + P('CI 상태 확인 중입니다.\n\n10초'), ); expect(channel.sendMessage).toHaveBeenCalledTimes(1); expect(channel.sendMessage).toHaveBeenCalledWith( @@ -779,11 +784,11 @@ describe('createMessageRuntime', () => { expect(result).toBe(true); expect(channel.sendAndTrack).toHaveBeenCalledWith( chatJid, - '진행 중입니다.\n\n0초', + P('진행 중입니다.\n\n0초'), ); expect(channel.sendMessage).toHaveBeenCalledWith( chatJid, - '진행 중입니다.\n\n0초', + P('진행 중입니다.\n\n0초'), ); }); @@ -851,11 +856,11 @@ describe('createMessageRuntime', () => { expect(result).toBe(true); expect(channel.sendAndTrack).toHaveBeenCalledWith( chatJid, - '진행 중입니다.\n\n0초', + P('진행 중입니다.\n\n0초'), ); expect(channel.sendMessage).toHaveBeenCalledWith( chatJid, - '진행 중입니다.\n\n0초', + P('진행 중입니다.\n\n0초'), ); }); @@ -949,17 +954,17 @@ describe('createMessageRuntime', () => { expect(channel.sendAndTrack).toHaveBeenCalledTimes(1); expect(channel.sendAndTrack).toHaveBeenCalledWith( chatJid, - '첫 번째 진행상황입니다.\n\n0초', + P('첫 번째 진행상황입니다.\n\n0초'), ); expect(channel.editMessage).toHaveBeenCalledWith( chatJid, 'progress-1', - '첫 번째 진행상황입니다.\n\n10초', + P('첫 번째 진행상황입니다.\n\n10초'), ); expect(channel.editMessage).not.toHaveBeenCalledWith( chatJid, 'progress-1', - '늦게 도착한 진행상황입니다.\n\n0초', + P('늦게 도착한 진행상황입니다.\n\n0초'), ); expect(channel.editMessage).not.toHaveBeenCalledWith( chatJid, @@ -1049,17 +1054,17 @@ describe('createMessageRuntime', () => { expect(result).toBe(true); expect(channel.sendAndTrack).toHaveBeenCalledWith( chatJid, - '오래 걸리는 작업입니다.\n\n0초', + P('오래 걸리는 작업입니다.\n\n0초'), ); expect(channel.editMessage).toHaveBeenCalledWith( chatJid, 'progress-1', - '오래 걸리는 작업입니다.\n\n1시간 0초', + P('오래 걸리는 작업입니다.\n\n1시간 0초'), ); expect(channel.editMessage).toHaveBeenCalledWith( chatJid, 'progress-1', - '오래 걸리는 작업입니다.\n\n1시간 10초', + P('오래 걸리는 작업입니다.\n\n1시간 10초'), ); expect(channel.sendMessage).toHaveBeenCalledWith( chatJid, @@ -1142,17 +1147,17 @@ describe('createMessageRuntime', () => { expect(result).toBe(true); expect(channel.sendAndTrack).toHaveBeenCalledWith( chatJid, - '테스트를 돌리는 중입니다.\n\n0초', + P('테스트를 돌리는 중입니다.\n\n0초'), ); expect(channel.editMessage).toHaveBeenCalledWith( chatJid, 'progress-1', - '테스트를 돌리는 중입니다.\n\n10초', + P('테스트를 돌리는 중입니다.\n\n10초'), ); expect(channel.editMessage).toHaveBeenCalledWith( chatJid, 'progress-1', - '테스트를 돌리는 중입니다.\n\n10초', + P('테스트를 돌리는 중입니다.\n\n10초'), ); expect(channel.sendMessage).toHaveBeenCalledTimes(1); expect(channel.sendMessage).toHaveBeenCalledWith( @@ -1313,29 +1318,29 @@ describe('createMessageRuntime', () => { expect(channel.sendAndTrack).toHaveBeenNthCalledWith( 1, chatJid, - '첫 번째 진행상황입니다.\n\n0초', + P('첫 번째 진행상황입니다.\n\n0초'), ); expect(channel.sendAndTrack).toHaveBeenNthCalledWith( 2, chatJid, - '두 번째 진행상황입니다.\n\n0초', + P('두 번째 진행상황입니다.\n\n0초'), ); expect(channel.editMessage).toHaveBeenNthCalledWith( 1, chatJid, 'progress-1', - '첫 번째 진행상황입니다.\n\n10초', + P('첫 번째 진행상황입니다.\n\n10초'), ); expect(channel.editMessage).toHaveBeenNthCalledWith( 2, chatJid, 'progress-2', - '두 번째 진행상황입니다.\n\n10초', + P('두 번째 진행상황입니다.\n\n10초'), ); expect(channel.editMessage).toHaveBeenCalledWith( chatJid, 'progress-2', - '두 번째 진행상황입니다.\n\n10초', + P('두 번째 진행상황입니다.\n\n10초'), ); expect(channel.sendMessage).toHaveBeenCalledTimes(1); expect(channel.sendMessage).toHaveBeenCalledWith( @@ -1425,17 +1430,17 @@ describe('createMessageRuntime', () => { expect(result).toBe(true); expect(channel.sendAndTrack).toHaveBeenCalledWith( chatJid, - '검증 중입니다.\n\n0초', + P('검증 중입니다.\n\n0초'), ); expect(channel.editMessage).toHaveBeenCalledWith( chatJid, 'progress-1', - '커밋은 정상 들어갔고 pre-commit도 통과했습니다.\n\n10초', + P('커밋은 정상 들어갔고 pre-commit도 통과했습니다.\n\n10초'), ); expect(channel.editMessage).toHaveBeenCalledWith( chatJid, 'progress-1', - '커밋은 정상 들어갔고 pre-commit도 통과했습니다.\n\n10초', + P('커밋은 정상 들어갔고 pre-commit도 통과했습니다.\n\n10초'), ); expect(channel.sendMessage).toHaveBeenCalledTimes(1); expect(channel.sendMessage).toHaveBeenCalledWith( @@ -1531,7 +1536,7 @@ describe('createMessageRuntime', () => { expect(channel.sendAndTrack).toHaveBeenCalledTimes(1); expect(channel.sendAndTrack).toHaveBeenCalledWith( chatJid, - '진행 중입니다.\n\n0초', + P('진행 중입니다.\n\n0초'), ); // 같은 progress 메시지는 유지하고, final은 별도 메시지로 보낸다. expect(channel.editMessage).toHaveBeenCalledWith( @@ -1542,7 +1547,7 @@ describe('createMessageRuntime', () => { expect(channel.editMessage).toHaveBeenLastCalledWith( chatJid, 'progress-1', - '아직 진행 중.\n\n10초', + P('아직 진행 중.\n\n10초'), ); expect(channel.sendMessage).toHaveBeenCalledTimes(1); expect(channel.sendMessage).toHaveBeenCalledWith( @@ -1764,7 +1769,7 @@ describe('createMessageRuntime', () => { expect(result).toBe(true); expect(channel.sendAndTrack).toHaveBeenCalledWith( chatJid, - '중간 진행상황입니다.\n\n0초', + P('중간 진행상황입니다.\n\n0초'), ); expect(channel.sendMessage).toHaveBeenCalledTimes(1); expect(channel.sendMessage).toHaveBeenCalledWith( diff --git a/src/message-turn-controller.ts b/src/message-turn-controller.ts index 7bcd2b2..863ed68 100644 --- a/src/message-turn-controller.ts +++ b/src/message-turn-controller.ts @@ -2,6 +2,7 @@ import { type AgentOutput } from './agent-runner.js'; import { logger } from './logger.js'; import { formatOutbound } from './router.js'; import { shouldResetSessionOnAgentFailure } from './session-recovery.js'; +import { TASK_STATUS_MESSAGE_PREFIX } from './task-watch-status.js'; import { type Channel, type RegisteredGroup } from './types.js'; export type VisiblePhase = 'silent' | 'progress' | 'final'; @@ -211,7 +212,7 @@ export class MessageTurnController { if (minutes > 0) elapsedParts.push(`${minutes}분`); elapsedParts.push(`${seconds}초`); - return `${text}\n\n${elapsedParts.join(' ')}`; + return `${TASK_STATUS_MESSAGE_PREFIX}${text}\n\n${elapsedParts.join(' ')}`; } private clearProgressTicker(): void { diff --git a/src/task-scheduler.ts b/src/task-scheduler.ts index 1167e1b..04895fc 100644 --- a/src/task-scheduler.ts +++ b/src/task-scheduler.ts @@ -29,6 +29,11 @@ import { } from './group-folder.js'; import { logger } from './logger.js'; import { createTaskStatusTracker } from './task-status-tracker.js'; +import { + evaluateTaskSuspension, + formatSuspensionNotice, + suspendTask, +} from './task-suspension.js'; import { getTaskQueueJid, getTaskRuntimeTaskId, @@ -305,8 +310,31 @@ async function runTask( return; } + // Clear suspension on success + if (!error && currentTask.suspended_until) { + updateTask(task.id, { suspended_until: null }); + } + + // Check for repeated quota/auth errors → auto-suspend + let suspended = false; if (error) { + const suspension = evaluateTaskSuspension(currentTask, error); + if (suspension.suspended && suspension.suspendedUntil) { + suspended = true; + suspendTask(task.id, suspension.suspendedUntil); + const notice = formatSuspensionNotice( + currentTask, + suspension.suspendedUntil, + suspension.reason || error.slice(0, 200), + ); + await deps.sendMessage(task.chat_jid, notice); + } + } + + if (error && !suspended) { await statusTracker.update('retrying', nextRun); + } else if (suspended) { + // Don't update status tracker — task is suspended, not retrying } else if (nextRun) { await statusTracker.update('waiting', nextRun); } else { diff --git a/src/task-suspension.ts b/src/task-suspension.ts new file mode 100644 index 0000000..514f808 --- /dev/null +++ b/src/task-suspension.ts @@ -0,0 +1,143 @@ +import { TIMEZONE } from './config.js'; +import { getRecentConsecutiveErrors, updateTask } from './db.js'; +import { logger } from './logger.js'; +import type { ScheduledTask } from './types.js'; + +const CONSECUTIVE_FAILURES_THRESHOLD = 3; +const DEFAULT_BACKOFF_STEPS_MS = [ + 3_600_000, // 1 hour + 14_400_000, // 4 hours + 43_200_000, // 12 hours +]; + +// Patterns that indicate a quota / auth / billing error (not transient). +const QUOTA_PATTERNS = [ + /usage limit/i, + /rate limit/i, + /quota exceeded/i, + /billing/i, + /insufficient.*(credit|fund|balance)/i, + /exceeded.*plan/i, + /purchase more credits/i, +]; + +export interface SuspensionResult { + suspended: boolean; + suspendedUntil: string | null; + reason: string | null; +} + +/** + * Detect if a quota/auth error message contains a human-readable + * retry-after date like "try again at Mar 26th, 2026 9:00 AM". + */ +export function parseRetryAfterDate(error: string): Date | null { + const match = error.match( + /try again (?:at|after)\s+(.+?\d{4}[\s,]+\d{1,2}:\d{2}\s*(?:AM|PM)?)/i, + ); + if (!match) return null; + + const raw = match[1] + .replace(/(\d+)(?:st|nd|rd|th)/g, '$1') + .replace(/,\s*/g, ', '); + + const parsed = new Date(raw); + if (Number.isNaN(parsed.getTime())) return null; + + // Sanity: must be in the future and within 30 days + const now = Date.now(); + if (parsed.getTime() <= now || parsed.getTime() > now + 30 * 86_400_000) { + return null; + } + return parsed; +} + +function isQuotaError(error: string): boolean { + return QUOTA_PATTERNS.some((p) => p.test(error)); +} + +function computeBackoffMs(consecutiveCount: number): number { + const idx = Math.min( + consecutiveCount - CONSECUTIVE_FAILURES_THRESHOLD, + DEFAULT_BACKOFF_STEPS_MS.length - 1, + ); + return DEFAULT_BACKOFF_STEPS_MS[Math.max(0, idx)]; +} + +/** + * Check recent run history and decide whether to suspend a task. + * Returns suspension info if the task should be suspended. + */ +export function evaluateTaskSuspension( + task: ScheduledTask, + latestError: string, +): SuspensionResult { + const noSuspension: SuspensionResult = { + suspended: false, + suspendedUntil: null, + reason: null, + }; + + if (!isQuotaError(latestError)) return noSuspension; + + const recentErrors = getRecentConsecutiveErrors( + task.id, + CONSECUTIVE_FAILURES_THRESHOLD, + ); + // +1 because the current error hasn't been logged yet + const totalConsecutive = recentErrors.length + 1; + + if (totalConsecutive < CONSECUTIVE_FAILURES_THRESHOLD) return noSuspension; + + // Try to parse an explicit retry-after date from the error + const retryDate = parseRetryAfterDate(latestError); + const suspendedUntil = retryDate + ? retryDate.toISOString() + : new Date(Date.now() + computeBackoffMs(totalConsecutive)).toISOString(); + + return { + suspended: true, + suspendedUntil, + reason: latestError.slice(0, 200), + }; +} + +/** + * Apply suspension to a task in the DB. + */ +export function suspendTask( + taskId: string, + suspendedUntil: string, +): void { + updateTask(taskId, { suspended_until: suspendedUntil }); + logger.info( + { taskId, suspendedUntil }, + 'Task suspended due to repeated quota/auth errors', + ); +} + +/** + * Format the suspension notification for Discord. + */ +export function formatSuspensionNotice( + task: ScheduledTask, + suspendedUntil: string, + reason: string, +): string { + const resumeLabel = new Intl.DateTimeFormat('ko-KR', { + month: 'numeric', + day: 'numeric', + hour: '2-digit', + minute: '2-digit', + hour12: false, + timeZone: TIMEZONE, + }).format(new Date(suspendedUntil)); + + const lines = [ + `⏸ 태스크 일시 중단`, + `- 사유: ${reason}`, + `- 자동 재개: ${resumeLabel}`, + `- 태스크 ID: \`${task.id}\``, + ]; + return lines.join('\n'); +} diff --git a/src/types.ts b/src/types.ts index 6780214..27c2ea6 100644 --- a/src/types.ts +++ b/src/types.ts @@ -48,6 +48,7 @@ export interface ScheduledTask { last_run: string | null; last_result: string | null; status: 'active' | 'paused' | 'completed'; + suspended_until?: string | null; created_at: string; }